taosdata / TDengine / #3780

coverage: 11.99% (-22.0%) from 34.001%
#3780

push

travis-ci

happyguoxy 
test:add case 2

40223 of 491333 branches covered (8.19%)

Branch coverage included in aggregate %.

70904 of 435487 relevant lines covered (16.28%)

1596.49 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



    }










421












422



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





×







423



      tDecoderClear(&dc);




×







424



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







425



    }










426



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





×







427



    tDecoderClear(&dc);




×







428












429



    SIdInfo info = {0};




×







430



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





×







431



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







432



    }










433



  }










434



  taosHashClear(ctx->idVersion);




×







435












436



  code = MoveToSnapShotVersion(ctx);




×







437



  if (code != 0) {





×







438



    return code;




×







439



  }










440



  while (1) {




×







441



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




×







442



    if (ret < 0) break;





×







443












444



    STbDbKey* tmp = (STbDbKey*)pKey;




×







445



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




×







446



    if (idData) {





×







447



      continue;




×







448



    }










449



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




×







450



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




×







451



    if (ret != 0) {





×







452



      return TAOS_GET_TERRNO(ret);





×







453



    }










454












455



    SDecoder   dc = {0};




×







456



    SMetaEntry me = {0};




×







457



    tDecoderInit(&dc, pVal, vLen);




×







458



    ret = metaDecodeEntry(&dc, &me);




×







459



    if (ret < 0) {





×







460



      tDecoderClear(&dc);




×







461



      return TAOS_GET_TERRNO(ret);





×







462



    }










463












464



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {





×







465



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





×







466



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





×







467



        tDecoderClear(&dc);




×







468



        continue;




×







469



      }










470



    }










471












472



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





×







473



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





×







474



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




×







475



      if (ret != 0) {





×







476



        tDecoderClear(&dc);




×







477



        return ret;




×







478



      }










479



    }










480



    tDecoderClear(&dc);




×







481



  }










482












483



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





×







484



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




×







485



    if (uid == NULL) {





×







486



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







487



    }










488



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




×







489



    if (!idData) {





×







490



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





×







491



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







492



    }










493












494



    idData->index = i;




×







495



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





×







496



              idData->index);










497



  }










498












499



  tdbFree(pKey);




×







500



  tdbFree(pVal);




×







501



  return TDB_CODE_SUCCESS;




×







502



}










503












504



void destroySnapContext(SSnapContext* ctx) {




×







505



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




×







506



  taosArrayDestroy(ctx->idList);




×







507



  taosHashCleanup(ctx->idVersion);




×







508



  taosHashCleanup(ctx->suidInfo);




×







509



  taosMemoryFree(ctx);





×







510



}




×







511












512



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




×







513



  int32_t            ret = 0;




×







514



  SVCreateTbBatchReq reqs = {0};




×







515












516



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




×







517



  if (NULL == reqs.pArray) {





×







518



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







519



    goto end;




×







520



  }










521



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





×







522



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







523



    goto end;




×







524



  }










525



  reqs.nReqs = 1;




×







526












527



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





×







528



  if (ret < 0) {





×







529



    ret = TAOS_GET_TERRNO(ret);





×







530



    goto end;




×







531



  }










532



  *contLen += sizeof(SMsgHead);




×







533



  *pBuf = taosMemoryMalloc(*contLen);





×







534



  if (NULL == *pBuf) {





×







535



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







536



    goto end;




×







537



  }










538



  SEncoder coder = {0};




×







539



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




×







540



  ret = tEncodeSVCreateTbBatchReq(&coder, &reqs);




×







541



  tEncoderClear(&coder);




×







542












543



  if (ret < 0) {





×







544



    taosMemoryFreeClear(*pBuf);





×







545



    ret = TAOS_GET_TERRNO(ret);





×







546



    goto end;




×







547



  }










548












549



end:




×







550



  taosArrayDestroy(reqs.pArray);




×







551



  return ret;




×







552



}










553












554



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




×







555



  int32_t ret = 0;




×







556



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





×







557



  if (ret < 0) {





×







558



    return TAOS_GET_TERRNO(ret);





×







559



  }










560












561



  *contLen += sizeof(SMsgHead);




×







562



  *pBuf = taosMemoryMalloc(*contLen);





×







563



  if (NULL == *pBuf) {





×







564



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







565



  }










566












567



  SEncoder encoder = {0};




×







568



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




×







569



  ret = tEncodeSVCreateStbReq(&encoder, req);




×







570



  tEncoderClear(&encoder);




×







571



  if (ret < 0) {





×







572



    taosMemoryFreeClear(*pBuf);





×







573



    return TAOS_GET_TERRNO(ret);





×







574



  }










575



  return 0;




×







576



}










577












578



int32_t setForSnapShot(SSnapContext* ctx, int64_t uid) {




×







579



  if (uid == 0) {





×







580



    ctx->index = 0;




×







581



    return 0;




×







582



  }










583












584



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




×







585



  if (idInfo == NULL) {





×







586



    return terrno;




×







587



  }










588












589



  ctx->index = idInfo->index;




×







590












591



  return 0;




×







592



}










593












594



void taosXSetTablePrimaryKey(SSnapContext* ctx, int64_t uid) {




×







595



  bool            ret = false;




×







596



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




×







597



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





×







598



    ret = true;




×







599



  }










600



  tDeleteSchemaWrapper(schema);










601



  ctx->hasPrimaryKey = ret;




×







602



}




×







603












604



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




×







605












606



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




×







607



  int32_t ret = 0;




×







608



  void*   pKey = NULL;




×







609



  void*   pVal = NULL;




×







610



  int     vLen = 0, kLen = 0;




×







611












612



  while (1) {




×







613



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





×







614



      metaDebug("tmqsnap get meta end");





×







615



      ctx->index = 0;




×







616



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




×







617



      return 0;




×







618



    }










619












620



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




×







621



    if (uidTmp == NULL) {





×







622



      metaError("tmqsnap get meta null uid");





×







623



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







624



    }










625



    ctx->index++;




×







626



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




×







627



    if (!idInfo) {





×







628



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





×







629



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







630



    }










631












632



    *uid = *uidTmp;




×







633



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




×







634



    if (ret == 0) {





×







635



      break;




×







636



    }










637



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





×







638



  }










639












640



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




×







641



  if (ret < 0) {





×







642



    return TAOS_GET_TERRNO(ret);





×







643



  }










644



  SDecoder   dc = {0};




×







645



  SMetaEntry me = {0};




×







646



  tDecoderInit(&dc, pVal, vLen);




×







647



  ret = metaDecodeEntry(&dc, &me);




×







648



  if (ret < 0) {





×







649



    tDecoderClear(&dc);




×







650



    ret = TAOS_GET_TERRNO(ret);





×







651



    goto END;




×







652



  }










653



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





×







654












655



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





×







656



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





×







657



    SVCreateStbReq req = {0};




×







658



    req.name = me.name;




×







659



    req.suid = me.uid;




×







660



    req.schemaRow = me.stbEntry.schemaRow;




×







661



    req.schemaTag = me.stbEntry.schemaTag;




×







662



    req.schemaRow.version = 1;




×







663



    req.schemaTag.version = 1;




×







664



    req.colCmpr = me.colCmpr;




×







665












666



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




×







667



    *type = TDMT_VND_CREATE_STB;




×







668



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





×







669



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





×







670



    STableInfoForChildTable* data =










671



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




×







672



    if (!data) {





×







673



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





×







674



      ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







675



      goto END;




×







676



    }










677












678



    SVCreateTbReq req = {0};




×







679












680



    req.type = TSDB_CHILD_TABLE;




×







681



    req.name = me.name;




×







682



    req.uid = me.uid;




×







683



    req.commentLen = -1;




×







684



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




×







685



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




×







686



    req.ctb.stbName = data->tableName;




×







687












688



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




×







689



    if (tagName == NULL) {





×







690



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





×







691



      ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







692



      goto END;




×







693



    }










694



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




×







695



    if (tTagIsJson(p)) {





×







696



      if (p->nTag != 0) {





×







697



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




×







698



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





×







699



          ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







700



          taosArrayDestroy(tagName);




×







701



          goto END;




×







702



        }










703



      }










704



    } else {










705



      SArray* pTagVals = NULL;




×







706



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




×







707



      if (ret != 0) {





×







708



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





×







709



        taosArrayDestroy(pTagVals);




×







710



        taosArrayDestroy(tagName);




×







711



        goto END;




×







712



      }










713



      int16_t nCols = taosArrayGetSize(pTagVals);




×







714



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





×







715



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




×







716



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





×







717



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




×







718



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





×







719



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





×







720



              ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







721



              taosArrayDestroy(pTagVals);




×







722



              taosArrayDestroy(tagName);




×







723



              goto END;




×







724



            }










725



          }










726



        }










727



      }










728



      taosArrayDestroy(pTagVals);




×







729



    }










730



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




×







731



    req.ctb.tagName = tagName;




×







732



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




×







733



    *type = TDMT_VND_CREATE_TABLE;




×







734



    taosArrayDestroy(tagName);




×







735



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





×







736



    SVCreateTbReq req = {0};




×







737



    req.type = TSDB_NORMAL_TABLE;




×







738



    req.name = me.name;




×







739



    req.uid = me.uid;




×







740



    req.commentLen = -1;




×







741



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




×







742



    req.colCmpr = me.colCmpr;




×







743



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




×







744



    *type = TDMT_VND_CREATE_TABLE;




×







745



  } else {










746



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





×







747



    ret = TSDB_CODE_SDB_INVALID_TABLE_TYPE;




×







748



  }










749












750



END:




×







751



  tDecoderClear(&dc);




×







752



  return ret;




×







753



}










754












755



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




×







756



  void* pKey = NULL;




×







757



  void* pVal = NULL;




×







758



  int   vLen, kLen;










759












760



  while (1) {




×







761



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





×







762



      metaDebug("tmqsnap get uid info end");





×







763



      return 0;




×







764



    }










765



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




×







766



    if (uidTmp == NULL) {





×







767



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







768



    }










769



    ctx->index++;




×







770



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




×







771



    if (!idInfo) {





×







772



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





×







773



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







774



    }










775












776



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




×







777



    if (ret != 0) {





×







778



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





×







779



                idInfo->version);










780



      continue;




×







781



    }










782



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




×







783



    if (ret != 0) {





×







784



      return TAOS_GET_TERRNO(ret);





×







785



    }










786



    SDecoder   dc = {0};




×







787



    SMetaEntry me = {0};




×







788



    tDecoderInit(&dc, pVal, vLen);




×







789



    ret = metaDecodeEntry(&dc, &me);




×







790



    if (ret != 0) {





×







791



      tDecoderClear(&dc);




×







792



      return TAOS_GET_TERRNO(ret);





×







793



    }










794



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





×







795












796



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





×







797



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





×







798



      STableInfoForChildTable* data =










799



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




×







800



      if (data == NULL) {





×







801



        tDecoderClear(&dc);




×







802



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





×







803



        return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







804



      }










805



      result->suid = me.ctbEntry.suid;




×







806



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





×







807



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





×







808



      result->suid = 0;





×







809



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




×







810



    } else {










811



      metaDebug("tmqsnap get uid continue");





×







812



      tDecoderClear(&dc);




×







813



      continue;




×







814



    }










815



    result->uid = me.uid;




×







816



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




×







817



    tDecoderClear(&dc);




×







818



    if (result->schema == NULL) {





×







819



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







820



    }










821



    break;




×







822



  }










823



  return 0;




×







824



}
























    

  • 

    •