taosdata / TDengine / #4937

coverage: 66.794% (+0.05%) from 66.746%
#4937

push

travis-ci

web-flow 
fix: case failuer caused by the modification of the error description (#34391)

204171 of 305671 relevant lines covered (66.79%)

127025939.33 hits per line

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

84.97
/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



  int32_t iLoop;










25



};










26












27



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




27,469✔





28



  int32_t          code = 0;




27,469✔





29



  int32_t          lino;










30



  int32_t          c = 0;




27,469✔





31



  SMetaSnapReader* pReader = NULL;




27,469✔





32












33



  // alloc










34



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




27,469✔





35



  if (pReader == NULL) {




27,469✔





36



    TSDB_CHECK_CODE(code = terrno, lino, _exit);




×







37



  }










38



  pReader->pMeta = pMeta;




27,469✔





39



  pReader->sver = sver;




27,469✔





40



  pReader->ever = ever;




27,469✔





41












42



  // impl










43



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




27,469✔





44



  TSDB_CHECK_CODE(code, lino, _exit);




27,469✔





45












46



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




27,469✔





47



  TSDB_CHECK_CODE(code, lino, _exit);




27,469✔





48












49



_exit:




27,469✔





50



  if (code) {




27,469✔





51



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




×







52



    metaSnapReaderClose(&pReader);




×







53



    *ppReader = NULL;




×







54



  } else {










55



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




27,469✔





56



    *ppReader = pReader;




27,469✔





57



  }










58



  return code;




27,469✔





59



}










60












61



void metaSnapReaderClose(SMetaSnapReader** ppReader) {




27,469✔





62



  if (ppReader && *ppReader) {




27,469✔





63



    tdbTbcClose((*ppReader)->pTbc);




27,469✔





64



    taosMemoryFree(*ppReader);




27,469✔





65



    *ppReader = NULL;




27,469✔





66



  }










67



}




27,469✔





68












69



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










70












71



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




316,430✔





72



  SDecoder decoder = {0};




316,430✔





73



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




316,430✔





74












75



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




316,430✔





76



  if (code) {




316,430✔





77



    tDecoderClear(&decoder);




×







78



    return code;




×







79



  }










80












81



  tDecoderClear(&decoder);




316,430✔





82



  return 0;




316,430✔





83



}










84












85



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




185,360✔





86



  int32_t     code = 0;




185,360✔





87



  const void* pKey = NULL;




185,360✔





88



  const void* pData = NULL;




185,360✔





89



  int32_t     nKey = 0;




185,360✔





90



  int32_t     nData = 0;




185,360✔





91



  STbDbKey    key;










92



  int32_t     c;




185,360✔





93












94



  *ppData = NULL;




185,360✔





95



  while (pReader->iLoop < 2) {




398,837✔





96



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




371,368✔





97



      pReader->iLoop++;




54,938✔





98












99



      // Reopen the cursor to read from the beginning










100



      tdbTbcClose(pReader->pTbc);




54,938✔





101



      pReader->pTbc = NULL;




54,938✔





102



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




54,938✔





103



      if (code) {




54,938✔





104



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




×







105



                  tstrerror(code));










106



        goto _exit;




×







107



      }










108












109



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




54,938✔





110



      if (code) {




54,938✔





111



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




×







112



                  tstrerror(code));










113



        goto _exit;




×







114



      }










115












116



      continue;




54,938✔





117



    }










118












119



    // Decode meta entry










120



    SMetaEntry entry = {0};




316,430✔





121



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




316,430✔





122



    if (code) {




316,430✔





123



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




×







124



                tstrerror(code));










125



      goto _exit;




×







126



    }










127












128



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




316,430✔





129



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




316,430✔





130



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




315,782✔





131



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




182,118✔





132



    ) {










133



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




158,539✔





134



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




×







135



      }










136



      continue;




158,539✔





137



    }










138












139



    if (!pData || !nData) {




157,891✔





140



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




×







141



      goto _exit;




×







142



    }










143












144



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




157,891✔





145



    if (*ppData == NULL) {




157,891✔





146



      code = terrno;




×







147



      goto _exit;




×







148



    }










149












150



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




157,891✔





151



    pHdr->type = SNAP_DATA_META;




157,891✔





152



    pHdr->size = nData;




157,891✔





153



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




157,891✔





154












155



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




157,891✔





156



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










157












158



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




157,891✔





159



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




×







160



    }










161



    break;




157,891✔





162



  }










163












164



_exit:




185,360✔





165



  if (code) {




185,360✔





166



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




×







167



              tstrerror(code));










168



  }










169



  return code;




185,360✔





170



}










171












172



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










173



struct SMetaSnapWriter {










174



  SMeta*  pMeta;










175



  int64_t sver;










176



  int64_t ever;










177



};










178












179



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




24,126✔





180



  int32_t          code = 0;




24,126✔





181



  int32_t          lino;










182



  SMetaSnapWriter* pWriter;










183












184



  // alloc










185



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




24,126✔





186



  if (pWriter == NULL) {




24,126✔





187



    TSDB_CHECK_CODE(code = terrno, lino, _exit);




×







188



  }










189



  pWriter->pMeta = pMeta;




24,126✔





190



  pWriter->sver = sver;




24,126✔





191



  pWriter->ever = ever;




24,126✔





192












193



  code = metaBegin(pMeta, META_BEGIN_HEAP_NIL);




24,126✔





194



  TSDB_CHECK_CODE(code, lino, _exit);




24,126✔





195












196



_exit:




24,126✔





197



  if (code) {




24,126✔





198



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




×







199



    taosMemoryFree(pWriter);




×







200



    *ppWriter = NULL;




×







201



  } else {










202



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




24,126✔





203



    *ppWriter = pWriter;




24,126✔





204



  }










205



  return code;




24,126✔





206



}










207












208



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




24,126✔





209



  int32_t          code = 0;




24,126✔





210



  SMetaSnapWriter* pWriter = *ppWriter;




24,126✔





211












212



  if (rollback) {




24,126✔





213



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




×







214



    code = metaAbort(pWriter->pMeta);




×







215



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




×







216



             code);










217



    if (code) goto _err;




×







218



  } else {










219



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




24,126✔





220



    if (code) goto _err;




24,126✔





221



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




24,126✔





222



    if (code) goto _err;




24,126✔





223



  }










224



  taosMemoryFree(pWriter);




24,126✔





225



  *ppWriter = NULL;




24,126✔





226












227



  return code;




24,126✔





228












229



_err:




×







230



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




×







231



  return code;




×







232



}










233












234



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




157,858✔





235



  int32_t    code = 0;




157,858✔





236



  int32_t    lino = 0;




157,858✔





237



  SMeta*     pMeta = pWriter->pMeta;




157,858✔





238



  SMetaEntry metaEntry = {0};




157,858✔





239



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




157,858✔





240












241



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




157,858✔





242



  code = metaDecodeEntry(pDecoder, &metaEntry);




157,858✔





243



  TSDB_CHECK_CODE(code, lino, _exit);




157,858✔





244












245



  metaHandleSyncEntry(pMeta, &metaEntry);




157,858✔





246












247



_exit:




157,858✔





248



  if (code) {




157,858✔





249



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




×







250



  }










251



  tDecoderClear(pDecoder);




157,858✔





252



  return code;




157,858✔





253



}










254












255



typedef struct STableInfoForChildTable {










256



  char*           tableName;










257



  SSchemaWrapper* schemaRow;










258



  SSchemaWrapper* tagRow;










259



  SExtSchema*     pExtSchemas;










260



} STableInfoForChildTable;










261












262



static void destroySTableInfoForChildTable(void* data) {




165,696✔





263



  STableInfoForChildTable* pData = (STableInfoForChildTable*)data;




165,696✔





264



  taosMemoryFree(pData->tableName);




165,696✔





265



  tDeleteSchemaWrapper(pData->schemaRow);




165,141✔





266



  tDeleteSchemaWrapper(pData->tagRow);




165,696✔





267



  taosMemoryFreeClear(pData->pExtSchemas);




165,696✔





268



}




165,446✔





269












270



static int32_t MoveToSnapShotVersion(SSnapContext* ctx) {




114,980✔





271



  int32_t code = 0;




114,980✔





272



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




114,980✔





273



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




114,980✔





274



  if (code != 0) {




114,690✔





275



    return TAOS_GET_TERRNO(code);




×







276



  }










277



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




114,690✔





278



  int      c = 0;




114,690✔





279



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




114,690✔





280



  if (code != 0) {




114,980✔





281



    return TAOS_GET_TERRNO(code);




×







282



  }










283



  if (c < 0) {




114,980✔





284



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




5,354✔





285



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




×







286



    }










287



  }










288



  return 0;




114,980✔





289



}










290












291



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




1,470,004✔





292



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




1,470,004✔





293



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




1,469,720✔





294



  if (code != 0) {




1,469,146✔





295



    return TAOS_GET_TERRNO(code);




×







296



  }










297



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




1,469,146✔





298



  int      c = 0;




1,469,436✔





299



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




1,470,004✔





300



  if (code != 0) {




1,469,720✔





301



    return TAOS_GET_TERRNO(code);




×







302



  }










303



  return c;




1,469,720✔





304



}










305












306



static int32_t MoveToFirst(SSnapContext* ctx) {




114,980✔





307



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




114,980✔





308



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




114,980✔





309



  if (code != 0) {




114,980✔





310



    return TAOS_GET_TERRNO(code);




×







311



  }










312



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




114,980✔





313



  if (code != 0) {




114,980✔





314



    return TAOS_GET_TERRNO(code);




×







315



  }










316



  return 0;




114,980✔





317



}










318












319



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




165,696✔





320



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




165,696✔





321



  if (data) {




165,696✔





322



    return 0;




×







323



  }










324



  int32_t                 code = 0;




165,696✔





325



  STableInfoForChildTable dataTmp = {0};




165,696✔





326



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




165,696✔





327



  if (dataTmp.tableName == NULL) {




165,696✔





328



    code = terrno;




×







329



    goto END;




×







330



  }










331



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




165,696✔





332



  if (dataTmp.schemaRow == NULL) {




165,696✔





333



    code = TSDB_CODE_OUT_OF_MEMORY;




×







334



    goto END;




×







335



  }










336



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




165,696✔





337



  if (dataTmp.tagRow == NULL) {




165,696✔





338



    code = TSDB_CODE_OUT_OF_MEMORY;




×







339



    goto END;




×







340



  }










341



  if (me->pExtSchemas != NULL) {




165,696✔





342



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




7,416✔





343



    if (dataTmp.pExtSchemas == NULL) {




7,416✔





344



      code = TSDB_CODE_OUT_OF_MEMORY;




×







345



      goto END;




×







346



    }










347



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




7,416✔





348



  }










349



  










350



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




165,696✔





351



  if (code != 0) {




165,696✔





352



    goto END;




×







353



  }










354



  return 0;




165,696✔





355












356



END:




×







357



  destroySTableInfoForChildTable(&dataTmp);




×







358



  return TAOS_GET_TERRNO(code);




×







359



}










360












361



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




114,354✔





362



                         SSnapContext** ctxRet) {










363



  int32_t code = 0;




114,354✔





364



  int32_t lino = 0;




114,354✔





365



  SDecoder   dc = {0};




114,354✔





366



  void* pKey = NULL;




114,980✔





367



  void* pVal = NULL;




114,673✔





368



  int   vLen = 0, kLen = 0;




114,673✔





369












370



  metaRLock(pVnode->pMeta);




114,980✔





371



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




114,980✔





372



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




114,980✔





373



  *ctxRet = ctx;




114,980✔





374



  ctx->pMeta = pVnode->pMeta;




114,980✔





375



  ctx->snapVersion = snapVersion;




114,980✔





376



  ctx->suid = suid;




114,980✔





377



  ctx->subType = subType;




114,980✔





378



  ctx->queryMeta = withMeta;




114,673✔





379



  ctx->withMeta = withMeta;




114,980✔





380



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




114,661✔





381



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




114,980✔





382



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




114,980✔





383



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




114,980✔





384



  taosHashSetFreeFp(ctx->suidInfo, destroySTableInfoForChildTable);




114,980✔





385












386



  ctx->index = 0;




114,980✔





387



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




114,980✔





388



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




114,980✔





389












390



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




114,980✔





391



  code = MoveToFirst(ctx);




114,980✔





392



  TSDB_CHECK_CODE(code, lino, END);




114,980✔





393



  while (1) {




11,552,533✔





394



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




11,667,513✔





395



    if (ret < 0) break;




11,670,874✔





396



    STbDbKey* tmp = (STbDbKey*)pKey;




11,561,850✔





397



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




11,561,850✔





398












399



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




11,556,497✔





400



    if (idData) {




11,556,713✔





401



      continue;




83,361✔





402



    }










403












404



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










405



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




11,473,352✔





406



      continue;




85,250✔





407



    }










408












409



    SMetaEntry me = {0};




11,377,060✔





410



    tDecoderInit(&dc, pVal, vLen);




11,378,285✔





411



    code = metaDecodeEntry(&dc, &me);




11,382,391✔





412



    TSDB_CHECK_CODE(code, lino, END);




11,358,130✔





413



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {




11,358,130✔





414



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




136,793✔





415



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




136,543✔





416



        tDecoderClear(&dc);




1,000✔





417



        continue;




1,000✔





418



      }










419



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




11,231,989✔





420



      if (me.type == TSDB_VIRTUAL_NORMAL_TABLE ||




11,240,196✔





421



          me.type == TSDB_VIRTUAL_CHILD_TABLE ||




11,236,572✔





422



          TABLE_IS_VIRTUAL(me.flags)) {




11,239,325✔





423



        tDecoderClear(&dc);




894✔





424



        continue;




×







425



      }










426



    }










427












428



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




22,746,422✔





429



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




11,378,007✔





430



    tDecoderClear(&dc);




11,362,013✔





431












432



    SIdInfo info = {0};




11,389,845✔





433



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




11,390,130✔





434



    TSDB_CHECK_CODE(code, lino, END);




11,385,568✔





435



  }










436



  taosHashClear(ctx->idVersion);




114,980✔





437












438



  code = MoveToSnapShotVersion(ctx);




114,980✔





439



  TSDB_CHECK_CODE(code, lino, END);




114,980✔





440












441



  while (1) {




11,556,023✔





442



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




11,671,003✔





443



    if (ret < 0) break;




11,669,527✔





444












445



    STbDbKey* tmp = (STbDbKey*)pKey;




11,554,547✔





446



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




11,554,547✔





447



    if (idData) {




11,544,848✔





448



      continue;




139,611✔





449



    }










450



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




11,405,237✔





451



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




11,402,519✔





452



    TSDB_CHECK_CODE(code, lino, END);




11,223,901✔





453












454



    SMetaEntry me = {0};




11,223,901✔





455



    tDecoderInit(&dc, pVal, vLen);




11,392,813✔





456



    code = metaDecodeEntry(&dc, &me);




11,413,365✔





457



    TSDB_CHECK_CODE(code, lino, END);




11,410,295✔





458












459



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {




11,410,295✔





460



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




136,793✔





461



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




136,543✔





462



        tDecoderClear(&dc);




1,000✔





463



        continue;




1,000✔





464



      }










465



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




11,274,653✔





466



      if (me.type == TSDB_VIRTUAL_NORMAL_TABLE ||




11,275,607✔





467



          me.type == TSDB_VIRTUAL_CHILD_TABLE ||




11,274,997✔





468



          TABLE_IS_VIRTUAL(me.flags)) {




11,275,912✔





469



        tDecoderClear(&dc);




×







470



        continue;




×







471



      }










472



    }










473












474



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




11,409,892✔





475



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




11,269,073✔





476



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




164,183✔





477



      TSDB_CHECK_CODE(code, lino, END);




165,696✔





478



    }










479



    tDecoderClear(&dc);




11,415,023✔





480












481



  }










482












483



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




11,505,603✔





484



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




11,390,654✔





485



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




11,390,089✔





486



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




11,390,089✔





487



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




11,391,539✔





488












489



    idData->index = i;




11,391,539✔





490



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




11,391,846✔





491



  }










492












493



END:




97,846✔





494



  tdbFree(pKey);




114,980✔





495



  tdbFree(pVal);




114,980✔





496



  tDecoderClear(&dc);




114,980✔





497












498



  if (ctx != NULL) {




114,980✔





499



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




114,980✔





500



    ctx->pCur = NULL;




114,980✔





501



  }










502



  metaULock(pVnode->pMeta);




114,980✔





503












504



  if(code != 0) {




114,980✔





505



    destroySnapContext(ctx);




×







506



    *ctxRet = NULL;




×







507



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




×







508



  }










509



  return code;




114,980✔





510



}










511












512



void destroySnapContext(SSnapContext* ctx) {




540,231✔





513



  if (ctx == NULL) {




540,231✔





514



    return;




426,420✔





515



  }










516



  taosArrayDestroy(ctx->idList);




113,811✔





517



  taosHashCleanup(ctx->idVersion);




114,730✔





518



  taosHashCleanup(ctx->suidInfo);




114,730✔





519



  taosMemoryFree(ctx);




114,980✔





520



}










521












522



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




23,380✔





523



  int32_t            ret = 0;




23,380✔





524



  SVCreateTbBatchReq reqs = {0};




23,380✔





525












526



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




23,380✔





527



  if (NULL == reqs.pArray) {




23,380✔





528



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);




×







529



    goto end;




×







530



  }










531



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




46,760✔





532



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);




×







533



    goto end;




×







534



  }










535



  reqs.nReqs = 1;




23,380✔





536












537



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




23,380✔





538



  if (ret < 0) {




23,380✔





539



    ret = TAOS_GET_TERRNO(ret);




×







540



    goto end;




×







541



  }










542



  *contLen += sizeof(SMsgHead);




23,380✔





543



  *pBuf = taosMemoryMalloc(*contLen);




23,380✔





544



  if (NULL == *pBuf) {




23,380✔





545



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);




×







546



    goto end;




×







547



  }










548



  SEncoder coder = {0};




23,380✔





549



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




23,380✔





550



  ret = tEncodeSVCreateTbBatchReq(&coder, &reqs);




23,380✔





551



  tEncoderClear(&coder);




23,380✔





552












553



  if (ret < 0) {




23,380✔





554



    taosMemoryFreeClear(*pBuf);




×







555



    ret = TAOS_GET_TERRNO(ret);




×







556



    goto end;




×







557



  }










558












559



end:




23,380✔





560



  taosArrayDestroy(reqs.pArray);




23,380✔





561



  return ret;




23,380✔





562



}










563












564



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




9,917✔





565



  int32_t ret = 0;




9,917✔





566



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




9,917✔





567



  if (ret < 0) {




9,917✔





568



    return TAOS_GET_TERRNO(ret);




×







569



  }










570












571



  *contLen += sizeof(SMsgHead);




9,917✔





572



  *pBuf = taosMemoryMalloc(*contLen);




9,917✔





573



  if (NULL == *pBuf) {




9,917✔





574



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);




×







575



  }










576












577



  SEncoder encoder = {0};




9,917✔





578



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




9,917✔





579



  ret = tEncodeSVCreateStbReq(&encoder, req);




9,917✔





580



  tEncoderClear(&encoder);




9,917✔





581



  if (ret < 0) {




9,917✔





582



    taosMemoryFreeClear(*pBuf);




×







583



    return TAOS_GET_TERRNO(ret);




×







584



  }










585



  return 0;




9,917✔





586



}










587












588



int32_t setForSnapShot(SSnapContext* ctx, int64_t uid) {




719,870✔





589



  if (uid == 0) {




719,870✔





590



    ctx->index = 0;




14,716✔





591



    return 0;




14,716✔





592



  }










593












594



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




705,154✔





595



  if (idInfo == NULL) {




705,154✔





596



    return terrno;




×







597



  }










598












599



  ctx->index = idInfo->index;




705,154✔





600












601



  return 0;




705,154✔





602



}










603












604



void taosXSetTablePrimaryKey(SSnapContext* ctx, int64_t uid) {




713,924✔





605



  bool            ret = false;




713,924✔





606



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




713,924✔





607



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




712,504✔





608



    ret = true;




1,268✔





609



  }










610



  tDeleteSchemaWrapper(schema);










611



  ctx->hasPrimaryKey = ret;




713,066✔





612



}




713,924✔





613












614



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




1,415,456✔





615












616



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




37,870✔





617



  int32_t ret = 0;




37,870✔





618



  int32_t lino = 0;




37,870✔





619



  void*   pKey = NULL;




37,870✔





620



  void*   pVal = NULL;




37,870✔





621



  int     vLen = 0, kLen = 0;




37,870✔





622



  SDecoder   dc = {0};




37,870✔





623



  SArray* tagName = NULL;




37,870✔





624



  SArray* pTagVals = NULL;




37,870✔





625












626



  metaRLock(ctx->pMeta);




37,870✔





627



  while (1) {




×







628



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




37,870✔





629



      metaDebug("tmqsnap get meta end");




4,573✔





630



      ctx->index = 0;




4,573✔





631



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




4,573✔





632



      goto END;




4,573✔





633



    }










634












635



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




33,297✔





636



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




33,297✔





637



    ctx->index++;




33,297✔





638



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




33,297✔





639



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




33,297✔





640












641



    *uid = *uidTmp;




33,297✔





642



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




33,297✔





643



    if (ret == 0) {




33,297✔





644



      break;




33,297✔





645



    }










646



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




×







647



  }










648












649



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




33,297✔





650



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




33,297✔





651



  SMetaEntry me = {0};




33,297✔





652



  tDecoderInit(&dc, pVal, vLen);




33,297✔





653



  ret = metaDecodeEntry(&dc, &me);




33,297✔





654



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




33,297✔





655



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




33,297✔





656












657



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




33,297✔





658



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




33,297✔





659



    SVCreateStbReq req = {0};




9,917✔





660



    req.name = me.name;




9,917✔





661



    req.suid = me.uid;




9,917✔





662



    req.schemaRow = me.stbEntry.schemaRow;




9,917✔





663



    req.schemaTag = me.stbEntry.schemaTag;




9,917✔





664



    req.schemaRow.version = 1;




9,917✔





665



    req.schemaTag.version = 1;




9,917✔





666



    req.colCmpr = me.colCmpr;




9,917✔





667



    req.pExtSchemas = me.pExtSchemas;




9,917✔





668












669



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




9,917✔





670



    *type = TDMT_VND_CREATE_STB;




9,917✔





671



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




23,380✔





672



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




23,380✔





673



    STableInfoForChildTable* data =










674



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




20,224✔





675



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




20,224✔





676












677



    SVCreateTbReq req = {0};




20,224✔





678












679



    req.type = TSDB_CHILD_TABLE;




20,224✔





680



    req.name = me.name;




20,224✔





681



    req.uid = me.uid;




20,224✔





682



    req.commentLen = -1;




20,224✔





683



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




20,224✔





684



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




20,224✔





685



    req.ctb.stbName = data->tableName;




20,224✔





686












687



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




20,224✔





688



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




20,224✔





689



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




20,224✔





690



    if (tTagIsJson(p)) {




20,224✔





691



      if (p->nTag != 0) {




3,000✔





692



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




1,500✔





693



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




3,000✔





694



      }










695



    } else {










696



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




17,224✔





697



      TSDB_CHECK_CODE(ret, lino, END);




17,224✔





698



      int16_t nCols = taosArrayGetSize(pTagVals);




17,224✔





699



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




59,448✔





700



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




42,224✔





701



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




171,448✔





702



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




129,224✔





703



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




129,224✔





704



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




84,448✔





705



          }










706



        }










707



      }










708



    }










709



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




20,224✔





710



    req.ctb.tagName = tagName;




20,224✔





711



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




20,224✔





712



    *type = TDMT_VND_CREATE_TABLE;




20,224✔





713



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




6,312✔





714



    SVCreateTbReq req = {0};




3,156✔





715



    req.type = TSDB_NORMAL_TABLE;




3,156✔





716



    req.name = me.name;




3,156✔





717



    req.uid = me.uid;




3,156✔





718



    req.commentLen = -1;




3,156✔





719



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




3,156✔





720



    req.colCmpr = me.colCmpr;




3,156✔





721



    req.pExtSchemas = me.pExtSchemas;




3,156✔





722



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




3,156✔





723



    *type = TDMT_VND_CREATE_TABLE;




3,156✔





724



  } else {










725



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




×







726



    ret = TSDB_CODE_SDB_INVALID_TABLE_TYPE;




×







727



  }










728












729



END:




37,870✔





730



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




37,870✔





731



  ctx->pCur = NULL;




37,870✔





732



  taosArrayDestroy(pTagVals);




37,870✔





733



  taosArrayDestroy(tagName);




37,870✔





734



  tDecoderClear(&dc);




37,870✔





735



  metaULock(ctx->pMeta);




37,870✔





736












737



  if(ret != 0) {




37,870✔





738



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




×







739



  }










740



  return ret;




37,870✔





741



}










742












743



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




1,428,822✔





744



  void* pKey = NULL;




1,428,822✔





745



  void* pVal = NULL;




1,428,822✔





746



  int   vLen = 0;




1,429,106✔





747



  int   kLen = 0;




1,428,822✔





748



  int32_t code = 0;




1,429,106✔





749



  int32_t lino = 0;




1,429,106✔





750



  SDecoder   dc = {0};




1,429,106✔





751












752



  metaRLock(ctx->pMeta);




1,429,106✔





753



  while (1) {




17,345✔





754



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




1,446,451✔





755



      metaDebug("tmqsnap get uid info end");




9,744✔





756



      goto END;




9,744✔





757



    }










758



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




1,436,707✔





759



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




1,436,707✔





760



    ctx->index++;




1,436,707✔





761



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




1,436,423✔





762



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




1,436,707✔





763












764



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




1,436,707✔





765



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




×







766



                idInfo->version);










767



      continue;




×







768



    }










769



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




1,436,423✔





770



    TSDB_CHECK_CODE(code, lino, END);




1,436,139✔





771



    SMetaEntry me = {0};




1,436,139✔





772



    tDecoderInit(&dc, pVal, vLen);




1,435,565✔





773



    code = metaDecodeEntry(&dc, &me);




1,436,133✔





774



    TSDB_CHECK_CODE(code, lino, END);




1,436,139✔





775



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




1,436,139✔





776












777



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




1,436,139✔





778



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




1,437,041✔





779



      STableInfoForChildTable* data =










780



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




280,330✔





781



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




280,904✔





782



      result->suid = me.ctbEntry.suid;




280,904✔





783



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




562,098✔





784



      if (data->pExtSchemas != NULL) {




281,194✔





785



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




7,806✔





786



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




7,806✔





787



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




7,806✔





788



      }










789



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




1,155,513✔





790



      result->suid = 0;




1,136,180✔





791



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




1,136,748✔





792



      if (me.pExtSchemas != NULL) {




1,136,748✔





793



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




×







794



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




×







795



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




×







796



      }










797



    } else {










798



      metaDebug("tmqsnap get uid continue");




18,481✔





799



      tDecoderClear(&dc);




18,481✔





800



      continue;




17,345✔





801



    }










802



    result->uid = me.uid;




1,417,942✔





803



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




1,417,374✔





804



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




1,415,380✔





805



    break;




1,415,090✔





806



  }










807












808



END:




1,427,402✔





809



  tDecoderClear(&dc);




1,427,686✔





810



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




1,428,248✔





811



  ctx->pCur = NULL;




1,426,260✔





812



  metaULock(ctx->pMeta);




1,427,396✔





813












814



  if (code != 0) {




1,425,124✔





815



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




×







816



  }










817



  return code;




1,425,124✔





818



}
























    

  • 

    •