taosdata / TDengine / #4143

coverage: 32.868% (-29.4%) from 62.238%
#4143

push

travis-ci

web-flow 
test: migrate stream cases (#31164)

76401 of 312956 branches covered (24.41%)

Branch coverage included in aggregate %.

128686 of 311012 relevant lines covered (41.38%)

579734.08 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



  int32_t iLoop;










25



};










26












27



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




×







28



  int32_t          code = 0;




×







29



  int32_t          lino;










30



  int32_t          c = 0;




×







31



  SMetaSnapReader* pReader = NULL;




×







32












33



  // alloc










34



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





×







35



  if (pReader == NULL) {





×







36



    TSDB_CHECK_CODE(code = terrno, lino, _exit);





×







37



  }










38



  pReader->pMeta = pMeta;




×







39



  pReader->sver = sver;




×







40



  pReader->ever = ever;




×







41












42



  // impl










43



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




×







44



  TSDB_CHECK_CODE(code, lino, _exit);





×







45












46



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




×







47



  TSDB_CHECK_CODE(code, lino, _exit);





×







48












49



_exit:




×







50



  if (code) {





×







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__);





×







56



    *ppReader = pReader;




×







57



  }










58



  return code;




×







59



}










60












61



void metaSnapReaderClose(SMetaSnapReader** ppReader) {




×







62



  if (ppReader && *ppReader) {





×







63



    tdbTbcClose((*ppReader)->pTbc);




×







64



    taosMemoryFree(*ppReader);





×







65



    *ppReader = NULL;




×







66



  }










67



}




×







68












69



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










70












71



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




×







72



  SDecoder decoder = {0};




×







73



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




×







74












75



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




×







76



  if (code) {





×







77



    tDecoderClear(&decoder);




×







78



    return code;




×







79



  }










80












81



  tDecoderClear(&decoder);




×







82



  return 0;




×







83



}










84












85



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




×







86



  int32_t     code = 0;




×







87



  const void* pKey = NULL;




×







88



  const void* pData = NULL;




×







89



  int32_t     nKey = 0;




×







90



  int32_t     nData = 0;




×







91



  STbDbKey    key;










92



  int32_t     c;










93












94



  *ppData = NULL;




×







95



  while (pReader->iLoop < 2) {





×







96



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





×







97



      pReader->iLoop++;




×







98












99



      // Reopen the cursor to read from the beginning










100



      tdbTbcClose(pReader->pTbc);




×







101



      pReader->pTbc = NULL;




×







102



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




×







103



      if (code) {





×







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);




×







110



      if (code) {





×







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;




×







117



    }










118












119



    // Decode meta entry










120



    SMetaEntry entry = {0};




×







121



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




×







122



    if (code) {





×







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];




×







129



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





×







130



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





×







131



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





×







132



    ) {










133



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





×







134



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





×







135



      }










136



      continue;




×







137



    }










138












139



    if (!pData || !nData) {





×







140



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





×







141



      goto _exit;




×







142



    }










143












144



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





×







145



    if (*ppData == NULL) {





×







146



      code = terrno;




×







147



      goto _exit;




×







148



    }










149












150



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




×







151



    pHdr->type = SNAP_DATA_META;




×







152



    pHdr->size = nData;




×







153



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




×







154












155



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





×







156



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










157












158



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





×







159



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





×







160



    }










161



    break;




×







162



  }










163












164



_exit:




×







165



  if (code) {





×







166



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





×







167



              tstrerror(code));










168



  }










169



  return code;




×







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) {




×







180



  int32_t          code = 0;




×







181



  int32_t          lino;










182



  SMetaSnapWriter* pWriter;










183












184



  // alloc










185



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





×







186



  if (pWriter == NULL) {





×







187



    TSDB_CHECK_CODE(code = terrno, lino, _exit);





×







188



  }










189



  pWriter->pMeta = pMeta;




×







190



  pWriter->sver = sver;




×







191



  pWriter->ever = ever;




×







192












193



  code = metaBegin(pMeta, META_BEGIN_HEAP_NIL);




×







194



  TSDB_CHECK_CODE(code, lino, _exit);





×







195












196



_exit:




×







197



  if (code) {





×







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__);





×







203



    *ppWriter = pWriter;




×







204



  }










205



  return code;




×







206



}










207












208



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




×







209



  int32_t          code = 0;




×







210



  SMetaSnapWriter* pWriter = *ppWriter;




×







211












212



  if (rollback) {





×







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);




×







220



    if (code) goto _err;





×







221



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




×







222



    if (code) goto _err;





×







223



  }










224



  taosMemoryFree(pWriter);





×







225



  *ppWriter = NULL;




×







226












227



  return code;




×







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) {




×







235



  int32_t    code = 0;




×







236



  int32_t    lino = 0;




×







237



  SMeta*     pMeta = pWriter->pMeta;




×







238



  SMetaEntry metaEntry = {0};




×







239



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




×







240












241



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




×







242



  code = metaDecodeEntry(pDecoder, &metaEntry);




×







243



  TSDB_CHECK_CODE(code, lino, _exit);





×







244












245



  metaHandleSyncEntry(pMeta, &metaEntry);




×







246












247



_exit:




×







248



  if (code) {





×







249



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





×







250



  }










251



  tDecoderClear(pDecoder);




×







252



  return code;




×







253



}










254












255



typedef struct STableInfoForChildTable {










256



  char*           tableName;










257



  SSchemaWrapper* schemaRow;










258



  SSchemaWrapper* tagRow;










259



} STableInfoForChildTable;










260












261



static void destroySTableInfoForChildTable(void* data) {




×







262



  STableInfoForChildTable* pData = (STableInfoForChildTable*)data;




×







263



  taosMemoryFree(pData->tableName);





×







264



  tDeleteSchemaWrapper(pData->schemaRow);





×







265



  tDeleteSchemaWrapper(pData->tagRow);





×







266



}




×







267












268



static int32_t MoveToSnapShotVersion(SSnapContext* ctx) {




×







269



  int32_t code = 0;




×







270



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




×







271



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




×







272



  if (code != 0) {





×







273



    return TAOS_GET_TERRNO(code);





×







274



  }










275



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




×







276



  int      c = 0;




×







277



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




×







278



  if (code != 0) {





×







279



    return TAOS_GET_TERRNO(code);





×







280



  }










281



  if (c < 0) {





×







282



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





×







283



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





×







284



    }










285



  }










286



  return 0;




×







287



}










288












289



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




×







290



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




×







291



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




×







292



  if (code != 0) {





×







293



    return TAOS_GET_TERRNO(code);





×







294



  }










295



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




×







296



  int      c = 0;




×







297



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




×







298



  if (code != 0) {





×







299



    return TAOS_GET_TERRNO(code);





×







300



  }










301



  return c;




×







302



}










303












304



static int32_t MoveToFirst(SSnapContext* ctx) {




×







305



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




×







306



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




×







307



  if (code != 0) {





×







308



    return TAOS_GET_TERRNO(code);





×







309



  }










310



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




×







311



  if (code != 0) {





×







312



    return TAOS_GET_TERRNO(code);





×







313



  }










314



  return 0;




×







315



}










316












317



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




×







318



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




×







319



  if (data) {





×







320



    return 0;




×







321



  }










322



  int32_t                 code = 0;




×







323



  STableInfoForChildTable dataTmp = {0};




×







324



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





×







325



  if (dataTmp.tableName == NULL) {





×







326



    code = terrno;




×







327



    goto END;




×







328



  }










329



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





×







330



  if (dataTmp.schemaRow == NULL) {





×







331



    code = TSDB_CODE_OUT_OF_MEMORY;




×







332



    goto END;




×







333



  }










334



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





×







335



  if (dataTmp.tagRow == NULL) {





×







336



    code = TSDB_CODE_OUT_OF_MEMORY;




×







337



    goto END;




×







338



  }










339



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




×







340



  if (code != 0) {





×







341



    goto END;




×







342



  }










343



  return 0;




×







344












345



END:




×







346



  destroySTableInfoForChildTable(&dataTmp);




×







347



  return TAOS_GET_TERRNO(code);





×







348



  ;










349



}










350












351



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




×







352



                         SSnapContext** ctxRet) {










353



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





×







354



  if (ctx == NULL) {





×







355



    return terrno;




×







356



  }










357



  *ctxRet = ctx;




×







358



  ctx->pMeta = pVnode->pMeta;




×







359



  ctx->snapVersion = snapVersion;




×







360



  ctx->suid = suid;




×







361



  ctx->subType = subType;




×







362



  ctx->queryMeta = withMeta;




×







363



  ctx->withMeta = withMeta;




×







364



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




×







365



  if (ctx->idVersion == NULL) {





×







366



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







367



  }










368












369



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




×







370



  if (ctx->suidInfo == NULL) {





×







371



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







372



  }










373



  taosHashSetFreeFp(ctx->suidInfo, destroySTableInfoForChildTable);




×







374












375



  ctx->index = 0;




×







376



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




×







377



  if (ctx->idList == NULL) {





×







378



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







379



    ;










380



  }










381



  void* pKey = NULL;




×







382



  void* pVal = NULL;




×







383



  int   vLen = 0, kLen = 0;




×







384












385



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





×







386



  int32_t code = MoveToFirst(ctx);




×







387



  if (code != 0) {





×







388



    return code;




×







389



  }










390



  while (1) {




×







391



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




×







392



    if (ret < 0) break;





×







393



    STbDbKey* tmp = (STbDbKey*)pKey;




×







394



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





×







395












396



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




×







397



    if (idData) {





×







398



      continue;




×







399



    }










400












401



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





×







402



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










403



      continue;




×







404



    }










405












406



    SDecoder   dc = {0};




×







407



    SMetaEntry me = {0};




×







408



    tDecoderInit(&dc, pVal, vLen);




×







409



    ret = metaDecodeEntry(&dc, &me);




×







410



    if (ret < 0) {





×







411



      tDecoderClear(&dc);




×







412



      return TAOS_GET_TERRNO(ret);





×







413



    }










414



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {





×







415



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





×







416



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





×







417



        tDecoderClear(&dc);




×







418



        continue;




×







419



      }










420



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





×







421



      if (me.type == TSDB_VIRTUAL_NORMAL_TABLE ||





×







422



          me.type == TSDB_VIRTUAL_CHILD_TABLE ||





×







423



          TABLE_IS_VIRTUAL(me.flags)) {





×







424



        tDecoderClear(&dc);




×







425



        continue;




×







426



      }










427



    }










428












429



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





×







430



      tDecoderClear(&dc);




×







431



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







432



    }










433



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





×







434



    tDecoderClear(&dc);




×







435












436



    SIdInfo info = {0};




×







437



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





×







438



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







439



    }










440



  }










441



  taosHashClear(ctx->idVersion);




×







442












443



  code = MoveToSnapShotVersion(ctx);




×







444



  if (code != 0) {





×







445



    return code;




×







446



  }










447



  while (1) {




×







448



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




×







449



    if (ret < 0) break;





×







450












451



    STbDbKey* tmp = (STbDbKey*)pKey;




×







452



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




×







453



    if (idData) {





×







454



      continue;




×







455



    }










456



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




×







457



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




×







458



    if (ret != 0) {





×







459



      return TAOS_GET_TERRNO(ret);





×







460



    }










461












462



    SDecoder   dc = {0};




×







463



    SMetaEntry me = {0};




×







464



    tDecoderInit(&dc, pVal, vLen);




×







465



    ret = metaDecodeEntry(&dc, &me);




×







466



    if (ret < 0) {





×







467



      tDecoderClear(&dc);




×







468



      return TAOS_GET_TERRNO(ret);





×







469



    }










470












471



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {





×







472



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





×







473



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





×







474



        tDecoderClear(&dc);




×







475



        continue;




×







476



      }










477



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





×







478



      if (me.type == TSDB_VIRTUAL_NORMAL_TABLE ||





×







479



          me.type == TSDB_VIRTUAL_CHILD_TABLE ||





×







480



          TABLE_IS_VIRTUAL(me.flags)) {





×







481



        tDecoderClear(&dc);




×







482



        continue;




×







483



      }










484



    }










485












486



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





×







487



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





×







488



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




×







489



      if (ret != 0) {





×







490



        tDecoderClear(&dc);




×







491



        return ret;




×







492



      }










493



    }










494



    tDecoderClear(&dc);




×







495



  }










496












497



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





×







498



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




×







499



    if (uid == NULL) {





×







500



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







501



    }










502



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




×







503



    if (!idData) {





×







504



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





×







505



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







506



    }










507












508



    idData->index = i;




×







509



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





×







510



              idData->index);










511



  }










512












513



  tdbFree(pKey);




×







514



  tdbFree(pVal);




×







515



  return TDB_CODE_SUCCESS;




×







516



}










517












518



void destroySnapContext(SSnapContext* ctx) {




×







519



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




×







520



  taosArrayDestroy(ctx->idList);




×







521



  taosHashCleanup(ctx->idVersion);




×







522



  taosHashCleanup(ctx->suidInfo);




×







523



  taosMemoryFree(ctx);





×







524



}




×







525












526



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




×







527



  int32_t            ret = 0;




×







528



  SVCreateTbBatchReq reqs = {0};




×







529












530



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




×







531



  if (NULL == reqs.pArray) {





×







532



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







533



    goto end;




×







534



  }










535



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





×







536



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







537



    goto end;




×







538



  }










539



  reqs.nReqs = 1;




×







540












541



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





×







542



  if (ret < 0) {





×







543



    ret = TAOS_GET_TERRNO(ret);





×







544



    goto end;




×







545



  }










546



  *contLen += sizeof(SMsgHead);




×







547



  *pBuf = taosMemoryMalloc(*contLen);





×







548



  if (NULL == *pBuf) {





×







549



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







550



    goto end;




×







551



  }










552



  SEncoder coder = {0};




×







553



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




×







554



  ret = tEncodeSVCreateTbBatchReq(&coder, &reqs);




×







555



  tEncoderClear(&coder);




×







556












557



  if (ret < 0) {





×







558



    taosMemoryFreeClear(*pBuf);





×







559



    ret = TAOS_GET_TERRNO(ret);





×







560



    goto end;




×







561



  }










562












563



end:




×







564



  taosArrayDestroy(reqs.pArray);




×







565



  return ret;




×







566



}










567












568



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




×







569



  int32_t ret = 0;




×







570



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





×







571



  if (ret < 0) {





×







572



    return TAOS_GET_TERRNO(ret);





×







573



  }










574












575



  *contLen += sizeof(SMsgHead);




×







576



  *pBuf = taosMemoryMalloc(*contLen);





×







577



  if (NULL == *pBuf) {





×







578



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







579



  }










580












581



  SEncoder encoder = {0};




×







582



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




×







583



  ret = tEncodeSVCreateStbReq(&encoder, req);




×







584



  tEncoderClear(&encoder);




×







585



  if (ret < 0) {





×







586



    taosMemoryFreeClear(*pBuf);





×







587



    return TAOS_GET_TERRNO(ret);





×







588



  }










589



  return 0;




×







590



}










591












592



int32_t setForSnapShot(SSnapContext* ctx, int64_t uid) {




×







593



  if (uid == 0) {





×







594



    ctx->index = 0;




×







595



    return 0;




×







596



  }










597












598



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




×







599



  if (idInfo == NULL) {





×







600



    return terrno;




×







601



  }










602












603



  ctx->index = idInfo->index;




×







604












605



  return 0;




×







606



}










607












608



void taosXSetTablePrimaryKey(SSnapContext* ctx, int64_t uid) {




×







609



  bool            ret = false;




×







610



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




×







611



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





×







612



    ret = true;




×







613



  }










614



  tDeleteSchemaWrapper(schema);










615



  ctx->hasPrimaryKey = ret;




×







616



}




×







617












618



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




×







619












620



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




×







621



  int32_t ret = 0;




×







622



  void*   pKey = NULL;




×







623



  void*   pVal = NULL;




×







624



  int     vLen = 0, kLen = 0;




×







625












626



  while (1) {




×







627



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





×







628



      metaDebug("tmqsnap get meta end");





×







629



      ctx->index = 0;




×







630



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




×







631



      return 0;




×







632



    }










633












634



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




×







635



    if (uidTmp == NULL) {





×







636



      metaError("tmqsnap get meta null uid");





×







637



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







638



    }










639



    ctx->index++;




×







640



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




×







641



    if (!idInfo) {





×







642



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





×







643



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







644



    }










645












646



    *uid = *uidTmp;




×







647



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




×







648



    if (ret == 0) {





×







649



      break;




×







650



    }










651



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





×







652



  }










653












654



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




×







655



  if (ret < 0) {





×







656



    return TAOS_GET_TERRNO(ret);





×







657



  }










658



  SDecoder   dc = {0};




×







659



  SMetaEntry me = {0};




×







660



  tDecoderInit(&dc, pVal, vLen);




×







661



  ret = metaDecodeEntry(&dc, &me);




×







662



  if (ret < 0) {





×







663



    tDecoderClear(&dc);




×







664



    ret = TAOS_GET_TERRNO(ret);





×







665



    goto END;




×







666



  }










667



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





×







668












669



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





×







670



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





×







671



    SVCreateStbReq req = {0};




×







672



    req.name = me.name;




×







673



    req.suid = me.uid;




×







674



    req.schemaRow = me.stbEntry.schemaRow;




×







675



    req.schemaTag = me.stbEntry.schemaTag;




×







676



    req.schemaRow.version = 1;




×







677



    req.schemaTag.version = 1;




×







678



    req.colCmpr = me.colCmpr;




×







679












680



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




×







681



    *type = TDMT_VND_CREATE_STB;




×







682



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





×







683



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





×







684



    STableInfoForChildTable* data =










685



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




×







686



    if (!data) {





×







687



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





×







688



      ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







689



      goto END;




×







690



    }










691












692



    SVCreateTbReq req = {0};




×







693












694



    req.type = TSDB_CHILD_TABLE;




×







695



    req.name = me.name;




×







696



    req.uid = me.uid;




×







697



    req.commentLen = -1;




×







698



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




×







699



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




×







700



    req.ctb.stbName = data->tableName;




×







701












702



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




×







703



    if (tagName == NULL) {





×







704



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





×







705



      ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







706



      goto END;




×







707



    }










708



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




×







709



    if (tTagIsJson(p)) {





×







710



      if (p->nTag != 0) {





×







711



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




×







712



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





×







713



          ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







714



          taosArrayDestroy(tagName);




×







715



          goto END;




×







716



        }










717



      }










718



    } else {










719



      SArray* pTagVals = NULL;




×







720



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




×







721



      if (ret != 0) {





×







722



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





×







723



        taosArrayDestroy(pTagVals);




×







724



        taosArrayDestroy(tagName);




×







725



        goto END;




×







726



      }










727



      int16_t nCols = taosArrayGetSize(pTagVals);




×







728



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





×







729



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




×







730



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





×







731



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




×







732



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





×







733



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





×







734



              ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







735



              taosArrayDestroy(pTagVals);




×







736



              taosArrayDestroy(tagName);




×







737



              goto END;




×







738



            }










739



          }










740



        }










741



      }










742



      taosArrayDestroy(pTagVals);




×







743



    }










744



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




×







745



    req.ctb.tagName = tagName;




×







746



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




×







747



    *type = TDMT_VND_CREATE_TABLE;




×







748



    taosArrayDestroy(tagName);




×







749



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





×







750



    SVCreateTbReq req = {0};




×







751



    req.type = TSDB_NORMAL_TABLE;




×







752



    req.name = me.name;




×







753



    req.uid = me.uid;




×







754



    req.commentLen = -1;




×







755



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




×







756



    req.colCmpr = me.colCmpr;




×







757



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




×







758



    *type = TDMT_VND_CREATE_TABLE;




×







759



  } else {










760



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





×







761



    ret = TSDB_CODE_SDB_INVALID_TABLE_TYPE;




×







762



  }










763












764



END:




×







765



  tDecoderClear(&dc);




×







766



  return ret;




×







767



}










768












769



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




×







770



  void* pKey = NULL;




×







771



  void* pVal = NULL;




×







772



  int   vLen, kLen;










773












774



  while (1) {




×







775



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





×







776



      metaDebug("tmqsnap get uid info end");





×







777



      return 0;




×







778



    }










779



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




×







780



    if (uidTmp == NULL) {





×







781



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







782



    }










783



    ctx->index++;




×







784



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




×







785



    if (!idInfo) {





×







786



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





×







787



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







788



    }










789












790



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




×







791



    if (ret != 0) {





×







792



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





×







793



                idInfo->version);










794



      continue;




×







795



    }










796



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




×







797



    if (ret != 0) {





×







798



      return TAOS_GET_TERRNO(ret);





×







799



    }










800



    SDecoder   dc = {0};




×







801



    SMetaEntry me = {0};




×







802



    tDecoderInit(&dc, pVal, vLen);




×







803



    ret = metaDecodeEntry(&dc, &me);




×







804



    if (ret != 0) {





×







805



      tDecoderClear(&dc);




×







806



      return TAOS_GET_TERRNO(ret);





×







807



    }










808



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





×







809












810



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





×







811



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





×







812



      STableInfoForChildTable* data =










813



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




×







814



      if (data == NULL) {





×







815



        tDecoderClear(&dc);




×







816



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





×







817



        return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







818



      }










819



      result->suid = me.ctbEntry.suid;




×







820



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





×







821



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





×







822



      result->suid = 0;





×







823



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




×







824



    } else {










825



      metaDebug("tmqsnap get uid continue");





×







826



      tDecoderClear(&dc);




×







827



      continue;




×







828



    }










829



    result->uid = me.uid;




×







830



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




×







831



    tDecoderClear(&dc);




×







832



    if (result->schema == NULL) {





×







833



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







834



    }










835



    break;




×







836



  }










837



  return 0;




×







838



}
























    

  • 

    •