taosdata / TDengine / #3562

coverage: 26.655% (-32.2%) from 58.812%
#3562

push

travis-ci

web-flow 
Merge pull request #29229 from taosdata/enh/TS-5749-3.0

enh: seperate tsdb async tasks to different thread pools

21498 of 109421 branches covered (19.65%)

Branch coverage included in aggregate %.

66 of 96 new or added lines in 7 files covered. (68.75%)

39441 existing lines in 157 files now uncovered.

35007 of 102566 relevant lines covered (34.13%)

53922.97 hits per line

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

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









16



#include "meta.h"










17












18



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










19



struct SMetaSnapReader {










20



  SMeta*  pMeta;










21



  int64_t sver;










22



  int64_t ever;










23



  TBC*    pTbc;










24



};










25













UNCOV

26



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




×








UNCOV

27



  int32_t          code = 0;




×







28



  int32_t          lino;











UNCOV

29



  int32_t          c = 0;




×








UNCOV

30



  SMetaSnapReader* pReader = NULL;




×







31












32



  // alloc











UNCOV

33



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





×








UNCOV

34



  if (pReader == NULL) {





×







35



    TSDB_CHECK_CODE(code = terrno, lino, _exit);





×







36



  }











UNCOV

37



  pReader->pMeta = pMeta;




×








UNCOV

38



  pReader->sver = sver;




×








UNCOV

39



  pReader->ever = ever;




×







40












41



  // impl











UNCOV

42



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




×








UNCOV

43



  TSDB_CHECK_CODE(code, lino, _exit);





×







44













UNCOV

45



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




×








UNCOV

46



  TSDB_CHECK_CODE(code, lino, _exit);





×







47













UNCOV

48



_exit:




×








UNCOV

49



  if (code) {





×







50



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





×







51



    metaSnapReaderClose(&pReader);




×







52



    *ppReader = NULL;




×







53



  } else {











UNCOV

54



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





×








UNCOV

55



    *ppReader = pReader;




×







56



  }











UNCOV

57



  return code;




×







58



}










59













UNCOV

60



void metaSnapReaderClose(SMetaSnapReader** ppReader) {




×








UNCOV

61



  if (ppReader && *ppReader) {





×








UNCOV

62



    tdbTbcClose((*ppReader)->pTbc);




×








UNCOV

63



    taosMemoryFree(*ppReader);





×








UNCOV

64



    *ppReader = NULL;




×







65



  }











UNCOV

66



}




×







67













UNCOV

68



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




×








UNCOV

69



  int32_t     code = 0;




×








UNCOV

70



  const void* pKey = NULL;




×








UNCOV

71



  const void* pData = NULL;




×








UNCOV

72



  int32_t     nKey = 0;




×








UNCOV

73



  int32_t     nData = 0;




×







74



  STbDbKey    key;










75













UNCOV

76



  *ppData = NULL;




×







77



  for (;;) {




×








UNCOV

78



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





×








UNCOV

79



      goto _exit;




×







80



    }










81













UNCOV

82



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




×








UNCOV

83



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





×







84



      goto _exit;




×







85



    }










86













UNCOV

87



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





×







88



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





×







89



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





×







90



      }










91



      continue;




×







92



    }










93













UNCOV

94



    if (!pData || !nData) {





×







95



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





×







96



      goto _exit;




×







97



    }










98













UNCOV

99



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





×








UNCOV

100



    if (*ppData == NULL) {





×







101



      code = terrno;




×







102



      goto _exit;




×







103



    }










104













UNCOV

105



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




×








UNCOV

106



    pHdr->type = SNAP_DATA_META;




×








UNCOV

107



    pHdr->size = nData;




×








UNCOV

108



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




×







109













UNCOV

110



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





×







111



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










112













UNCOV

113



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





×







114



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





×







115



    }











UNCOV

116



    break;




×







117



  }










118













UNCOV

119



_exit:




×








UNCOV

120



  if (code) {





×







121



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





×







122



              tstrerror(code));










123



  }











UNCOV

124



  return code;




×







125



}










126












127



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










128



struct SMetaSnapWriter {










129



  SMeta*  pMeta;










130



  int64_t sver;










131



  int64_t ever;










132



};










133













UNCOV

134



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




×








UNCOV

135



  int32_t          code = 0;




×







136



  int32_t          lino;










137



  SMetaSnapWriter* pWriter;










138












139



  // alloc











UNCOV

140



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





×








UNCOV

141



  if (pWriter == NULL) {





×







142



    TSDB_CHECK_CODE(code = terrno, lino, _exit);





×







143



  }











UNCOV

144



  pWriter->pMeta = pMeta;




×








UNCOV

145



  pWriter->sver = sver;




×








UNCOV

146



  pWriter->ever = ever;




×







147













UNCOV

148



  code = metaBegin(pMeta, META_BEGIN_HEAP_NIL);




×








UNCOV

149



  TSDB_CHECK_CODE(code, lino, _exit);





×







150













UNCOV

151



_exit:




×








UNCOV

152



  if (code) {





×







153



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





×







154



    taosMemoryFree(pWriter);





×







155



    *ppWriter = NULL;




×







156



  } else {











UNCOV

157



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





×








UNCOV

158



    *ppWriter = pWriter;




×







159



  }











UNCOV

160



  return code;




×







161



}










162













UNCOV

163



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




×








UNCOV

164



  int32_t          code = 0;




×








UNCOV

165



  SMetaSnapWriter* pWriter = *ppWriter;




×







166













UNCOV

167



  if (rollback) {





×







168



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





×







169



    code = metaAbort(pWriter->pMeta);




×







170



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





×







171



             code);










172



    if (code) goto _err;





×







173



  } else {











UNCOV

174



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




×








UNCOV

175



    if (code) goto _err;





×








UNCOV

176



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




×








UNCOV

177



    if (code) goto _err;





×







178



  }











UNCOV

179



  taosMemoryFree(pWriter);





×








UNCOV

180



  *ppWriter = NULL;




×







181













UNCOV

182



  return code;




×







183












184



_err:




×







185



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





×







186



  return code;




×







187



}










188













UNCOV

189



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




×








UNCOV

190



  int32_t    code = 0;




×








UNCOV

191



  int32_t    lino = 0;




×








UNCOV

192



  SMeta*     pMeta = pWriter->pMeta;




×








UNCOV

193



  SMetaEntry metaEntry = {0};




×








UNCOV

194



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




×







195













UNCOV

196



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




×








UNCOV

197



  code = metaDecodeEntry(pDecoder, &metaEntry);




×








UNCOV

198



  TSDB_CHECK_CODE(code, lino, _exit);





×







199













UNCOV

200



  code = metaHandleEntry2(pMeta, &metaEntry);




×








UNCOV

201



  TSDB_CHECK_CODE(code, lino, _exit);





×







202













UNCOV

203



_exit:




×








UNCOV

204



  if (code) {





×







205



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





×







206



  }











UNCOV

207



  tDecoderClear(pDecoder);




×








UNCOV

208



  return code;




×







209



}










210












211



typedef struct STableInfoForChildTable {










212



  char*           tableName;










213



  SSchemaWrapper* schemaRow;










214



  SSchemaWrapper* tagRow;










215



} STableInfoForChildTable;










216













UNCOV

217



static void destroySTableInfoForChildTable(void* data) {




×








UNCOV

218



  STableInfoForChildTable* pData = (STableInfoForChildTable*)data;




×








UNCOV

219



  taosMemoryFree(pData->tableName);





×








UNCOV

220



  tDeleteSchemaWrapper(pData->schemaRow);





×








UNCOV

221



  tDeleteSchemaWrapper(pData->tagRow);





×








UNCOV

222



}




×







223













UNCOV

224



static int32_t MoveToSnapShotVersion(SSnapContext* ctx) {




×








UNCOV

225



  int32_t code = 0;




×








UNCOV

226



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




×








UNCOV

227



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




×








UNCOV

228



  if (code != 0) {





×







229



    return TAOS_GET_TERRNO(code);





×







230



  }











UNCOV

231



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




×








UNCOV

232



  int      c = 0;




×








UNCOV

233



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




×








UNCOV

234



  if (code != 0) {





×







235



    return TAOS_GET_TERRNO(code);





×







236



  }











UNCOV

237



  if (c < 0) {





×








UNCOV

238



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





×







239



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





×







240



    }










241



  }











UNCOV

242



  return 0;




×







243



}










244













UNCOV

245



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




×








UNCOV

246



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




×








UNCOV

247



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




×








UNCOV

248



  if (code != 0) {





×







249



    return TAOS_GET_TERRNO(code);





×







250



  }











UNCOV

251



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




×








UNCOV

252



  int      c = 0;




×








UNCOV

253



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




×








UNCOV

254



  if (code != 0) {





×







255



    return TAOS_GET_TERRNO(code);





×







256



  }











UNCOV

257



  return c;




×







258



}










259













UNCOV

260



static int32_t MoveToFirst(SSnapContext* ctx) {




×








UNCOV

261



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




×








UNCOV

262



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




×








UNCOV

263



  if (code != 0) {





×







264



    return TAOS_GET_TERRNO(code);





×







265



  }











UNCOV

266



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




×








UNCOV

267



  if (code != 0) {





×







268



    return TAOS_GET_TERRNO(code);





×







269



  }











UNCOV

270



  return 0;




×







271



}










272













UNCOV

273



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




×








UNCOV

274



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




×








UNCOV

275



  if (data) {





×







276



    return 0;




×







277



  }











UNCOV

278



  int32_t                 code = 0;




×








UNCOV

279



  STableInfoForChildTable dataTmp = {0};




×








UNCOV

280



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





×








UNCOV

281



  if (dataTmp.tableName == NULL) {





×







282



    code = terrno;




×







283



    goto END;




×







284



  }











UNCOV

285



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





×








UNCOV

286



  if (dataTmp.schemaRow == NULL) {





×







287



    code = TSDB_CODE_OUT_OF_MEMORY;




×







288



    goto END;




×







289



  }











UNCOV

290



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





×








UNCOV

291



  if (dataTmp.tagRow == NULL) {





×







292



    code = TSDB_CODE_OUT_OF_MEMORY;




×







293



    goto END;




×







294



  }











UNCOV

295



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




×








UNCOV

296



  if (code != 0) {





×







297



    goto END;




×







298



  }











UNCOV

299



  return 0;




×







300












301



END:




×







302



  destroySTableInfoForChildTable(&dataTmp);




×







303



  return TAOS_GET_TERRNO(code);





×







304



  ;










305



}










306













UNCOV

307



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




×







308



                         SSnapContext** ctxRet) {











UNCOV

309



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





×








UNCOV

310



  if (ctx == NULL) {





×







311



    return terrno;




×







312



  }











UNCOV

313



  *ctxRet = ctx;




×








UNCOV

314



  ctx->pMeta = pVnode->pMeta;




×








UNCOV

315



  ctx->snapVersion = snapVersion;




×








UNCOV

316



  ctx->suid = suid;




×








UNCOV

317



  ctx->subType = subType;




×








UNCOV

318



  ctx->queryMeta = withMeta;




×








UNCOV

319



  ctx->withMeta = withMeta;




×








UNCOV

320



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




×








UNCOV

321



  if (ctx->idVersion == NULL) {





×







322



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







323



  }










324













UNCOV

325



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




×








UNCOV

326



  if (ctx->suidInfo == NULL) {





×







327



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







328



    ;










329



  }











UNCOV

330



  taosHashSetFreeFp(ctx->suidInfo, destroySTableInfoForChildTable);




×







331













UNCOV

332



  ctx->index = 0;




×








UNCOV

333



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




×








UNCOV

334



  if (ctx->idList == NULL) {





×







335



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







336



    ;










337



  }











UNCOV

338



  void* pKey = NULL;




×








UNCOV

339



  void* pVal = NULL;




×








UNCOV

340



  int   vLen = 0, kLen = 0;




×







341













UNCOV

342



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





×








UNCOV

343



  int32_t code = MoveToFirst(ctx);




×








UNCOV

344



  if (code != 0) {





×







345



    return code;




×







346



  }











UNCOV

347



  while (1) {




×








UNCOV

348



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




×








UNCOV

349



    if (ret < 0) break;





×








UNCOV

350



    STbDbKey* tmp = (STbDbKey*)pKey;




×








UNCOV

351



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





×







352













UNCOV

353



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




×








UNCOV

354



    if (idData) {





×








UNCOV

355



      continue;




×







356



    }










357













UNCOV

358



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





×







359



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











UNCOV

360



      continue;




×







361



    }










362













UNCOV

363



    SDecoder   dc = {0};




×








UNCOV

364



    SMetaEntry me = {0};




×








UNCOV

365



    tDecoderInit(&dc, pVal, vLen);




×








UNCOV

366



    ret = metaDecodeEntry(&dc, &me);




×








UNCOV

367



    if (ret < 0) {





×







368



      tDecoderClear(&dc);




×







369



      return TAOS_GET_TERRNO(ret);





×







370



    }











UNCOV

371



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {





×








UNCOV

372



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





×








UNCOV

373



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





×








UNCOV

374



        tDecoderClear(&dc);




×








UNCOV

375



        continue;




×







376



      }










377



    }










378













UNCOV

379



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





×







380



      tDecoderClear(&dc);




×







381



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







382



    }











UNCOV

383



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





×








UNCOV

384



    tDecoderClear(&dc);




×







385













UNCOV

386



    SIdInfo info = {0};




×








UNCOV

387



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





×







388



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







389



    }










390



  }











UNCOV

391



  taosHashClear(ctx->idVersion);




×







392













UNCOV

393



  code = MoveToSnapShotVersion(ctx);




×








UNCOV

394



  if (code != 0) {





×







395



    return code;




×







396



  }











UNCOV

397



  while (1) {




×








UNCOV

398



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




×








UNCOV

399



    if (ret < 0) break;





×







400













UNCOV

401



    STbDbKey* tmp = (STbDbKey*)pKey;




×








UNCOV

402



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




×








UNCOV

403



    if (idData) {





×








UNCOV

404



      continue;




×







405



    }











UNCOV

406



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




×








UNCOV

407



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




×








UNCOV

408



    if (ret != 0) {





×







409



      return TAOS_GET_TERRNO(ret);





×







410



    }










411













UNCOV

412



    SDecoder   dc = {0};




×








UNCOV

413



    SMetaEntry me = {0};




×








UNCOV

414



    tDecoderInit(&dc, pVal, vLen);




×








UNCOV

415



    ret = metaDecodeEntry(&dc, &me);




×








UNCOV

416



    if (ret < 0) {





×







417



      tDecoderClear(&dc);




×







418



      return TAOS_GET_TERRNO(ret);





×







419



    }










420













UNCOV

421



    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {





×








UNCOV

422



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





×








UNCOV

423



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





×








UNCOV

424



        tDecoderClear(&dc);




×








UNCOV

425



        continue;




×







426



      }










427



    }










428













UNCOV

429



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





×








UNCOV

430



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





×








UNCOV

431



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




×








UNCOV

432



      if (ret != 0) {





×







433



        tDecoderClear(&dc);




×







434



        return ret;




×







435



      }










436



    }











UNCOV

437



    tDecoderClear(&dc);




×







438



  }










439













UNCOV

440



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





×








UNCOV

441



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




×








UNCOV

442



    if (uid == NULL) {





×







443



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







444



    }











UNCOV

445



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




×








UNCOV

446



    if (!idData) {





×







447



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





×







448



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







449



    }










450













UNCOV

451



    idData->index = i;




×








UNCOV

452



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





×







453



              idData->index);










454



  }










455













UNCOV

456



  tdbFree(pKey);




×








UNCOV

457



  tdbFree(pVal);




×








UNCOV

458



  return TDB_CODE_SUCCESS;




×







459



}










460













UNCOV

461



void destroySnapContext(SSnapContext* ctx) {




×








UNCOV

462



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




×








UNCOV

463



  taosArrayDestroy(ctx->idList);




×








UNCOV

464



  taosHashCleanup(ctx->idVersion);




×








UNCOV

465



  taosHashCleanup(ctx->suidInfo);




×








UNCOV

466



  taosMemoryFree(ctx);





×








UNCOV

467



}




×







468













UNCOV

469



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




×








UNCOV

470



  int32_t            ret = 0;




×








UNCOV

471



  SVCreateTbBatchReq reqs = {0};




×







472













UNCOV

473



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




×








UNCOV

474



  if (NULL == reqs.pArray) {





×







475



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







476



    goto end;




×







477



  }











UNCOV

478



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





×







479



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







480



    goto end;




×







481



  }











UNCOV

482



  reqs.nReqs = 1;




×







483













UNCOV

484



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





×








UNCOV

485



  if (ret < 0) {





×







486



    ret = TAOS_GET_TERRNO(ret);





×







487



    goto end;




×







488



  }











UNCOV

489



  *contLen += sizeof(SMsgHead);




×








UNCOV

490



  *pBuf = taosMemoryMalloc(*contLen);





×








UNCOV

491



  if (NULL == *pBuf) {





×







492



    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







493



    goto end;




×







494



  }











UNCOV

495



  SEncoder coder = {0};




×








UNCOV

496



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




×








UNCOV

497



  ret = tEncodeSVCreateTbBatchReq(&coder, &reqs);




×








UNCOV

498



  tEncoderClear(&coder);




×







499













UNCOV

500



  if (ret < 0) {





×







501



    taosMemoryFreeClear(*pBuf);





×







502



    ret = TAOS_GET_TERRNO(ret);





×







503



    goto end;




×







504



  }










505













UNCOV

506



end:




×








UNCOV

507



  taosArrayDestroy(reqs.pArray);




×








UNCOV

508



  return ret;




×







509



}










510













UNCOV

511



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




×








UNCOV

512



  int32_t ret = 0;




×








UNCOV

513



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





×








UNCOV

514



  if (ret < 0) {





×







515



    return TAOS_GET_TERRNO(ret);





×







516



  }










517













UNCOV

518



  *contLen += sizeof(SMsgHead);




×








UNCOV

519



  *pBuf = taosMemoryMalloc(*contLen);





×








UNCOV

520



  if (NULL == *pBuf) {





×







521



    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







522



  }










523













UNCOV

524



  SEncoder encoder = {0};




×








UNCOV

525



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




×








UNCOV

526



  ret = tEncodeSVCreateStbReq(&encoder, req);




×








UNCOV

527



  tEncoderClear(&encoder);




×








UNCOV

528



  if (ret < 0) {





×







529



    taosMemoryFreeClear(*pBuf);





×







530



    return TAOS_GET_TERRNO(ret);





×







531



  }











UNCOV

532



  return 0;




×







533



}










534













UNCOV

535



int32_t setForSnapShot(SSnapContext* ctx, int64_t uid) {




×








UNCOV

536



  if (uid == 0) {





×








UNCOV

537



    ctx->index = 0;




×








UNCOV

538



    return 0;




×







539



  }










540













UNCOV

541



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




×








UNCOV

542



  if (idInfo == NULL) {





×







543



    return terrno;




×







544



  }










545













UNCOV

546



  ctx->index = idInfo->index;




×







547













UNCOV

548



  return 0;




×







549



}










550













UNCOV

551



void taosXSetTablePrimaryKey(SSnapContext* ctx, int64_t uid) {




×








UNCOV

552



  bool            ret = false;




×








UNCOV

553



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




×








UNCOV

554



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





×








UNCOV

555



    ret = true;




×







556



  }










557



  tDeleteSchemaWrapper(schema);











UNCOV

558



  ctx->hasPrimaryKey = ret;




×








UNCOV

559



}




×







560













UNCOV

561



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




×







562













UNCOV

563



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




×








UNCOV

564



  int32_t ret = 0;




×








UNCOV

565



  void*   pKey = NULL;




×








UNCOV

566



  void*   pVal = NULL;




×








UNCOV

567



  int     vLen = 0, kLen = 0;




×







568












569



  while (1) {




×








UNCOV

570



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





×








UNCOV

571



      metaDebug("tmqsnap get meta end");





×








UNCOV

572



      ctx->index = 0;




×








UNCOV

573



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




×








UNCOV

574



      return 0;




×







575



    }










576













UNCOV

577



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




×








UNCOV

578



    if (uidTmp == NULL) {





×







579



      metaError("tmqsnap get meta null uid");





×







580



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







581



    }











UNCOV

582



    ctx->index++;




×








UNCOV

583



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




×








UNCOV

584



    if (!idInfo) {





×







585



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





×







586



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







587



    }










588













UNCOV

589



    *uid = *uidTmp;




×








UNCOV

590



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




×








UNCOV

591



    if (ret == 0) {





×








UNCOV

592



      break;




×







593



    }










594



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





×







595



  }










596













UNCOV

597



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




×








UNCOV

598



  if (ret < 0) {





×







599



    return TAOS_GET_TERRNO(ret);





×







600



  }











UNCOV

601



  SDecoder   dc = {0};




×








UNCOV

602



  SMetaEntry me = {0};




×








UNCOV

603



  tDecoderInit(&dc, pVal, vLen);




×








UNCOV

604



  ret = metaDecodeEntry(&dc, &me);




×








UNCOV

605



  if (ret < 0) {





×







606



    tDecoderClear(&dc);




×







607



    ret = TAOS_GET_TERRNO(ret);





×







608



    goto END;




×







609



  }











UNCOV

610



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





×







611













UNCOV

612



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





×








UNCOV

613



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





×








UNCOV

614



    SVCreateStbReq req = {0};




×








UNCOV

615



    req.name = me.name;




×








UNCOV

616



    req.suid = me.uid;




×








UNCOV

617



    req.schemaRow = me.stbEntry.schemaRow;




×








UNCOV

618



    req.schemaTag = me.stbEntry.schemaTag;




×








UNCOV

619



    req.schemaRow.version = 1;




×








UNCOV

620



    req.schemaTag.version = 1;




×








UNCOV

621



    req.colCmpr = me.colCmpr;




×







622













UNCOV

623



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




×








UNCOV

624



    *type = TDMT_VND_CREATE_STB;




×








UNCOV

625



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





×








UNCOV

626



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





×







627



    STableInfoForChildTable* data =











UNCOV

628



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




×








UNCOV

629



    if (!data) {





×







630



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





×







631



      ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







632



      goto END;




×







633



    }










634













UNCOV

635



    SVCreateTbReq req = {0};




×







636













UNCOV

637



    req.type = TSDB_CHILD_TABLE;




×








UNCOV

638



    req.name = me.name;




×








UNCOV

639



    req.uid = me.uid;




×








UNCOV

640



    req.commentLen = -1;




×








UNCOV

641



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




×








UNCOV

642



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




×








UNCOV

643



    req.ctb.stbName = data->tableName;




×







644













UNCOV

645



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




×








UNCOV

646



    if (tagName == NULL) {





×







647



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





×







648



      ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







649



      goto END;




×







650



    }











UNCOV

651



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




×








UNCOV

652



    if (tTagIsJson(p)) {





×








UNCOV

653



      if (p->nTag != 0) {





×








UNCOV

654



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




×








UNCOV

655



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





×







656



          ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







657



          taosArrayDestroy(tagName);




×







658



          goto END;




×







659



        }










660



      }










661



    } else {











UNCOV

662



      SArray* pTagVals = NULL;




×








UNCOV

663



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




×








UNCOV

664



      if (ret != 0) {





×







665



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





×







666



        taosArrayDestroy(pTagVals);




×







667



        taosArrayDestroy(tagName);




×







668



        goto END;




×







669



      }











UNCOV

670



      int16_t nCols = taosArrayGetSize(pTagVals);




×








UNCOV

671



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





×








UNCOV

672



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




×








UNCOV

673



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





×








UNCOV

674



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




×








UNCOV

675



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





×








UNCOV

676



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





×







677



              ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







678



              taosArrayDestroy(pTagVals);




×







679



              taosArrayDestroy(tagName);




×







680



              goto END;




×







681



            }










682



          }










683



        }










684



      }











UNCOV

685



      taosArrayDestroy(pTagVals);




×







686



    }











UNCOV

687



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




×








UNCOV

688



    req.ctb.tagName = tagName;




×








UNCOV

689



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




×








UNCOV

690



    *type = TDMT_VND_CREATE_TABLE;




×








UNCOV

691



    taosArrayDestroy(tagName);




×








UNCOV

692



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





×








UNCOV

693



    SVCreateTbReq req = {0};




×








UNCOV

694



    req.type = TSDB_NORMAL_TABLE;




×








UNCOV

695



    req.name = me.name;




×








UNCOV

696



    req.uid = me.uid;




×








UNCOV

697



    req.commentLen = -1;




×








UNCOV

698



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




×








UNCOV

699



    req.colCmpr = me.colCmpr;




×








UNCOV

700



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




×








UNCOV

701



    *type = TDMT_VND_CREATE_TABLE;




×







702



  } else {










703



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





×







704



    ret = TSDB_CODE_SDB_INVALID_TABLE_TYPE;




×







705



  }










706













UNCOV

707



END:




×








UNCOV

708



  tDecoderClear(&dc);




×








UNCOV

709



  return ret;




×







710



}










711













UNCOV

712



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




×








UNCOV

713



  void* pKey = NULL;




×








UNCOV

714



  void* pVal = NULL;




×







715



  int   vLen, kLen;










716













UNCOV

717



  while (1) {




×








UNCOV

718



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





×








UNCOV

719



      metaDebug("tmqsnap get uid info end");





×








UNCOV

720



      return 0;




×







721



    }











UNCOV

722



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




×








UNCOV

723



    if (uidTmp == NULL) {





×







724



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







725



    }











UNCOV

726



    ctx->index++;




×








UNCOV

727



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




×








UNCOV

728



    if (!idInfo) {





×







729



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





×







730



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







731



    }










732













UNCOV

733



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




×








UNCOV

734



    if (ret != 0) {





×







735



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





×







736



                idInfo->version);











UNCOV

737



      continue;




×







738



    }











UNCOV

739



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




×








UNCOV

740



    if (ret != 0) {





×







741



      return TAOS_GET_TERRNO(ret);





×







742



    }











UNCOV

743



    SDecoder   dc = {0};




×








UNCOV

744



    SMetaEntry me = {0};




×








UNCOV

745



    tDecoderInit(&dc, pVal, vLen);




×








UNCOV

746



    ret = metaDecodeEntry(&dc, &me);




×








UNCOV

747



    if (ret != 0) {





×







748



      tDecoderClear(&dc);




×







749



      return TAOS_GET_TERRNO(ret);





×







750



    }











UNCOV

751



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





×







752













UNCOV

753



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





×








UNCOV

754



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





×







755



      STableInfoForChildTable* data =











UNCOV

756



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




×








UNCOV

757



      if (data == NULL) {





×







758



        tDecoderClear(&dc);




×







759



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





×







760



        return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







761



      }











UNCOV

762



      result->suid = me.ctbEntry.suid;




×








UNCOV

763



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





×








UNCOV

764



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





×








UNCOV

765



      result->suid = 0;





×








UNCOV

766



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




×







767



    } else {











UNCOV

768



      metaDebug("tmqsnap get uid continue");





×








UNCOV

769



      tDecoderClear(&dc);




×








UNCOV

770



      continue;




×







771



    }











UNCOV

772



    result->uid = me.uid;




×








UNCOV

773



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




×








UNCOV

774



    tDecoderClear(&dc);




×








UNCOV

775



    if (result->schema == NULL) {





×







776



      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);





×







777



    }











UNCOV

778



    break;




×







779



  }











UNCOV

780



  return 0;




×







781



}
























    

  • 

    •