taosdata / TDengine / #3599

coverage: 1.77% (-61.6%) from 63.396%
#3599

push

travis-ci

web-flow 
Merge pull request #29712 from taosdata/fix/TD-33652-3.0

fix: reduce write rows from 30w to 3w

3776 of 278949 branches covered (1.35%)

Branch coverage included in aggregate %.

6012 of 274147 relevant lines covered (2.19%)

1642.73 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0
/source/dnode/vnode/src/meta/metaSnapshot.c
1 
/*
2 
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3 
 *
4 
 * This program is free software: you can use, redistribute, and/or modify
5 
 * it under the terms of the GNU Affero General Public License, version 3
6 
 * or later ("AGPL"), as published by the Free Software Foundation.
7 
 *
8 
 * This program is distributed in the hope that it will be useful, but WITHOUT
9 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 
 * FITNESS FOR A PARTICULAR PURPOSE.
11 
 *
12 
 * You should have received a copy of the GNU Affero General Public License
13 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14 
 */
15 









16



#include "meta.h"










17












18



// SMetaSnapReader ========================================










19



struct SMetaSnapReader {










20



  SMeta*  pMeta;










21



  int64_t sver;










22



  int64_t ever;










23



  TBC*    pTbc;










24



};










25












26



int32_t metaSnapReaderOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapReader** ppReader) {




×







27



  int32_t          code = 0;




×







28



  int32_t          lino;










29



  int32_t          c = 0;




×







30



  SMetaSnapReader* pReader = NULL;




×







31












32



  // alloc










33



  pReader = (SMetaSnapReader*)taosMemoryCalloc(1, sizeof(*pReader));





×







34



  if (pReader == NULL) {





×







35



    TSDB_CHECK_CODE(code = terrno, lino, _exit);





×







36



  }










37



  pReader->pMeta = pMeta;




×







38



  pReader->sver = sver;




×







39



  pReader->ever = ever;




×







40












41



  // impl










42



  code = tdbTbcOpen(pMeta->pTbDb, &pReader->pTbc, NULL);




×







43



  TSDB_CHECK_CODE(code, lino, _exit);





×







44












45



  code = tdbTbcMoveTo(pReader->pTbc, &(STbDbKey){.version = sver, .uid = INT64_MIN}, sizeof(STbDbKey), &c);




×







46



  TSDB_CHECK_CODE(code, lino, _exit);





×







47












48



_exit:




×







49



  if (code) {





×







50



    metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, lino, tstrerror(code));





×







51



    metaSnapReaderClose(&pReader);




×







52



    *ppReader = NULL;




×







53



  } else {










54



    metaInfo("vgId:%d, %s success", TD_VID(pMeta->pVnode), __func__);





×







55



    *ppReader = pReader;




×







56



  }










57



  return code;




×







58



}










59












60



void metaSnapReaderClose(SMetaSnapReader** ppReader) {




×







61



  if (ppReader && *ppReader) {





×







62



    tdbTbcClose((*ppReader)->pTbc);




×







63



    taosMemoryFree(*ppReader);





×







64



    *ppReader = NULL;




×







65



  }










66



}




×







67












68



int32_t metaSnapRead(SMetaSnapReader* pReader, uint8_t** ppData) {




×







69



  int32_t     code = 0;




×







70



  const void* pKey = NULL;




×







71



  const void* pData = NULL;




×







72



  int32_t     nKey = 0;




×







73



  int32_t     nData = 0;




×







74



  STbDbKey    key;










75












76



  *ppData = NULL;




×







77



  for (;;) {




×







78



    if (tdbTbcGet(pReader->pTbc, &pKey, &nKey, &pData, &nData)) {





×







79



      goto _exit;




×







80



    }










81












82



    key = ((STbDbKey*)pKey)[0];




×







83



    if (key.version > pReader->ever) {





×







84



      goto _exit;




×







85



    }










86












87



    if (key.version < pReader->sver) {





×







88



      if (tdbTbcMoveToNext(pReader->pTbc) != 0) {





×







89



        metaTrace("vgId:%d, vnode snapshot meta read data done", TD_VID(pReader->pMeta->pVnode));





×







90



      }










91



      continue;




×







92



    }










93












94



    if (!pData || !nData) {





×







95



      metaError("meta/snap: invalide nData: %" PRId32 " meta snap read failed.", nData);





×







96



      goto _exit;




×







97



    }










98












99



    *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + nData);





×







100



    if (*ppData == NULL) {





×







101



      code = terrno;




×







102



      goto _exit;




×







103



    }










104












105



    SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppData);




×







106



    pHdr->type = SNAP_DATA_META;




×







107



    pHdr->size = nData;




×







108



    memcpy(pHdr->data, pData, nData);




×







109












110



    metaDebug("vgId:%d, vnode snapshot meta read data, version:%" PRId64 " uid:%" PRId64 " blockLen:%d",





×







111



              TD_VID(pReader->pMeta->pVnode), key.version, key.uid, nData);










112












113



    if (tdbTbcMoveToNext(pReader->pTbc) != 0) {





×







114



      metaTrace("vgId:%d, vnode snapshot meta read data done", TD_VID(pReader->pMeta->pVnode));





×







115



    }










116



    break;




×







117



  }










118












119



_exit:




×







120



  if (code) {





×







121



    metaError("vgId:%d, vnode snapshot meta read data failed since %s", TD_VID(pReader->pMeta->pVnode),





×







122



              tstrerror(code));










123



  }










124



  return code;




×







125



}










126












127



// SMetaSnapWriter ========================================










128



struct SMetaSnapWriter {










129



  SMeta*  pMeta;










130



  int64_t sver;










131



  int64_t ever;










132



};










133












134



int32_t metaSnapWriterOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapWriter** ppWriter) {




×







135



  int32_t          code = 0;




×







136



  int32_t          lino;










137



  SMetaSnapWriter* pWriter;










138












139



  // alloc










140



  pWriter = (SMetaSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter));





×







141



  if (pWriter == NULL) {





×







142



    TSDB_CHECK_CODE(code = terrno, lino, _exit);





×







143



  }










144



  pWriter->pMeta = pMeta;




×







145



  pWriter->sver = sver;




×







146



  pWriter->ever = ever;




×







147












148



  code = metaBegin(pMeta, META_BEGIN_HEAP_NIL);




×







149



  TSDB_CHECK_CODE(code, lino, _exit);





×







150












151



_exit:




×







152



  if (code) {





×







153



    metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, lino, tstrerror(code));





×







154



    taosMemoryFree(pWriter);





×







155



    *ppWriter = NULL;




×







156



  } else {










157



    metaDebug("vgId:%d, %s success", TD_VID(pMeta->pVnode), __func__);





×







158



    *ppWriter = pWriter;




×







159



  }










160



  return code;




×







161



}










162












163



int32_t metaSnapWriterClose(SMetaSnapWriter** ppWriter, int8_t rollback) {




×







164



  int32_t          code = 0;




×







165



  SMetaSnapWriter* pWriter = *ppWriter;




×







166












167



  if (rollback) {





×







168



    metaInfo("vgId:%d, meta snapshot writer close and rollback start ", TD_VID(pWriter->pMeta->pVnode));





×







169



    code = metaAbort(pWriter->pMeta);




×







170



    metaInfo("vgId:%d, meta snapshot writer close and rollback finished, code:0x%x", TD_VID(pWriter->pMeta->pVnode),





×







171



             code);










172



    if (code) goto _err;





×







173



  } else {










174



    code = metaCommit(pWriter->pMeta, pWriter->pMeta->txn);




×







175



    if (code) goto _err;





×







176



    code = metaFinishCommit(pWriter->pMeta, pWriter->pMeta->txn);




×







177



    if (code) goto _err;





×







178



  }










179



  taosMemoryFree(pWriter);





×







180



  *ppWriter = NULL;




×







181












182



  return code;




×







183












184



_err:




×







185



  metaError("vgId:%d, meta snapshot writer close failed since %s", TD_VID(pWriter->pMeta->pVnode), tstrerror(code));





×







186



  return code;




×







187



}










188












189



int32_t metaSnapWrite(SMetaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {




×







190



  int32_t    code = 0;




×







191



  int32_t    lino = 0;




×







192



  SMeta*     pMeta = pWriter->pMeta;




×







193



  SMetaEntry metaEntry = {0};




×







194



  SDecoder*  pDecoder = &(SDecoder){0};




×







195












196



  tDecoderInit(pDecoder, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));




×







197



  code = metaDecodeEntry(pDecoder, &metaEntry);




×







198



  TSDB_CHECK_CODE(code, lino, _exit);





×







199












200



  metaHandleSyncEntry(pMeta, &metaEntry);




×







201












202



_exit:




×







203



  if (code) {





×







204



    metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, lino, tstrerror(code));





×







205



  }










206



  tDecoderClear(pDecoder);




×







207



  return code;




×







208



}










209












210



typedef struct STableInfoForChildTable {










211



  char*           tableName;










212



  SSchemaWrapper* schemaRow;










213



  SSchemaWrapper* tagRow;










214



} STableInfoForChildTable;










215












216



static void destroySTableInfoForChildTable(void* data) {




×







217



  STableInfoForChildTable* pData = (STableInfoForChildTable*)data;




×







218



  taosMemoryFree(pData->tableName);





×







219



  tDeleteSchemaWrapper(pData->schemaRow);





×







220



  tDeleteSchemaWrapper(pData->tagRow);





×







221



}




×







222












223



static int32_t MoveToSnapShotVersion(SSnapContext* ctx) {




×







224



  int32_t code = 0;




×







225



  tdbTbcClose((TBC*)ctx->pCur);




×







226



  code = tdbTbcOpen(ctx->pMeta->pTbDb, (TBC**)&ctx->pCur, NULL);




×







227



  if (code != 0) {





×







228



    return TAOS_GET_TERRNO(code);





×







229



  }










230



  STbDbKey key = {.version = ctx->snapVersion, .uid = INT64_MAX};




×







231



  int      c = 0;




×







232



  code = tdbTbcMoveTo((TBC*)ctx->pCur, &key, sizeof(key), &c);




×







233



  if (code != 0) {





×







234



    return TAOS_GET_TERRNO(code);





×







235



  }










236



  if (c < 0) {





×







237



    if (tdbTbcMoveToPrev((TBC*)ctx->pCur) != 0) {





×







238



      metaTrace("vgId:%d, vnode snapshot move to prev failed", TD_VID(ctx->pMeta->pVnode));





×







239



    }










240



  }










241



  return 0;




×







242



}










243












244



static int32_t MoveToPosition(SSnapContext* ctx, int64_t ver, int64_t uid) {




×







245



  tdbTbcClose((TBC*)ctx->pCur);




×







246



  int32_t code = tdbTbcOpen(ctx->pMeta->pTbDb, (TBC**)&ctx->pCur, NULL);




×







247



  if (code != 0) {





×







248



    return TAOS_GET_TERRNO(code);





×







249



  }










250



  STbDbKey key = {.version = ver, .uid = uid};




×







251



  int      c = 0;




×







252



  code = tdbTbcMoveTo((TBC*)ctx->pCur, &key, sizeof(key), &c);




×







253



  if (code != 0) {





×







254



    return TAOS_GET_TERRNO(code);





×







255



  }










256



  return c;




×







257



}










258












259



static int32_t MoveToFirst(SSnapContext* ctx) {




×







260



  tdbTbcClose((TBC*)ctx->pCur);




×







261



  int32_t code = tdbTbcOpen(ctx->pMeta->pTbDb, (TBC**)&ctx->pCur, NULL);




×







262



  if (code != 0) {





×







263



    return TAOS_GET_TERRNO(code);





×







264



  }










265



  code = tdbTbcMoveToFirst((TBC*)ctx->pCur);




×







266



  if (code != 0) {





×







267



    return TAOS_GET_TERRNO(code);





×







268



  }










269



  return 0;




×







270



}










271












272



static int32_t saveSuperTableInfoForChildTable(SMetaEntry* me, SHashObj* suidInfo) {




×







273



  STableInfoForChildTable* data = (STableInfoForChildTable*)taosHashGet(suidInfo, &me->uid, sizeof(tb_uid_t));




×







274



  if (data) {





×







275



    return 0;




×







276



  }










277



  int32_t                 code = 0;




×







278



  STableInfoForChildTable dataTmp = {0};




×







279



  dataTmp.tableName = taosStrdup(me->name);





×







280



  if (dataTmp.tableName == NULL) {





×







281



    code = terrno;




×







282



    goto END;




×







283



  }










284



  dataTmp.schemaRow = tCloneSSchemaWrapper(&me->stbEntry.schemaRow);





×







285



  if (dataTmp.schemaRow == NULL) {





×







286



    code = TSDB_CODE_OUT_OF_MEMORY;




×







287



    goto END;




×







288



  }










289



  dataTmp.tagRow = tCloneSSchemaWrapper(&me->stbEntry.schemaTag);





×







290



  if (dataTmp.tagRow == NULL) {





×







291



    code = TSDB_CODE_OUT_OF_MEMORY;




×







292



    goto END;




×







293



  }










294



  code = taosHashPut(suidInfo, &me->uid, sizeof(tb_uid_t), &dataTmp, sizeof(STableInfoForChildTable));




×







295



  if (code != 0) {





×







296



    goto END;




×







297



  }










298



  return 0;




×







299












300



END:




×







301



  destroySTableInfoForChildTable(&dataTmp);




×







302



  return TAOS_GET_TERRNO(code);





×







303



  ;










304



}










305












306



int32_t buildSnapContext(SVnode* pVnode, int64_t snapVersion, int64_t suid, int8_t subType, int8_t withMeta,




×







307



                         SSnapContext** ctxRet) {










308



  SSnapContext* ctx = taosMemoryCalloc(1, sizeof(SSnapContext));





×







309



  if (ctx == NULL) {





×







310



    return terrno;




×







311



  }










312



  *ctxRet = ctx;




×







313



  ctx->pMeta = pVnode->pMeta;




×







314



  ctx->snapVersion = snapVersion;




×







315



  ctx->suid = suid;




×







316



  ctx->subType = subType;




×







317



  ctx->queryMeta = withMeta;




×







318



  ctx->withMeta = withMeta;




×







319



  ctx->idVersion = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);




×







320



  if (ctx->idVersion == NULL) {





×







321



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







322



  }










323












324



  ctx->suidInfo = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);




×







325



  if (ctx->suidInfo == NULL) {





×







326



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







327



  }










328



  taosHashSetFreeFp(ctx->suidInfo, destroySTableInfoForChildTable);




×







329












330



  ctx->index = 0;




×







331



  ctx->idList = taosArrayInit(100, sizeof(int64_t));




×







332



  if (ctx->idList == NULL) {





×







333



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







334



    ;










335



  }










336



  void* pKey = NULL;




×







337



  void* pVal = NULL;




×







338



  int   vLen = 0, kLen = 0;




×







339












340



  metaDebug("tmqsnap init snapVersion:%" PRIi64, ctx->snapVersion);





×







341



  int32_t code = MoveToFirst(ctx);




×







342



  if (code != 0) {





×







343



    return code;




×







344



  }










345



  while (1) {




×







346



    int32_t ret = tdbTbcNext((TBC*)ctx->pCur, &pKey, &kLen, &pVal, &vLen);




×







347



    if (ret < 0) break;





×







348



    STbDbKey* tmp = (STbDbKey*)pKey;




×







349



    if (tmp->version > ctx->snapVersion) break;





×







350












351



    SIdInfo* idData = (SIdInfo*)taosHashGet(ctx->idVersion, &tmp->uid, sizeof(tb_uid_t));




×







352



    if (idData) {





×







353



      continue;




×







354



    }










355












356



    if (tdbTbGet(ctx->pMeta->pUidIdx, &tmp->uid, sizeof(tb_uid_t), NULL, NULL) <





×







357



        0) {  // check if table exist for now, need optimize later










358



      continue;




×







359



    }










360












361



    SDecoder   dc = {0};




×







362



    SMetaEntry me = {0};




×







363



    tDecoderInit(&dc, pVal, vLen);




×







364



    ret = metaDecodeEntry(&dc, &me);




×







365



    if (ret < 0) {





×







366



      tDecoderClear(&dc);




×







367



      return TAOS_GET_TERRNO(ret);





×







368



    }










369



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {





×







370



      if ((me.uid != ctx->suid && me.type == TSDB_SUPER_TABLE) ||





×







371



          (me.ctbEntry.suid != ctx->suid && me.type == TSDB_CHILD_TABLE)) {





×







372



        tDecoderClear(&dc);




×







373



        continue;




×







374



      }










375



    }










376












377



    if (taosArrayPush(ctx->idList, &tmp->uid) == NULL) {





×







378



      tDecoderClear(&dc);




×







379



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







380



    }










381



    metaDebug("tmqsnap init idlist name:%s, uid:%" PRIi64, me.name, tmp->uid);





×







382



    tDecoderClear(&dc);




×







383












384



    SIdInfo info = {0};




×







385



    if (taosHashPut(ctx->idVersion, &tmp->uid, sizeof(tb_uid_t), &info, sizeof(SIdInfo)) != 0) {





×







386



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







387



    }










388



  }










389



  taosHashClear(ctx->idVersion);




×







390












391



  code = MoveToSnapShotVersion(ctx);




×







392



  if (code != 0) {





×







393



    return code;




×







394



  }










395



  while (1) {




×







396



    int32_t ret = tdbTbcPrev((TBC*)ctx->pCur, &pKey, &kLen, &pVal, &vLen);




×







397



    if (ret < 0) break;





×







398












399



    STbDbKey* tmp = (STbDbKey*)pKey;




×







400



    SIdInfo*  idData = (SIdInfo*)taosHashGet(ctx->idVersion, &tmp->uid, sizeof(tb_uid_t));




×







401



    if (idData) {





×







402



      continue;




×







403



    }










404



    SIdInfo info = {.version = tmp->version, .index = 0};




×







405



    ret = taosHashPut(ctx->idVersion, &tmp->uid, sizeof(tb_uid_t), &info, sizeof(SIdInfo));




×







406



    if (ret != 0) {





×







407



      return TAOS_GET_TERRNO(ret);





×







408



    }










409












410



    SDecoder   dc = {0};




×







411



    SMetaEntry me = {0};




×







412



    tDecoderInit(&dc, pVal, vLen);




×







413



    ret = metaDecodeEntry(&dc, &me);




×







414



    if (ret < 0) {





×







415



      tDecoderClear(&dc);




×







416



      return TAOS_GET_TERRNO(ret);





×







417



    }










418












419



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {





×







420



      if ((me.uid != ctx->suid && me.type == TSDB_SUPER_TABLE) ||





×







421



          (me.ctbEntry.suid != ctx->suid && me.type == TSDB_CHILD_TABLE)) {





×







422



        tDecoderClear(&dc);




×







423



        continue;




×







424



      }










425



    }










426












427



    if ((ctx->subType == TOPIC_SUB_TYPE__DB && me.type == TSDB_SUPER_TABLE) ||





×







428



        (ctx->subType == TOPIC_SUB_TYPE__TABLE && me.uid == ctx->suid)) {





×







429



      ret = saveSuperTableInfoForChildTable(&me, ctx->suidInfo);




×







430



      if (ret != 0) {





×







431



        tDecoderClear(&dc);




×







432



        return ret;




×







433



      }










434



    }










435



    tDecoderClear(&dc);




×







436



  }










437












438



  for (int i = 0; i < taosArrayGetSize(ctx->idList); i++) {





×







439



    int64_t* uid = taosArrayGet(ctx->idList, i);




×







440



    if (uid == NULL) {





×







441



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







442



    }










443



    SIdInfo* idData = (SIdInfo*)taosHashGet(ctx->idVersion, uid, sizeof(int64_t));




×







444



    if (!idData) {





×







445



      metaError("meta/snap: null idData");





×







446



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







447



    }










448












449



    idData->index = i;




×







450



    metaDebug("tmqsnap init idVersion uid:%" PRIi64 " version:%" PRIi64 " index:%d", *uid, idData->version,





×







451



              idData->index);










452



  }










453












454



  tdbFree(pKey);




×







455



  tdbFree(pVal);




×







456



  return TDB_CODE_SUCCESS;




×







457



}










458












459



void destroySnapContext(SSnapContext* ctx) {




×







460



  tdbTbcClose((TBC*)ctx->pCur);




×







461



  taosArrayDestroy(ctx->idList);




×







462



  taosHashCleanup(ctx->idVersion);




×







463



  taosHashCleanup(ctx->suidInfo);




×







464



  taosMemoryFree(ctx);





×







465



}




×







466












467



static int32_t buildNormalChildTableInfo(SVCreateTbReq* req, void** pBuf, int32_t* contLen) {




×







468



  int32_t            ret = 0;




×







469



  SVCreateTbBatchReq reqs = {0};




×







470












471



  reqs.pArray = taosArrayInit(1, sizeof(struct SVCreateTbReq));




×







472



  if (NULL == reqs.pArray) {





×







473



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







474



    goto end;




×







475



  }










476



  if (taosArrayPush(reqs.pArray, req) == NULL) {





×







477



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







478



    goto end;




×







479



  }










480



  reqs.nReqs = 1;




×







481












482



  tEncodeSize(tEncodeSVCreateTbBatchReq, &reqs, *contLen, ret);





×







483



  if (ret < 0) {





×







484



    ret = TAOS_GET_TERRNO(ret);





×







485



    goto end;




×







486



  }










487



  *contLen += sizeof(SMsgHead);




×







488



  *pBuf = taosMemoryMalloc(*contLen);





×







489



  if (NULL == *pBuf) {





×







490



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







491



    goto end;




×







492



  }










493



  SEncoder coder = {0};




×







494



  tEncoderInit(&coder, POINTER_SHIFT(*pBuf, sizeof(SMsgHead)), *contLen);




×







495



  ret = tEncodeSVCreateTbBatchReq(&coder, &reqs);




×







496



  tEncoderClear(&coder);




×







497












498



  if (ret < 0) {





×







499



    taosMemoryFreeClear(*pBuf);





×







500



    ret = TAOS_GET_TERRNO(ret);





×







501



    goto end;




×







502



  }










503












504



end:




×







505



  taosArrayDestroy(reqs.pArray);




×







506



  return ret;




×







507



}










508












509



static int32_t buildSuperTableInfo(SVCreateStbReq* req, void** pBuf, int32_t* contLen) {




×







510



  int32_t ret = 0;




×







511



  tEncodeSize(tEncodeSVCreateStbReq, req, *contLen, ret);





×







512



  if (ret < 0) {





×







513



    return TAOS_GET_TERRNO(ret);





×







514



  }










515












516



  *contLen += sizeof(SMsgHead);




×







517



  *pBuf = taosMemoryMalloc(*contLen);





×







518



  if (NULL == *pBuf) {





×







519



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







520



  }










521












522



  SEncoder encoder = {0};




×







523



  tEncoderInit(&encoder, POINTER_SHIFT(*pBuf, sizeof(SMsgHead)), *contLen);




×







524



  ret = tEncodeSVCreateStbReq(&encoder, req);




×







525



  tEncoderClear(&encoder);




×







526



  if (ret < 0) {





×







527



    taosMemoryFreeClear(*pBuf);





×







528



    return TAOS_GET_TERRNO(ret);





×







529



  }










530



  return 0;




×







531



}










532












533



int32_t setForSnapShot(SSnapContext* ctx, int64_t uid) {




×







534



  if (uid == 0) {





×







535



    ctx->index = 0;




×







536



    return 0;




×







537



  }










538












539



  SIdInfo* idInfo = (SIdInfo*)taosHashGet(ctx->idVersion, &uid, sizeof(tb_uid_t));




×







540



  if (idInfo == NULL) {





×







541



    return terrno;




×







542



  }










543












544



  ctx->index = idInfo->index;




×







545












546



  return 0;




×







547



}










548












549



void taosXSetTablePrimaryKey(SSnapContext* ctx, int64_t uid) {




×







550



  bool            ret = false;




×







551



  SSchemaWrapper* schema = metaGetTableSchema(ctx->pMeta, uid, -1, 1, NULL);




×







552



  if (schema && schema->nCols >= 2 && schema->pSchema[1].flags & COL_IS_KEY) {





×







553



    ret = true;




×







554



  }










555



  tDeleteSchemaWrapper(schema);










556



  ctx->hasPrimaryKey = ret;




×







557



}




×







558












559



bool taosXGetTablePrimaryKey(SSnapContext* ctx) { return ctx->hasPrimaryKey; }




×







560












561



int32_t getTableInfoFromSnapshot(SSnapContext* ctx, void** pBuf, int32_t* contLen, int16_t* type, int64_t* uid) {




×







562



  int32_t ret = 0;




×







563



  void*   pKey = NULL;




×







564



  void*   pVal = NULL;




×







565



  int     vLen = 0, kLen = 0;




×







566












567



  while (1) {




×







568



    if (ctx->index >= taosArrayGetSize(ctx->idList)) {





×







569



      metaDebug("tmqsnap get meta end");





×







570



      ctx->index = 0;




×







571



      ctx->queryMeta = 0;  // change to get data




×







572



      return 0;




×







573



    }










574












575



    int64_t* uidTmp = taosArrayGet(ctx->idList, ctx->index);




×







576



    if (uidTmp == NULL) {





×







577



      metaError("tmqsnap get meta null uid");





×







578



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







579



    }










580



    ctx->index++;




×







581



    SIdInfo* idInfo = (SIdInfo*)taosHashGet(ctx->idVersion, uidTmp, sizeof(tb_uid_t));




×







582



    if (!idInfo) {





×







583



      metaError("meta/snap: null idInfo");





×







584



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







585



    }










586












587



    *uid = *uidTmp;




×







588



    ret = MoveToPosition(ctx, idInfo->version, *uidTmp);




×







589



    if (ret == 0) {





×







590



      break;




×







591



    }










592



    metaDebug("tmqsnap get meta not exist uid:%" PRIi64 " version:%" PRIi64, *uid, idInfo->version);





×







593



  }










594












595



  ret = tdbTbcGet((TBC*)ctx->pCur, (const void**)&pKey, &kLen, (const void**)&pVal, &vLen);




×







596



  if (ret < 0) {





×







597



    return TAOS_GET_TERRNO(ret);





×







598



  }










599



  SDecoder   dc = {0};




×







600



  SMetaEntry me = {0};




×







601



  tDecoderInit(&dc, pVal, vLen);




×







602



  ret = metaDecodeEntry(&dc, &me);




×







603



  if (ret < 0) {





×







604



    tDecoderClear(&dc);




×







605



    ret = TAOS_GET_TERRNO(ret);





×







606



    goto END;




×







607



  }










608



  metaDebug("tmqsnap get meta uid:%" PRIi64 " name:%s index:%d", *uid, me.name, ctx->index - 1);





×







609












610



  if ((ctx->subType == TOPIC_SUB_TYPE__DB && me.type == TSDB_SUPER_TABLE) ||





×







611



      (ctx->subType == TOPIC_SUB_TYPE__TABLE && me.uid == ctx->suid)) {





×







612



    SVCreateStbReq req = {0};




×







613



    req.name = me.name;




×







614



    req.suid = me.uid;




×







615



    req.schemaRow = me.stbEntry.schemaRow;




×







616



    req.schemaTag = me.stbEntry.schemaTag;




×







617



    req.schemaRow.version = 1;




×







618



    req.schemaTag.version = 1;




×







619



    req.colCmpr = me.colCmpr;




×







620












621



    ret = buildSuperTableInfo(&req, pBuf, contLen);




×







622



    *type = TDMT_VND_CREATE_STB;




×







623



  } else if ((ctx->subType == TOPIC_SUB_TYPE__DB && me.type == TSDB_CHILD_TABLE) ||





×







624



             (ctx->subType == TOPIC_SUB_TYPE__TABLE && me.type == TSDB_CHILD_TABLE && me.ctbEntry.suid == ctx->suid)) {





×







625



    STableInfoForChildTable* data =










626



        (STableInfoForChildTable*)taosHashGet(ctx->suidInfo, &me.ctbEntry.suid, sizeof(tb_uid_t));




×







627



    if (!data) {





×







628



      metaError("meta/snap: null data");





×







629



      ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







630



      goto END;




×







631



    }










632












633



    SVCreateTbReq req = {0};




×







634












635



    req.type = TSDB_CHILD_TABLE;




×







636



    req.name = me.name;




×







637



    req.uid = me.uid;




×







638



    req.commentLen = -1;




×







639



    req.ctb.suid = me.ctbEntry.suid;




×







640



    req.ctb.tagNum = data->tagRow->nCols;




×







641



    req.ctb.stbName = data->tableName;




×







642












643



    SArray* tagName = taosArrayInit(req.ctb.tagNum, TSDB_COL_NAME_LEN);




×







644



    if (tagName == NULL) {





×







645



      metaError("meta/snap: init tag name failed.");





×







646



      ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







647



      goto END;




×







648



    }










649



    STag* p = (STag*)me.ctbEntry.pTags;




×







650



    if (tTagIsJson(p)) {





×







651



      if (p->nTag != 0) {





×







652



        SSchema* schema = &data->tagRow->pSchema[0];




×







653



        if (taosArrayPush(tagName, schema->name) == NULL) {





×







654



          ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







655



          taosArrayDestroy(tagName);




×







656



          goto END;




×







657



        }










658



      }










659



    } else {










660



      SArray* pTagVals = NULL;




×







661



      ret = tTagToValArray((const STag*)p, &pTagVals);




×







662



      if (ret != 0) {





×







663



        metaError("meta/snap: tag to val array failed.");





×







664



        taosArrayDestroy(pTagVals);




×







665



        taosArrayDestroy(tagName);




×







666



        goto END;




×







667



      }










668



      int16_t nCols = taosArrayGetSize(pTagVals);




×







669



      for (int j = 0; j < nCols; ++j) {





×







670



        STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, j);




×







671



        for (int i = 0; pTagVal && i < data->tagRow->nCols; i++) {





×







672



          SSchema* schema = &data->tagRow->pSchema[i];




×







673



          if (schema->colId == pTagVal->cid) {





×







674



            if (taosArrayPush(tagName, schema->name) == NULL) {





×







675



              ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







676



              taosArrayDestroy(pTagVals);




×







677



              taosArrayDestroy(tagName);




×







678



              goto END;




×







679



            }










680



          }










681



        }










682



      }










683



      taosArrayDestroy(pTagVals);




×







684



    }










685



    req.ctb.pTag = me.ctbEntry.pTags;




×







686



    req.ctb.tagName = tagName;




×







687



    ret = buildNormalChildTableInfo(&req, pBuf, contLen);




×







688



    *type = TDMT_VND_CREATE_TABLE;




×







689



    taosArrayDestroy(tagName);




×







690



  } else if (ctx->subType == TOPIC_SUB_TYPE__DB) {





×







691



    SVCreateTbReq req = {0};




×







692



    req.type = TSDB_NORMAL_TABLE;




×







693



    req.name = me.name;




×







694



    req.uid = me.uid;




×







695



    req.commentLen = -1;




×







696



    req.ntb.schemaRow = me.ntbEntry.schemaRow;




×







697



    req.colCmpr = me.colCmpr;




×







698



    ret = buildNormalChildTableInfo(&req, pBuf, contLen);




×







699



    *type = TDMT_VND_CREATE_TABLE;




×







700



  } else {










701



    metaError("meta/snap: invalid topic sub type: %" PRId8 " get meta from snap failed.", ctx->subType);





×







702



    ret = TSDB_CODE_SDB_INVALID_TABLE_TYPE;




×







703



  }










704












705



END:




×







706



  tDecoderClear(&dc);




×







707



  return ret;




×







708



}










709












710



int32_t getMetaTableInfoFromSnapshot(SSnapContext* ctx, SMetaTableInfo* result) {




×







711



  void* pKey = NULL;




×







712



  void* pVal = NULL;




×







713



  int   vLen, kLen;










714












715



  while (1) {




×







716



    if (ctx->index >= taosArrayGetSize(ctx->idList)) {





×







717



      metaDebug("tmqsnap get uid info end");





×







718



      return 0;




×







719



    }










720



    int64_t* uidTmp = taosArrayGet(ctx->idList, ctx->index);




×







721



    if (uidTmp == NULL) {





×







722



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







723



    }










724



    ctx->index++;




×







725



    SIdInfo* idInfo = (SIdInfo*)taosHashGet(ctx->idVersion, uidTmp, sizeof(tb_uid_t));




×







726



    if (!idInfo) {





×







727



      metaError("meta/snap: null idInfo");





×







728



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







729



    }










730












731



    int32_t ret = MoveToPosition(ctx, idInfo->version, *uidTmp);




×







732



    if (ret != 0) {





×







733



      metaDebug("tmqsnap getMetaTableInfoFromSnapshot not exist uid:%" PRIi64 " version:%" PRIi64, *uidTmp,





×







734



                idInfo->version);










735



      continue;




×







736



    }










737



    ret = tdbTbcGet((TBC*)ctx->pCur, (const void**)&pKey, &kLen, (const void**)&pVal, &vLen);




×







738



    if (ret != 0) {





×







739



      return TAOS_GET_TERRNO(ret);





×







740



    }










741



    SDecoder   dc = {0};




×







742



    SMetaEntry me = {0};




×







743



    tDecoderInit(&dc, pVal, vLen);




×







744



    ret = metaDecodeEntry(&dc, &me);




×







745



    if (ret != 0) {





×







746



      tDecoderClear(&dc);




×







747



      return TAOS_GET_TERRNO(ret);





×







748



    }










749



    metaDebug("tmqsnap get uid info uid:%" PRIi64 " name:%s index:%d", me.uid, me.name, ctx->index - 1);





×







750












751



    if ((ctx->subType == TOPIC_SUB_TYPE__DB && me.type == TSDB_CHILD_TABLE) ||





×







752



        (ctx->subType == TOPIC_SUB_TYPE__TABLE && me.type == TSDB_CHILD_TABLE && me.ctbEntry.suid == ctx->suid)) {





×







753



      STableInfoForChildTable* data =










754



          (STableInfoForChildTable*)taosHashGet(ctx->suidInfo, &me.ctbEntry.suid, sizeof(tb_uid_t));




×







755



      if (data == NULL) {





×







756



        tDecoderClear(&dc);




×







757



        metaError("meta/snap: null data");





×







758



        return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







759



      }










760



      result->suid = me.ctbEntry.suid;




×







761



      result->schema = tCloneSSchemaWrapper(data->schemaRow);





×







762



    } else if (ctx->subType == TOPIC_SUB_TYPE__DB && me.type == TSDB_NORMAL_TABLE) {





×







763



      result->suid = 0;





×







764



      result->schema = tCloneSSchemaWrapper(&me.ntbEntry.schemaRow);




×







765



    } else {










766



      metaDebug("tmqsnap get uid continue");





×







767



      tDecoderClear(&dc);




×







768



      continue;




×







769



    }










770



    result->uid = me.uid;




×







771



    tstrncpy(result->tbName, me.name, TSDB_TABLE_NAME_LEN);




×







772



    tDecoderClear(&dc);




×







773



    if (result->schema == NULL) {





×







774



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







775



    }










776



    break;




×







777



  }










778



  return 0;




×







779



}
























    

  • 

    •