taosdata / TDengine / #3720

coverage: 30.242% (-31.7%) from 61.936%
#3720

push

travis-ci

web-flow 
feat(taosBenchmark): supports decimal data type (#30456)

* feat: taosBenchmark supports decimal data type

* build: decimal script not use pytest.sh

* fix: fix typo for decimal script

* test: insertBasic.py debug

71234 of 313946 branches covered (22.69%)

Branch coverage included in aggregate %.

38 of 423 new or added lines in 8 files covered. (8.98%)

120240 existing lines in 447 files now uncovered.

118188 of 312400 relevant lines covered (37.83%)

1450220.33 hits per line

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

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










17



#include "functionResInfo.h"










18



#include "taoserror.h"










19



#include "tarray.h"










20



#include "tcommon.h"










21



#include "tsdb.h"










22



#include "tsdbDataFileRW.h"










23



#include "tsdbReadUtil.h"










24












25



#define HASTYPE(_type, _t) (((_type) & (_t)) == (_t))










26













UNCOV

27



static int32_t setFirstLastResColToNull(SColumnInfoData* pCol, int32_t row) {




×








UNCOV

28



  int32_t        code = TSDB_CODE_SUCCESS;




×








UNCOV

29



  int32_t        lino = 0;




×








UNCOV

30



  char*          buf = NULL;




×








UNCOV

31



  SFirstLastRes* pRes = NULL;




×







32













UNCOV

33



  TSDB_CHECK_NULL(pCol, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







34













UNCOV

35



  buf = taosMemoryCalloc(1, pCol->info.bytes);





×








UNCOV

36



  TSDB_CHECK_NULL(buf, code, lino, _end, terrno);





×







37













UNCOV

38



  pRes = (SFirstLastRes*)((char*)buf + VARSTR_HEADER_SIZE);




×








UNCOV

39



  pRes->bytes = 0;




×








UNCOV

40



  pRes->hasResult = true;




×








UNCOV

41



  pRes->isNull = true;




×








UNCOV

42



  varDataSetLen(buf, pCol->info.bytes - VARSTR_HEADER_SIZE);




×








UNCOV

43



  code = colDataSetVal(pCol, row, buf, false);




×








UNCOV

44



  TSDB_CHECK_CODE(code, lino, _end);





×







45













UNCOV

46



_end:




×








UNCOV

47



  if (code != TSDB_CODE_SUCCESS) {





×







48



    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







49



  }











UNCOV

50



  if (buf != NULL) {





×








UNCOV

51



    taosMemoryFreeClear(buf);





×







52



  }











UNCOV

53



  return code;




×







54



}










55













UNCOV

56



static int32_t saveOneRowForLastRaw(SLastCol* pColVal, SCacheRowsReader* pReader, const int32_t slotId,




×







57



                                    SColumnInfoData* pColInfoData, int32_t numOfRows) {











UNCOV

58



  int32_t  code = TSDB_CODE_SUCCESS;




×








UNCOV

59



  int32_t  lino = 0;




×








UNCOV

60



  SColVal* pVal = NULL;




×







61













UNCOV

62



  TSDB_CHECK_NULL(pColVal, code, lino, _end, TSDB_CODE_INVALID_PARA);





×








UNCOV

63



  TSDB_CHECK_NULL(pColInfoData, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







64













UNCOV

65



  pVal = &pColVal->colVal;




×







66












67



  // allNullRow = false;











UNCOV

68



  if (IS_VAR_DATA_TYPE(pColVal->colVal.value.type)) {





×








UNCOV

69



    if (!COL_VAL_IS_VALUE(&pColVal->colVal)) {





×








UNCOV

70



      colDataSetNULL(pColInfoData, numOfRows);





×







71



    } else {











UNCOV

72



      TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);





×








UNCOV

73



      varDataSetLen(pReader->transferBuf[slotId], pVal->value.nData);




×







74













UNCOV

75



      memcpy(varDataVal(pReader->transferBuf[slotId]), pVal->value.pData, pVal->value.nData);




×








UNCOV

76



      code = colDataSetVal(pColInfoData, numOfRows, pReader->transferBuf[slotId], false);




×








UNCOV

77



      TSDB_CHECK_CODE(code, lino, _end);





×







78



    }










79



  } else {











UNCOV

80



    code = colDataSetVal(pColInfoData, numOfRows, VALUE_GET_DATUM(&pVal->value, pColVal->colVal.value.type),





×








UNCOV

81



                         !COL_VAL_IS_VALUE(pVal));





×








UNCOV

82



    TSDB_CHECK_CODE(code, lino, _end);





×







83



  }










84













UNCOV

85



_end:




×








UNCOV

86



  if (code != TSDB_CODE_SUCCESS) {





×







87



    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







88



  }











UNCOV

89



  return code;




×







90



}










91













UNCOV

92



static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* pReader, const int32_t* slotIds,




×







93



                          const int32_t* dstSlotIds, void** pRes, const char* idStr) {











UNCOV

94



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

95



  int32_t lino = 0;




×








UNCOV

96



  int32_t numOfRows = 0;




×








UNCOV

97



  SArray* funcTypeBlockArray = NULL;




×







98













UNCOV

99



  TSDB_CHECK_NULL(pBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);





×








UNCOV

100



  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);





×








UNCOV

101



  if (pReader->numOfCols > 0) {





×








UNCOV

102



    TSDB_CHECK_NULL(slotIds, code, lino, _end, TSDB_CODE_INVALID_PARA);





×








UNCOV

103



    TSDB_CHECK_NULL(dstSlotIds, code, lino, _end, TSDB_CODE_INVALID_PARA);





×








UNCOV

104



    TSDB_CHECK_NULL(pRes, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







105



  }










106













UNCOV

107



  numOfRows = pBlock->info.rows;




×







108













UNCOV

109



  if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST)) {





×








UNCOV

110



    uint64_t       ts = TSKEY_MIN;




×








UNCOV

111



    SFirstLastRes* p = NULL;




×








UNCOV

112



    col_id_t       colId = -1;




×







113













UNCOV

114



    funcTypeBlockArray = taosArrayInit(pReader->numOfCols, sizeof(int32_t));




×








UNCOV

115



    TSDB_CHECK_NULL(funcTypeBlockArray, code, lino, _end, terrno);





×







116













UNCOV

117



    for (int32_t i = 0; i < pReader->numOfCols; ++i) {





×








UNCOV

118



      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotIds[i]);




×








UNCOV

119



      TSDB_CHECK_NULL(pColInfoData, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







120













UNCOV

121



      int32_t funcType = FUNCTION_TYPE_CACHE_LAST;




×








UNCOV

122



      if (pReader->pFuncTypeList != NULL && taosArrayGetSize(pReader->pFuncTypeList) > i) {





×








UNCOV

123



        void* pVal = taosArrayGet(pReader->pFuncTypeList, i);




×








UNCOV

124



        TSDB_CHECK_NULL(pVal, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







125













UNCOV

126



        funcType = *(int32_t*)pVal;




×








UNCOV

127



        pVal = taosArrayGet(pReader->pFuncTypeList, i);




×








UNCOV

128



        TSDB_CHECK_NULL(pVal, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







129













UNCOV

130



        void* px = taosArrayInsert(funcTypeBlockArray, dstSlotIds[i], pVal);




×








UNCOV

131



        TSDB_CHECK_NULL(px, code, lino, _end, terrno);





×







132



      }










133













UNCOV

134



      if (slotIds[i] == -1) {





×








UNCOV

135



        if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {





×







136



          colDataSetNULL(pColInfoData, numOfRows);





×







137



          continue;




×







138



        }










139













UNCOV

140



        code = setFirstLastResColToNull(pColInfoData, numOfRows);




×








UNCOV

141



        TSDB_CHECK_CODE(code, lino, _end);





×








UNCOV

142



        continue;




×







143



      }










144













UNCOV

145



      int32_t   slotId = slotIds[i];




×








UNCOV

146



      SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, i);




×








UNCOV

147



      TSDB_CHECK_NULL(pColVal, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







148













UNCOV

149



      colId = pColVal->colVal.cid;




×








UNCOV

150



      if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {





×








UNCOV

151



        code = saveOneRowForLastRaw(pColVal, pReader, slotId, pColInfoData, numOfRows);




×








UNCOV

152



        TSDB_CHECK_CODE(code, lino, _end);





×







153













UNCOV

154



        continue;




×







155



      }










156













UNCOV

157



      p = (SFirstLastRes*)varDataVal(pRes[i]);




×







158













UNCOV

159



      p->ts = pColVal->rowKey.ts;




×








UNCOV

160



      ts = p->ts;




×








UNCOV

161



      p->isNull = !COL_VAL_IS_VALUE(&pColVal->colVal);




×







162



      // allNullRow = p->isNull & allNullRow;











UNCOV

163



      if (!p->isNull) {





×








UNCOV

164



        if (IS_VAR_DATA_TYPE(pColVal->colVal.value.type)) {





×








UNCOV

165



          varDataSetLen(p->buf, pColVal->colVal.value.nData);




×







166













UNCOV

167



          memcpy(varDataVal(p->buf), pColVal->colVal.value.pData, pColVal->colVal.value.nData);




×








UNCOV

168



          p->bytes = pColVal->colVal.value.nData + VARSTR_HEADER_SIZE;  // binary needs to plus the header size




×







169



        } else {











UNCOV

170



          memcpy(p->buf, VALUE_GET_DATUM(&pColVal->colVal.value, pColVal->colVal.value.type),





×








UNCOV

171



                 pReader->pSchema->columns[slotId].bytes);





×








UNCOV

172



          p->bytes = pReader->pSchema->columns[slotId].bytes;




×







173



        }










174



      }










175












176



      // pColInfoData->info.bytes includes the VARSTR_HEADER_SIZE, need to subtract it











UNCOV

177



      p->hasResult = true;




×








UNCOV

178



      varDataSetLen(pRes[i], pColInfoData->info.bytes - VARSTR_HEADER_SIZE);




×








UNCOV

179



      code = colDataSetVal(pColInfoData, numOfRows, (const char*)pRes[i], false);




×








UNCOV

180



      TSDB_CHECK_CODE(code, lino, _end);





×







181



    }










182













UNCOV

183



    for (int32_t idx = 0; idx < taosArrayGetSize(pBlock->pDataBlock); ++idx) {





×








UNCOV

184



      SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, idx);




×








UNCOV

185



      TSDB_CHECK_NULL(pCol, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







186













UNCOV

187



      if (idx < funcTypeBlockArray->size) {





×








UNCOV

188



        void* pVal = taosArrayGet(funcTypeBlockArray, idx);




×








UNCOV

189



        TSDB_CHECK_NULL(pVal, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







190













UNCOV

191



        int32_t funcType = *(int32_t*)pVal;




×








UNCOV

192



        if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {





×








UNCOV

193



          continue;




×







194



        }










195



      }










196













UNCOV

197



      if (pCol->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID && pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP) {





×








UNCOV

198



        if (ts == TSKEY_MIN) {





×








UNCOV

199



          colDataSetNULL(pCol, numOfRows);





×







200



        } else {











UNCOV

201



          code = colDataSetVal(pCol, numOfRows, (const char*)&ts, false);




×








UNCOV

202



          TSDB_CHECK_CODE(code, lino, _end);





×







203



        }











UNCOV

204



        continue;




×








UNCOV

205



      } else if (pReader->numOfCols == 1 && idx != dstSlotIds[0] && (pCol->info.colId == colId || colId == -1)) {





×








UNCOV

206



        if (p && !p->isNull) {





×








UNCOV

207



          code = colDataSetVal(pCol, numOfRows, p->buf, false);




×








UNCOV

208



          TSDB_CHECK_CODE(code, lino, _end);





×







209



        } else {











UNCOV

210



          colDataSetNULL(pCol, numOfRows);





×







211



        }










212



      }










213



    }










214












215



    // pBlock->info.rows += allNullRow ? 0 : 1;











UNCOV

216



    ++pBlock->info.rows;




×








UNCOV

217



  } else if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST_ROW)) {





×








UNCOV

218



    for (int32_t i = 0; i < pReader->numOfCols; ++i) {





×








UNCOV

219



      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotIds[i]);




×








UNCOV

220



      TSDB_CHECK_NULL(pColInfoData, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







221













UNCOV

222



      int32_t slotId = slotIds[i];




×








UNCOV

223



      if (slotId == -1) {





×








UNCOV

224



        colDataSetNULL(pColInfoData, numOfRows);





×








UNCOV

225



        continue;




×







226



      }










227













UNCOV

228



      SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, i);




×








UNCOV

229



      TSDB_CHECK_NULL(pColVal, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







230













UNCOV

231



      code = saveOneRowForLastRaw(pColVal, pReader, slotId, pColInfoData, numOfRows);




×








UNCOV

232



      TSDB_CHECK_CODE(code, lino, _end);





×







233



    }










234












235



    // pBlock->info.rows += allNullRow ? 0 : 1;











UNCOV

236



    ++pBlock->info.rows;




×







237



  } else {











UNCOV

238



    tsdbError("invalid retrieve type:%d, %s", pReader->type, idStr);





×







239



    code = TSDB_CODE_INVALID_PARA;




×







240



    TSDB_CHECK_CODE(code, lino, _end);





×







241



  }










242












243



_end:




×








UNCOV

244



  if (code != TSDB_CODE_SUCCESS) {





×







245



    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







246



  }











UNCOV

247



  if (funcTypeBlockArray != NULL) {





×








UNCOV

248



    taosArrayDestroy(funcTypeBlockArray);




×







249



  }











UNCOV

250



  return code;




×







251



}










252













UNCOV

253



static int32_t setTableSchema(SCacheRowsReader* p, uint64_t suid, const char* idstr) {




×








UNCOV

254



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

255



  int32_t lino = 0;




×








UNCOV

256



  int32_t numOfTables = 0;




×







257













UNCOV

258



  TSDB_CHECK_NULL(p, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







259













UNCOV

260



  numOfTables = p->numOfTables;




×







261













UNCOV

262



  if (suid != 0) {





×








UNCOV

263



    code = metaGetTbTSchemaNotNull(p->pVnode->pMeta, suid, -1, 1, &p->pSchema);




×








UNCOV

264



    if (TSDB_CODE_SUCCESS != code) {





×







265



      tsdbWarn("stable:%" PRIu64 " has been dropped, failed to retrieve cached rows, %s", suid, idstr);





×







266



      if (code == TSDB_CODE_NOT_FOUND) {





×







267



        code = TSDB_CODE_PAR_TABLE_NOT_EXIST;




×







268



      }










269



      TSDB_CHECK_CODE(code, lino, _end);





×







270



    }










271



  } else {











UNCOV

272



    for (int32_t i = 0; i < numOfTables; ++i) {





×








UNCOV

273



      uint64_t uid = p->pTableList[i].uid;




×








UNCOV

274



      code = metaGetTbTSchemaMaybeNull(p->pVnode->pMeta, uid, -1, 1, &p->pSchema);




×








UNCOV

275



      TSDB_CHECK_CODE(code, lino, _end);





×








UNCOV

276



      if (p->pSchema != NULL) {





×








UNCOV

277



        break;




×







278



      }










279












280



      tsdbWarn("table:%" PRIu64 " has been dropped, failed to retrieve cached rows, %s", uid, idstr);





×







281



    }










282












283



    // all queried tables have been dropped already, return immediately.











UNCOV

284



    if (p->pSchema == NULL) {





×







285



      tsdbWarn("all queried tables has been dropped, try next group, %s", idstr);





×







286



      code = TSDB_CODE_PAR_TABLE_NOT_EXIST;




×







287



      TSDB_CHECK_CODE(code, lino, _end);





×







288



    }










289



  }










290













UNCOV

291



_end:




×








UNCOV

292



  if (code != TSDB_CODE_SUCCESS) {





×







293



    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







294



  }











UNCOV

295



  return code;




×







296



}










297













UNCOV

298



int32_t tsdbReuseCacherowsReader(void* reader, void* pTableIdList, int32_t numOfTables) {




×








UNCOV

299



  int32_t           code = TSDB_CODE_SUCCESS;




×








UNCOV

300



  int32_t           lino = 0;




×








UNCOV

301



  SCacheRowsReader* pReader = (SCacheRowsReader*)reader;




×







302













UNCOV

303



  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







304













UNCOV

305



  pReader->pTableList = pTableIdList;




×








UNCOV

306



  pReader->numOfTables = numOfTables;




×








UNCOV

307



  pReader->lastTs = INT64_MIN;




×








UNCOV

308



  destroySttBlockReader(pReader->pLDataIterArray, NULL);




×








UNCOV

309



  pReader->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);




×








UNCOV

310



  TSDB_CHECK_NULL(pReader->pLDataIterArray, code, lino, _end, terrno);





×







311













UNCOV

312



_end:




×








UNCOV

313



  if (code != TSDB_CODE_SUCCESS) {





×







314



    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







315



  }











UNCOV

316



  return code;




×







317



}










318













UNCOV

319



int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList, int32_t numOfTables, int32_t numOfCols,




×







320



                                SArray* pCidList, int32_t* pSlotIds, uint64_t suid, void** pReader, const char* idstr,










321



                                SArray* pFuncTypeList, SColumnInfo* pPkCol, int32_t numOfPks) {











UNCOV

322



  int32_t           code = TSDB_CODE_SUCCESS;




×








UNCOV

323



  int32_t           lino = 0;




×








UNCOV

324



  SCacheRowsReader* p = NULL;




×







325













UNCOV

326



  TSDB_CHECK_NULL(pVnode, code, lino, _end, TSDB_CODE_INVALID_PARA);





×








UNCOV

327



  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







328













UNCOV

329



  *pReader = NULL;




×








UNCOV

330



  p = taosMemoryCalloc(1, sizeof(SCacheRowsReader));





×








UNCOV

331



  TSDB_CHECK_NULL(p, code, lino, _end, terrno);





×







332













UNCOV

333



  p->type = type;




×








UNCOV

334



  p->pVnode = pVnode;




×








UNCOV

335



  p->pTsdb = p->pVnode->pTsdb;




×








UNCOV

336



  p->info.verRange = (SVersionRange){.minVer = 0, .maxVer = INT64_MAX};




×








UNCOV

337



  p->info.suid = suid;




×








UNCOV

338



  p->numOfCols = numOfCols;




×








UNCOV

339



  p->pCidList = pCidList;




×








UNCOV

340



  p->pSlotIds = pSlotIds;




×








UNCOV

341



  p->pFuncTypeList = pFuncTypeList;




×







342













UNCOV

343



  p->rowKey.numOfPKs = numOfPks;




×








UNCOV

344



  if (numOfPks > 0) {





×








UNCOV

345



    TSDB_CHECK_NULL(pPkCol, code, lino, _end, TSDB_CODE_INVALID_PARA);





×








UNCOV

346



    p->rowKey.pks[0].type = pPkCol->type;




×








UNCOV

347



    if (IS_VAR_DATA_TYPE(pPkCol->type)) {





×







348



      p->rowKey.pks[0].pData = taosMemoryCalloc(1, pPkCol->bytes);





×







349



      if (p->rowKey.pks[0].pData == NULL) {





×







350



        taosMemoryFreeClear(p);





×







351



        code = terrno;




×







352



        TSDB_CHECK_CODE(code, lino, _end);





×







353



      }










354



    }










355













UNCOV

356



    p->pkColumn = *pPkCol;




×







357



  }










358













UNCOV

359



  if (numOfTables == 0) {





×







360



    *pReader = p;




×







361



    p = NULL;




×







362



    goto _end;




×







363



  }










364













UNCOV

365



  p->pTableList = pTableIdList;




×








UNCOV

366



  p->numOfTables = numOfTables;




×







367













UNCOV

368



  code = setTableSchema(p, suid, idstr);




×








UNCOV

369



  TSDB_CHECK_CODE(code, lino, _end);





×







370













UNCOV

371



  p->transferBuf = taosMemoryCalloc(p->pSchema->numOfCols, POINTER_BYTES);





×








UNCOV

372



  TSDB_CHECK_NULL(p->transferBuf, code, lino, _end, terrno);





×







373













UNCOV

374



  for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) {





×








UNCOV

375



    if (IS_VAR_DATA_TYPE(p->pSchema->columns[i].type)) {





×








UNCOV

376



      p->transferBuf[i] = taosMemoryMalloc(p->pSchema->columns[i].bytes);





×








UNCOV

377



      TSDB_CHECK_NULL(p->transferBuf[i], code, lino, _end, terrno);





×







378



    }










379



  }










380













UNCOV

381



  if (idstr != NULL) {





×








UNCOV

382



    p->idstr = taosStrdup(idstr);





×








UNCOV

383



    TSDB_CHECK_NULL(p->idstr, code, lino, _end, terrno);





×







384



  }











UNCOV

385



  code = taosThreadMutexInit(&p->readerMutex, NULL);




×








UNCOV

386



  TSDB_CHECK_CODE(code, lino, _end);





×







387













UNCOV

388



  p->lastTs = INT64_MIN;




×







389













UNCOV

390



  *pReader = p;




×








UNCOV

391



  p = NULL;




×







392













UNCOV

393



_end:




×








UNCOV

394



  if (code != TSDB_CODE_SUCCESS) {





×







395



    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







396



    *pReader = NULL;




×







397



  }











UNCOV

398



  if (p != NULL) {





×







399



    tsdbCacherowsReaderClose(p);




×







400



  }











UNCOV

401



  return code;




×







402



}










403













UNCOV

404



void tsdbCacherowsReaderClose(void* pReader) {




×








UNCOV

405



  SCacheRowsReader* p = pReader;




×








UNCOV

406



  if (p == NULL) {





×







407



    return;




×







408



  }










409













UNCOV

410



  if (p->pSchema != NULL && p->transferBuf != NULL) {





×








UNCOV

411



    for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) {





×








UNCOV

412



      taosMemoryFreeClear(p->transferBuf[i]);





×







413



    }










414













UNCOV

415



    taosMemoryFree(p->transferBuf);





×








UNCOV

416



    taosMemoryFree(p->pSchema);





×







417



  }










418













UNCOV

419



  taosMemoryFree(p->pCurrSchema);





×







420













UNCOV

421



  if (p->rowKey.numOfPKs > 0) {





×








UNCOV

422



    for (int32_t i = 0; i < p->rowKey.numOfPKs; i++) {





×








UNCOV

423



      if (IS_VAR_DATA_TYPE(p->rowKey.pks[i].type)) {





×







424



        taosMemoryFree(p->rowKey.pks[i].pData);





×







425



      }










426



    }










427



  }










428













UNCOV

429



  if (p->pLDataIterArray) {





×








UNCOV

430



    destroySttBlockReader(p->pLDataIterArray, NULL);




×








UNCOV

431



    p->pLDataIterArray = NULL;




×







432



  }










433













UNCOV

434



  if (p->pFileReader) {





×







435



    tsdbDataFileReaderClose(&p->pFileReader);




×







436



    p->pFileReader = NULL;




×







437



  }










438













UNCOV

439



  taosMemoryFree((void*)p->idstr);





×








UNCOV

440



  (void)taosThreadMutexDestroy(&p->readerMutex);




×







441













UNCOV

442



  if (p->pTableMap) {





×








UNCOV

443



    void*   pe = NULL;




×








UNCOV

444



    int32_t iter = 0;




×








UNCOV

445



    while ((pe = tSimpleHashIterate(p->pTableMap, pe, &iter)) != NULL) {





×








UNCOV

446



      STableLoadInfo* pInfo = *(STableLoadInfo**)pe;




×








UNCOV

447



      taosArrayDestroy(pInfo->pTombData);




×








UNCOV

448



      pInfo->pTombData = NULL;




×







449



    }










450













UNCOV

451



    tSimpleHashCleanup(p->pTableMap);




×







452



  }











UNCOV

453



  if (p->uidList) {





×








UNCOV

454



    taosMemoryFree(p->uidList);





×







455



  }










456













UNCOV

457



  taosMemoryFree(pReader);





×







458



}










459












460



static int32_t tsdbCacheQueryReseek(void* pQHandle) {




×







461



  int32_t           code = 0;




×







462



  SCacheRowsReader* pReader = pQHandle;




×







463












464



  code = taosThreadMutexTryLock(&pReader->readerMutex);




×







465



  if (code == 0) {





×







466



    // pause current reader's state if not paused, save ts & version for resuming










467



    // just wait for the big all tables' snapshot untaking for now










468












469



    code = TSDB_CODE_VND_QUERY_BUSY;




×







470



    (void)taosThreadMutexUnlock(&pReader->readerMutex);




×







471












472



    return code;




×







473



  } else if (code == EBUSY) {





×







474



    return TSDB_CODE_VND_QUERY_BUSY;




×







475



  } else {










476



    return -1;




×







477



  }










478



}










479













UNCOV

480



int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32_t* slotIds, const int32_t* dstSlotIds,




×







481



                              SArray* pTableUidList, bool* pGotAll) {











UNCOV

482



  int32_t           code = TSDB_CODE_SUCCESS;




×








UNCOV

483



  int32_t           lino = 0;




×








UNCOV

484



  bool              hasRes = false;




×








UNCOV

485



  SArray*           pRow = NULL;




×








UNCOV

486



  void**            pRes = NULL;




×








UNCOV

487



  SCacheRowsReader* pr = NULL;




×








UNCOV

488



  int32_t           pkBufLen = 0;




×







489













UNCOV

490



  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);





×








UNCOV

491



  TSDB_CHECK_NULL(pResBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);





×







492













UNCOV

493



  pr = pReader;




×







494













UNCOV

495



  pr->pReadSnap = NULL;




×








UNCOV

496



  pRow = taosArrayInit(TARRAY_SIZE(pr->pCidList), sizeof(SLastCol));




×








UNCOV

497



  TSDB_CHECK_NULL(pRow, code, lino, _end, terrno);





×







498













UNCOV

499



  pRes = taosMemoryCalloc(pr->numOfCols, POINTER_BYTES);





×








UNCOV

500



  TSDB_CHECK_NULL(pRes, code, lino, _end, terrno);





×







501













UNCOV

502



  pkBufLen = (pr->rowKey.numOfPKs > 0) ? pr->pkColumn.bytes : 0;





×








UNCOV

503



  for (int32_t j = 0; j < pr->numOfCols; ++j) {





×








UNCOV

504



    int32_t bytes = (slotIds[j] == -1) ? 1 : pr->pSchema->columns[slotIds[j]].bytes;





×







505













UNCOV

506



    pRes[j] = taosMemoryCalloc(1, sizeof(SFirstLastRes) + bytes + pkBufLen + VARSTR_HEADER_SIZE);





×








UNCOV

507



    TSDB_CHECK_NULL(pRes[j], code, lino, _end, terrno);





×







508













UNCOV

509



    SFirstLastRes* p = (SFirstLastRes*)varDataVal(pRes[j]);




×








UNCOV

510



    p->ts = INT64_MIN;




×







511



  }










512













UNCOV

513



  (void)taosThreadMutexLock(&pr->readerMutex);




×








UNCOV

514



  code = tsdbTakeReadSnap2((STsdbReader*)pr, tsdbCacheQueryReseek, &pr->pReadSnap, pr->idstr);




×








UNCOV

515



  TSDB_CHECK_CODE(code, lino, _end);





×







516













UNCOV

517



  int8_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3;




×







518













UNCOV

519



  STableKeyInfo* pTableList = pr->pTableList;




×







520












521



  // retrieve the only one last row of all tables in the uid list.











UNCOV

522



  if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_SINGLE)) {





×








UNCOV

523



    SArray* pLastCols = taosArrayInit(pr->numOfCols, sizeof(SLastCol));




×








UNCOV

524



    TSDB_CHECK_NULL(pLastCols, code, lino, _end, terrno);





×







525













UNCOV

526



    for (int32_t i = 0; i < pr->numOfCols; ++i) {





×








UNCOV

527



      int32_t slotId = slotIds[i];




×








UNCOV

528



      if (slotId == -1) {





×








UNCOV

529



        SLastCol p = {.rowKey.ts = INT64_MIN, .colVal.value.type = TSDB_DATA_TYPE_BOOL, .colVal.flag = CV_FLAG_NULL};




×








UNCOV

530



        void*    px = taosArrayPush(pLastCols, &p);




×








UNCOV

531



        TSDB_CHECK_NULL(px, code, lino, _end, terrno);





×








UNCOV

532



        continue;




×







533



      }











UNCOV

534



      struct STColumn* pCol = &pr->pSchema->columns[slotId];




×








UNCOV

535



      SLastCol         p = {.rowKey.ts = INT64_MIN, .colVal.value.type = pCol->type, .colVal.flag = CV_FLAG_NULL};




×







536













UNCOV

537



      if (pr->rowKey.numOfPKs > 0) {





×







538



        p.rowKey.numOfPKs = pr->rowKey.numOfPKs;




×







539



        for (int32_t j = 0; j < pr->rowKey.numOfPKs; j++) {





×







540



          p.rowKey.pks[j].type = pr->pkColumn.type;




×







541



          if (IS_VAR_DATA_TYPE(pr->pkColumn.type)) {





×







542



            p.rowKey.pks[j].pData = taosMemoryCalloc(1, pr->pkColumn.bytes);





×







543



            TSDB_CHECK_NULL(p.rowKey.pks[j].pData, code, lino, _end, terrno);





×







544



          }










545



        }










546



      }










547













UNCOV

548



      if (IS_VAR_DATA_TYPE(pCol->type) || pCol->type == TSDB_DATA_TYPE_DECIMAL) {





×








UNCOV

549



        p.colVal.value.pData = taosMemoryCalloc(pCol->bytes, sizeof(char));





×








UNCOV

550



        TSDB_CHECK_NULL(p.colVal.value.pData, code, lino, _end, terrno);





×







551



      }










552













UNCOV

553



      void* px = taosArrayPush(pLastCols, &p);




×








UNCOV

554



      TSDB_CHECK_NULL(px, code, lino, _end, terrno);





×







555



    }










556













UNCOV

557



    int64_t st = taosGetTimestampUs();




×








UNCOV

558



    int64_t totalLastTs = INT64_MAX;




×








UNCOV

559



    for (int32_t i = 0; i < pr->numOfTables; ++i) {





×








UNCOV

560



      tb_uid_t uid = pTableList[i].uid;




×







561













UNCOV

562



      code = tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype);




×








UNCOV

563



      if (code == -1) {  // fix the invalid return code





×







564



        code = 0;




×







565



      }











UNCOV

566



      TSDB_CHECK_CODE(code, lino, _end);





×







567













UNCOV

568



      if (TARRAY_SIZE(pRow) <= 0 || COL_VAL_IS_NONE(&((SLastCol*)TARRAY_DATA(pRow))[0].colVal)) {





×








UNCOV

569



        taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);




×








UNCOV

570



        continue;




×







571



      }










572












573



      {











UNCOV

574



        bool    hasNotNullRow = true;




×








UNCOV

575



        int64_t singleTableLastTs = INT64_MAX;




×








UNCOV

576



        for (int32_t k = 0; k < pr->numOfCols; ++k) {





×








UNCOV

577



          if (slotIds[k] == -1) continue;





×








UNCOV

578



          SLastCol* p = taosArrayGet(pLastCols, k);




×








UNCOV

579



          if (p == NULL) {





×







580



            return TSDB_CODE_INVALID_PARA;




×







581



          }










582













UNCOV

583



          SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, k);




×








UNCOV

584



          if (pColVal == NULL) {





×







585



            return TSDB_CODE_INVALID_PARA;




×







586



          }










587













UNCOV

588



          if (tRowKeyCompare(&pColVal->rowKey, &p->rowKey) > 0) {





×








UNCOV

589



            if (!COL_VAL_IS_VALUE(&pColVal->colVal) && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {





×








UNCOV

590



              if (!COL_VAL_IS_VALUE(&p->colVal)) {





×








UNCOV

591



                hasNotNullRow = false;




×







592



              }










593



              // For all of cols is null, the last null col of last table will be save











UNCOV

594



              if (i != pr->numOfTables - 1 || k != pr->numOfCols - 1 || hasRes) {





×








UNCOV

595



                continue;




×







596



              }










597



            }










598













UNCOV

599



            hasRes = true;




×








UNCOV

600



            p->rowKey.ts = pColVal->rowKey.ts;




×








UNCOV

601



            for (int32_t j = 0; j < p->rowKey.numOfPKs; j++) {





×







602



              if (IS_VAR_DATA_TYPE(p->rowKey.pks[j].type)) {





×







603



                memcpy(p->rowKey.pks[j].pData, pColVal->rowKey.pks[j].pData, pColVal->rowKey.pks[j].nData);




×







604



                p->rowKey.pks[j].nData = pColVal->rowKey.pks[j].nData;




×







605



              } else {










606



                valueCloneDatum(p->rowKey.pks + j, pColVal->rowKey.pks + j, p->rowKey.pks[j].type);




×







607



              }










608



            }










609













UNCOV

610



            if (k == 0) {





×








UNCOV

611



              if (TARRAY_SIZE(pTableUidList) == 0) {





×








UNCOV

612



                void* px = taosArrayPush(pTableUidList, &uid);




×








UNCOV

613



                TSDB_CHECK_NULL(px, code, lino, _end, terrno);





×







614



              } else {











UNCOV

615



                taosArraySet(pTableUidList, 0, &uid);




×







616



              }










617



            }










618













UNCOV

619



            if (pColVal->rowKey.ts < singleTableLastTs && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {





×








UNCOV

620



              singleTableLastTs = pColVal->rowKey.ts;




×







621



            }










622













UNCOV

623



            if (p->colVal.value.type != pColVal->colVal.value.type) {





×







624



              // check for type/cid mismatch










625



              tsdbError("last cache type mismatch, uid:%" PRIu64





×







626



                        ", schema-type:%d, slotId:%d, cache-type:%d, cache-col:%d",










627



                        uid, p->colVal.value.type, slotIds[k], pColVal->colVal.value.type, pColVal->colVal.cid);










628



              taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);




×







629



              code = TSDB_CODE_INVALID_PARA;




×







630



              goto _end;




×







631



            }










632













UNCOV

633



            if (!IS_VAR_DATA_TYPE(pColVal->colVal.value.type) && pColVal->colVal.value.type != TSDB_DATA_TYPE_DECIMAL) {





×








UNCOV

634



              p->colVal = pColVal->colVal;




×







635



            } else {











UNCOV

636



              if (COL_VAL_IS_VALUE(&pColVal->colVal)) {





×








UNCOV

637



                memcpy(p->colVal.value.pData, pColVal->colVal.value.pData, pColVal->colVal.value.nData);




×







638



              }










639













UNCOV

640



              p->colVal.value.nData = pColVal->colVal.value.nData;




×








UNCOV

641



              p->colVal.value.type = pColVal->colVal.value.type;




×








UNCOV

642



              p->colVal.flag = pColVal->colVal.flag;




×








UNCOV

643



              p->colVal.cid = pColVal->colVal.cid;




×







644



            }










645



          }










646



        }










647













UNCOV

648



        if (hasNotNullRow) {





×








UNCOV

649



          if (INT64_MAX == totalLastTs || (INT64_MAX != singleTableLastTs && totalLastTs < singleTableLastTs)) {





×








UNCOV

650



            totalLastTs = singleTableLastTs;




×







651



          }











UNCOV

652



          double cost = (taosGetTimestampUs() - st) / 1000.0;




×








UNCOV

653



          if (cost > tsCacheLazyLoadThreshold) {




×







654



            // pr->lastTs = totalLastTs;










655



          }










656



        }










657



      }










658













UNCOV

659



      taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);




×







660



    }










661













UNCOV

662



    if (hasRes) {





×








UNCOV

663



      code = saveOneRow(pLastCols, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);




×








UNCOV

664



      TSDB_CHECK_CODE(code, lino, _end);





×







665



    }










666













UNCOV

667



    taosArrayDestroyEx(pLastCols, tsdbCacheFreeSLastColItem);




×








UNCOV

668



  } else if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_ALL)) {





×








UNCOV

669



    int32_t i = pr->tableIndex;




×








UNCOV

670



    for (; i < pr->numOfTables; ++i) {





×








UNCOV

671



      tb_uid_t uid = pTableList[i].uid;




×







672













UNCOV

673



      if ((code = tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype)) != 0) {





×







674



        if (code == -1) {  // fix the invalid return code





×







675



          code = 0;




×







676



        }










677



        TSDB_CHECK_CODE(code, lino, _end);





×







678



      }










679













UNCOV

680



      if (TARRAY_SIZE(pRow) <= 0 || COL_VAL_IS_NONE(&((SLastCol*)TARRAY_DATA(pRow))[0].colVal)) {





×








UNCOV

681



        taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);




×








UNCOV

682



        continue;




×







683



      }










684













UNCOV

685



      code = saveOneRow(pRow, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);




×








UNCOV

686



      TSDB_CHECK_CODE(code, lino, _end);





×







687













UNCOV

688



      taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);




×







689













UNCOV

690



      void* px = taosArrayPush(pTableUidList, &uid);




×








UNCOV

691



      TSDB_CHECK_NULL(px, code, lino, _end, terrno);





×







692













UNCOV

693



      ++pr->tableIndex;




×








UNCOV

694



      if (pResBlock->info.rows >= pResBlock->info.capacity) {





×








UNCOV

695



        break;




×







696



      }










697



    }










698













UNCOV

699



    if (pGotAll && i == pr->numOfTables) {





×








UNCOV

700



      *pGotAll = true;




×







701



    }










702



  } else {










703



    code = TSDB_CODE_INVALID_PARA;




×







704



    TSDB_CHECK_CODE(code, lino, _end);





×







705



  }










706












707



_end:




×








UNCOV

708



  tsdbUntakeReadSnap2((STsdbReader*)pr, pr->pReadSnap, true);




×








UNCOV

709



  pr->pReadSnap = NULL;




×







710













UNCOV

711



  if (pr->pCurFileSet) {





×








UNCOV

712



    pr->pCurFileSet = NULL;




×







713



  }










714













UNCOV

715



  (void)taosThreadMutexUnlock(&pr->readerMutex);




×







716













UNCOV

717



  if (pRes != NULL) {





×








UNCOV

718



    for (int32_t j = 0; j < pr->numOfCols; ++j) {





×








UNCOV

719



      taosMemoryFree(pRes[j]);





×







720



    }










721



  }










722













UNCOV

723



  taosMemoryFree(pRes);





×








UNCOV

724



  taosArrayDestroy(pRow);




×







725













UNCOV

726



  if (code != TSDB_CODE_SUCCESS) {





×







727



    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







728



  }











UNCOV

729



  return code;




×







730



}
























    

  • 

    •