taosdata / TDengine / #4980

coverage: 68.492% (-0.02%) from 68.512%
#4980

push

travis-ci

web-flow 
fix: add retry while exec ci case test_stable_keep_compact.py. (#34729)

211901 of 309380 relevant lines covered (68.49%)

135171938.8 hits per line

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

85.54
/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 <string.h>










17



#include "meta.h"










18



#include "tdataformat.h"










19












20



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










21



struct SMetaSnapReader {










22



  SMeta*  pMeta;










23



  int64_t sver;










24



  int64_t ever;










25



  TBC*    pTbc;










26



  int32_t iLoop;










27



};










28












29



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




29,496✔





30



  int32_t          code = 0;




29,496✔





31



  int32_t          lino;










32



  int32_t          c = 0;




29,496✔





33



  SMetaSnapReader* pReader = NULL;




29,496✔





34












35



  // alloc










36



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




29,496✔





37



  if (pReader == NULL) {




29,496✔





38



    TSDB_CHECK_CODE(code = terrno, lino, _exit);




×







39



  }










40



  pReader->pMeta = pMeta;




29,496✔





41



  pReader->sver = sver;




29,496✔





42



  pReader->ever = ever;




29,496✔





43












44



  // impl










45



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




29,496✔





46



  TSDB_CHECK_CODE(code, lino, _exit);




29,496✔





47












48



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




29,496✔





49



  TSDB_CHECK_CODE(code, lino, _exit);




29,496✔





50












51



_exit:




29,496✔





52



  if (code) {




29,496✔





53



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




×







54



    metaSnapReaderClose(&pReader);




×







55



    *ppReader = NULL;




×







56



  } else {










57



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




29,496✔





58



    *ppReader = pReader;




29,496✔





59



  }










60



  return code;




29,496✔





61



}










62












63



void metaSnapReaderClose(SMetaSnapReader** ppReader) {




29,496✔





64



  if (ppReader && *ppReader) {




29,496✔





65



    tdbTbcClose((*ppReader)->pTbc);




29,496✔





66



    taosMemoryFree(*ppReader);




29,496✔





67



    *ppReader = NULL;




29,496✔





68



  }










69



}




29,496✔





70












71



extern int metaDecodeEntryImpl(SDecoder* pCoder, SMetaEntry* pME, bool headerOnly);










72












73



static int32_t metaDecodeEntryHeader(void* data, int32_t size, SMetaEntry* entry) {




340,848✔





74



  SDecoder decoder = {0};




340,848✔





75



  tDecoderInit(&decoder, (uint8_t*)data, size);




340,848✔





76












77



  int32_t code = metaDecodeEntryImpl(&decoder, entry, true);




340,848✔





78



  if (code) {




340,848✔





79



    tDecoderClear(&decoder);




×







80



    return code;




×







81



  }










82












83



  tDecoderClear(&decoder);




340,848✔





84



  return 0;




340,848✔





85



}










86












87



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




199,580✔





88



  int32_t     code = 0;




199,580✔





89



  const void* pKey = NULL;




199,580✔





90



  const void* pData = NULL;




199,580✔





91



  int32_t     nKey = 0;




199,580✔





92



  int32_t     nData = 0;




199,580✔





93



  STbDbKey    key;










94



  int32_t     c;




199,580✔





95












96



  *ppData = NULL;




199,580✔





97



  while (pReader->iLoop < 2) {




429,336✔





98



    if (tdbTbcGet(pReader->pTbc, &pKey, &nKey, &pData, &nData) != 0 || ((STbDbKey*)pKey)->version > pReader->ever) {




399,840✔





99



      pReader->iLoop++;




58,992✔





100












101



      // Reopen the cursor to read from the beginning










102



      tdbTbcClose(pReader->pTbc);




58,992✔





103



      pReader->pTbc = NULL;




58,992✔





104



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




58,992✔





105



      if (code) {




58,992✔





106



        metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pReader->pMeta->pVnode), __func__, __FILE__, __LINE__,




×







107



                  tstrerror(code));










108



        goto _exit;




×







109



      }










110












111



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




58,992✔





112



      if (code) {




58,992✔





113



        metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pReader->pMeta->pVnode), __func__, __FILE__, __LINE__,




×







114



                  tstrerror(code));










115



        goto _exit;




×







116



      }










117












118



      continue;




58,992✔





119



    }










120












121



    // Decode meta entry










122



    SMetaEntry entry = {0};




340,848✔





123



    code = metaDecodeEntryHeader((void*)pData, nData, &entry);




340,848✔





124



    if (code) {




340,848✔





125



      metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pReader->pMeta->pVnode), __func__, __FILE__, __LINE__,




×







126



                tstrerror(code));










127



      goto _exit;




×







128



    }










129












130



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




340,848✔





131



    if (key.version < pReader->sver                                       //




340,848✔





132



        || (pReader->iLoop == 0 && TABS(entry.type) != TSDB_SUPER_TABLE)  // First loop send super table entry




340,168✔





133



        || (pReader->iLoop == 1 && TABS(entry.type) == TSDB_SUPER_TABLE)  // Second loop send non-super table entry




196,138✔





134



    ) {










135



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




170,764✔





136



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




×







137



      }










138



      continue;




170,764✔





139



    }










140












141



    if (!pData || !nData) {




170,084✔





142



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




×







143



      goto _exit;




×







144



    }










145












146



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




170,084✔





147



    if (*ppData == NULL) {




170,084✔





148



      code = terrno;




×







149



      goto _exit;




×







150



    }










151












152



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




170,084✔





153



    pHdr->type = SNAP_DATA_META;




170,084✔





154



    pHdr->size = nData;




170,084✔





155



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




170,084✔





156












157



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




170,084✔





158



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










159












160



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




170,084✔





161



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




×







162



    }










163



    break;




170,084✔





164



  }










165












166



_exit:




199,580✔





167



  if (code) {




199,580✔





168



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




×







169



              tstrerror(code));










170



  }










171



  return code;




199,580✔





172



}










173












174



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










175



struct SMetaSnapWriter {










176



  SMeta*  pMeta;










177



  int64_t sver;










178



  int64_t ever;










179



};










180












181



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




25,921✔





182



  int32_t          code = 0;




25,921✔





183



  int32_t          lino;










184



  SMetaSnapWriter* pWriter;










185












186



  // alloc










187



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




25,921✔





188



  if (pWriter == NULL) {




25,921✔





189



    TSDB_CHECK_CODE(code = terrno, lino, _exit);




×







190



  }










191



  pWriter->pMeta = pMeta;




25,921✔





192



  pWriter->sver = sver;




25,921✔





193



  pWriter->ever = ever;




25,921✔





194












195



  code = metaBegin(pMeta, META_BEGIN_HEAP_NIL);




25,921✔





196



  TSDB_CHECK_CODE(code, lino, _exit);




25,921✔





197












198



_exit:




25,921✔





199



  if (code) {




25,921✔





200



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




×







201



    taosMemoryFree(pWriter);




×







202



    *ppWriter = NULL;




×







203



  } else {










204



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




25,921✔





205



    *ppWriter = pWriter;




25,921✔





206



  }










207



  return code;




25,921✔





208



}










209












210



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




25,921✔





211



  int32_t          code = 0;




25,921✔





212



  SMetaSnapWriter* pWriter = *ppWriter;




25,921✔





213












214



  if (rollback) {




25,921✔





215



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




×







216



    code = metaAbort(pWriter->pMeta);




×







217



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




×







218



             code);










219



    if (code) goto _err;




×







220



  } else {










221



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




25,921✔





222



    if (code) goto _err;




25,921✔





223



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




25,921✔





224



    if (code) goto _err;




25,921✔





225



  }










226



  taosMemoryFree(pWriter);




25,921✔





227



  *ppWriter = NULL;




25,921✔





228












229



  return code;




25,921✔





230












231



_err:




×







232



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




×







233



  return code;




×







234



}










235












236



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




169,998✔





237



  int32_t    code = 0;




169,998✔





238



  int32_t    lino = 0;




169,998✔





239



  SMeta*     pMeta = pWriter->pMeta;




169,998✔





240



  SMetaEntry metaEntry = {0};




169,998✔





241



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




169,998✔





242












243



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




169,998✔





244



  code = metaDecodeEntry(pDecoder, &metaEntry);




169,998✔





245



  TSDB_CHECK_CODE(code, lino, _exit);




169,998✔





246












247



  metaHandleSyncEntry(pMeta, &metaEntry);




169,998✔





248












249



_exit:




169,998✔





250



  if (code) {




169,998✔





251



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




×







252



  }










253



  tDecoderClear(pDecoder);




169,998✔





254



  return code;




169,998✔





255



}










256












257



typedef struct STableInfoForChildTable {










258



  char*           tableName;










259



  SSchemaWrapper* schemaRow;










260



  SSchemaWrapper* tagRow;










261



  SExtSchema*     pExtSchemas;










262



} STableInfoForChildTable;










263












264



static void destroySTableInfoForChildTable(void* data) {




182,197✔





265



  STableInfoForChildTable* pData = (STableInfoForChildTable*)data;




182,197✔





266



  taosMemoryFree(pData->tableName);




182,197✔





267



  tDeleteSchemaWrapper(pData->schemaRow);




181,926✔





268



  tDeleteSchemaWrapper(pData->tagRow);




181,926✔





269



  taosMemoryFreeClear(pData->pExtSchemas);




181,926✔





270



}




181,926✔





271












272



static int32_t MoveToSnapShotVersion(SSnapContext* ctx) {




127,879✔





273



  int32_t code = 0;




127,879✔





274



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




127,879✔





275



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




127,191✔





276



  if (code != 0) {




127,879✔





277



    return TAOS_GET_TERRNO(code);




×







278



  }










279



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




127,879✔





280



  int      c = 0;




127,879✔





281



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




127,879✔





282



  if (code != 0) {




127,547✔





283



    return TAOS_GET_TERRNO(code);




×







284



  }










285



  if (c < 0) {




127,547✔





286



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




2,385✔





287



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




×







288



    }










289



  }










290



  return 0;




127,547✔





291



}










292












293



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




1,631,138✔





294



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




1,631,138✔





295



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




1,631,138✔





296



  if (code != 0) {




1,631,138✔





297



    return TAOS_GET_TERRNO(code);




×







298



  }










299



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




1,631,138✔





300



  int      c = 0;




1,631,138✔





301



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




1,631,764✔





302



  if (code != 0) {




1,630,825✔





303



    return TAOS_GET_TERRNO(code);




×







304



  }










305



  return c;




1,630,825✔





306



}










307












308



static int32_t MoveToFirst(SSnapContext* ctx) {




127,879✔





309



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




127,879✔





310



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




127,547✔





311



  if (code != 0) {




127,547✔





312



    return TAOS_GET_TERRNO(code);




×







313



  }










314



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




127,547✔





315



  if (code != 0) {




127,547✔





316



    return TAOS_GET_TERRNO(code);




×







317



  }










318



  return 0;




127,547✔





319



}










320












321



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




181,865✔





322



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




181,865✔





323



  if (data) {




182,197✔





324



    return 0;




×







325



  }










326



  int32_t                 code = 0;




182,197✔





327



  STableInfoForChildTable dataTmp = {0};




182,197✔





328



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




182,197✔





329



  if (dataTmp.tableName == NULL) {




181,841✔





330



    code = terrno;




×







331



    goto END;




×







332



  }










333



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




181,841✔





334



  if (dataTmp.schemaRow == NULL) {




181,509✔





335



    code = TSDB_CODE_OUT_OF_MEMORY;




×







336



    goto END;




×







337



  }










338



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




181,509✔





339



  if (dataTmp.tagRow == NULL) {




181,865✔





340



    code = TSDB_CODE_OUT_OF_MEMORY;




×







341



    goto END;




×







342



  }










343



  if (me->pExtSchemas != NULL) {




181,865✔





344



    dataTmp.pExtSchemas = taosMemoryMalloc(sizeof(SExtSchema) * me->stbEntry.schemaRow.nCols);




8,018✔





345



    if (dataTmp.pExtSchemas == NULL) {




8,018✔





346



      code = TSDB_CODE_OUT_OF_MEMORY;




×







347



      goto END;




×







348



    }










349



    memcpy(dataTmp.pExtSchemas, me->pExtSchemas, sizeof(SExtSchema) * me->stbEntry.schemaRow.nCols);




8,018✔





350



  }










351



  










352



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




182,197✔





353



  if (code != 0) {




182,197✔





354



    goto END;




×







355



  }










356



  return 0;




182,197✔





357












358



END:




×







359



  destroySTableInfoForChildTable(&dataTmp);




×







360



  return TAOS_GET_TERRNO(code);




×







361



}










362












363



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




127,879✔





364



                         SSnapContext** ctxRet) {










365



  int32_t code = 0;




127,879✔





366



  int32_t lino = 0;




127,879✔





367



  SDecoder   dc = {0};




127,879✔





368



  void* pKey = NULL;




127,879✔





369



  void* pVal = NULL;




127,879✔





370



  int   vLen = 0, kLen = 0;




127,879✔





371












372



  metaRLock(pVnode->pMeta);




127,879✔





373



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




127,879✔





374



  TSDB_CHECK_NULL(ctx, code, lino, END, terrno);




127,879✔





375



  *ctxRet = ctx;




127,879✔





376



  ctx->pMeta = pVnode->pMeta;




127,879✔





377



  ctx->snapVersion = snapVersion;




127,879✔





378



  ctx->suid = suid;




127,879✔





379



  ctx->subType = subType;




127,879✔





380



  ctx->queryMeta = withMeta;




127,879✔





381



  ctx->withMeta = withMeta;




127,879✔





382



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




127,879✔





383



  TSDB_CHECK_NULL(ctx->idVersion, code, lino, END, terrno);




127,879✔





384



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




127,879✔





385



  TSDB_CHECK_NULL(ctx->suidInfo, code, lino, END, terrno);




127,879✔





386



  taosHashSetFreeFp(ctx->suidInfo, destroySTableInfoForChildTable);




127,879✔





387












388



  ctx->index = 0;




127,879✔





389



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




127,879✔





390



  TSDB_CHECK_NULL(ctx->idList, code, lino, END, terrno);




127,547✔





391












392



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




127,547✔





393



  code = MoveToFirst(ctx);




127,547✔





394



  TSDB_CHECK_CODE(code, lino, END);




127,547✔





395



  while (1) {




12,481,411✔





396



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




12,608,958✔





397



    if (ret < 0) break;




12,604,371✔





398



    STbDbKey* tmp = (STbDbKey*)pKey;




12,479,522✔





399



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




12,479,522✔





400












401



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




12,479,092✔





402



    if (idData) {




12,479,490✔





403



      continue;




97,675✔





404



    }










405












406



    // check if table exist for now, need optimize later










407



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




12,381,815✔





408



      continue;




92,875✔





409



    }










410












411



    SMetaEntry me = {0};




12,281,159✔





412



    tDecoderInit(&dc, pVal, vLen);




12,274,993✔





413



    code = metaDecodeEntry(&dc, &me);




12,281,065✔





414



    TSDB_CHECK_CODE(code, lino, END);




12,258,166✔





415



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {




12,258,166✔





416



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




280,883✔





417



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




257,224✔





418



        tDecoderClear(&dc);




23,952✔





419



        continue;




23,596✔





420



      }










421



    }










422












423



    TSDB_CHECK_NULL(taosArrayPush(ctx->idList, &tmp->uid), code, lino, END, terrno);




24,491,346✔





424



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




12,248,135✔





425



    tDecoderClear(&dc);




12,267,007✔





426












427



    SIdInfo info = {0};




12,267,247✔





428



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




12,267,882✔





429



    TSDB_CHECK_CODE(code, lino, END);




12,267,908✔





430



  }










431



  taosHashClear(ctx->idVersion);




127,879✔





432












433



  code = MoveToSnapShotVersion(ctx);




127,879✔





434



  TSDB_CHECK_CODE(code, lino, END);




127,547✔





435












436



  while (1) {




12,416,912✔





437



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




12,544,459✔





438



    if (ret < 0) break;




12,590,806✔





439












440



    STbDbKey* tmp = (STbDbKey*)pKey;




12,463,259✔





441



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




12,463,259✔





442



    if (idData) {




12,443,744✔





443



      continue;




166,042✔





444



    }










445



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




12,277,702✔





446



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




12,276,406✔





447



    TSDB_CHECK_CODE(code, lino, END);




12,288,078✔





448












449



    SMetaEntry me = {0};




12,288,078✔





450



    tDecoderInit(&dc, pVal, vLen);




12,236,562✔





451



    code = metaDecodeEntry(&dc, &me);




12,272,878✔





452



    TSDB_CHECK_CODE(code, lino, END);




12,261,329✔





453












454



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {




12,261,329✔





455



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




273,723✔





456



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




250,064✔





457



        tDecoderClear(&dc);




16,436✔





458



        continue;




16,436✔





459



      }










460



    }










461












462



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




12,264,981✔





463



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




12,117,082✔





464



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




173,449✔





465



      TSDB_CHECK_CODE(code, lino, END);




182,197✔





466



    }










467



    tDecoderClear(&dc);




12,285,998✔





468












469



  }










470












471



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




12,396,617✔





472



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




12,269,062✔





473



    TSDB_CHECK_NULL(uid, code, lino, END, terrno);




12,268,122✔





474



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




12,268,122✔





475



    TSDB_CHECK_NULL(idData, code, lino, END, terrno);




12,268,778✔





476












477



    idData->index = i;




12,268,778✔





478



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




12,268,446✔





479



  }










480












481



END:




90,918✔





482



  tdbFree(pKey);




127,879✔





483



  tdbFree(pVal);




127,547✔





484



  tDecoderClear(&dc);




127,879✔





485












486



  if (ctx != NULL) {




127,879✔





487



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




127,879✔





488



    ctx->pCur = NULL;




127,547✔





489



  }










490



  metaULock(pVnode->pMeta);




127,879✔





491












492



  if(code != 0) {




127,608✔





493



    destroySnapContext(ctx);




×







494



    *ctxRet = NULL;




×







495



    metaError("tmqsnap build snap context failed line:%d since %s", lino, tstrerror(code));




×







496



  }










497



  return code;




127,608✔





498



}










499












500



void destroySnapContext(SSnapContext* ctx) {




594,422✔





501



  if (ctx == NULL) {




594,422✔





502



    return;




468,396✔





503



  }










504



  taosArrayDestroy(ctx->idList);




126,026✔





505



  taosHashCleanup(ctx->idVersion);




127,879✔





506



  taosHashCleanup(ctx->suidInfo);




127,879✔





507



  taosMemoryFree(ctx);




127,879✔





508



}










509












510



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




30,322✔





511



  int32_t            ret = 0;




30,322✔





512



  SVCreateTbBatchReq reqs = {0};




30,322✔





513












514



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




30,322✔





515



  if (NULL == reqs.pArray) {




30,322✔





516



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);




×







517



    goto end;




×







518



  }










519



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




60,644✔





520



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);




×







521



    goto end;




×







522



  }










523



  reqs.nReqs = 1;




30,322✔





524












525



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




30,322✔





526



  if (ret < 0) {




30,322✔





527



    ret = TAOS_GET_TERRNO(ret);




×







528



    goto end;




×







529



  }










530



  *contLen += sizeof(SMsgHead);




30,322✔





531



  *pBuf = taosMemoryMalloc(*contLen);




30,322✔





532



  if (NULL == *pBuf) {




30,322✔





533



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);




×







534



    goto end;




×







535



  }










536



  SEncoder coder = {0};




30,322✔





537



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




30,322✔





538



  ret = tEncodeSVCreateTbBatchReq(&coder, &reqs);




30,322✔





539



  tEncoderClear(&coder);




30,322✔





540












541



  if (ret < 0) {




30,322✔





542



    taosMemoryFreeClear(*pBuf);




×







543



    ret = TAOS_GET_TERRNO(ret);




×







544



    goto end;




×







545



  }










546












547



end:




30,322✔





548



  taosArrayDestroy(reqs.pArray);




30,322✔





549



  return ret;




30,322✔





550



}










551












552



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




12,160✔





553



  int32_t ret = 0;




12,160✔





554



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




12,160✔





555



  if (ret < 0) {




12,160✔





556



    return TAOS_GET_TERRNO(ret);




×







557



  }










558












559



  *contLen += sizeof(SMsgHead);




12,160✔





560



  *pBuf = taosMemoryMalloc(*contLen);




12,160✔





561



  if (NULL == *pBuf) {




12,160✔





562



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);




×







563



  }










564












565



  SEncoder encoder = {0};




12,160✔





566



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




12,160✔





567



  ret = tEncodeSVCreateStbReq(&encoder, req);




12,160✔





568



  tEncoderClear(&encoder);




12,160✔





569



  if (ret < 0) {




12,160✔





570



    taosMemoryFreeClear(*pBuf);




×







571



    return TAOS_GET_TERRNO(ret);




×







572



  }










573



  return 0;




12,160✔





574



}










575












576



int32_t setForSnapShot(SSnapContext* ctx, int64_t uid) {




794,673✔





577



  if (uid == 0) {




794,673✔





578



    ctx->index = 0;




17,418✔





579



    return 0;




17,418✔





580



  }










581












582



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




777,255✔





583



  if (idInfo == NULL) {




777,568✔





584



    return terrno;




×







585



  }










586












587



  ctx->index = idInfo->index;




777,568✔





588












589



  return 0;




777,255✔





590



}










591












592



void taosXSetTablePrimaryKey(SSnapContext* ctx, int64_t uid) {




787,801✔





593



  bool            ret = false;




787,801✔





594



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




787,801✔





595



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




787,488✔





596



    ret = true;




1,360✔





597



  }










598



  tDeleteSchemaWrapper(schema);










599



  ctx->hasPrimaryKey = ret;




786,862✔





600



}




787,175✔





601












602



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




1,568,108✔





603












604



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




48,130✔





605



  int32_t ret = 0;




48,130✔





606



  int32_t lino = 0;




48,130✔





607



  void*   pKey = NULL;




48,130✔





608



  void*   pVal = NULL;




48,130✔





609



  int     vLen = 0, kLen = 0;




48,130✔





610



  SDecoder   dc = {0};




48,130✔





611



  SArray* tagName = NULL;




48,130✔





612



  SArray* pTagVals = NULL;




48,130✔





613












614



  metaRLock(ctx->pMeta);




48,130✔





615



  while (1) {




×







616



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




48,130✔





617



      metaDebug("tmqsnap get meta end");




5,648✔





618



      ctx->index = 0;




5,648✔





619



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




5,648✔





620



      goto END;




5,648✔





621



    }










622












623



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




42,482✔





624



    TSDB_CHECK_NULL(uidTmp, ret, lino, END, terrno);




42,482✔





625



    ctx->index++;




42,482✔





626



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




42,482✔





627



    TSDB_CHECK_NULL(idInfo, ret, lino, END, terrno);




42,482✔





628












629



    *uid = *uidTmp;




42,482✔





630



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




42,482✔





631



    if (ret == 0) {




42,482✔





632



      break;




42,482✔





633



    }










634



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




×







635



  }










636












637



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




42,482✔





638



  TSDB_CHECK_CONDITION(ret >= 0, ret, lino, END, TAOS_GET_TERRNO(ret));




42,482✔





639



  SMetaEntry me = {0};




42,482✔





640



  tDecoderInit(&dc, pVal, vLen);




42,482✔





641



  ret = metaDecodeEntry(&dc, &me);




42,482✔





642



  TSDB_CHECK_CONDITION(ret >= 0, ret, lino, END, TAOS_GET_TERRNO(ret));




42,482✔





643



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




42,482✔





644












645



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




42,482✔





646



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




42,840✔





647



    SVCreateStbReq req = {0};




12,160✔





648



    req.name = me.name;




12,160✔





649



    req.suid = me.uid;




12,160✔





650



    req.schemaRow = me.stbEntry.schemaRow;




12,160✔





651



    req.schemaTag = me.stbEntry.schemaTag;




12,160✔





652



    req.schemaRow.version = 1;




12,160✔





653



    req.schemaTag.version = 1;




12,160✔





654



    req.colCmpr = me.colCmpr;




12,160✔





655



    req.pExtSchemas = me.pExtSchemas;




12,160✔





656



    req.virtualStb = TABLE_IS_VIRTUAL(me.flags);




12,160✔





657












658



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




12,160✔





659



    *type = TDMT_VND_CREATE_STB;




12,160✔





660



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




30,322✔





661



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




31,038✔





662



    STableInfoForChildTable* data =










663



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




24,756✔





664



    TSDB_CHECK_NULL(data, ret, lino, END, terrno);




24,756✔





665












666



    SVCreateTbReq req = {0};




24,756✔





667












668



    req.type = me.type;




24,756✔





669



    req.name = me.name;




24,756✔





670



    req.uid = me.uid;




24,756✔





671



    req.commentLen = -1;




24,756✔





672



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




24,756✔





673



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




24,756✔





674



    req.ctb.stbName = data->tableName;




24,756✔





675












676



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




24,756✔





677



    TSDB_CHECK_NULL(tagName, ret, lino, END, terrno);




24,756✔





678



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




24,756✔





679



    if (tTagIsJson(p)) {




24,756✔





680



      if (p->nTag != 0) {




3,252✔





681



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




1,626✔





682



        TSDB_CHECK_NULL(taosArrayPush(tagName, schema->name), ret, lino, END, terrno);




3,252✔





683



      }










684



    } else {










685



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




21,504✔





686



      TSDB_CHECK_CODE(ret, lino, END);




21,504✔





687



      int16_t nCols = taosArrayGetSize(pTagVals);




21,504✔





688



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




70,108✔





689



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




48,604✔





690



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




191,516✔





691



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




142,912✔





692



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




142,912✔





693



            TSDB_CHECK_NULL(taosArrayPush(tagName, schema->name), ret, lino, END, terrno);




97,208✔





694



          }










695



        }










696



      }










697



    }










698



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




24,756✔





699



    req.ctb.tagName = tagName;




24,756✔





700



    req.colRef = me.colRef;




24,756✔





701



    if (me.type == TSDB_VIRTUAL_CHILD_TABLE) {




24,756✔





702



      SMetaEntry *pSuper = NULL;




2,148✔





703



      int32_t code = metaFetchEntryByUid(ctx->pMeta, me.ctbEntry.suid, &pSuper);




2,148✔





704



      if (code == 0) {




2,148✔





705



        for (int i = 0; i < req.colRef.nCols && i < pSuper->stbEntry.schemaRow.nCols; i++) {




10,740✔





706



          SColRef *p = &req.colRef.pColRef[i];




8,592✔





707



          if (p->hasRef) {




8,592✔





708



            SSchema *schema = &pSuper->stbEntry.schemaRow.pSchema[i];




4,296✔





709



            tstrncpy(p->colName, schema->name, TSDB_COL_NAME_LEN);




4,296✔





710



          }










711



        }










712



      }










713



      metaFetchEntryFree(&pSuper);




2,148✔





714



    }










715



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




24,756✔





716



    *type = TDMT_VND_CREATE_TABLE;




24,756✔





717



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




11,132✔





718



    SVCreateTbReq req = {0};




5,566✔





719



    req.type = me.type;




5,566✔





720



    req.name = me.name;




5,566✔





721



    req.uid = me.uid;




5,566✔





722



    req.commentLen = -1;




5,566✔





723



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




5,566✔





724



    req.colCmpr = me.colCmpr;




5,566✔





725



    req.pExtSchemas = me.pExtSchemas;




5,566✔





726



    req.colRef = me.colRef;




5,566✔





727



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




5,566✔





728



    *type = TDMT_VND_CREATE_TABLE;




5,566✔





729



  } else {










730



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




×







731



    ret = TSDB_CODE_SDB_INVALID_TABLE_TYPE;




×







732



  }










733












734



END:




48,130✔





735



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




48,130✔





736



  ctx->pCur = NULL;




48,130✔





737



  taosArrayDestroy(pTagVals);




48,130✔





738



  taosArrayDestroy(tagName);




48,130✔





739



  tDecoderClear(&dc);




48,130✔





740



  metaULock(ctx->pMeta);




48,130✔





741












742



  if(ret != 0) {




48,130✔





743



    metaError("tmqsnap get table info from snapshot failed line:%d since %s", lino, tstrerror(ret));




×







744



  }










745



  return ret;




48,130✔





746



}










747












748



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




1,576,420✔





749



  void* pKey = NULL;




1,576,420✔





750



  void* pVal = NULL;




1,576,420✔





751



  int   vLen = 0;




1,576,420✔





752



  int   kLen = 0;




1,576,733✔





753



  int32_t code = 0;




1,576,733✔





754



  int32_t lino = 0;




1,576,733✔





755



  SDecoder   dc = {0};




1,576,733✔





756












757



  metaRLock(ctx->pMeta);




1,576,733✔





758



  while (1) {




23,898✔





759



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




1,600,318✔





760



      metaDebug("tmqsnap get uid info end");




11,662✔





761



      goto END;




11,662✔





762



    }










763



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




1,589,282✔





764



    TSDB_CHECK_NULL(uidTmp, code, lino, END, terrno);




1,589,282✔





765



    ctx->index++;




1,589,282✔





766



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




1,588,030✔





767



    TSDB_CHECK_NULL(idInfo, code, lino, END, terrno);




1,588,650✔





768












769



    if (MoveToPosition(ctx, idInfo->version, *uidTmp) != 0) {




1,588,650✔





770



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




×







771



                idInfo->version);










772



      continue;




×







773



    }










774



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




1,588,030✔





775



    TSDB_CHECK_CODE(code, lino, END);




1,589,282✔





776



    SMetaEntry me = {0};




1,589,282✔





777



    tDecoderInit(&dc, pVal, vLen);




1,589,282✔





778



    code = metaDecodeEntry(&dc, &me);




1,588,969✔





779



    TSDB_CHECK_CODE(code, lino, END);




1,587,349✔





780



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




1,587,349✔





781












782



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




1,587,349✔





783



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




1,587,126✔





784



      STableInfoForChildTable* data =










785



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




309,963✔





786



      TSDB_CHECK_NULL(data, code, lino, END, terrno);




309,337✔





787



      result->suid = me.ctbEntry.suid;




309,337✔





788



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




618,993✔





789



      if (data->pExtSchemas != NULL) {




309,656✔





790



        result->pExtSchemas = taosMemoryMalloc(sizeof(SExtSchema) * data->schemaRow->nCols);




8,454✔





791



        TSDB_CHECK_NULL(result->pExtSchemas, code, lino, END, terrno);




8,454✔





792



        memcpy(result->pExtSchemas, data->pExtSchemas, sizeof(SExtSchema) * data->schemaRow->nCols);




8,454✔





793



      }










794



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




1,278,687✔





795



      result->suid = 0;




1,254,789✔





796



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




1,252,911✔





797



      if (me.pExtSchemas != NULL) {




1,251,659✔





798



        result->pExtSchemas = taosMemoryMalloc(sizeof(SExtSchema) * me.ntbEntry.schemaRow.nCols);




×







799



        TSDB_CHECK_NULL(result->pExtSchemas, code, lino, END, terrno);




×







800



        memcpy(result->pExtSchemas, me.pExtSchemas, sizeof(SExtSchema) * me.ntbEntry.schemaRow.nCols);




×







801



      }










802



    } else {










803



      metaDebug("tmqsnap get uid continue");




23,272✔





804



      tDecoderClear(&dc);




23,272✔





805



      continue;




23,898✔





806



    }










807



    result->uid = me.uid;




1,560,996✔





808



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




1,560,327✔





809



    TSDB_CHECK_NULL(result->schema, code, lino, END, TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY));




1,561,892✔





810



    break;




1,559,376✔





811



  }










812












813



END:




1,574,769✔





814



  tDecoderClear(&dc);




1,575,745✔





815



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




1,575,481✔





816



  ctx->pCur = NULL;




1,574,229✔





817



  metaULock(ctx->pMeta);




1,575,481✔





818












819



  if (code != 0) {




1,571,719✔





820



    metaError("tmqsnap get meta table info from snapshot failed line:%d since %s", lino, tstrerror(code));




×







821



  }










822



  return code;




1,571,719✔





823



}
























    

  • 

    •