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/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, (const char*)&pVal->value.val, !COL_VAL_IS_VALUE(pVal));




×








UNCOV

81



    TSDB_CHECK_CODE(code, lino, _end);





×







82



  }










83













UNCOV

84



_end:




×








UNCOV

85



  if (code != TSDB_CODE_SUCCESS) {





×







86



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





×







87



  }











UNCOV

88



  return code;




×







89



}










90













UNCOV

91



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




×







92



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











UNCOV

93



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

94



  int32_t lino = 0;




×








UNCOV

95



  int32_t numOfRows = 0;




×








UNCOV

96



  SArray* funcTypeBlockArray = NULL;




×







97













UNCOV

98



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





×








UNCOV

99



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





×








UNCOV

100



  if (pReader->numOfCols > 0) {





×








UNCOV

101



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





×








UNCOV

102



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





×








UNCOV

103



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





×







104



  }










105













UNCOV

106



  numOfRows = pBlock->info.rows;




×







107













UNCOV

108



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





×








UNCOV

109



    uint64_t       ts = TSKEY_MIN;




×








UNCOV

110



    SFirstLastRes* p = NULL;




×








UNCOV

111



    col_id_t       colId = -1;




×







112













UNCOV

113



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




×








UNCOV

114



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





×







115













UNCOV

116



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





×








UNCOV

117



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




×








UNCOV

118



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





×







119













UNCOV

120



      int32_t funcType = FUNCTION_TYPE_CACHE_LAST;




×








UNCOV

121



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





×








UNCOV

122



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




×








UNCOV

123



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





×







124













UNCOV

125



        funcType = *(int32_t*)pVal;




×








UNCOV

126



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




×








UNCOV

127



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





×







128













UNCOV

129



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




×








UNCOV

130



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





×







131



      }










132













UNCOV

133



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





×








UNCOV

134



        if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {





×







135



          colDataSetNULL(pColInfoData, numOfRows);





×







136



          continue;




×







137



        }










138













UNCOV

139



        code = setFirstLastResColToNull(pColInfoData, numOfRows);




×








UNCOV

140



        TSDB_CHECK_CODE(code, lino, _end);





×








UNCOV

141



        continue;




×







142



      }










143













UNCOV

144



      int32_t   slotId = slotIds[i];




×








UNCOV

145



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




×








UNCOV

146



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





×







147













UNCOV

148



      colId = pColVal->colVal.cid;




×








UNCOV

149



      if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {





×








UNCOV

150



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




×








UNCOV

151



        TSDB_CHECK_CODE(code, lino, _end);





×







152













UNCOV

153



        continue;




×







154



      }










155













UNCOV

156



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




×







157













UNCOV

158



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




×








UNCOV

159



      ts = p->ts;




×








UNCOV

160



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




×







161



      // allNullRow = p->isNull & allNullRow;











UNCOV

162



      if (!p->isNull) {





×








UNCOV

163



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





×








UNCOV

164



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




×







165













UNCOV

166



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




×








UNCOV

167



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




×







168



        } else {











UNCOV

169



          memcpy(p->buf, &pColVal->colVal.value.val, pReader->pSchema->columns[slotId].bytes);




×








UNCOV

170



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




×







171



        }










172



      }










173












174



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











UNCOV

175



      p->hasResult = true;




×








UNCOV

176



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




×








UNCOV

177



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




×








UNCOV

178



      TSDB_CHECK_CODE(code, lino, _end);





×







179



    }










180













UNCOV

181



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





×








UNCOV

182



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




×








UNCOV

183



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





×







184













UNCOV

185



      if (idx < funcTypeBlockArray->size) {





×








UNCOV

186



        void* pVal = taosArrayGet(funcTypeBlockArray, idx);




×








UNCOV

187



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





×







188













UNCOV

189



        int32_t funcType = *(int32_t*)pVal;




×








UNCOV

190



        if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {





×








UNCOV

191



          continue;




×







192



        }










193



      }










194













UNCOV

195



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





×








UNCOV

196



        if (ts == TSKEY_MIN) {





×








UNCOV

197



          colDataSetNULL(pCol, numOfRows);





×







198



        } else {











UNCOV

199



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




×








UNCOV

200



          TSDB_CHECK_CODE(code, lino, _end);





×







201



        }











UNCOV

202



        continue;




×








UNCOV

203



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





×








UNCOV

204



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





×








UNCOV

205



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




×








UNCOV

206



          TSDB_CHECK_CODE(code, lino, _end);





×







207



        } else {











UNCOV

208



          colDataSetNULL(pCol, numOfRows);





×







209



        }










210



      }










211



    }










212












213



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











UNCOV

214



    ++pBlock->info.rows;




×








UNCOV

215



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





×








UNCOV

216



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





×








UNCOV

217



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




×








UNCOV

218



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





×







219













UNCOV

220



      int32_t slotId = slotIds[i];




×








UNCOV

221



      if (slotId == -1) {





×








UNCOV

222



        colDataSetNULL(pColInfoData, numOfRows);





×








UNCOV

223



        continue;




×







224



      }










225













UNCOV

226



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




×








UNCOV

227



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





×







228













UNCOV

229



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




×








UNCOV

230



      TSDB_CHECK_CODE(code, lino, _end);





×







231



    }










232












233



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











UNCOV

234



    ++pBlock->info.rows;




×







235



  } else {










236



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





×







237



    code = TSDB_CODE_INVALID_PARA;




×







238



    TSDB_CHECK_CODE(code, lino, _end);





×







239



  }










240












241



_end:




×








UNCOV

242



  if (code != TSDB_CODE_SUCCESS) {





×







243



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





×







244



  }











UNCOV

245



  if (funcTypeBlockArray != NULL) {





×








UNCOV

246



    taosArrayDestroy(funcTypeBlockArray);




×







247



  }











UNCOV

248



  return code;




×







249



}










250













UNCOV

251



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




×








UNCOV

252



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

253



  int32_t lino = 0;




×








UNCOV

254



  int32_t numOfTables = 0;




×







255













UNCOV

256



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





×







257













UNCOV

258



  numOfTables = p->numOfTables;




×







259













UNCOV

260



  if (suid != 0) {





×








UNCOV

261



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




×








UNCOV

262



    if (TSDB_CODE_SUCCESS != code) {





×







263



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





×







264



      if (code == TSDB_CODE_NOT_FOUND) {





×







265



        code = TSDB_CODE_PAR_TABLE_NOT_EXIST;




×







266



      }










267



      TSDB_CHECK_CODE(code, lino, _end);





×







268



    }










269



  } else {











UNCOV

270



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





×








UNCOV

271



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




×








UNCOV

272



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




×








UNCOV

273



      TSDB_CHECK_CODE(code, lino, _end);





×








UNCOV

274



      if (p->pSchema != NULL) {





×








UNCOV

275



        break;




×







276



      }










277












278



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





×







279



    }










280












281



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











UNCOV

282



    if (p->pSchema == NULL) {





×







283



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





×







284



      code = TSDB_CODE_PAR_TABLE_NOT_EXIST;




×







285



      TSDB_CHECK_CODE(code, lino, _end);





×







286



    }










287



  }










288













UNCOV

289



_end:




×








UNCOV

290



  if (code != TSDB_CODE_SUCCESS) {





×







291



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





×







292



  }











UNCOV

293



  return code;




×







294



}










295













UNCOV

296



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




×








UNCOV

297



  int32_t           code = TSDB_CODE_SUCCESS;




×








UNCOV

298



  int32_t           lino = 0;




×








UNCOV

299



  SCacheRowsReader* pReader = (SCacheRowsReader*)reader;




×







300













UNCOV

301



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





×







302













UNCOV

303



  pReader->pTableList = pTableIdList;




×








UNCOV

304



  pReader->numOfTables = numOfTables;




×








UNCOV

305



  pReader->lastTs = INT64_MIN;




×








UNCOV

306



  destroySttBlockReader(pReader->pLDataIterArray, NULL);




×








UNCOV

307



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




×








UNCOV

308



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





×







309













UNCOV

310



_end:




×








UNCOV

311



  if (code != TSDB_CODE_SUCCESS) {





×







312



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





×







313



  }











UNCOV

314



  return code;




×







315



}










316













UNCOV

317



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




×







318



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










319



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











UNCOV

320



  int32_t           code = TSDB_CODE_SUCCESS;




×








UNCOV

321



  int32_t           lino = 0;




×








UNCOV

322



  SCacheRowsReader* p = NULL;




×







323













UNCOV

324



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





×








UNCOV

325



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





×







326













UNCOV

327



  *pReader = NULL;




×








UNCOV

328



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





×








UNCOV

329



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





×







330













UNCOV

331



  p->type = type;




×








UNCOV

332



  p->pVnode = pVnode;




×








UNCOV

333



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




×








UNCOV

334



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




×








UNCOV

335



  p->info.suid = suid;




×








UNCOV

336



  p->numOfCols = numOfCols;




×








UNCOV

337



  p->pCidList = pCidList;




×








UNCOV

338



  p->pSlotIds = pSlotIds;




×








UNCOV

339



  p->pFuncTypeList = pFuncTypeList;




×







340













UNCOV

341



  p->rowKey.numOfPKs = numOfPks;




×








UNCOV

342



  if (numOfPks > 0) {





×








UNCOV

343



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





×








UNCOV

344



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




×








UNCOV

345



    if (IS_VAR_DATA_TYPE(pPkCol->type)) {





×







346



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





×







347



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





×







348



        taosMemoryFreeClear(p);





×







349



        code = terrno;




×







350



        TSDB_CHECK_CODE(code, lino, _end);





×







351



      }










352



    }










353













UNCOV

354



    p->pkColumn = *pPkCol;




×







355



  }










356













UNCOV

357



  if (numOfTables == 0) {





×







358



    *pReader = p;




×







359



    p = NULL;




×







360



    goto _end;




×







361



  }










362













UNCOV

363



  p->pTableList = pTableIdList;




×








UNCOV

364



  p->numOfTables = numOfTables;




×







365













UNCOV

366



  code = setTableSchema(p, suid, idstr);




×








UNCOV

367



  TSDB_CHECK_CODE(code, lino, _end);





×







368













UNCOV

369



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





×








UNCOV

370



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





×







371













UNCOV

372



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





×








UNCOV

373



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





×








UNCOV

374



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





×








UNCOV

375



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





×







376



    }










377



  }










378













UNCOV

379



  if (idstr != NULL) {





×








UNCOV

380



    p->idstr = taosStrdup(idstr);





×








UNCOV

381



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





×







382



  }











UNCOV

383



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




×








UNCOV

384



  TSDB_CHECK_CODE(code, lino, _end);





×







385













UNCOV

386



  p->lastTs = INT64_MIN;




×







387













UNCOV

388



  *pReader = p;




×








UNCOV

389



  p = NULL;




×







390













UNCOV

391



_end:




×








UNCOV

392



  if (code != TSDB_CODE_SUCCESS) {





×







393



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





×







394



    *pReader = NULL;




×







395



  }











UNCOV

396



  if (p != NULL) {





×







397



    tsdbCacherowsReaderClose(p);




×







398



  }











UNCOV

399



  return code;




×







400



}










401













UNCOV

402



void tsdbCacherowsReaderClose(void* pReader) {




×








UNCOV

403



  SCacheRowsReader* p = pReader;




×








UNCOV

404



  if (p == NULL) {





×







405



    return;




×







406



  }










407













UNCOV

408



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





×








UNCOV

409



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





×








UNCOV

410



      taosMemoryFreeClear(p->transferBuf[i]);





×







411



    }










412













UNCOV

413



    taosMemoryFree(p->transferBuf);





×








UNCOV

414



    taosMemoryFree(p->pSchema);





×







415



  }










416













UNCOV

417



  taosMemoryFree(p->pCurrSchema);





×







418













UNCOV

419



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





×








UNCOV

420



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





×








UNCOV

421



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





×







422



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





×







423



      }










424



    }










425



  }










426













UNCOV

427



  if (p->pLDataIterArray) {





×








UNCOV

428



    destroySttBlockReader(p->pLDataIterArray, NULL);




×








UNCOV

429



    p->pLDataIterArray = NULL;




×







430



  }










431













UNCOV

432



  if (p->pFileReader) {





×







433



    tsdbDataFileReaderClose(&p->pFileReader);




×







434



    p->pFileReader = NULL;




×







435



  }










436













UNCOV

437



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





×








UNCOV

438



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




×







439













UNCOV

440



  if (p->pTableMap) {





×








UNCOV

441



    void*   pe = NULL;




×








UNCOV

442



    int32_t iter = 0;




×








UNCOV

443



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





×








UNCOV

444



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




×








UNCOV

445



      taosArrayDestroy(pInfo->pTombData);




×








UNCOV

446



      pInfo->pTombData = NULL;




×







447



    }










448













UNCOV

449



    tSimpleHashCleanup(p->pTableMap);




×







450



  }











UNCOV

451



  if (p->uidList) {





×








UNCOV

452



    taosMemoryFree(p->uidList);





×







453



  }










454













UNCOV

455



  taosMemoryFree(pReader);





×







456



}










457












458



static int32_t tsdbCacheQueryReseek(void* pQHandle) {




×







459



  int32_t           code = 0;




×







460



  SCacheRowsReader* pReader = pQHandle;




×







461












462



  code = taosThreadMutexTryLock(&pReader->readerMutex);




×







463



  if (code == 0) {





×







464



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










465



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










466












467



    code = TSDB_CODE_VND_QUERY_BUSY;




×







468



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




×







469












470



    return code;




×







471



  } else if (code == EBUSY) {





×







472



    return TSDB_CODE_VND_QUERY_BUSY;




×







473



  } else {










474



    return -1;




×







475



  }










476



}










477













UNCOV

478



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




×







479



                              SArray* pTableUidList, bool* pGotAll) {











UNCOV

480



  int32_t           code = TSDB_CODE_SUCCESS;




×








UNCOV

481



  int32_t           lino = 0;




×








UNCOV

482



  bool              hasRes = false;




×








UNCOV

483



  SArray*           pRow = NULL;




×








UNCOV

484



  void**            pRes = NULL;




×








UNCOV

485



  SCacheRowsReader* pr = NULL;




×








UNCOV

486



  int32_t           pkBufLen = 0;




×







487













UNCOV

488



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





×








UNCOV

489



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





×







490













UNCOV

491



  pr = pReader;




×







492













UNCOV

493



  pr->pReadSnap = NULL;




×








UNCOV

494



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




×








UNCOV

495



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





×







496













UNCOV

497



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





×








UNCOV

498



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





×







499













UNCOV

500



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





×








UNCOV

501



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





×








UNCOV

502



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





×







503













UNCOV

504



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





×








UNCOV

505



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





×







506













UNCOV

507



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




×








UNCOV

508



    p->ts = INT64_MIN;




×







509



  }










510













UNCOV

511



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




×








UNCOV

512



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




×








UNCOV

513



  TSDB_CHECK_CODE(code, lino, _end);





×







514













UNCOV

515



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




×







516













UNCOV

517



  STableKeyInfo* pTableList = pr->pTableList;




×







518












519



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











UNCOV

520



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





×








UNCOV

521



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




×








UNCOV

522



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





×







523













UNCOV

524



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





×








UNCOV

525



      int32_t slotId = slotIds[i];




×








UNCOV

526



      if (slotId == -1) {





×








UNCOV

527



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




×








UNCOV

528



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




×








UNCOV

529



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





×








UNCOV

530



        continue;




×







531



      }











UNCOV

532



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




×








UNCOV

533



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




×







534













UNCOV

535



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





×







536



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




×







537



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





×







538



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




×







539



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





×







540



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





×







541



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





×







542



          }










543



        }










544



      }










545













UNCOV

546



      if (IS_VAR_DATA_TYPE(pCol->type)) {





×








UNCOV

547



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





×








UNCOV

548



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





×







549



      }










550













UNCOV

551



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




×








UNCOV

552



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





×







553



    }










554













UNCOV

555



    int64_t st = taosGetTimestampUs();




×








UNCOV

556



    int64_t totalLastTs = INT64_MAX;




×








UNCOV

557



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





×








UNCOV

558



      tb_uid_t uid = pTableList[i].uid;




×







559













UNCOV

560



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




×








UNCOV

561



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





×







562



        code = 0;




×







563



      }











UNCOV

564



      TSDB_CHECK_CODE(code, lino, _end);





×







565













UNCOV

566



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





×








UNCOV

567



        taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);




×








UNCOV

568



        continue;




×







569



      }










570












571



      {











UNCOV

572



        bool    hasNotNullRow = true;




×








UNCOV

573



        int64_t singleTableLastTs = INT64_MAX;




×








UNCOV

574



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





×








UNCOV

575



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





×








UNCOV

576



          SLastCol* p = taosArrayGet(pLastCols, k);




×








UNCOV

577



          if (p == NULL) {





×







578



            return TSDB_CODE_INVALID_PARA;




×







579



          }










580













UNCOV

581



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




×








UNCOV

582



          if (pColVal == NULL) {





×







583



            return TSDB_CODE_INVALID_PARA;




×







584



          }










585













UNCOV

586



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





×








UNCOV

587



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





×








UNCOV

588



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





×








UNCOV

589



                hasNotNullRow = false;




×







590



              }










591



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











UNCOV

592



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





×








UNCOV

593



                continue;




×







594



              }










595



            }










596













UNCOV

597



            hasRes = true;




×








UNCOV

598



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




×








UNCOV

599



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





×







600



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





×







601



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




×







602



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




×







603



              } else {










604



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




×







605



              }










606



            }










607













UNCOV

608



            if (k == 0) {





×








UNCOV

609



              if (TARRAY_SIZE(pTableUidList) == 0) {





×








UNCOV

610



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




×








UNCOV

611



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





×







612



              } else {











UNCOV

613



                taosArraySet(pTableUidList, 0, &uid);




×







614



              }










615



            }










616













UNCOV

617



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





×








UNCOV

618



              singleTableLastTs = pColVal->rowKey.ts;




×







619



            }










620













UNCOV

621



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





×







622



              // check for type/cid mismatch










623



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





×







624



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










625



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










626



              taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);




×







627



              code = TSDB_CODE_INVALID_PARA;




×







628



              goto _end;




×







629



            }










630













UNCOV

631



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





×








UNCOV

632



              p->colVal = pColVal->colVal;




×







633



            } else {











UNCOV

634



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





×








UNCOV

635



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




×







636



              }










637













UNCOV

638



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




×








UNCOV

639



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




×








UNCOV

640



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




×








UNCOV

641



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




×







642



            }










643



          }










644



        }










645













UNCOV

646



        if (hasNotNullRow) {





×








UNCOV

647



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





×








UNCOV

648



            totalLastTs = singleTableLastTs;




×







649



          }











UNCOV

650



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




×








UNCOV

651



          if (cost > tsCacheLazyLoadThreshold) {




×







652



            // pr->lastTs = totalLastTs;










653



          }










654



        }










655



      }










656













UNCOV

657



      taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);




×







658



    }










659













UNCOV

660



    if (hasRes) {





×








UNCOV

661



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




×








UNCOV

662



      TSDB_CHECK_CODE(code, lino, _end);





×







663



    }










664













UNCOV

665



    taosArrayDestroyEx(pLastCols, tsdbCacheFreeSLastColItem);




×








UNCOV

666



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





×








UNCOV

667



    int32_t i = pr->tableIndex;




×








UNCOV

668



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





×








UNCOV

669



      tb_uid_t uid = pTableList[i].uid;




×







670













UNCOV

671



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





×







672



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





×







673



          code = 0;




×







674



        }










675



        TSDB_CHECK_CODE(code, lino, _end);





×







676



      }










677













UNCOV

678



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





×








UNCOV

679



        taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);




×








UNCOV

680



        continue;




×







681



      }










682













UNCOV

683



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




×








UNCOV

684



      TSDB_CHECK_CODE(code, lino, _end);





×







685













UNCOV

686



      taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);




×







687













UNCOV

688



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




×








UNCOV

689



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





×







690













UNCOV

691



      ++pr->tableIndex;




×








UNCOV

692



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





×








UNCOV

693



        break;




×







694



      }










695



    }










696













UNCOV

697



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





×








UNCOV

698



      *pGotAll = true;




×







699



    }










700



  } else {










701



    code = TSDB_CODE_INVALID_PARA;




×







702



    TSDB_CHECK_CODE(code, lino, _end);





×







703



  }










704












705



_end:




×








UNCOV

706



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




×








UNCOV

707



  pr->pReadSnap = NULL;




×







708













UNCOV

709



  if (pr->pCurFileSet) {





×








UNCOV

710



    pr->pCurFileSet = NULL;




×







711



  }










712













UNCOV

713



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




×







714













UNCOV

715



  if (pRes != NULL) {





×








UNCOV

716



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





×








UNCOV

717



      taosMemoryFree(pRes[j]);





×







718



    }










719



  }










720













UNCOV

721



  taosMemoryFree(pRes);





×








UNCOV

722



  taosArrayDestroy(pRow);




×







723













UNCOV

724



  if (code != TSDB_CODE_SUCCESS) {





×







725



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





×







726



  }











UNCOV

727



  return code;




×







728



}
























    

  • 

    •