taosdata / TDengine / #3621

coverage: 2.037% (-61.5%) from 63.573%
#3621

push

travis-ci

web-flow 
Merge pull request #29874 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

4357 of 287032 branches covered (1.52%)

Branch coverage included in aggregate %.

0 of 174 new or added lines in 18 files covered. (0.0%)

213359 existing lines in 469 files now uncovered.

7260 of 283369 relevant lines covered (2.56%)

23737.72 hits per line

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

0.0
/source/dnode/vnode/src/tsdb/tsdbCommit2.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 "tsdbCommit2.h"










17












18



// extern dependencies










19



typedef struct {










20



  int32_t    fid;










21



  STFileSet *fset;










22



} SFileSetCommitInfo;










23












24



typedef struct {










25



  STsdb  *tsdb;










26



  int32_t minutes;










27



  int8_t  precision;










28



  int32_t minRow;










29



  int32_t maxRow;










30



  int8_t  cmprAlg;










31



  int32_t sttTrigger;










32



  int32_t szPage;










33



  int64_t compactVersion;










34



  int64_t cid;










35



  int64_t now;










36












37



  struct {










38



    SFileSetCommitInfo *info;










39












40



    int32_t expLevel;










41



    TSKEY   minKey;










42



    TSKEY   maxKey;










43



    TABLEID tbid[1];










44



    bool    hasTSData;










45












46



    bool      skipTsRow;










47



    SHashObj *pColCmprObj;










48



  } ctx[1];










49












50



  // reader










51



  TSttFileReaderArray sttReaderArray[1];










52



  // iter










53



  TTsdbIterArray dataIterArray[1];










54



  SIterMerger   *dataIterMerger;










55



  TTsdbIterArray tombIterArray[1];










56



  SIterMerger   *tombIterMerger;










57



  // writer










58



  SFSetWriter *writer;










59












60



  TFileOpArray fopArray[1];










61



} SCommitter2;










62













UNCOV

63



static int32_t tsdbCommitOpenWriter(SCommitter2 *committer) {




×








UNCOV

64



  int32_t code = 0;




×








UNCOV

65



  int32_t lino = 0;




×







66













UNCOV

67



  SFSetWriterConfig config = {




×








UNCOV

68



      .tsdb = committer->tsdb,




×







69



      .toSttOnly = true,











UNCOV

70



      .compactVersion = committer->compactVersion,




×








UNCOV

71



      .minRow = committer->minRow,




×








UNCOV

72



      .maxRow = committer->maxRow,




×








UNCOV

73



      .szPage = committer->szPage,




×








UNCOV

74



      .cmprAlg = committer->cmprAlg,




×








UNCOV

75



      .fid = committer->ctx->info->fid,




×








UNCOV

76



      .cid = committer->cid,




×








UNCOV

77



      .expLevel = committer->ctx->expLevel,




×







78



      .level = 0,










79



  };










80













UNCOV

81



  if (committer->sttTrigger == 1) {





×








UNCOV

82



    config.toSttOnly = false;




×







83













UNCOV

84



    if (committer->ctx->info->fset) {





×








UNCOV

85



      for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ftype++) {





×








UNCOV

86



        if (committer->ctx->info->fset->farr[ftype] != NULL) {





×








UNCOV

87



          config.files[ftype].exist = true;




×








UNCOV

88



          config.files[ftype].file = committer->ctx->info->fset->farr[ftype]->f[0];




×







89



        }










90



      }










91



    }










92



  }










93













UNCOV

94



  TAOS_CHECK_GOTO(tsdbFSetWriterOpen(&config, &committer->writer), &lino, _exit);





×







95













UNCOV

96



_exit:




×








UNCOV

97



  if (code) {





×







98



    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(committer->tsdb->pVnode), __func__, __FILE__, lino,





×







99



              tstrerror(code));










100



  }











UNCOV

101



  return code;




×







102



}










103













UNCOV

104



static int32_t tsdbCommitCloseWriter(SCommitter2 *committer) {




×








UNCOV

105



  return tsdbFSetWriterClose(&committer->writer, 0, committer->fopArray);




×







106



}










107













UNCOV

108



static int32_t tsdbCommitTSData(SCommitter2 *committer) {




×








UNCOV

109



  int32_t   code = 0;




×








UNCOV

110



  int32_t   lino = 0;




×








UNCOV

111



  int64_t   numOfRow = 0;




×







112



  SMetaInfo info;










113













UNCOV

114



  committer->ctx->hasTSData = false;




×







115













UNCOV

116



  committer->ctx->tbid->suid = 0;




×








UNCOV

117



  committer->ctx->tbid->uid = 0;




×








UNCOV

118



  for (SRowInfo *row; (row = tsdbIterMergerGetData(committer->dataIterMerger)) != NULL;) {





×








UNCOV

119



    if (row->uid != committer->ctx->tbid->uid) {





×








UNCOV

120



      committer->ctx->tbid->suid = row->suid;




×








UNCOV

121



      committer->ctx->tbid->uid = row->uid;




×







122













UNCOV

123



      if (metaGetInfo(committer->tsdb->pVnode->pMeta, row->uid, &info, NULL) != 0) {





×







124



        TAOS_CHECK_GOTO(tsdbIterMergerSkipTableData(committer->dataIterMerger, committer->ctx->tbid), &lino, _exit);





×








UNCOV

125



        continue;




×







126



      }










127



    }










128













UNCOV

129



    int64_t ts = TSDBROW_TS(&row->row);





×








UNCOV

130



    if (ts > committer->ctx->maxKey) {





×








UNCOV

131



      TAOS_CHECK_GOTO(tsdbIterMergerSkipTableData(committer->dataIterMerger, committer->ctx->tbid), &lino, _exit);





×








UNCOV

132



      continue;




×







133



    }










134













UNCOV

135



    committer->ctx->hasTSData = true;




×








UNCOV

136



    numOfRow++;




×







137













UNCOV

138



    TAOS_CHECK_GOTO(tsdbFSetWriteRow(committer->writer, row), &lino, _exit);





×








UNCOV

139



    TAOS_CHECK_GOTO(tsdbIterMergerNext(committer->dataIterMerger), &lino, _exit);





×







140



  }










141













UNCOV

142



_exit:




×








UNCOV

143



  if (code) {





×








UNCOV

144



    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(committer->tsdb->pVnode), __func__, __FILE__, lino,





×







145



              tstrerror(code));










146



  } else {











UNCOV

147



    tsdbDebug("vgId:%d fid:%d commit %" PRId64 " rows", TD_VID(committer->tsdb->pVnode), committer->ctx->info->fid,





×







148



              numOfRow);










149



  }











UNCOV

150



  return code;




×







151



}










152













UNCOV

153



static int32_t tsdbCommitTombData(SCommitter2 *committer) {




×








UNCOV

154



  int32_t   code = 0;




×








UNCOV

155



  int32_t   lino = 0;




×








UNCOV

156



  int64_t   numRecord = 0;




×







157



  SMetaInfo info;










158












159



  // if no history data and no new timestamp data, skip tomb data











UNCOV

160



  if (committer->ctx->info->fset || committer->ctx->hasTSData) {





×








UNCOV

161



    committer->ctx->tbid->suid = 0;




×








UNCOV

162



    committer->ctx->tbid->uid = 0;




×








UNCOV

163



    for (STombRecord *record; (record = tsdbIterMergerGetTombRecord(committer->tombIterMerger));) {





×








UNCOV

164



      if (record->uid != committer->ctx->tbid->uid) {





×








UNCOV

165



        committer->ctx->tbid->suid = record->suid;




×








UNCOV

166



        committer->ctx->tbid->uid = record->uid;




×







167













UNCOV

168



        if (metaGetInfo(committer->tsdb->pVnode->pMeta, record->uid, &info, NULL) != 0) {





×








UNCOV

169



          TAOS_CHECK_GOTO(tsdbIterMergerSkipTableData(committer->tombIterMerger, committer->ctx->tbid), &lino, _exit);





×








UNCOV

170



          continue;




×







171



        }










172



      }










173













UNCOV

174



      if (record->ekey < committer->ctx->minKey) {





×







175



        // do nothing











UNCOV

176



      } else if (record->skey > committer->ctx->maxKey) {





×







177



        // committer->ctx->nextKey = TMIN(record->skey, committer->ctx->nextKey);










178



      } else {











UNCOV

179



        record->skey = TMAX(record->skey, committer->ctx->minKey);




×








UNCOV

180



        record->ekey = TMIN(record->ekey, committer->ctx->maxKey);




×







181













UNCOV

182



        numRecord++;




×








UNCOV

183



        TAOS_CHECK_GOTO(tsdbFSetWriteTombRecord(committer->writer, record), &lino, _exit);





×







184



      }










185













UNCOV

186



      TAOS_CHECK_GOTO(tsdbIterMergerNext(committer->tombIterMerger), &lino, _exit);





×







187



    }










188



  }










189













UNCOV

190



_exit:




×








UNCOV

191



  if (code) {





×







192



    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(committer->tsdb->pVnode), __func__, __FILE__, lino,





×







193



              tstrerror(code));










194



  } else {











UNCOV

195



    tsdbDebug("vgId:%d fid:%d commit %" PRId64 " tomb records", TD_VID(committer->tsdb->pVnode),





×







196



              committer->ctx->info->fid, numRecord);










197



  }











UNCOV

198



  return code;




×







199



}










200













UNCOV

201



static void tsdbCommitCloseReader(SCommitter2 *committer) {




×








UNCOV

202



  TARRAY2_CLEAR(committer->sttReaderArray, tsdbSttFileReaderClose);





×








UNCOV

203



  return;




×







204



}










205













UNCOV

206



static int32_t tsdbCommitOpenReader(SCommitter2 *committer) {




×








UNCOV

207



  int32_t code = 0;




×








UNCOV

208



  int32_t lino = 0;




×







209













UNCOV

210



  if (committer->ctx->info->fset == NULL                        //





×








UNCOV

211



      || committer->sttTrigger > 1                              //





×








UNCOV

212



      || TARRAY2_SIZE(committer->ctx->info->fset->lvlArr) == 0  //





×







213



  ) {











UNCOV

214



    return 0;




×







215



  }










216












217



  SSttLvl *lvl;











UNCOV

218



  TARRAY2_FOREACH(committer->ctx->info->fset->lvlArr, lvl) {





×








UNCOV

219



    STFileObj *fobj = NULL;




×








UNCOV

220



    TARRAY2_FOREACH(lvl->fobjArr, fobj) {





×







221



      SSttFileReader *sttReader;










222













UNCOV

223



      SSttFileReaderConfig config = {




×








UNCOV

224



          .tsdb = committer->tsdb,




×








UNCOV

225



          .szPage = committer->szPage,




×







226



          .file = fobj->f[0],










227



      };










228













UNCOV

229



      TAOS_CHECK_GOTO(tsdbSttFileReaderOpen(fobj->fname, &config, &sttReader), &lino, _exit);





×







230













UNCOV

231



      TAOS_CHECK_GOTO(TARRAY2_APPEND(committer->sttReaderArray, sttReader), &lino, _exit);





×







232













UNCOV

233



      STFileOp op = {




×







234



          .optype = TSDB_FOP_REMOVE,











UNCOV

235



          .fid = fobj->f->fid,




×







236



          .of = fobj->f[0],










237



      };










238













UNCOV

239



      TAOS_CHECK_GOTO(TARRAY2_APPEND(committer->fopArray, op), &lino, _exit);





×







240



    }










241



  }










242













UNCOV

243



_exit:




×








UNCOV

244



  if (code) {





×







245



    tsdbCommitCloseReader(committer);




×







246



    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(committer->tsdb->pVnode), __func__, __FILE__, lino,





×







247



              tstrerror(code));










248



  }











UNCOV

249



  return code;




×







250



}










251













UNCOV

252



static void tsdbCommitCloseIter(SCommitter2 *committer) {




×








UNCOV

253



  tsdbIterMergerClose(&committer->tombIterMerger);




×








UNCOV

254



  tsdbIterMergerClose(&committer->dataIterMerger);




×








UNCOV

255



  TARRAY2_CLEAR(committer->tombIterArray, tsdbIterClose);





×








UNCOV

256



  TARRAY2_CLEAR(committer->dataIterArray, tsdbIterClose);





×








UNCOV

257



  return;




×







258



}










259













UNCOV

260



static int32_t tsdbCommitOpenIter(SCommitter2 *committer) {




×








UNCOV

261



  int32_t code = 0;




×








UNCOV

262



  int32_t lino = 0;




×







263












264



  STsdbIter      *iter;











UNCOV

265



  STsdbIterConfig config = {0};




×







266












267



  // mem data iter











UNCOV

268



  config.type = TSDB_ITER_TYPE_MEMT;




×








UNCOV

269



  config.memt = committer->tsdb->imem;




×








UNCOV

270



  config.from->version = VERSION_MIN;




×








UNCOV

271



  config.from->key = (SRowKey){




×








UNCOV

272



      .ts = committer->ctx->minKey,




×







273



      .numOfPKs = 0,










274



  };










275













UNCOV

276



  TAOS_CHECK_GOTO(tsdbIterOpen(&config, &iter), &lino, _exit);





×








UNCOV

277



  TAOS_CHECK_GOTO(TARRAY2_APPEND(committer->dataIterArray, iter), &lino, _exit);





×







278












279



  // mem tomb iter











UNCOV

280



  config.type = TSDB_ITER_TYPE_MEMT_TOMB;




×








UNCOV

281



  config.memt = committer->tsdb->imem;




×







282













UNCOV

283



  TAOS_CHECK_GOTO(tsdbIterOpen(&config, &iter), &lino, _exit);





×








UNCOV

284



  TAOS_CHECK_GOTO(TARRAY2_APPEND(committer->tombIterArray, iter), &lino, _exit);





×







285












286



  // STT










287



  SSttFileReader *sttReader;











UNCOV

288



  TARRAY2_FOREACH(committer->sttReaderArray, sttReader) {





×







289



    // data iter











UNCOV

290



    config.type = TSDB_ITER_TYPE_STT;




×








UNCOV

291



    config.sttReader = sttReader;




×







292













UNCOV

293



    TAOS_CHECK_GOTO(tsdbIterOpen(&config, &iter), &lino, _exit);





×








UNCOV

294



    TAOS_CHECK_GOTO(TARRAY2_APPEND(committer->dataIterArray, iter), &lino, _exit);





×







295












296



    // tomb iter











UNCOV

297



    config.type = TSDB_ITER_TYPE_STT_TOMB;




×








UNCOV

298



    config.sttReader = sttReader;




×







299













UNCOV

300



    TAOS_CHECK_GOTO(tsdbIterOpen(&config, &iter), &lino, _exit);





×







301













UNCOV

302



    TAOS_CHECK_GOTO(TARRAY2_APPEND(committer->tombIterArray, iter), &lino, _exit);





×







303



  }










304












305



  // open merger











UNCOV

306



  TAOS_CHECK_GOTO(tsdbIterMergerOpen(committer->dataIterArray, &committer->dataIterMerger, false), &lino, _exit);





×








UNCOV

307



  TAOS_CHECK_GOTO(tsdbIterMergerOpen(committer->tombIterArray, &committer->tombIterMerger, true), &lino, _exit);





×







308













UNCOV

309



_exit:




×








UNCOV

310



  if (code) {





×







311



    tsdbCommitCloseIter(committer);




×







312



    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(committer->tsdb->pVnode), __func__, __FILE__, lino,





×







313



              tstrerror(code));










314



  }











UNCOV

315



  return code;




×







316



}










317













UNCOV

318



static int32_t tsdbCommitFileSetBegin(SCommitter2 *committer) {




×








UNCOV

319



  int32_t code = 0;




×








UNCOV

320



  int32_t lino = 0;




×








UNCOV

321



  STsdb  *tsdb = committer->tsdb;




×







322












323



  // check if can commit











UNCOV

324



  tsdbFSCheckCommit(tsdb, committer->ctx->info->fid);




×







325













UNCOV

326



  committer->ctx->expLevel = tsdbFidLevel(committer->ctx->info->fid, &tsdb->keepCfg, committer->now);




×








UNCOV

327



  tsdbFidKeyRange(committer->ctx->info->fid, committer->minutes, committer->precision, &committer->ctx->minKey,




×







328



                  &committer->ctx->maxKey);










329













UNCOV

330



  committer->ctx->tbid->suid = 0;




×








UNCOV

331



  committer->ctx->tbid->uid = 0;




×







332













UNCOV

333



  TAOS_CHECK_GOTO(tsdbCommitOpenReader(committer), &lino, _exit);





×








UNCOV

334



  TAOS_CHECK_GOTO(tsdbCommitOpenIter(committer), &lino, _exit);





×








UNCOV

335



  TAOS_CHECK_GOTO(tsdbCommitOpenWriter(committer), &lino, _exit);





×







336













UNCOV

337



_exit:




×








UNCOV

338



  if (code) {





×







339



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





×







340



  } else {











UNCOV

341



    tsdbDebug("vgId:%d %s done, fid:%d minKey:%" PRId64 " maxKey:%" PRId64 " expLevel:%d", TD_VID(tsdb->pVnode),





×







342



              __func__, committer->ctx->info->fid, committer->ctx->minKey, committer->ctx->maxKey,










343



              committer->ctx->expLevel);










344



  }











UNCOV

345



  return code;




×







346



}










347













UNCOV

348



static int32_t tsdbCommitFileSetEnd(SCommitter2 *committer) {




×








UNCOV

349



  int32_t code = 0;




×








UNCOV

350



  int32_t lino = 0;




×







351













UNCOV

352



  TAOS_CHECK_GOTO(tsdbCommitCloseWriter(committer), &lino, _exit);





×








UNCOV

353



  tsdbCommitCloseIter(committer);




×








UNCOV

354



  tsdbCommitCloseReader(committer);




×







355













UNCOV

356



_exit:




×








UNCOV

357



  if (code) {





×







358



    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(committer->tsdb->pVnode), __func__, __FILE__, lino,





×







359



              tstrerror(code));










360



  } else {











UNCOV

361



    tsdbDebug("vgId:%d %s done, fid:%d", TD_VID(committer->tsdb->pVnode), __func__, committer->ctx->info->fid);





×







362



  }











UNCOV

363



  return code;




×







364



}










365













UNCOV

366



static int32_t tsdbCommitFileSet(SCommitter2 *committer) {




×








UNCOV

367



  int32_t code = 0;




×








UNCOV

368



  int32_t lino = 0;




×







369













UNCOV

370



  TAOS_CHECK_GOTO(tsdbCommitFileSetBegin(committer), &lino, _exit);





×








UNCOV

371



  TAOS_CHECK_GOTO(tsdbCommitTSData(committer), &lino, _exit);





×








UNCOV

372



  TAOS_CHECK_GOTO(tsdbCommitTombData(committer), &lino, _exit);





×








UNCOV

373



  TAOS_CHECK_GOTO(tsdbCommitFileSetEnd(committer), &lino, _exit);





×







374













UNCOV

375



_exit:




×








UNCOV

376



  if (code) {





×








UNCOV

377



    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(committer->tsdb->pVnode), __func__, __FILE__, lino,





×







378



              tstrerror(code));










379



  } else {











UNCOV

380



    tsdbDebug("vgId:%d %s done, fid:%d", TD_VID(committer->tsdb->pVnode), __func__, committer->ctx->info->fid);





×







381



  }











UNCOV

382



  return code;




×







383



}










384













UNCOV

385



static int32_t tFileSetCommitInfoCompare(const void *arg1, const void *arg2) {




×








UNCOV

386



  SFileSetCommitInfo *info1 = (SFileSetCommitInfo *)arg1;




×








UNCOV

387



  SFileSetCommitInfo *info2 = (SFileSetCommitInfo *)arg2;




×







388













UNCOV

389



  if (info1->fid < info2->fid) {





×








UNCOV

390



    return -1;




×








UNCOV

391



  } else if (info1->fid > info2->fid) {





×








UNCOV

392



    return 1;




×







393



  } else {











UNCOV

394



    return 0;




×







395



  }










396



}










397













UNCOV

398



static int32_t tFileSetCommitInfoPCompare(const void *arg1, const void *arg2) {




×








UNCOV

399



  return tFileSetCommitInfoCompare(*(SFileSetCommitInfo **)arg1, *(SFileSetCommitInfo **)arg2);




×







400



}










401













UNCOV

402



static uint32_t tFileSetCommitInfoHash(const void *arg) {




×








UNCOV

403



  SFileSetCommitInfo *info = (SFileSetCommitInfo *)arg;




×








UNCOV

404



  return MurmurHash3_32((const char *)&info->fid, sizeof(info->fid));




×







405



}










406













UNCOV

407



static void tsdbCommitInfoDestroy(STsdb *pTsdb) {




×








UNCOV

408



  if (pTsdb->commitInfo) {





×








UNCOV

409



    for (int32_t i = 0; i < taosArrayGetSize(pTsdb->commitInfo->arr); i++) {





×








UNCOV

410



      SFileSetCommitInfo *info = *(SFileSetCommitInfo **)taosArrayGet(pTsdb->commitInfo->arr, i);




×








UNCOV

411



      int32_t             ret = vHashDrop(pTsdb->commitInfo->ht, info);




×








UNCOV

412



      tsdbTFileSetClear(&info->fset);




×








UNCOV

413



      taosMemoryFree(info);





×







414



    }










415













UNCOV

416



    vHashDestroy(&pTsdb->commitInfo->ht);




×








UNCOV

417



    taosArrayDestroy(pTsdb->commitInfo->arr);




×








UNCOV

418



    pTsdb->commitInfo->arr = NULL;




×








UNCOV

419



    taosMemoryFreeClear(pTsdb->commitInfo);





×







420



  }











UNCOV

421



  return;




×







422



}










423













UNCOV

424



static int32_t tsdbCommitInfoInit(STsdb *pTsdb) {




×








UNCOV

425



  int32_t code = 0;




×








UNCOV

426



  int32_t lino = 0;




×







427













UNCOV

428



  pTsdb->commitInfo = taosMemoryCalloc(1, sizeof(*pTsdb->commitInfo));





×








UNCOV

429



  if (pTsdb->commitInfo == NULL) {





×







430



    TAOS_CHECK_GOTO(terrno, &lino, _exit);





×







431



  }










432













UNCOV

433



  TAOS_CHECK_GOTO(vHashInit(&pTsdb->commitInfo->ht, tFileSetCommitInfoHash, tFileSetCommitInfoCompare), &lino, _exit);





×







434













UNCOV

435



  pTsdb->commitInfo->arr = taosArrayInit(0, sizeof(SFileSetCommitInfo *));




×








UNCOV

436



  if (pTsdb->commitInfo->arr == NULL) {





×







437



    TSDB_CHECK_CODE(code = terrno, lino, _exit);





×







438



  }










439













UNCOV

440



_exit:




×








UNCOV

441



  if (code) {





×







442



    tsdbCommitInfoDestroy(pTsdb);




×







443



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





×







444



  }











UNCOV

445



  return code;




×







446



}










447













UNCOV

448



static int32_t tsdbCommitInfoAdd(STsdb *tsdb, int32_t fid) {




×








UNCOV

449



  int32_t code = 0;




×








UNCOV

450



  int32_t lino = 0;




×







451












452



  SFileSetCommitInfo *tinfo;










453













UNCOV

454



  if ((tinfo = taosMemoryMalloc(sizeof(*tinfo))) == NULL) {





×







455



    TAOS_CHECK_GOTO(terrno, &lino, _exit);





×







456



  }











UNCOV

457



  tinfo->fid = fid;




×








UNCOV

458



  tinfo->fset = NULL;




×







459













UNCOV

460



  TAOS_CHECK_GOTO(vHashPut(tsdb->commitInfo->ht, tinfo), &lino, _exit);





×







461













UNCOV

462



  if ((taosArrayPush(tsdb->commitInfo->arr, &tinfo)) == NULL) {





×







463



    TAOS_CHECK_GOTO(terrno, &lino, _exit);





×







464



  }











UNCOV

465



  taosArraySort(tsdb->commitInfo->arr, tFileSetCommitInfoPCompare);




×







466













UNCOV

467



_exit:




×








UNCOV

468



  if (code) {





×







469



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





×







470



  }











UNCOV

471



  return code;




×







472



}










473













UNCOV

474



static int32_t tsdbCommitInfoBuild(STsdb *tsdb) {




×








UNCOV

475



  int32_t code = 0;




×








UNCOV

476



  int32_t lino = 0;




×







477













UNCOV

478



  STFileSet  *fset = NULL;




×







479



  SRBTreeIter iter;










480













UNCOV

481



  TAOS_CHECK_GOTO(tsdbCommitInfoInit(tsdb), &lino, _exit);





×







482












483



  // scan time-series data











UNCOV

484



  iter = tRBTreeIterCreate(tsdb->imem->tbDataTree, 1);




×








UNCOV

485



  for (SRBTreeNode *node = tRBTreeIterNext(&iter); node; node = tRBTreeIterNext(&iter)) {





×








UNCOV

486



    STbData *pTbData = TCONTAINER_OF(node, STbData, rbtn);




×







487












488



    // scan time-series data











UNCOV

489



    STsdbRowKey from = {




×







490



        .key.ts = INT64_MIN,










491



        .key.numOfPKs = 0,










492



        .version = INT64_MIN,










493



    };











UNCOV

494



    for (;;) {




×







495



      int64_t     minKey, maxKey;











UNCOV

496



      STbDataIter tbDataIter = {0};




×







497



      TSDBROW    *row;










498



      int32_t     fid;










499













UNCOV

500



      tsdbTbDataIterOpen(pTbData, &from, 0, &tbDataIter);




×








UNCOV

501



      if ((row = tsdbTbDataIterGet(&tbDataIter)) == NULL) {





×








UNCOV

502



        break;




×







503



      }










504













UNCOV

505



      fid = tsdbKeyFid(TSDBROW_TS(row), tsdb->keepCfg.days, tsdb->keepCfg.precision);





×








UNCOV

506



      tsdbFidKeyRange(fid, tsdb->keepCfg.days, tsdb->keepCfg.precision, &minKey, &maxKey);




×







507












508



      SFileSetCommitInfo *info;











UNCOV

509



      SFileSetCommitInfo  tinfo = {




×







510



           .fid = fid,










511



      };











UNCOV

512



      int32_t ret = vHashGet(tsdb->commitInfo->ht, &tinfo, (void **)&info);




×








UNCOV

513



      if (info == NULL) {





×








UNCOV

514



        TAOS_CHECK_GOTO(tsdbCommitInfoAdd(tsdb, fid), &lino, _exit);





×







515



      }










516













UNCOV

517



      from.key.ts = maxKey + 1;




×







518



    }










519



  }










520













UNCOV

521



  (void)taosThreadMutexLock(&tsdb->mutex);




×







522












523



  // scan tomb data











UNCOV

524



  if (tsdb->imem->nDel > 0) {





×








UNCOV

525



    TARRAY2_FOREACH(tsdb->pFS->fSetArr, fset) {





×








UNCOV

526



      if (tsdbTFileSetIsEmpty(fset)) {





×








UNCOV

527



        continue;




×







528



      }










529












530



      SFileSetCommitInfo *info;











UNCOV

531



      SFileSetCommitInfo  tinfo = {




×








UNCOV

532



           .fid = fset->fid,




×







533



      };










534












535



      // check if the file set already on the commit list











UNCOV

536



      int32_t ret = vHashGet(tsdb->commitInfo->ht, &tinfo, (void **)&info);




×








UNCOV

537



      if (info != NULL) {





×








UNCOV

538



        continue;




×







539



      }










540












541



      int64_t minKey, maxKey;











UNCOV

542



      bool    hasDataToCommit = false;




×








UNCOV

543



      tsdbFidKeyRange(fset->fid, tsdb->keepCfg.days, tsdb->keepCfg.precision, &minKey, &maxKey);




×








UNCOV

544



      iter = tRBTreeIterCreate(tsdb->imem->tbDataTree, 1);




×








UNCOV

545



      for (SRBTreeNode *node = tRBTreeIterNext(&iter); node; node = tRBTreeIterNext(&iter)) {





×








UNCOV

546



        STbData *pTbData = TCONTAINER_OF(node, STbData, rbtn);




×








UNCOV

547



        for (SDelData *pDelData = pTbData->pHead; pDelData; pDelData = pDelData->pNext) {





×








UNCOV

548



          if (pDelData->sKey > maxKey || pDelData->eKey < minKey) {





×








UNCOV

549



            continue;




×







550



          } else {











UNCOV

551



            hasDataToCommit = true;




×








UNCOV

552



            if ((code = tsdbCommitInfoAdd(tsdb, fset->fid))) {





×







553



              (void)taosThreadMutexUnlock(&tsdb->mutex);




×







554



              TSDB_CHECK_CODE(code, lino, _exit);





×







555



            }











UNCOV

556



            break;




×







557



          }










558



        }










559













UNCOV

560



        if (hasDataToCommit) {





×








UNCOV

561



          break;




×







562



        }










563



      }










564



    }










565



  }










566












567



  // begin tasks on file set











UNCOV

568



  for (int i = 0; i < taosArrayGetSize(tsdb->commitInfo->arr); i++) {





×








UNCOV

569



    SFileSetCommitInfo *info = *(SFileSetCommitInfo **)taosArrayGet(tsdb->commitInfo->arr, i);




×








UNCOV

570



    tsdbBeginTaskOnFileSet(tsdb, info->fid, EVA_TASK_COMMIT, &fset);




×








UNCOV

571



    if (fset) {





×








UNCOV

572



      code = tsdbTFileSetInitCopy(tsdb, fset, &info->fset);




×








UNCOV

573



      if (code) {





×







574



        (void)taosThreadMutexUnlock(&tsdb->mutex);




×







575



        TAOS_CHECK_GOTO(code, &lino, _exit);





×







576



      }










577



    }










578



  }










579













UNCOV

580



  (void)taosThreadMutexUnlock(&tsdb->mutex);




×







581













UNCOV

582



_exit:




×








UNCOV

583



  if (code) {





×







584



    tsdbCommitInfoDestroy(tsdb);




×







585



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





×







586



  }











UNCOV

587



  return code;




×







588



}










589













UNCOV

590



static int32_t tsdbOpenCommitter(STsdb *tsdb, SCommitInfo *info, SCommitter2 *committer) {




×








UNCOV

591



  int32_t code = 0;




×








UNCOV

592



  int32_t lino = 0;




×







593













UNCOV

594



  committer->tsdb = tsdb;




×








UNCOV

595



  committer->minutes = tsdb->keepCfg.days;




×








UNCOV

596



  committer->precision = tsdb->keepCfg.precision;




×








UNCOV

597



  committer->minRow = info->info.config.tsdbCfg.minRows;




×








UNCOV

598



  committer->maxRow = info->info.config.tsdbCfg.maxRows;




×








UNCOV

599



  committer->cmprAlg = info->info.config.tsdbCfg.compression;




×








UNCOV

600



  committer->sttTrigger = info->info.config.sttTrigger;




×








UNCOV

601



  committer->szPage = info->info.config.tsdbPageSize;




×








UNCOV

602



  committer->compactVersion = INT64_MAX;




×








UNCOV

603



  committer->cid = tsdbFSAllocEid(tsdb->pFS);




×








UNCOV

604



  committer->now = taosGetTimestampSec();




×







605













UNCOV

606



  TAOS_CHECK_GOTO(tsdbCommitInfoBuild(tsdb), &lino, _exit);





×







607













UNCOV

608



_exit:




×








UNCOV

609



  if (code) {





×







610



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





×







611



  } else {











UNCOV

612



    tsdbDebug("vgId:%d %s done", TD_VID(tsdb->pVnode), __func__);





×







613



  }











UNCOV

614



  return code;




×







615



}










616













UNCOV

617



static int32_t tsdbCloseCommitter(SCommitter2 *committer, int32_t eno) {




×








UNCOV

618



  int32_t code = 0;




×








UNCOV

619



  int32_t lino = 0;




×







620













UNCOV

621



  if (eno == 0) {





×








UNCOV

622



    TAOS_CHECK_GOTO(tsdbFSEditBegin(committer->tsdb->pFS, committer->fopArray, TSDB_FEDIT_COMMIT), &lino, _exit);





×







623



  } else {










624



    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(committer->tsdb->pVnode), __func__, __FILE__, lino,





×







625



              tstrerror(eno));










626



  }










627













UNCOV

628



  TARRAY2_DESTROY(committer->dataIterArray, NULL);





×








UNCOV

629



  TARRAY2_DESTROY(committer->tombIterArray, NULL);





×








UNCOV

630



  TARRAY2_DESTROY(committer->sttReaderArray, NULL);





×








UNCOV

631



  TARRAY2_DESTROY(committer->fopArray, NULL);





×








UNCOV

632



  TARRAY2_DESTROY(committer->sttReaderArray, NULL);





×







633













UNCOV

634



_exit:




×








UNCOV

635



  if (code) {





×







636



    tsdbError("vgId:%d %s failed at %s:%d since %s, eid:%" PRId64, TD_VID(committer->tsdb->pVnode), __func__, __FILE__,





×







637



              lino, tstrerror(code), committer->cid);










638



  } else {











UNCOV

639



    tsdbDebug("vgId:%d %s done, eid:%" PRId64, TD_VID(committer->tsdb->pVnode), __func__, committer->cid);





×







640



  }











UNCOV

641



  return code;




×







642



}










643













UNCOV

644



int32_t tsdbPreCommit(STsdb *tsdb) {




×








UNCOV

645



  (void)taosThreadMutexLock(&tsdb->mutex);




×








UNCOV

646



  ASSERT_CORE(tsdb->imem == NULL, "imem should be null to commit mem");





×








UNCOV

647



  tsdb->imem = tsdb->mem;




×








UNCOV

648



  tsdb->mem = NULL;




×








UNCOV

649



  (void)taosThreadMutexUnlock(&tsdb->mutex);




×








UNCOV

650



  return 0;




×







651



}










652













UNCOV

653



int32_t tsdbCommitBegin(STsdb *tsdb, SCommitInfo *info) {




×








UNCOV

654



  if (!tsdb) return 0;





×







655













UNCOV

656



  int32_t code = 0;




×








UNCOV

657



  int32_t lino = 0;




×







658













UNCOV

659



  SMemTable *imem = tsdb->imem;




×








UNCOV

660



  int64_t    nRow = imem->nRow;




×








UNCOV

661



  int64_t    nDel = imem->nDel;




×







662













UNCOV

663



  if ((nRow == 0 && nDel == 0) || (tsBypassFlag & TSDB_BYPASS_RB_TSDB_COMMIT)) {





×








UNCOV

664



    (void)taosThreadMutexLock(&tsdb->mutex);




×








UNCOV

665



    tsdb->imem = NULL;




×








UNCOV

666



    (void)taosThreadMutexUnlock(&tsdb->mutex);




×








UNCOV

667



    tsdbUnrefMemTable(imem, NULL, true);




×







668



  } else {











UNCOV

669



    SCommitter2 committer = {0};




×







670













UNCOV

671



    TAOS_CHECK_GOTO(tsdbOpenCommitter(tsdb, info, &committer), &lino, _exit);





×







672













UNCOV

673



    for (int32_t i = 0; i < taosArrayGetSize(tsdb->commitInfo->arr); i++) {





×








UNCOV

674



      committer.ctx->info = *(SFileSetCommitInfo **)taosArrayGet(tsdb->commitInfo->arr, i);




×








UNCOV

675



      TAOS_CHECK_GOTO(tsdbCommitFileSet(&committer), &lino, _exit);





×







676



    }










677













UNCOV

678



    TAOS_CHECK_GOTO(tsdbCloseCommitter(&committer, code), &lino, _exit);





×







679



  }










680













UNCOV

681



_exit:




×








UNCOV

682



  if (code) {





×








UNCOV

683



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





×







684



  } else {











UNCOV

685



    tsdbInfo("vgId:%d %s done, nRow:%" PRId64 " nDel:%" PRId64, TD_VID(tsdb->pVnode), __func__, nRow, nDel);





×







686



  }











UNCOV

687



  return code;




×







688



}










689













UNCOV

690



int32_t tsdbCommitCommit(STsdb *tsdb) {




×








UNCOV

691



  int32_t code = 0;




×








UNCOV

692



  int32_t lino = 0;




×







693













UNCOV

694



  if (tsdb->imem) {





×








UNCOV

695



    SMemTable *pMemTable = tsdb->imem;




×







696













UNCOV

697



    (void)taosThreadMutexLock(&tsdb->mutex);




×







698













UNCOV

699



    if ((code = tsdbFSEditCommit(tsdb->pFS))) {





×







700



      (void)taosThreadMutexUnlock(&tsdb->mutex);




×







701



      TSDB_CHECK_CODE(code, lino, _exit);





×







702



    }











UNCOV

703



    tsdb->imem = NULL;




×







704













UNCOV

705



    for (int32_t i = 0; i < taosArrayGetSize(tsdb->commitInfo->arr); i++) {





×








UNCOV

706



      SFileSetCommitInfo *info = *(SFileSetCommitInfo **)taosArrayGet(tsdb->commitInfo->arr, i);




×








UNCOV

707



      if (info->fset) {





×








UNCOV

708



        tsdbFinishTaskOnFileSet(tsdb, info->fid, EVA_TASK_COMMIT);




×







709



      }










710



    }










711













UNCOV

712



    (void)taosThreadMutexUnlock(&tsdb->mutex);




×







713













UNCOV

714



    tsdbCommitInfoDestroy(tsdb);




×








UNCOV

715



    tsdbUnrefMemTable(pMemTable, NULL, true);




×







716



  }










717













UNCOV

718



_exit:




×








UNCOV

719



  if (code) {





×







720



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





×







721



  } else {











UNCOV

722



    tsdbInfo("vgId:%d %s done", TD_VID(tsdb->pVnode), __func__);





×







723



  }











UNCOV

724



  return code;




×







725



}










726












727



int32_t tsdbCommitAbort(STsdb *pTsdb) {




×







728



  int32_t code = 0;




×







729



  int32_t lino = 0;




×







730












731



  if (pTsdb->imem == NULL) goto _exit;





×







732












733



  TAOS_CHECK_GOTO(tsdbFSEditAbort(pTsdb->pFS), &lino, _exit);





×







734












735



  (void)taosThreadMutexLock(&pTsdb->mutex);




×







736



  for (int32_t i = 0; i < taosArrayGetSize(pTsdb->commitInfo->arr); i++) {





×







737



    SFileSetCommitInfo *info = *(SFileSetCommitInfo **)taosArrayGet(pTsdb->commitInfo->arr, i);




×







738



    if (info->fset) {





×







739



      tsdbFinishTaskOnFileSet(pTsdb, info->fid, EVA_TASK_COMMIT);




×







740



    }










741



  }










742



  (void)taosThreadMutexUnlock(&pTsdb->mutex);




×







743



  tsdbCommitInfoDestroy(pTsdb);




×







744












745



_exit:




×







746



  if (code) {





×







747



    tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));





×







748



  } else {










749



    tsdbInfo("vgId:%d %s done", TD_VID(pTsdb->pVnode), __func__);





×







750



  }










751



  return code;




×







752



}
























    

  • 

    •