taosdata / TDengine / #3384

coverage: 20.146% (-45.9%) from 66.017%
#3384

push

travis-ci

web-flow 
Merge pull request #25150 from taosdata/fix/rpc_portMainTo30

Fix/rpc port main to30

24170 of 172239 branches covered (14.03%)

Branch coverage included in aggregate %.

17 of 20 new or added lines in 1 file covered. (85.0%)

109571 existing lines in 395 files now uncovered.

53175 of 211679 relevant lines covered (25.12%)

21661.65 hits per line

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

5.17
/source/libs/executor/src/scanoperator.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 "executorInt.h"










17



#include "filter.h"










18



#include "function.h"










19



#include "functionMgt.h"










20



#include "os.h"










21



#include "querynodes.h"










22



#include "systable.h"










23



#include "tname.h"










24












25



#include "tdatablock.h"










26



#include "tmsg.h"










27












28



#include "query.h"










29



#include "tcompare.h"










30



#include "thash.h"










31



#include "ttypes.h"










32



#include "operator.h"










33



#include "querytask.h"










34












35



#include "storageapi.h"










36



#include "wal.h"










37












38



int32_t scanDebug = 0;










39












40



#define MULTI_READER_MAX_TABLE_NUM        5000










41



#define SET_REVERSE_SCAN_FLAG(_info)      ((_info)->scanFlag = REVERSE_SCAN)










42



#define SWITCH_ORDER(n)                   (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC))










43



#define STREAM_SCAN_OP_NAME               "StreamScanOperator"










44



#define STREAM_SCAN_OP_STATE_NAME         "StreamScanFillHistoryState"










45



#define STREAM_SCAN_OP_CHECKPOINT_NAME    "StreamScanOperator_Checkpoint"










46












47



typedef struct STableMergeScanExecInfo {










48



  SFileBlockLoadRecorder blockRecorder;










49



  SSortExecInfo          sortExecInfo;










50



} STableMergeScanExecInfo;










51












52



typedef struct STableMergeScanSortSourceParam {










53



  SOperatorInfo* pOperator;










54



  int32_t        readerIdx;










55



  uint64_t       uid;










56



  STsdbReader*   reader;










57



} STableMergeScanSortSourceParam;










58












59



typedef struct STableCountScanOperatorInfo {










60



  SReadHandle  readHandle;










61



  SSDataBlock* pRes;










62












63



  STableCountScanSupp supp;










64












65



  int32_t currGrpIdx;










66



  SArray* stbUidList;  // when group by db_name and/or stable_name










67



} STableCountScanOperatorInfo;










68












69



static bool processBlockWithProbability(const SSampleExecInfo* pInfo);










70













UNCOV

71



bool processBlockWithProbability(const SSampleExecInfo* pInfo) {




×







72



#if 0










73



  if (pInfo->sampleRatio == 1) {










74



    return true;










75



  }










76












77



  uint32_t val = taosRandR((uint32_t*) &pInfo->seed);










78



  return (val % ((uint32_t)(1/pInfo->sampleRatio))) == 0;










79



#else











UNCOV

80



  return true;




×







81



#endif










82



}










83













UNCOV

84



static void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) {




×








UNCOV

85



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





×







86



    SWITCH_ORDER(pCtx[i].order);





×







87



  }











UNCOV

88



}




×







89













UNCOV

90



static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockInfo, int32_t order) {




×








UNCOV

91



  STimeWindow w = {0};




×







92












93



  // 0 by default, which means it is not a interval operator of the upstream operator.











UNCOV

94



  if (pInterval->interval == 0) {





×








UNCOV

95



    return false;




×







96



  }










97













UNCOV

98



  if (order == TSDB_ORDER_ASC) {





×








UNCOV

99



    w = getAlignQueryTimeWindow(pInterval, pBlockInfo->window.skey);




×








UNCOV

100



    ASSERT(w.ekey >= pBlockInfo->window.skey);





×







101













UNCOV

102



    if (w.ekey < pBlockInfo->window.ekey) {





×








UNCOV

103



      return true;




×







104



    }










105












106



    while (1) {











UNCOV

107



      getNextTimeWindow(pInterval, &w, order);




×








UNCOV

108



      if (w.skey > pBlockInfo->window.ekey) {





×








UNCOV

109



        break;




×







110



      }










111













UNCOV

112



      ASSERT(w.ekey > pBlockInfo->window.ekey);





×








UNCOV

113



      if (TMAX(w.skey, pBlockInfo->window.skey) <= pBlockInfo->window.ekey) {





×








UNCOV

114



        return true;




×







115



      }










116



    }










117



  } else {











UNCOV

118



    w = getAlignQueryTimeWindow(pInterval, pBlockInfo->window.ekey);




×








UNCOV

119



    ASSERT(w.skey <= pBlockInfo->window.ekey);





×







120













UNCOV

121



    if (w.skey > pBlockInfo->window.skey) {





×








UNCOV

122



      return true;




×







123



    }










124












125



    while (1) {











UNCOV

126



      getNextTimeWindow(pInterval, &w, order);




×








UNCOV

127



      if (w.ekey < pBlockInfo->window.skey) {





×








UNCOV

128



        break;




×







129



      }










130












131



      ASSERT(w.skey < pBlockInfo->window.skey);





×







132



      if (pBlockInfo->window.skey <= TMIN(w.ekey, pBlockInfo->window.ekey)) {





×







133



        return true;




×







134



      }










135



    }










136



  }










137













UNCOV

138



  return false;




×







139



}










140












141



// this function is for table scanner to extract temporary results of upstream aggregate results.











UNCOV

142



static SResultRow* getTableGroupOutputBuf(SOperatorInfo* pOperator, uint64_t groupId, SFilePage** pPage) {




×








UNCOV

143



  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {





×







144



    return NULL;




×







145



  }










146













UNCOV

147



  int64_t buf[2] = {0};




×








UNCOV

148



  SET_RES_WINDOW_KEY((char*)buf, &groupId, sizeof(groupId), groupId);




×







149













UNCOV

150



  STableScanInfo* pTableScanInfo = pOperator->info;




×







151













UNCOV

152



  SResultRowPosition* p1 = (SResultRowPosition*)tSimpleHashGet(pTableScanInfo->base.pdInfo.pAggSup->pResultRowHashTable,




×







153



                                                               buf, GET_RES_WINDOW_KEY_LEN(sizeof(groupId)));










154













UNCOV

155



  if (p1 == NULL) {





×








UNCOV

156



    return NULL;




×







157



  }










158













UNCOV

159



  *pPage = getBufPage(pTableScanInfo->base.pdInfo.pAggSup->pResultBuf, p1->pageId);




×








UNCOV

160



  if (NULL == *pPage) {





×







161



    return NULL;




×







162



  }










163













UNCOV

164



  return (SResultRow*)((char*)(*pPage) + p1->offset);




×







165



}










166













UNCOV

167



static int32_t insertTableToScanIgnoreList(STableScanInfo* pTableScanInfo, uint64_t uid) {




×








UNCOV

168



  if (NULL == pTableScanInfo->pIgnoreTables) {





×








UNCOV

169



    int32_t tableNum = taosArrayGetSize(pTableScanInfo->base.pTableListInfo->pTableList);




×








UNCOV

170



    pTableScanInfo->pIgnoreTables = taosHashInit(tableNum,  taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);




×








UNCOV

171



    if (NULL == pTableScanInfo->pIgnoreTables) {





×







172



      return TSDB_CODE_OUT_OF_MEMORY;




×







173



    }










174



  }










175













UNCOV

176



  taosHashPut(pTableScanInfo->pIgnoreTables, &uid, sizeof(uid), &pTableScanInfo->scanTimes, sizeof(pTableScanInfo->scanTimes));




×







177













UNCOV

178



  return TSDB_CODE_SUCCESS;




×







179



}










180













UNCOV

181



static int32_t doDynamicPruneDataBlock(SOperatorInfo* pOperator, SDataBlockInfo* pBlockInfo, uint32_t* status) {




×








UNCOV

182



  STableScanInfo* pTableScanInfo = pOperator->info;




×








UNCOV

183



  int32_t code = TSDB_CODE_SUCCESS;




×







184













UNCOV

185



  if (pTableScanInfo->base.pdInfo.pExprSup == NULL) {





×








UNCOV

186



    return TSDB_CODE_SUCCESS;




×







187



  }










188













UNCOV

189



  SExprSupp* pSup1 = pTableScanInfo->base.pdInfo.pExprSup;




×







190













UNCOV

191



  SFilePage*  pPage = NULL;




×








UNCOV

192



  SResultRow* pRow = getTableGroupOutputBuf(pOperator, pBlockInfo->id.groupId, &pPage);




×







193













UNCOV

194



  if (pRow == NULL) {





×








UNCOV

195



    return TSDB_CODE_SUCCESS;




×







196



  }










197













UNCOV

198



  bool notLoadBlock = true;




×








UNCOV

199



  for (int32_t i = 0; i < pSup1->numOfExprs; ++i) {





×








UNCOV

200



    int32_t functionId = pSup1->pCtx[i].functionId;




×







201













UNCOV

202



    SResultRowEntryInfo* pEntry = getResultEntryInfo(pRow, i, pTableScanInfo->base.pdInfo.pExprSup->rowEntryInfoOffset);




×







203













UNCOV

204



    int32_t reqStatus = fmFuncDynDataRequired(functionId, pEntry, &pBlockInfo->window);




×








UNCOV

205



    if (reqStatus != FUNC_DATA_REQUIRED_NOT_LOAD) {





×








UNCOV

206



      notLoadBlock = false;




×








UNCOV

207



      break;




×







208



    }










209



  }










210












211



  // release buffer pages











UNCOV

212



  releaseBufPage(pTableScanInfo->base.pdInfo.pAggSup->pResultBuf, pPage);




×







213













UNCOV

214



  if (notLoadBlock) {





×








UNCOV

215



    *status = FUNC_DATA_REQUIRED_NOT_LOAD;




×








UNCOV

216



    code = insertTableToScanIgnoreList(pTableScanInfo, pBlockInfo->id.uid);




×







217



  }










218













UNCOV

219



  return code;




×







220



}










221












222



static bool doFilterByBlockSMA(SFilterInfo* pFilterInfo, SColumnDataAgg** pColsAgg, int32_t numOfCols,




×







223



                               int32_t numOfRows) {










224



  if (pColsAgg == NULL || pFilterInfo == NULL) {





×







225



    return true;




×







226



  }










227












228



  bool keep = filterRangeExecute(pFilterInfo, pColsAgg, numOfCols, numOfRows);




×







229



  return keep;




×







230



}










231













UNCOV

232



static bool doLoadBlockSMA(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) {




×








UNCOV

233



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×







234













UNCOV

235



  bool    allColumnsHaveAgg = true;




×








UNCOV

236



  bool    hasNullSMA = false;




×








UNCOV

237



  int32_t code = pAPI->tsdReader.tsdReaderRetrieveBlockSMAInfo(pTableScanInfo->dataReader, pBlock, &allColumnsHaveAgg, &hasNullSMA);




×








UNCOV

238



  if (code != TSDB_CODE_SUCCESS) {





×







239



    T_LONG_JMP(pTaskInfo->env, code);





×







240



  }










241













UNCOV

242



  if (!allColumnsHaveAgg || hasNullSMA) {





×








UNCOV

243



    return false;




×







244



  }











UNCOV

245



  return true;




×







246



}










247













UNCOV

248



static void doSetTagColumnData(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo,




×







249



                               int32_t rows) {











UNCOV

250



  if (pTableScanInfo->pseudoSup.numOfExprs > 0) {





×








UNCOV

251



    SExprSupp* pSup = &pTableScanInfo->pseudoSup;




×







252













UNCOV

253



    int32_t code = addTagPseudoColumnData(&pTableScanInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pBlock, rows,




×







254



                                          pTaskInfo, &pTableScanInfo->metaCache);










255



    // ignore the table not exists error, since this table may have been dropped during the scan procedure.











UNCOV

256



    if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_PAR_TABLE_NOT_EXIST) {





×







257



      T_LONG_JMP(pTaskInfo->env, code);





×







258



    }










259












260



    // reset the error code.











UNCOV

261



    terrno = 0;




×







262



  }











UNCOV

263



}




×







264












265



bool applyLimitOffset(SLimitInfo* pLimitInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) {




4✔





266



  SLimit*     pLimit = &pLimitInfo->limit;




4✔





267



  const char* id = GET_TASKID(pTaskInfo);




4✔





268












269



  if (pLimitInfo->remainOffset > 0) {





4!






UNCOV

270



    if (pLimitInfo->remainOffset >= pBlock->info.rows) {





×








UNCOV

271



      pLimitInfo->remainOffset -= pBlock->info.rows;




×








UNCOV

272



      blockDataEmpty(pBlock);




×








UNCOV

273



      qDebug("current block ignore due to offset, current:%" PRId64 ", %s", pLimitInfo->remainOffset, id);





×








UNCOV

274



      return false;




×







275



    } else {











UNCOV

276



      blockDataTrimFirstRows(pBlock, pLimitInfo->remainOffset);




×








UNCOV

277



      pLimitInfo->remainOffset = 0;




×







278



    }










279



  }










280












281



  if (pLimit->limit != -1 && pLimit->limit <= (pLimitInfo->numOfOutputRows + pBlock->info.rows)) {





4!





282



    // limit the output rows











UNCOV

283



    int32_t keep = (int32_t)(pLimit->limit - pLimitInfo->numOfOutputRows);




×








UNCOV

284



    blockDataKeepFirstNRows(pBlock, keep);




×







285













UNCOV

286



    pLimitInfo->numOfOutputRows += pBlock->info.rows;




×








UNCOV

287



    qDebug("output limit %" PRId64 " has reached, %s", pLimit->limit, id);





×








UNCOV

288



    return true;




×







289



  }










290












291



  pLimitInfo->numOfOutputRows += pBlock->info.rows;




4✔





292



  return false;




4✔





293



}










294













UNCOV

295



static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableScanInfo, SSDataBlock* pBlock,




×







296



                             uint32_t* status) {











UNCOV

297



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

298



  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;




×







299













UNCOV

300



  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;




×







301













UNCOV

302



  pCost->totalBlocks += 1;




×








UNCOV

303



  pCost->totalRows += pBlock->info.rows;




×







304













UNCOV

305



  bool loadSMA = false;




×








UNCOV

306



  *status = pTableScanInfo->dataBlockLoadFlag;




×








UNCOV

307



  if (pOperator->exprSupp.pFilterInfo != NULL ||





×








UNCOV

308



      overlapWithTimeWindow(&pTableScanInfo->pdInfo.interval, &pBlock->info, pTableScanInfo->cond.order)) {




×








UNCOV

309



    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;




×







310



  }










311













UNCOV

312



  SDataBlockInfo* pBlockInfo = &pBlock->info;




×








UNCOV

313



  taosMemoryFreeClear(pBlock->pBlockAgg);





×







314













UNCOV

315



  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {





×







316



    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64, GET_TASKID(pTaskInfo),





×







317



           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);










318



    pCost->filterOutBlocks += 1;




×







319



    pCost->totalRows += pBlock->info.rows;




×







320



    pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);




×







321



    return TSDB_CODE_SUCCESS;




×








UNCOV

322



  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {





×








UNCOV

323



    qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64 ", uid:%" PRIu64,





×







324



           GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows,










325



           pBlockInfo->id.uid);











UNCOV

326



    doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);




×








UNCOV

327



    pCost->skipBlocks += 1;




×








UNCOV

328



    pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);




×








UNCOV

329



    return TSDB_CODE_SUCCESS;




×








UNCOV

330



  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {





×








UNCOV

331



    pCost->loadBlockStatis += 1;




×








UNCOV

332



    loadSMA = true;  // mark the operation of load sma;




×








UNCOV

333



    bool success = doLoadBlockSMA(pTableScanInfo, pBlock, pTaskInfo);




×








UNCOV

334



    if (success) {  // failed to load the block sma data, data block statistics does not exist, load data block instead





×








UNCOV

335



      qDebug("%s data block SMA loaded, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64, GET_TASKID(pTaskInfo),





×







336



             pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);











UNCOV

337



      doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);




×








UNCOV

338



      pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);




×








UNCOV

339



      return TSDB_CODE_SUCCESS;




×







340



    } else {











UNCOV

341



      qDebug("%s failed to load SMA, since not all columns have SMA", GET_TASKID(pTaskInfo));





×








UNCOV

342



      *status = FUNC_DATA_REQUIRED_DATA_LOAD;




×







343



    }










344



  }










345













UNCOV

346



  ASSERT(*status == FUNC_DATA_REQUIRED_DATA_LOAD);





×







347












348



  // try to filter data block according to sma info











UNCOV

349



  if (pOperator->exprSupp.pFilterInfo != NULL && (!loadSMA)) {





×








UNCOV

350



    bool success = doLoadBlockSMA(pTableScanInfo, pBlock, pTaskInfo);




×








UNCOV

351



    if (success) {





×







352



      size_t size = taosArrayGetSize(pBlock->pDataBlock);




×







353



      bool   keep = doFilterByBlockSMA(pOperator->exprSupp.pFilterInfo, pBlock->pBlockAgg, size, pBlockInfo->rows);




×







354



      if (!keep) {





×







355



        qDebug("%s data block filter out by block SMA, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64,





×







356



               GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);










357



        pCost->filterOutBlocks += 1;




×







358



        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;




×







359












360



        pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);




×







361



        return TSDB_CODE_SUCCESS;




×







362



      }










363



    }










364



  }










365












366



  // free the sma info, since it should not be involved in later computing process.











UNCOV

367



  taosMemoryFreeClear(pBlock->pBlockAgg);





×







368












369



  // try to filter data block according to current results











UNCOV

370



  doDynamicPruneDataBlock(pOperator, pBlockInfo, status);




×








UNCOV

371



  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {





×








UNCOV

372



    qDebug("%s data block skipped due to dynamic prune, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64,





×







373



           GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);











UNCOV

374



    pCost->skipBlocks += 1;




×








UNCOV

375



    pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);




×







376













UNCOV

377



    STableScanInfo* p1 = pOperator->info;




×








UNCOV

378



    if (taosHashGetSize(p1->pIgnoreTables) == taosArrayGetSize(p1->base.pTableListInfo->pTableList)) {





×








UNCOV

379



      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;




×







380



    } else {











UNCOV

381



      *status = FUNC_DATA_REQUIRED_FILTEROUT;




×







382



    }











UNCOV

383



    return TSDB_CODE_SUCCESS;




×







384



  }










385













UNCOV

386



  pCost->totalCheckedRows += pBlock->info.rows;




×








UNCOV

387



  pCost->loadBlocks += 1;




×







388













UNCOV

389



  SSDataBlock* p = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pTableScanInfo->dataReader, NULL);




×








UNCOV

390



  if (p == NULL) {





×







391



    return terrno;




×







392



  }










393













UNCOV

394



  ASSERT(p == pBlock);





×








UNCOV

395



  doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);




×







396












397



  // restore the previous value











UNCOV

398



  pCost->totalRows -= pBlock->info.rows;




×







399













UNCOV

400



  if (pOperator->exprSupp.pFilterInfo != NULL) {





×








UNCOV

401



    int32_t code = doFilter(pBlock, pOperator->exprSupp.pFilterInfo, &pTableScanInfo->matchInfo);




×








UNCOV

402



    if (code != TSDB_CODE_SUCCESS) return code;





×







403













UNCOV

404



    int64_t st = taosGetTimestampUs();




×








UNCOV

405



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




×








UNCOV

406



    pTableScanInfo->readRecorder.filterTime += el;




×







407













UNCOV

408



    if (pBlock->info.rows == 0) {





×








UNCOV

409



      pCost->filterOutBlocks += 1;




×








UNCOV

410



      qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64 ", elapsed time:%.2f ms",





×







411



             GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, el);










412



    } else {











UNCOV

413



      qDebug("%s data block filter applied, elapsed time:%.2f ms", GET_TASKID(pTaskInfo), el);





×







414



    }










415



  }










416













UNCOV

417



  bool limitReached = applyLimitOffset(&pTableScanInfo->limitInfo, pBlock, pTaskInfo);




×








UNCOV

418



  if (limitReached) {  // set operator flag is done





×








UNCOV

419



    setOperatorCompleted(pOperator);




×







420



  }










421













UNCOV

422



  pCost->totalRows += pBlock->info.rows;




×








UNCOV

423



  return TSDB_CODE_SUCCESS;




×







424



}










425













UNCOV

426



static void prepareForDescendingScan(STableScanBase* pTableScanInfo, SqlFunctionCtx* pCtx, int32_t numOfOutput) {




×








UNCOV

427



  SET_REVERSE_SCAN_FLAG(pTableScanInfo);




×







428













UNCOV

429



  switchCtxOrder(pCtx, numOfOutput);




×








UNCOV

430



  pTableScanInfo->cond.order = TSDB_ORDER_DESC;




×








UNCOV

431



  STimeWindow* pTWindow = &pTableScanInfo->cond.twindows;




×








UNCOV

432



  TSWAP(pTWindow->skey, pTWindow->ekey);




×








UNCOV

433



}




×







434












435



typedef struct STableCachedVal {










436



  const char* pName;










437



  STag*       pTags;










438



} STableCachedVal;










439













UNCOV

440



static void freeTableCachedVal(void* param) {




×








UNCOV

441



  if (param == NULL) {





×







442



    return;




×







443



  }










444













UNCOV

445



  STableCachedVal* pVal = param;




×








UNCOV

446



  taosMemoryFree((void*)pVal->pName);




×








UNCOV

447



  taosMemoryFree(pVal->pTags);




×








UNCOV

448



  taosMemoryFree(pVal);




×







449



}










450













UNCOV

451



static STableCachedVal* createTableCacheVal(const SMetaReader* pMetaReader) {




×








UNCOV

452



  STableCachedVal* pVal = taosMemoryMalloc(sizeof(STableCachedVal));




×








UNCOV

453



  pVal->pName = taosStrdup(pMetaReader->me.name);




×








UNCOV

454



  pVal->pTags = NULL;




×







455












456



  // only child table has tag value











UNCOV

457



  if (pMetaReader->me.type == TSDB_CHILD_TABLE) {





×








UNCOV

458



    STag* pTag = (STag*)pMetaReader->me.ctbEntry.pTags;




×








UNCOV

459



    pVal->pTags = taosMemoryMalloc(pTag->len);




×








UNCOV

460



    memcpy(pVal->pTags, pTag, pTag->len);




×







461



  }










462













UNCOV

463



  return pVal;




×







464



}










465












466



// const void *key, size_t keyLen, void *value











UNCOV

467



static void freeCachedMetaItem(const void* key, size_t keyLen, void* value, void* ud) {




×







468



  (void)key;










469



  (void)keyLen;










470



  (void)ud;











UNCOV

471



  freeTableCachedVal(value);




×








UNCOV

472



}




×







473












474



static void doSetNullValue(SSDataBlock* pBlock, const SExprInfo* pExpr, int32_t numOfExpr) {




×







475



  for (int32_t j = 0; j < numOfExpr; ++j) {





×







476



    int32_t dstSlotId = pExpr[j].base.resSchema.slotId;




×







477












478



    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotId);




×







479



    colDataSetNNULL(pColInfoData, 0, pBlock->info.rows);





×







480



  }










481



}




×







482













UNCOV

483



int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock,




×







484



                               int32_t rows, SExecTaskInfo* pTask, STableMetaCacheInfo* pCache) {











UNCOV

485



  int32_t         code = 0;




×








UNCOV

486



  bool            freeReader = false;




×








UNCOV

487



  LRUHandle*      h = NULL;




×








UNCOV

488



  STableCachedVal val = {0};




×








UNCOV

489



  SMetaReader     mr = {0};




×








UNCOV

490



  const char*     idStr = pTask->id.str;




×







491












492



  // currently only the tbname pseudo column











UNCOV

493



  if (numOfExpr <= 0) {





×








UNCOV

494



    return TSDB_CODE_SUCCESS;




×







495



  }










496












497



  // todo: opt if only require the vgId and the vgVer;










498












499



  // backup the rows











UNCOV

500



  int32_t backupRows = pBlock->info.rows;




×








UNCOV

501



  pBlock->info.rows = rows;




×







502












503



  // todo refactor: extract method










504



  // the handling of the null data should be packed in the extracted method










505












506



  // 1. check if it is existed in meta cache











UNCOV

507



  if (pCache == NULL) {





×








UNCOV

508



    pHandle->api.metaReaderFn.initReader(&mr, pHandle->vnode, META_READER_LOCK, &pHandle->api.metaFn);




×








UNCOV

509



    code = pHandle->api.metaReaderFn.getEntryGetUidCache(&mr, pBlock->info.id.uid);




×








UNCOV

510



    if (code != TSDB_CODE_SUCCESS) {





×







511



      // when encounter the TSDB_CODE_PAR_TABLE_NOT_EXIST error, we proceed.










512



      if (terrno == TSDB_CODE_PAR_TABLE_NOT_EXIST) {





×







513



        qWarn("failed to get table meta, table may have been dropped, uid:0x%" PRIx64 ", code:%s, %s",





×







514



              pBlock->info.id.uid, tstrerror(terrno), idStr);










515












516



        // append null value before return to caller, since the caller will ignore this error code and proceed










517



        doSetNullValue(pBlock, pExpr, numOfExpr);




×







518



      } else {










519



        qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid, tstrerror(terrno),





×







520



               idStr);










521



      }










522



      pHandle->api.metaReaderFn.clearReader(&mr);




×







523



      return terrno;




×







524



    }










525













UNCOV

526



    pHandle->api.metaReaderFn.readerReleaseLock(&mr);




×







527













UNCOV

528



    val.pName = mr.me.name;




×








UNCOV

529



    val.pTags = (STag*)mr.me.ctbEntry.pTags;




×







530













UNCOV

531



    freeReader = true;




×







532



  } else {











UNCOV

533



    pCache->metaFetch += 1;




×







534













UNCOV

535



    h = taosLRUCacheLookup(pCache->pTableMetaEntryCache, &pBlock->info.id.uid, sizeof(pBlock->info.id.uid));




×








UNCOV

536



    if (h == NULL) {





×








UNCOV

537



      pHandle->api.metaReaderFn.initReader(&mr, pHandle->vnode, META_READER_LOCK, &pHandle->api.metaFn);




×








UNCOV

538



      code = pHandle->api.metaReaderFn.getEntryGetUidCache(&mr, pBlock->info.id.uid);




×








UNCOV

539



      if (code != TSDB_CODE_SUCCESS) {





×







540



        if (terrno == TSDB_CODE_PAR_TABLE_NOT_EXIST) {





×







541



          qWarn("failed to get table meta, table may have been dropped, uid:0x%" PRIx64 ", code:%s, %s",





×







542



                pBlock->info.id.uid, tstrerror(terrno), idStr);










543



          // append null value before return to caller, since the caller will ignore this error code and proceed










544



          doSetNullValue(pBlock, pExpr, numOfExpr);




×







545



        } else {










546



          qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid, tstrerror(terrno),





×







547



                 idStr);










548



        }










549



        pHandle->api.metaReaderFn.clearReader(&mr);




×







550



        return terrno;




×







551



      }










552













UNCOV

553



      pHandle->api.metaReaderFn.readerReleaseLock(&mr);




×







554













UNCOV

555



      STableCachedVal* pVal = createTableCacheVal(&mr);




×







556













UNCOV

557



      val = *pVal;




×








UNCOV

558



      freeReader = true;




×







559













UNCOV

560



      int32_t ret = taosLRUCacheInsert(pCache->pTableMetaEntryCache, &pBlock->info.id.uid, sizeof(uint64_t), pVal,




×







561



                                       sizeof(STableCachedVal), freeCachedMetaItem, NULL, TAOS_LRU_PRIORITY_LOW, NULL);











UNCOV

562



      if (ret != TAOS_LRU_STATUS_OK) {





×







563



        qError("failed to put meta into lru cache, code:%d, %s", ret, idStr);





×







564



        freeTableCachedVal(pVal);




×







565



      }










566



    } else {











UNCOV

567



      pCache->cacheHit += 1;




×








UNCOV

568



      STableCachedVal* pVal = taosLRUCacheValue(pCache->pTableMetaEntryCache, h);




×








UNCOV

569



      val = *pVal;




×







570













UNCOV

571



      taosLRUCacheRelease(pCache->pTableMetaEntryCache, h, false);




×







572



    }










573













UNCOV

574



    qDebug("retrieve table meta from cache:%" PRIu64 ", hit:%" PRIu64 " miss:%" PRIu64 ", %s", pCache->metaFetch,





×







575



           pCache->cacheHit, (pCache->metaFetch - pCache->cacheHit), idStr);










576



  }










577













UNCOV

578



  for (int32_t j = 0; j < numOfExpr; ++j) {





×








UNCOV

579



    const SExprInfo* pExpr1 = &pExpr[j];




×








UNCOV

580



    int32_t          dstSlotId = pExpr1->base.resSchema.slotId;




×







581













UNCOV

582



    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotId);




×








UNCOV

583



    colInfoDataCleanup(pColInfoData, pBlock->info.rows);




×







584













UNCOV

585



    int32_t functionId = pExpr1->pExpr->_function.functionId;




×







586












587



    // this is to handle the tbname











UNCOV

588



    if (fmIsScanPseudoColumnFunc(functionId)) {





×








UNCOV

589



      int32_t fType = pExpr1->pExpr->_function.functionType;




×








UNCOV

590



      if (fType == FUNCTION_TYPE_TBNAME) {





×








UNCOV

591



        setTbNameColData(pBlock, pColInfoData, functionId, val.pName);




×








UNCOV

592



      } else if (fType == FUNCTION_TYPE_VGID) {





×








UNCOV

593



        setVgIdColData(pBlock, pColInfoData, functionId, pTask->id.vgId);




×








UNCOV

594



      } else if (fType == FUNCTION_TYPE_VGVER) {





×








UNCOV

595



        setVgVerColData(pBlock, pColInfoData, functionId, pBlock->info.version);




×







596



      }










597



    } else {  // these are tags











UNCOV

598



      STagVal tagVal = {0};




×








UNCOV

599



      tagVal.cid = pExpr1->base.pParam[0].pCol->colId;




×








UNCOV

600



      const char* p = pHandle->api.metaFn.extractTagVal(val.pTags, pColInfoData->info.type, &tagVal);




×







601













UNCOV

602



      char* data = NULL;




×








UNCOV

603



      if (pColInfoData->info.type != TSDB_DATA_TYPE_JSON && p != NULL) {





×








UNCOV

604



        data = tTagValToData((const STagVal*)p, false);




×







605



      } else {











UNCOV

606



        data = (char*)p;




×







607



      }










608













UNCOV

609



      bool isNullVal = (data == NULL) || (pColInfoData->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data));





×








UNCOV

610



      if (isNullVal) {





×








UNCOV

611



        colDataSetNNULL(pColInfoData, 0, pBlock->info.rows);





×








UNCOV

612



      } else if (pColInfoData->info.type != TSDB_DATA_TYPE_JSON) {





×








UNCOV

613



        code = colDataSetNItems(pColInfoData, 0, data, pBlock->info.rows, false);




×








UNCOV

614



        if (IS_VAR_DATA_TYPE(((const STagVal*)p)->type)) {





×








UNCOV

615



          taosMemoryFree(data);




×







616



        }











UNCOV

617



        if (code) {





×







618



          if (freeReader) {





×







619



            pHandle->api.metaReaderFn.clearReader(&mr);




×







620



          }










621



          return code;




×







622



        }










623



      } else {  // todo opt for json tag











UNCOV

624



        for (int32_t i = 0; i < pBlock->info.rows; ++i) {





×








UNCOV

625



          colDataSetVal(pColInfoData, i, data, false);




×







626



        }










627



      }










628



    }










629



  }










630












631



  // restore the rows











UNCOV

632



  pBlock->info.rows = backupRows;




×








UNCOV

633



  if (freeReader) {





×








UNCOV

634



    pHandle->api.metaReaderFn.clearReader(&mr);




×







635



  }










636













UNCOV

637



  return TSDB_CODE_SUCCESS;




×







638



}










639













UNCOV

640



void setTbNameColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, const char* name) {




×








UNCOV

641



  struct SScalarFuncExecFuncs fpSet = {0};




×








UNCOV

642



  fmGetScalarFuncExecFuncs(functionId, &fpSet);




×







643













UNCOV

644



  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;




×








UNCOV

645



  char   buf[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};




×








UNCOV

646



  STR_TO_VARSTR(buf, name)




×







647













UNCOV

648



  SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, len, 1);




×







649













UNCOV

650



  colInfoDataEnsureCapacity(&infoData, 1, false);




×








UNCOV

651



  colDataSetVal(&infoData, 0, buf, false);




×







652













UNCOV

653



  SScalarParam srcParam = {.numOfRows = pBlock->info.rows, .columnData = &infoData};




×








UNCOV

654



  SScalarParam param = {.columnData = pColInfoData};




×







655













UNCOV

656



  if (fpSet.process != NULL) {





×








UNCOV

657



    fpSet.process(&srcParam, 1, &param);




×







658



  } else {










659



    qError("failed to get the corresponding callback function, functionId:%d", functionId);





×







660



  }










661













UNCOV

662



  colDataDestroy(&infoData);




×








UNCOV

663



}




×







664













UNCOV

665



void setVgIdColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, int32_t vgId) {




×








UNCOV

666



  struct SScalarFuncExecFuncs fpSet = {0};




×








UNCOV

667



  fmGetScalarFuncExecFuncs(functionId, &fpSet);




×







668













UNCOV

669



  SColumnInfoData infoData = createColumnInfoData(pColInfoData->info.type, pColInfoData->info.bytes, 1);




×







670













UNCOV

671



  colInfoDataEnsureCapacity(&infoData, 1, false);




×








UNCOV

672



  colDataSetVal(&infoData, 0, (const char*)&vgId, false);




×







673













UNCOV

674



  SScalarParam srcParam = {.numOfRows = pBlock->info.rows, .columnData = &infoData};




×








UNCOV

675



  SScalarParam param = {.columnData = pColInfoData};




×







676













UNCOV

677



  if (fpSet.process != NULL) {





×








UNCOV

678



    fpSet.process(&srcParam, 1, &param);




×







679



  } else {










680



    qError("failed to get the corresponding callback function, functionId:%d", functionId);





×







681



  }










682













UNCOV

683



  colDataDestroy(&infoData);




×








UNCOV

684



}




×







685













UNCOV

686



void setVgVerColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, int64_t vgVer) {




×








UNCOV

687



  struct SScalarFuncExecFuncs fpSet = {0};




×








UNCOV

688



  fmGetScalarFuncExecFuncs(functionId, &fpSet);




×







689













UNCOV

690



  SColumnInfoData infoData = createColumnInfoData(pColInfoData->info.type, pColInfoData->info.bytes, 1);




×







691













UNCOV

692



  colInfoDataEnsureCapacity(&infoData, 1, false);




×








UNCOV

693



  colDataSetVal(&infoData, 0, (const char*)&vgVer, false);




×







694













UNCOV

695



  SScalarParam srcParam = {.numOfRows = pBlock->info.rows, .columnData = &infoData};




×








UNCOV

696



  SScalarParam param = {.columnData = pColInfoData};




×







697













UNCOV

698



  if (fpSet.process != NULL) {





×








UNCOV

699



    fpSet.process(&srcParam, 1, &param);




×







700



  } else {










701



    qError("failed to get the corresponding callback function, functionId:%d", functionId);





×







702



  }










703













UNCOV

704



  colDataDestroy(&infoData);




×








UNCOV

705



}




×







706













UNCOV

707



static void initNextGroupScan(STableScanInfo* pInfo, STableKeyInfo** pKeyInfo, int32_t* size) {




×








UNCOV

708



  tableListGetGroupList(pInfo->base.pTableListInfo, pInfo->currentGroupId, pKeyInfo, size);




×







709













UNCOV

710



  pInfo->tableStartIndex = TARRAY_ELEM_IDX(pInfo->base.pTableListInfo->pTableList, *pKeyInfo);




×







711













UNCOV

712



  pInfo->tableEndIndex = (pInfo->tableStartIndex + (*size) - 1);




×







713













UNCOV

714



  pInfo->pResBlock->info.blankFill = false;




×







715













UNCOV

716



  if (!pInfo->needCountEmptyTable) {





×








UNCOV

717



    pInfo->countState = TABLE_COUNT_STATE_END;




×







718



  } else {











UNCOV

719



    pInfo->countState = TABLE_COUNT_STATE_SCAN;




×







720



  }











UNCOV

721



}




×







722













UNCOV

723



void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {




×








UNCOV

724



  if (pInfo->countState ==  TABLE_COUNT_STATE_END) {





×








UNCOV

725



    return;




×







726



  }











UNCOV

727



  if (pInfo->base.pTableListInfo->oneTableForEachGroup || pInfo->base.pTableListInfo->groupOffset) {





×








UNCOV

728



    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;




×







729



  } else {











UNCOV

730



    taosHashRemove(pInfo->base.pTableListInfo->remainGroups, &groupId, sizeof(groupId));




×







731



  }










732



}










733













UNCOV

734



static SSDataBlock* getOneRowResultBlock(SExecTaskInfo* pTaskInfo, STableScanBase* pBase, SSDataBlock* pBlock,




×







735



                                         const STableKeyInfo* tbInfo) {











UNCOV

736



  blockDataEmpty(pBlock);




×








UNCOV

737



  pBlock->info.rows = 1;




×








UNCOV

738



  pBlock->info.id.uid = tbInfo->uid;




×








UNCOV

739



  pBlock->info.id.groupId = tbInfo->groupId;




×








UNCOV

740



  pBlock->info.blankFill = true;




×







741












742



  // only one row: set all col data to null & hasNull











UNCOV

743



  int32_t col_num = blockDataGetNumOfCols(pBlock);




×








UNCOV

744



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





×








UNCOV

745



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




×







746



    colDataSetNULL(pColInfoData, 0);










747



  }










748












749



  // set tag/tbname











UNCOV

750



  doSetTagColumnData(pBase, pBlock, pTaskInfo, 1);




×








UNCOV

751



  return pBlock;




×







752



}










753













UNCOV

754



static SSDataBlock* getBlockForEmptyTable(SOperatorInfo* pOperator, const STableKeyInfo* tbInfo) {




×








UNCOV

755



  STableScanInfo* pTableScanInfo = pOperator->info;




×







756



  SSDataBlock*    pBlock =











UNCOV

757



      getOneRowResultBlock(pOperator->pTaskInfo, &pTableScanInfo->base, pTableScanInfo->pResBlock, tbInfo);




×







758













UNCOV

759



  pOperator->resultInfo.totalRows++;




×








UNCOV

760



  return pBlock;




×







761



}










762













UNCOV

763



static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {




×








UNCOV

764



  STableScanInfo* pTableScanInfo = pOperator->info;




×








UNCOV

765



  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

766



  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;




×







767













UNCOV

768



  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;




×








UNCOV

769



  bool            hasNext = false;




×








UNCOV

770



  int32_t         code = TSDB_CODE_SUCCESS;




×







771













UNCOV

772



  int64_t st = taosGetTimestampUs();




×







773













UNCOV

774



  while (true) {




×








UNCOV

775



    code = pAPI->tsdReader.tsdNextDataBlock(pTableScanInfo->base.dataReader, &hasNext);




×








UNCOV

776



    if (code) {





×







777



      pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->base.dataReader);




×







778



      T_LONG_JMP(pTaskInfo->env, code);





×







779



    }










780













UNCOV

781



    if (!hasNext) {





×








UNCOV

782



      break;




×







783



    }










784













UNCOV

785



    if (isTaskKilled(pTaskInfo)) {





×








UNCOV

786



      pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->base.dataReader);




×








UNCOV

787



      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);





×







788



    }










789













UNCOV

790



    if (pOperator->status == OP_EXEC_DONE) {





×







791



      pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->base.dataReader);




×







792



      break;




×







793



    }










794












795



    // process this data block based on the probabilities











UNCOV

796



    bool processThisBlock = processBlockWithProbability(&pTableScanInfo->sample);




×








UNCOV

797



    if (!processThisBlock) {





×








UNCOV

798



      continue;




×







799



    }










800













UNCOV

801



    if (pBlock->info.id.uid) {





×








UNCOV

802



      pBlock->info.id.groupId = tableListGetTableGroupId(pTableScanInfo->base.pTableListInfo, pBlock->info.id.uid);




×







803



    }










804













UNCOV

805



    uint32_t status = 0;




×








UNCOV

806



    code = loadDataBlock(pOperator, &pTableScanInfo->base, pBlock, &status);




×








UNCOV

807



    if (code != TSDB_CODE_SUCCESS) {





×







808



      T_LONG_JMP(pTaskInfo->env, code);





×







809



    }










810













UNCOV

811



    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {





×








UNCOV

812



      break;




×







813



    }










814












815



    // current block is filter out according to filter condition, continue load the next block











UNCOV

816



    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {





×








UNCOV

817



      continue;




×







818



    }










819













UNCOV

820



    pOperator->resultInfo.totalRows = pTableScanInfo->base.readRecorder.totalRows;




×








UNCOV

821



    pTableScanInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;




×







822













UNCOV

823



    pOperator->cost.totalCost = pTableScanInfo->base.readRecorder.elapsedTime;




×








UNCOV

824



    pBlock->info.scanFlag = pTableScanInfo->base.scanFlag;




×








UNCOV

825



    return pBlock;




×







826



  }











UNCOV

827



  return NULL;




×







828



}










829













UNCOV

830



static SSDataBlock* doGroupedTableScan(SOperatorInfo* pOperator) {




×








UNCOV

831



  STableScanInfo* pTableScanInfo = pOperator->info;




×








UNCOV

832



  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

833



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×







834












835



  // The read handle is not initialized yet, since no qualified tables exists











UNCOV

836



  if (pTableScanInfo->base.dataReader == NULL || pOperator->status == OP_EXEC_DONE) {





×








UNCOV

837



    return NULL;




×







838



  }










839












840



  // do the ascending order traverse in the first place.











UNCOV

841



  while (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {





×








UNCOV

842



    SSDataBlock* p = doTableScanImpl(pOperator);




×








UNCOV

843



    if (p != NULL) {





×








UNCOV

844



      markGroupProcessed(pTableScanInfo, p->info.id.groupId);




×








UNCOV

845



      return p;




×







846



    }










847













UNCOV

848



    pTableScanInfo->scanTimes += 1;




×








UNCOV

849



    taosHashClear(pTableScanInfo->pIgnoreTables);




×







850













UNCOV

851



    if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {





×








UNCOV

852



      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);




×








UNCOV

853



      pTableScanInfo->base.scanFlag = MAIN_SCAN;




×








UNCOV

854



      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;




×








UNCOV

855



      qDebug("start to repeat ascending order scan data blocks due to query func required, %s", GET_TASKID(pTaskInfo));





×







856












857



      // do prepare for the next round table scan operation











UNCOV

858



      pAPI->tsdReader.tsdReaderResetStatus(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond);




×







859



    }










860



  }










861













UNCOV

862



  int32_t total = pTableScanInfo->scanInfo.numOfAsc + pTableScanInfo->scanInfo.numOfDesc;




×








UNCOV

863



  if (pTableScanInfo->scanTimes < total) {





×








UNCOV

864



    if (pTableScanInfo->base.cond.order == TSDB_ORDER_ASC) {





×








UNCOV

865



      prepareForDescendingScan(&pTableScanInfo->base, pOperator->exprSupp.pCtx, 0);




×








UNCOV

866



      pAPI->tsdReader.tsdReaderResetStatus(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond);




×








UNCOV

867



      qDebug("%s start to descending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));





×







868



    }










869













UNCOV

870



    while (pTableScanInfo->scanTimes < total) {





×








UNCOV

871



      SSDataBlock* p = doTableScanImpl(pOperator);




×








UNCOV

872



      if (p != NULL) {





×








UNCOV

873



        markGroupProcessed(pTableScanInfo, p->info.id.groupId);




×








UNCOV

874



        return p;




×







875



      }










876













UNCOV

877



      pTableScanInfo->scanTimes += 1;




×








UNCOV

878



      taosHashClear(pTableScanInfo->pIgnoreTables);




×







879













UNCOV

880



      if (pTableScanInfo->scanTimes < total) {





×







881



        setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);




×







882



        pTableScanInfo->base.scanFlag = MAIN_SCAN;




×







883












884



        qDebug("%s start to repeat descending order scan data blocks", GET_TASKID(pTaskInfo));





×







885



        pAPI->tsdReader.tsdReaderResetStatus(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond);




×







886



      }










887



    }










888



  }










889













UNCOV

890



  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {





×








UNCOV

891



    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;




×








UNCOV

892



    if (pTableListInfo->oneTableForEachGroup || pTableListInfo->groupOffset) { // group by tbname, group by tag + sort





×








UNCOV

893



      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {





×








UNCOV

894



        pTableScanInfo->countState = TABLE_COUNT_STATE_PROCESSED;




×







895



        STableKeyInfo* pStart =











UNCOV

896



            (STableKeyInfo*)tableListGetInfo(pTableScanInfo->base.pTableListInfo, pTableScanInfo->tableStartIndex);




×








UNCOV

897



        return getBlockForEmptyTable(pOperator, pStart);




×







898



      }










899



    } else { // group by tag + no sort











UNCOV

900



      int32_t numOfTables = tableListGetSize(pTableListInfo);




×








UNCOV

901



      if (pTableScanInfo->tableEndIndex + 1 >= numOfTables) {





×







902



        // get empty group, mark processed & rm from hash











UNCOV

903



        void* pIte = taosHashIterate(pTableListInfo->remainGroups, NULL);




×








UNCOV

904



        if (pIte != NULL) {





×








UNCOV

905



          size_t        keySize = 0;




×








UNCOV

906



          uint64_t*     pGroupId = taosHashGetKey(pIte, &keySize);




×








UNCOV

907



          STableKeyInfo info = {.uid = *(uint64_t*)pIte, .groupId = *pGroupId};




×








UNCOV

908



          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);




×








UNCOV

909



          markGroupProcessed(pTableScanInfo, *pGroupId);




×








UNCOV

910



          return getBlockForEmptyTable(pOperator, &info);




×







911



        }










912



      }










913



    }











UNCOV

914



    pTableScanInfo->countState = TABLE_COUNT_STATE_END;




×







915



  }










916













UNCOV

917



  return NULL;




×







918



}










919













UNCOV

920



static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {




×








UNCOV

921



  STableScanInfo* pInfo = pOperator->info;




×








UNCOV

922



  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;  




×








UNCOV

923



  int32_t code = 0;




×








UNCOV

924



  STableListInfo* pListInfo = pInfo->base.pTableListInfo;




×








UNCOV

925



  STableScanOperatorParam* pParam = (STableScanOperatorParam*)pOperator->pOperatorGetParam->value;




×








UNCOV

926



  int32_t num = taosArrayGetSize(pParam->pUidList);




×








UNCOV

927



  if (num <= 0) {





×







928



    qError("empty table scan uid list");





×







929



    return TSDB_CODE_INVALID_PARA;




×







930



  }










931



  











UNCOV

932



  qError("vgId:%d add total %d dynamic tables to scan, tableSeq:%d, exist num:%" PRId64 ", operator status:%d", 





×







933



      pTaskInfo->id.vgId, num, pParam->tableSeq, (int64_t)taosArrayGetSize(pListInfo->pTableList), pOperator->status);










934













UNCOV

935



  if (pParam->tableSeq) {





×







936



    pListInfo->oneTableForEachGroup = true;




×







937



    if (taosArrayGetSize(pListInfo->pTableList) > 0) {





×







938



      taosHashClear(pListInfo->map);




×







939



      taosArrayClear(pListInfo->pTableList);




×







940



      pOperator->status = OP_EXEC_DONE;




×







941



    }










942



  } else {











UNCOV

943



    pListInfo->oneTableForEachGroup = false;




×








UNCOV

944



    pListInfo->numOfOuputGroups = 1;




×







945



  }










946













UNCOV

947



  STableKeyInfo info = {.groupId = 0};  




×








UNCOV

948



  int32_t tableIdx = 0;




×








UNCOV

949



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





×








UNCOV

950



    uint64_t* pUid = taosArrayGet(pParam->pUidList, i);




×







951













UNCOV

952



    if (taosHashPut(pListInfo->map, pUid, sizeof(uint64_t), &tableIdx, sizeof(int32_t))) {





×








UNCOV

953



      if (TSDB_CODE_DUP_KEY == terrno) {





×








UNCOV

954



        continue;




×







955



      }










956



      return TSDB_CODE_OUT_OF_MEMORY;




×







957



    }










958













UNCOV

959



    info.uid = *pUid;




×








UNCOV

960



    void* p = taosArrayPush(pListInfo->pTableList, &info);




×








UNCOV

961



    if (p == NULL) {





×







962



      return TSDB_CODE_OUT_OF_MEMORY;




×







963



    }










964













UNCOV

965



    tableIdx++;




×








UNCOV

966



    qError("add dynamic table scan uid:%" PRIu64 ", %s", info.uid, GET_TASKID(pTaskInfo));





×







967



  }










968



  











UNCOV

969



  return code;




×







970



}










971













UNCOV

972



static SSDataBlock* startNextGroupScan(SOperatorInfo* pOperator) {




×








UNCOV

973



  STableScanInfo* pInfo = pOperator->info;




×








UNCOV

974



  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

975



  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;




×








UNCOV

976



  int32_t numOfTables = tableListGetSize(pInfo->base.pTableListInfo);




×








UNCOV

977



  if ((++pInfo->currentGroupId) >= tableListGetOutputGroups(pInfo->base.pTableListInfo)) {





×








UNCOV

978



    setOperatorCompleted(pOperator);




×








UNCOV

979



    if (pOperator->dynamicTask) {





×








UNCOV

980



      taosArrayClear(pInfo->base.pTableListInfo->pTableList);




×








UNCOV

981



      taosHashClear(pInfo->base.pTableListInfo->map);




×







982



    }











UNCOV

983



    return NULL;




×







984



  }










985












986



  // reset value for the next group data output











UNCOV

987



  pOperator->status = OP_OPENED;




×








UNCOV

988



  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);




×







989



  











UNCOV

990



  int32_t        num = 0;




×








UNCOV

991



  STableKeyInfo* pList = NULL;




×








UNCOV

992



  initNextGroupScan(pInfo, &pList, &num);




×







993













UNCOV

994



  pAPI->tsdReader.tsdSetQueryTableList(pInfo->base.dataReader, pList, num);




×








UNCOV

995



  pAPI->tsdReader.tsdReaderResetStatus(pInfo->base.dataReader, &pInfo->base.cond);




×








UNCOV

996



  pInfo->scanTimes = 0;




×







997



  











UNCOV

998



  SSDataBlock* result = doGroupedTableScan(pOperator);




×








UNCOV

999



  if (result != NULL) {





×








UNCOV

1000



    if (pOperator->dynamicTask) {





×







1001



      result->info.id.groupId = result->info.id.uid;




×







1002



    }











UNCOV

1003



    return result;




×







1004



  }










1005



  











UNCOV

1006



  return NULL;




×







1007



}










1008













UNCOV

1009



static SSDataBlock* groupSeqTableScan(SOperatorInfo* pOperator) {




×








UNCOV

1010



  STableScanInfo* pInfo = pOperator->info;




×








UNCOV

1011



  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

1012



  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;




×








UNCOV

1013



  int32_t        num = 0;




×








UNCOV

1014



  STableKeyInfo* pList = NULL;




×







1015













UNCOV

1016



  if (pInfo->currentGroupId == -1) {





×








UNCOV

1017



    if ((++pInfo->currentGroupId) >= tableListGetOutputGroups(pInfo->base.pTableListInfo)) {





×







1018



      setOperatorCompleted(pOperator);




×







1019



      return NULL;




×







1020



    }










1021













UNCOV

1022



    initNextGroupScan(pInfo, &pList, &num);




×








UNCOV

1023



    ASSERT(pInfo->base.dataReader == NULL);





×







1024



  











UNCOV

1025



    int32_t code = pAPI->tsdReader.tsdReaderOpen(pInfo->base.readHandle.vnode, &pInfo->base.cond, pList, num, pInfo->pResBlock,




×








UNCOV

1026



                                  (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->pIgnoreTables);




×








UNCOV

1027



    if (code != TSDB_CODE_SUCCESS) {





×







1028



      T_LONG_JMP(pTaskInfo->env, code);





×







1029



    }











UNCOV

1030



    if (pInfo->filesetDelimited) {





×








UNCOV

1031



      pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);




×







1032



    }











UNCOV

1033



    if (pInfo->pResBlock->info.capacity > pOperator->resultInfo.capacity) {





×







1034



      pOperator->resultInfo.capacity = pInfo->pResBlock->info.capacity;




×







1035



    }










1036



  }










1037













UNCOV

1038



  SSDataBlock* result = doGroupedTableScan(pOperator);




×








UNCOV

1039



  if (result != NULL) {





×








UNCOV

1040



    if (pOperator->dynamicTask) {





×








UNCOV

1041



      result->info.id.groupId = result->info.id.uid;




×







1042



    }











UNCOV

1043



    return result;




×







1044



  }










1045












1046



  while (true) {











UNCOV

1047



    result = startNextGroupScan(pOperator);




×








UNCOV

1048



    if (result || pOperator->status == OP_EXEC_DONE) {





×








UNCOV

1049



      return result;




×







1050



    }










1051



  }










1052












1053



  return result;










1054



}










1055













UNCOV

1056



static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {




×








UNCOV

1057



  STableScanInfo* pInfo = pOperator->info;




×








UNCOV

1058



  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

1059



  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;




×







1060













UNCOV

1061



  if (pOperator->pOperatorGetParam) {





×








UNCOV

1062



    pOperator->dynamicTask = true;




×








UNCOV

1063



    int32_t code = createTableListInfoFromParam(pOperator);




×








UNCOV

1064



    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);




×








UNCOV

1065



    pOperator->pOperatorGetParam = NULL;




×








UNCOV

1066



    if (code != TSDB_CODE_SUCCESS) {





×







1067



      pTaskInfo->code = code;




×







1068



      T_LONG_JMP(pTaskInfo->env, code);





×







1069



    }











UNCOV

1070



    if (pOperator->status == OP_EXEC_DONE) {





×







1071



      pInfo->currentGroupId = -1;




×







1072



      pOperator->status = OP_OPENED;




×







1073



      SSDataBlock* result = NULL;




×







1074



      while (true) {










1075



        result = startNextGroupScan(pOperator);




×







1076



        if (result || pOperator->status == OP_EXEC_DONE) {





×







1077



          return result;




×







1078



        }










1079



      }










1080



    }










1081



  }










1082












1083



  // scan table one by one sequentially











UNCOV

1084



  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {





×








UNCOV

1085



    int32_t       numOfTables = 0;  // tableListGetSize(pTaskInfo->pTableListInfo);




×








UNCOV

1086



    STableKeyInfo tInfo = {0};




×








UNCOV

1087



    pInfo->countState = TABLE_COUNT_STATE_END;




×







1088













UNCOV

1089



    while (1) {




×








UNCOV

1090



      SSDataBlock* result = doGroupedTableScan(pOperator);




×








UNCOV

1091



      if (result || (pOperator->status == OP_EXEC_DONE) || isTaskKilled(pTaskInfo)) {





×








UNCOV

1092



        return result;




×







1093



      }










1094












1095



      // if no data, switch to next table and continue scan











UNCOV

1096



      pInfo->currentTable++;




×







1097













UNCOV

1098



      taosRLockLatch(&pTaskInfo->lock);




×








UNCOV

1099



      numOfTables = tableListGetSize(pInfo->base.pTableListInfo);




×







1100













UNCOV

1101



      if (pInfo->currentTable >= numOfTables) {





×








UNCOV

1102



        qDebug("all table checked in table list, total:%d, return NULL, %s", numOfTables, GET_TASKID(pTaskInfo));





×








UNCOV

1103



        taosRUnLockLatch(&pTaskInfo->lock);




×








UNCOV

1104



        return NULL;




×







1105



      }










1106













UNCOV

1107



      tInfo = *(STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->currentTable);




×








UNCOV

1108



      taosRUnLockLatch(&pTaskInfo->lock);




×







1109













UNCOV

1110



      pAPI->tsdReader.tsdSetQueryTableList(pInfo->base.dataReader, &tInfo, 1);




×








UNCOV

1111



      qDebug("set uid:%" PRIu64 " into scanner, total tables:%d, index:%d/%d %s", tInfo.uid, numOfTables,





×







1112



             pInfo->currentTable, numOfTables, GET_TASKID(pTaskInfo));










1113













UNCOV

1114



      pAPI->tsdReader.tsdReaderResetStatus(pInfo->base.dataReader, &pInfo->base.cond);




×








UNCOV

1115



      pInfo->scanTimes = 0;




×







1116



    }










1117



  } else {  // scan table group by group sequentially











UNCOV

1118



    return groupSeqTableScan(pOperator);




×







1119



  }










1120



}










1121













UNCOV

1122



static int32_t getTableScannerExecInfo(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {




×








UNCOV

1123



  SFileBlockLoadRecorder* pRecorder = taosMemoryCalloc(1, sizeof(SFileBlockLoadRecorder));




×








UNCOV

1124



  STableScanInfo*         pTableScanInfo = pOptr->info;




×








UNCOV

1125



  *pRecorder = pTableScanInfo->base.readRecorder;




×








UNCOV

1126



  *pOptrExplain = pRecorder;




×








UNCOV

1127



  *len = sizeof(SFileBlockLoadRecorder);




×








UNCOV

1128



  return 0;




×







1129



}










1130












1131



static void destroyTableScanBase(STableScanBase* pBase, TsdReader* pAPI) {




12✔





1132



  cleanupQueryTableDataCond(&pBase->cond);




12✔





1133












1134



  pAPI->tsdReaderClose(pBase->dataReader);




12✔





1135



  pBase->dataReader = NULL;




12✔





1136












1137



  if (pBase->matchInfo.pList != NULL) {





12!





1138



    taosArrayDestroy(pBase->matchInfo.pList);




12✔





1139



  }










1140












1141



  tableListDestroy(pBase->pTableListInfo);




12✔





1142



  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);




12✔





1143



  cleanupExprSupp(&pBase->pseudoSup);




12✔





1144



}




12✔





1145












1146



static void destroyTableScanOperatorInfo(void* param) {




12✔





1147



  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;




12✔





1148



  blockDataDestroy(pTableScanInfo->pResBlock);




12✔





1149



  taosHashCleanup(pTableScanInfo->pIgnoreTables);




12✔





1150



  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);




12✔





1151



  taosMemoryFreeClear(param);





12!





1152



}




12✔





1153












1154



SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,




12✔





1155



                                           STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo) {










1156



  int32_t         code = 0;




12✔





1157



  STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));




12✔





1158



  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));




12✔





1159



  if (pInfo == NULL || pOperator == NULL) {





12!





1160



    code = TSDB_CODE_OUT_OF_MEMORY;




×







1161



    goto _error;




×







1162



  }










1163












1164



  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;




12✔





1165



  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;




12✔





1166












1167



  int32_t numOfCols = 0;




12✔





1168



  code =










1169



      extractColMatchInfo(pScanNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->base.matchInfo);




12✔





1170



  if (code != TSDB_CODE_SUCCESS) {





12!





1171



    goto _error;




×







1172



  }










1173












1174



  initLimitInfo(pScanNode->node.pLimit, pScanNode->node.pSlimit, &pInfo->base.limitInfo);




12✔





1175



  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle);




12✔





1176



  if (code != TSDB_CODE_SUCCESS) {





12!





1177



    goto _error;




×







1178



  }










1179












1180



  if (pScanNode->pScanPseudoCols != NULL) {





12!






UNCOV

1181



    SExprSupp* pSup = &pInfo->base.pseudoSup;




×








UNCOV

1182



    pSup->pExprInfo = createExprInfo(pScanNode->pScanPseudoCols, NULL, &pSup->numOfExprs);




×








UNCOV

1183



    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset, &pTaskInfo->storageAPI.functionStore);




×







1184



  }










1185












1186



  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};




12✔





1187



  pInfo->base.scanFlag = (pInfo->scanInfo.numOfAsc > 1) ? PRE_SCAN : MAIN_SCAN;





12!





1188












1189



  pInfo->base.pdInfo.interval = extractIntervalInfo(pTableScanNode);




12✔





1190



  pInfo->base.readHandle = *readHandle;




12✔





1191



  pInfo->base.dataBlockLoadFlag = pTableScanNode->dataRequired;




12✔





1192












1193



  pInfo->sample.sampleRatio = pTableScanNode->ratio;




12✔





1194



  pInfo->sample.seed = taosGetTimestampSec();




12✔





1195












1196



  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;




12✔





1197



  initResultSizeInfo(&pOperator->resultInfo, 4096);




12✔





1198



  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);




12✔





1199












1200



  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);




12✔





1201



  if (code != TSDB_CODE_SUCCESS) {





12!





1202



    goto _error;




×







1203



  }










1204



  










1205



  pInfo->currentGroupId = -1;




12✔





1206












1207



  pInfo->tableEndIndex = -1;




12✔





1208



  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;




12✔





1209



  pInfo->hasGroupByTag = pTableScanNode->pGroupTags ? true : false;




12✔





1210












1211



  setOperatorInfo(pOperator, "TableScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, false, OP_NOT_OPENED, pInfo,




12✔





1212



                  pTaskInfo);










1213



  pOperator->exprSupp.numOfExprs = numOfCols;




12✔





1214












1215



  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;





12!





1216












1217



  pInfo->base.pTableListInfo = pTableListInfo;




12✔





1218



  pInfo->base.metaCache.pTableMetaEntryCache = taosLRUCacheInit(1024 * 128, -1, .5);




12✔





1219



  if (pInfo->base.metaCache.pTableMetaEntryCache == NULL) {





12!





1220



    code = terrno;




×







1221



    goto _error;




×







1222



  }










1223












1224



  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;




12✔





1225












1226



  taosLRUCacheSetStrictCapacity(pInfo->base.metaCache.pTableMetaEntryCache, false);




12✔





1227



  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableScan, NULL, destroyTableScanOperatorInfo,




12✔





1228



                                         optrDefaultBufFn, getTableScannerExecInfo, optrDefaultGetNextExtFn, NULL);










1229












1230



  // for non-blocking operator, the open cost is always 0










1231



  pOperator->cost.openCost = 0;




12✔





1232



  return pOperator;




12✔





1233












1234



_error:




×







1235



  if (pInfo != NULL) {





×







1236



    destroyTableScanOperatorInfo(pInfo);




×







1237



  }










1238












1239



  taosMemoryFreeClear(pOperator);





×







1240



  pTaskInfo->code = code;




×







1241



  return NULL;




×







1242



}










1243












1244



SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* pTaskInfo) {




×







1245



  STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));




×







1246



  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));




×







1247












1248



  pInfo->base.dataReader = pReadHandle;




×







1249



  //  pInfo->prevGroupId       = -1;










1250












1251



  setOperatorInfo(pOperator, "TableSeqScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN, false, OP_NOT_OPENED,




×







1252



                  pInfo, pTaskInfo);










1253



  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableScanImpl, NULL, NULL, optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);




×







1254



  return pOperator;




×







1255



}










1256












1257



FORCE_INLINE void doClearBufferedBlocks(SStreamScanInfo* pInfo) {




×








UNCOV

1258



  qDebug("clear buff blocks:%d", (int32_t)taosArrayGetSize(pInfo->pBlockLists));





×








UNCOV

1259



  taosArrayClear(pInfo->pBlockLists);




×








UNCOV

1260



  pInfo->validBlockIndex = 0;




×








UNCOV

1261



}




×







1262













UNCOV

1263



static bool isSessionWindow(SStreamScanInfo* pInfo) {




×








UNCOV

1264



  return pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;




×







1265



}










1266













UNCOV

1267



static bool isStateWindow(SStreamScanInfo* pInfo) {




×








UNCOV

1268



  return pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE ||





×








UNCOV

1269



         pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT;





×







1270



}










1271













UNCOV

1272



static bool isIntervalWindow(SStreamScanInfo* pInfo) {




×








UNCOV

1273



  return pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL ||




×








UNCOV

1274



         pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL ||





×








UNCOV

1275



         pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL ||





×








UNCOV

1276



         pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL;





×







1277



}










1278













UNCOV

1279



static bool isSignleIntervalWindow(SStreamScanInfo* pInfo) {




×








UNCOV

1280



  return pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL;




×







1281



}










1282













UNCOV

1283



static bool isSlidingWindow(SStreamScanInfo* pInfo) {




×








UNCOV

1284



  return isIntervalWindow(pInfo) && pInfo->interval.interval != pInfo->interval.sliding;





×







1285



}










1286













UNCOV

1287



static bool isCountSlidingWindow(SStreamScanInfo* pInfo) {




×








UNCOV

1288



  return pInfo->windowSup.pStreamAggSup && (pInfo->windowSup.pStreamAggSup->windowCount != pInfo->windowSup.pStreamAggSup->windowSliding);





×







1289



}










1290













UNCOV

1291



static bool isCountWindow(SStreamScanInfo* pInfo) {




×








UNCOV

1292



  return pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT;




×







1293



}










1294













UNCOV

1295



static void setGroupId(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t groupColIndex, int32_t rowIndex) {




×








UNCOV

1296



  SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, groupColIndex);




×








UNCOV

1297



  uint64_t*        groupCol = (uint64_t*)pColInfo->pData;




×








UNCOV

1298



  ASSERT(rowIndex < pBlock->info.rows);





×








UNCOV

1299



  pInfo->groupId = groupCol[rowIndex];




×








UNCOV

1300



}




×







1301













UNCOV

1302



void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin, uint64_t ver) {




×








UNCOV

1303



  pTableScanInfo->base.cond.twindows = *pWin;




×








UNCOV

1304



  pTableScanInfo->base.cond.startVersion = 0;




×








UNCOV

1305



  pTableScanInfo->base.cond.endVersion = ver;




×








UNCOV

1306



  pTableScanInfo->scanTimes = 0;




×








UNCOV

1307



  pTableScanInfo->currentGroupId = -1;




×








UNCOV

1308



  pTableScanInfo->tableEndIndex = -1;




×








UNCOV

1309



  pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);




×








UNCOV

1310



  pTableScanInfo->base.dataReader = NULL;




×








UNCOV

1311



}




×







1312













UNCOV

1313



static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbUid, TSKEY startTs, TSKEY endTs,




×







1314



                                       int64_t maxVersion) {











UNCOV

1315



  STableKeyInfo tblInfo = {.uid = tbUid, .groupId = 0};




×







1316













UNCOV

1317



  STableScanInfo*     pTableScanInfo = pTableScanOp->info;




×








UNCOV

1318



  SQueryTableDataCond cond = pTableScanInfo->base.cond;




×







1319













UNCOV

1320



  cond.startVersion = -1;




×








UNCOV

1321



  cond.endVersion = maxVersion;




×








UNCOV

1322



  cond.twindows = (STimeWindow){.skey = startTs, .ekey = endTs};




×







1323













UNCOV

1324



  SExecTaskInfo* pTaskInfo = pTableScanOp->pTaskInfo;




×








UNCOV

1325



  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;




×







1326













UNCOV

1327



  SSDataBlock* pBlock = pTableScanInfo->pResBlock;




×








UNCOV

1328



  STsdbReader* pReader = NULL;




×








UNCOV

1329



  int32_t      code = pAPI->tsdReader.tsdReaderOpen(pTableScanInfo->base.readHandle.vnode, &cond, &tblInfo, 1, pBlock,




×








UNCOV

1330



                                     (void**)&pReader, GET_TASKID(pTaskInfo), NULL);




×








UNCOV

1331



  if (code != TSDB_CODE_SUCCESS) {





×







1332



    terrno = code;




×







1333



    T_LONG_JMP(pTaskInfo->env, code);





×







1334



    return NULL;










1335



  }










1336













UNCOV

1337



  bool hasNext = false;




×








UNCOV

1338



  code = pAPI->tsdReader.tsdNextDataBlock(pReader, &hasNext);




×








UNCOV

1339



  if (code != TSDB_CODE_SUCCESS) {





×







1340



    terrno = code;




×







1341



    T_LONG_JMP(pTaskInfo->env, code);





×







1342



    return NULL;










1343



  }










1344













UNCOV

1345



  if (hasNext) {





×








UNCOV

1346



    /*SSDataBlock* p = */ pAPI->tsdReader.tsdReaderRetrieveDataBlock(pReader, NULL);




×








UNCOV

1347



    doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, pBlock->info.rows);




×








UNCOV

1348



    pBlock->info.id.groupId = tableListGetTableGroupId(pTableScanInfo->base.pTableListInfo, pBlock->info.id.uid);




×







1349



  }










1350













UNCOV

1351



  pAPI->tsdReader.tsdReaderClose(pReader);




×








UNCOV

1352



  qDebug("retrieve prev rows:%" PRId64 ", skey:%" PRId64 ", ekey:%" PRId64 " uid:%" PRIu64 ", max ver:%" PRId64





×







1353



         ", suid:%" PRIu64,










1354



         pBlock->info.rows, startTs, endTs, tbUid, maxVersion, cond.suid);










1355













UNCOV

1356



  return pBlock->info.rows > 0 ? pBlock : NULL;





×







1357



}










1358













UNCOV

1359



static uint64_t getGroupIdByCol(SStreamScanInfo* pInfo, uint64_t uid, TSKEY ts, int64_t maxVersion) {




×








UNCOV

1360



  SSDataBlock* pPreRes = readPreVersionData(pInfo->pTableScanOp, uid, ts, ts, maxVersion);




×








UNCOV

1361



  if (!pPreRes || pPreRes->info.rows == 0) {





×








UNCOV

1362



    return 0;




×







1363



  }











UNCOV

1364



  ASSERT(pPreRes->info.rows == 1);





×








UNCOV

1365



  return calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pPreRes, 0);




×







1366



}










1367













UNCOV

1368



static uint64_t getGroupIdByUid(SStreamScanInfo* pInfo, uint64_t uid) {




×








UNCOV

1369



  STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;




×








UNCOV

1370



  return tableListGetTableGroupId(pTableScanInfo->base.pTableListInfo, uid);




×







1371



}










1372













UNCOV

1373



static uint64_t getGroupIdByData(SStreamScanInfo* pInfo, uint64_t uid, TSKEY ts, int64_t maxVersion) {




×








UNCOV

1374



  if (pInfo->partitionSup.needCalc) {





×








UNCOV

1375



    return getGroupIdByCol(pInfo, uid, ts, maxVersion);




×







1376



  }










1377













UNCOV

1378



  return getGroupIdByUid(pInfo, uid);




×







1379



}










1380













UNCOV

1381



static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t* pRowIndex) {




×








UNCOV

1382



  if (pBlock->info.rows == 0) {





×








UNCOV

1383



    return false;




×







1384



  }











UNCOV

1385



  if ((*pRowIndex) == pBlock->info.rows) {





×








UNCOV

1386



    return false;




×







1387



  }










1388













UNCOV

1389



  ASSERT(taosArrayGetSize(pBlock->pDataBlock) >= 3);





×








UNCOV

1390



  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

1391



  TSKEY*           startData = (TSKEY*)pStartTsCol->pData;




×








UNCOV

1392



  SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

1393



  TSKEY*           endData = (TSKEY*)pEndTsCol->pData;




×








UNCOV

1394



  STimeWindow      win = {.skey = startData[*pRowIndex], .ekey = endData[*pRowIndex]};




×








UNCOV

1395



  SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);




×








UNCOV

1396



  uint64_t*        gpData = (uint64_t*)pGpCol->pData;




×








UNCOV

1397



  uint64_t         groupId = gpData[*pRowIndex];




×







1398













UNCOV

1399



  SColumnInfoData* pCalStartTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);




×








UNCOV

1400



  TSKEY*           calStartData = (TSKEY*)pCalStartTsCol->pData;




×








UNCOV

1401



  SColumnInfoData* pCalEndTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);




×








UNCOV

1402



  TSKEY*           calEndData = (TSKEY*)pCalEndTsCol->pData;




×







1403













UNCOV

1404



  setGroupId(pInfo, pBlock, GROUPID_COLUMN_INDEX, *pRowIndex);




×








UNCOV

1405



  if (isSlidingWindow(pInfo) || isCountSlidingWindow(pInfo)) {





×








UNCOV

1406



    pInfo->updateWin.skey = calStartData[*pRowIndex];




×








UNCOV

1407



    pInfo->updateWin.ekey = calEndData[*pRowIndex];




×







1408



  }











UNCOV

1409



  (*pRowIndex)++;




×







1410













UNCOV

1411



  for (; *pRowIndex < pBlock->info.rows; (*pRowIndex)++) {





×








UNCOV

1412



    if (win.skey == startData[*pRowIndex] && groupId == gpData[*pRowIndex]) {





×








UNCOV

1413



      win.ekey = TMAX(win.ekey, endData[*pRowIndex]);




×








UNCOV

1414



      continue;




×







1415



    }










1416













UNCOV

1417



    if (win.skey == endData[*pRowIndex] && groupId == gpData[*pRowIndex]) {





×







1418



      win.skey = TMIN(win.skey, startData[*pRowIndex]);




×







1419



      continue;




×







1420



    }










1421













UNCOV

1422



    ASSERT(!(win.skey > startData[*pRowIndex] && win.ekey < endData[*pRowIndex]) ||





×







1423



           !(isInTimeWindow(&win, startData[*pRowIndex], 0) || isInTimeWindow(&win, endData[*pRowIndex], 0)));











UNCOV

1424



    break;




×







1425



  }










1426













UNCOV

1427



  STableScanInfo* pTScanInfo = pInfo->pTableScanOp->info;




×








UNCOV

1428



  qDebug("prepare range scan start:%" PRId64 ",end:%" PRId64 ",maxVer:%" PRIu64, win.skey, win.ekey, pInfo->pUpdateInfo->maxDataVersion);





×








UNCOV

1429



  resetTableScanInfo(pInfo->pTableScanOp->info, &win, pInfo->pUpdateInfo->maxDataVersion);




×








UNCOV

1430



  pInfo->pTableScanOp->status = OP_OPENED;




×








UNCOV

1431



  return true;




×







1432



}










1433













UNCOV

1434



static STimeWindow getSlidingWindow(TSKEY* startTsCol, TSKEY* endTsCol, uint64_t* gpIdCol, SInterval* pInterval,




×







1435



                                    SDataBlockInfo* pDataBlockInfo, int32_t* pRowIndex, bool hasGroup) {











UNCOV

1436



  SResultRowInfo dumyInfo = {0};




×








UNCOV

1437



  dumyInfo.cur.pageId = -1;




×








UNCOV

1438



  STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, startTsCol[*pRowIndex], pInterval, TSDB_ORDER_ASC);




×








UNCOV

1439



  STimeWindow endWin = win;




×








UNCOV

1440



  STimeWindow preWin = win;




×








UNCOV

1441



  uint64_t    groupId = gpIdCol[*pRowIndex];




×







1442












1443



  while (1) {











UNCOV

1444



    if (hasGroup) {





×








UNCOV

1445



      (*pRowIndex) += 1;




×







1446



    } else {











UNCOV

1447



      while ((groupId == gpIdCol[(*pRowIndex)] && startTsCol[*pRowIndex] <= endWin.ekey)) {





×








UNCOV

1448



        (*pRowIndex) += 1;




×








UNCOV

1449



        if ((*pRowIndex) == pDataBlockInfo->rows) {





×








UNCOV

1450



          break;




×







1451



        }










1452



      }










1453



    }










1454












1455



    do {











UNCOV

1456



      preWin = endWin;




×








UNCOV

1457



      getNextTimeWindow(pInterval, &endWin, TSDB_ORDER_ASC);




×








UNCOV

1458



    } while (endTsCol[(*pRowIndex) - 1] >= endWin.skey);





×








UNCOV

1459



    endWin = preWin;




×








UNCOV

1460



    if (win.ekey == endWin.ekey || (*pRowIndex) == pDataBlockInfo->rows || groupId != gpIdCol[*pRowIndex]) {





×








UNCOV

1461



      win.ekey = endWin.ekey;




×








UNCOV

1462



      return win;




×







1463



    }










1464



    win.ekey = endWin.ekey;




×







1465



  }










1466



}










1467













UNCOV

1468



static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {




×








UNCOV

1469



  qDebug("do stream range scan. windows index:%d", *pRowIndex);





×








UNCOV

1470



  bool prepareRes = true;




×







1471













UNCOV

1472



  while (1) {




×








UNCOV

1473



    SSDataBlock* pResult = NULL;




×








UNCOV

1474



    pResult = doTableScan(pInfo->pTableScanOp);




×








UNCOV

1475



    if (!pResult) {





×








UNCOV

1476



      prepareRes = prepareRangeScan(pInfo, pSDB, pRowIndex);




×







1477



      // scan next window data











UNCOV

1478



      pResult = doTableScan(pInfo->pTableScanOp);




×







1479



    }











UNCOV

1480



    if (!pResult) {





×








UNCOV

1481



      if (prepareRes) {





×








UNCOV

1482



        continue;




×







1483



      }











UNCOV

1484



      blockDataCleanup(pSDB);




×








UNCOV

1485



      *pRowIndex = 0;




×








UNCOV

1486



      pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};




×








UNCOV

1487



      STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;




×








UNCOV

1488



      pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);




×








UNCOV

1489



      pTableScanInfo->base.dataReader = NULL;




×








UNCOV

1490



      return NULL;




×







1491



    }










1492













UNCOV

1493



    doFilter(pResult, pInfo->pTableScanOp->exprSupp.pFilterInfo, NULL);




×








UNCOV

1494



    if (pResult->info.rows == 0) {





×







1495



      continue;




×







1496



    }










1497













UNCOV

1498



    if (pInfo->partitionSup.needCalc) {





×








UNCOV

1499



      SSDataBlock* tmpBlock = createOneDataBlock(pResult, true);




×








UNCOV

1500



      blockDataCleanup(pResult);




×








UNCOV

1501



      for (int32_t i = 0; i < tmpBlock->info.rows; i++) {





×








UNCOV

1502



        if (calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, tmpBlock, i) == pInfo->groupId) {





×








UNCOV

1503



          for (int32_t j = 0; j < pInfo->pTableScanOp->exprSupp.numOfExprs; j++) {





×








UNCOV

1504



            SColumnInfoData* pSrcCol = taosArrayGet(tmpBlock->pDataBlock, j);




×








UNCOV

1505



            SColumnInfoData* pDestCol = taosArrayGet(pResult->pDataBlock, j);




×








UNCOV

1506



            bool             isNull = colDataIsNull(pSrcCol, tmpBlock->info.rows, i, NULL);





×








UNCOV

1507



            char*            pSrcData = NULL;




×








UNCOV

1508



            if (!isNull) pSrcData = colDataGetData(pSrcCol, i);





×








UNCOV

1509



            colDataSetVal(pDestCol, pResult->info.rows, pSrcData, isNull);




×







1510



          }











UNCOV

1511



          pResult->info.rows++;




×







1512



        }










1513



      }










1514













UNCOV

1515



      blockDataDestroy(tmpBlock);




×







1516













UNCOV

1517



      if (pResult->info.rows > 0) {





×








UNCOV

1518



        pResult->info.calWin = pInfo->updateWin;




×








UNCOV

1519



        return pResult;




×







1520



      }











UNCOV

1521



    } else if (pResult->info.id.groupId == pInfo->groupId) {





×








UNCOV

1522



      pResult->info.calWin = pInfo->updateWin;




×








UNCOV

1523



      return pResult;




×







1524



    }










1525



  }










1526



}










1527













UNCOV

1528



static int32_t getPreSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,




×







1529



                                   SSessionKey* pKey) {











UNCOV

1530



  pKey->win.skey = startTs;




×








UNCOV

1531



  pKey->win.ekey = endTs;




×








UNCOV

1532



  pKey->groupId = groupId;




×







1533













UNCOV

1534



  void* pCur = pAggSup->stateStore.streamStateSessionSeekKeyCurrentPrev(pAggSup->pState, pKey);




×








UNCOV

1535



  int32_t          code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, pKey, NULL, 0);




×








UNCOV

1536



  if (code != TSDB_CODE_SUCCESS) {





×







1537



    SET_SESSION_WIN_KEY_INVALID(pKey);




×







1538



  }










1539













UNCOV

1540



  taosMemoryFree(pCur);




×








UNCOV

1541



  return code;




×







1542



}










1543













UNCOV

1544



static int32_t generateSessionScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) {




×








UNCOV

1545



  blockDataCleanup(pDestBlock);




×








UNCOV

1546



  if (pSrcBlock->info.rows == 0) {





×







1547



    return TSDB_CODE_SUCCESS;




×







1548



  }











UNCOV

1549



  int32_t code = blockDataEnsureCapacity(pDestBlock, pSrcBlock->info.rows);




×








UNCOV

1550



  if (code != TSDB_CODE_SUCCESS) {





×







1551



    return code;




×







1552



  }











UNCOV

1553



  ASSERT(taosArrayGetSize(pSrcBlock->pDataBlock) >= 3);





×








UNCOV

1554



  SColumnInfoData* pStartTsCol = taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

1555



  TSKEY*           startData = (TSKEY*)pStartTsCol->pData;




×








UNCOV

1556



  SColumnInfoData* pEndTsCol = taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

1557



  TSKEY*           endData = (TSKEY*)pEndTsCol->pData;




×








UNCOV

1558



  SColumnInfoData* pUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX);




×








UNCOV

1559



  uint64_t*        uidCol = (uint64_t*)pUidCol->pData;




×







1560













UNCOV

1561



  SColumnInfoData* pDestStartCol = taosArrayGet(pDestBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

1562



  SColumnInfoData* pDestEndCol = taosArrayGet(pDestBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

1563



  SColumnInfoData* pDestUidCol = taosArrayGet(pDestBlock->pDataBlock, UID_COLUMN_INDEX);




×








UNCOV

1564



  SColumnInfoData* pDestGpCol = taosArrayGet(pDestBlock->pDataBlock, GROUPID_COLUMN_INDEX);




×








UNCOV

1565



  SColumnInfoData* pDestCalStartTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);




×








UNCOV

1566



  SColumnInfoData* pDestCalEndTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);




×








UNCOV

1567



  int64_t          ver = pSrcBlock->info.version - 1;




×








UNCOV

1568



  for (int32_t i = 0; i < pSrcBlock->info.rows; i++) {





×








UNCOV

1569



    uint64_t groupId = getGroupIdByData(pInfo, uidCol[i], startData[i], ver);




×







1570



    // gap must be 0.











UNCOV

1571



    SSessionKey startWin = {0};




×








UNCOV

1572



    getCurSessionWindow(pInfo->windowSup.pStreamAggSup, startData[i], startData[i], groupId, &startWin);




×








UNCOV

1573



    if (IS_INVALID_SESSION_WIN_KEY(startWin)) {





×







1574



      // window has been closed.











UNCOV

1575



      continue;




×







1576



    }











UNCOV

1577



    SSessionKey endWin = {0};




×








UNCOV

1578



    getCurSessionWindow(pInfo->windowSup.pStreamAggSup, endData[i], endData[i], groupId, &endWin);




×








UNCOV

1579



    if (IS_INVALID_SESSION_WIN_KEY(endWin)) {





×








UNCOV

1580



      getPreSessionWindow(pInfo->windowSup.pStreamAggSup, endData[i], endData[i], groupId, &endWin);




×







1581



    }











UNCOV

1582



    if (IS_INVALID_SESSION_WIN_KEY(startWin)) {





×







1583



      // window has been closed.










1584



      qError("generate session scan range failed. rang start:%" PRIx64 ", end:%" PRIx64, startData[i], endData[i]);





×







1585



      continue;




×







1586



    }











UNCOV

1587



    colDataSetVal(pDestStartCol, i, (const char*)&startWin.win.skey, false);




×








UNCOV

1588



    colDataSetVal(pDestEndCol, i, (const char*)&endWin.win.ekey, false);




×







1589













UNCOV

1590



    colDataSetNULL(pDestUidCol, i);





×








UNCOV

1591



    colDataSetVal(pDestGpCol, i, (const char*)&groupId, false);




×








UNCOV

1592



    colDataSetNULL(pDestCalStartTsCol, i);





×








UNCOV

1593



    colDataSetNULL(pDestCalEndTsCol, i);





×








UNCOV

1594



    pDestBlock->info.rows++;




×







1595



  }











UNCOV

1596



  return TSDB_CODE_SUCCESS;




×







1597



}










1598













UNCOV

1599



static int32_t generateCountScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock, EStreamType mode) {




×








UNCOV

1600



  blockDataCleanup(pDestBlock);




×








UNCOV

1601



  if (pSrcBlock->info.rows == 0) {





×







1602



    return TSDB_CODE_SUCCESS;




×







1603



  }











UNCOV

1604



  int32_t code = blockDataEnsureCapacity(pDestBlock, pSrcBlock->info.rows);




×








UNCOV

1605



  if (code != TSDB_CODE_SUCCESS) {





×







1606



    return code;




×







1607



  }











UNCOV

1608



  ASSERT(taosArrayGetSize(pSrcBlock->pDataBlock) >= 3);





×








UNCOV

1609



  SColumnInfoData* pStartTsCol = taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

1610



  TSKEY*           startData = (TSKEY*)pStartTsCol->pData;




×








UNCOV

1611



  SColumnInfoData* pEndTsCol = taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

1612



  TSKEY*           endData = (TSKEY*)pEndTsCol->pData;




×








UNCOV

1613



  SColumnInfoData* pUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX);




×








UNCOV

1614



  uint64_t*        uidCol = (uint64_t*)pUidCol->pData;




×







1615













UNCOV

1616



  SColumnInfoData* pDestStartCol = taosArrayGet(pDestBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

1617



  SColumnInfoData* pDestEndCol = taosArrayGet(pDestBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

1618



  SColumnInfoData* pDestUidCol = taosArrayGet(pDestBlock->pDataBlock, UID_COLUMN_INDEX);




×








UNCOV

1619



  SColumnInfoData* pDestGpCol = taosArrayGet(pDestBlock->pDataBlock, GROUPID_COLUMN_INDEX);




×








UNCOV

1620



  SColumnInfoData* pDestCalStartTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);




×








UNCOV

1621



  SColumnInfoData* pDestCalEndTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);




×








UNCOV

1622



  int64_t          ver = pSrcBlock->info.version - 1;




×








UNCOV

1623



  for (int32_t i = 0; i < pSrcBlock->info.rows; i++) {





×








UNCOV

1624



    uint64_t groupId = getGroupIdByData(pInfo, uidCol[i], startData[i], ver);




×








UNCOV

1625



    SSessionKey startWin = {.win.skey = startData[i], .win.ekey = endData[i], .groupId = groupId};




×








UNCOV

1626



    SSessionKey range = {0};




×








UNCOV

1627



    getCountWinRange(pInfo->windowSup.pStreamAggSup, &startWin, mode, &range);




×








UNCOV

1628



    colDataSetVal(pDestStartCol, i, (const char*)&range.win.skey, false);




×








UNCOV

1629



    colDataSetVal(pDestEndCol, i, (const char*)&range.win.ekey, false);




×







1630













UNCOV

1631



    colDataSetNULL(pDestUidCol, i);





×








UNCOV

1632



    colDataSetVal(pDestGpCol, i, (const char*)&groupId, false);




×








UNCOV

1633



    colDataSetVal(pDestCalStartTsCol, i, (const char*)&range.win.skey, false);




×








UNCOV

1634



    colDataSetVal(pDestCalEndTsCol, i, (const char*)&range.win.ekey, false);




×








UNCOV

1635



    pDestBlock->info.rows++;




×







1636



  }











UNCOV

1637



  return TSDB_CODE_SUCCESS;




×







1638



}










1639













UNCOV

1640



static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) {




×








UNCOV

1641



  blockDataCleanup(pDestBlock);




×








UNCOV

1642



  int32_t rows = pSrcBlock->info.rows;




×








UNCOV

1643



  if (rows == 0) {





×







1644



    return TSDB_CODE_SUCCESS;




×







1645



  }











UNCOV

1646



  SExecTaskInfo*   pTaskInfo = pInfo->pStreamScanOp->pTaskInfo;




×








UNCOV

1647



  SColumnInfoData* pSrcStartTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

1648



  SColumnInfoData* pSrcEndTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

1649



  SColumnInfoData* pSrcUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX);




×








UNCOV

1650



  SColumnInfoData* pSrcGpCol = taosArrayGet(pSrcBlock->pDataBlock, GROUPID_COLUMN_INDEX);




×







1651













UNCOV

1652



  uint64_t* srcUidData = (uint64_t*)pSrcUidCol->pData;




×








UNCOV

1653



  ASSERT(pSrcStartTsCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);





×








UNCOV

1654



  TSKEY*  srcStartTsCol = (TSKEY*)pSrcStartTsCol->pData;




×








UNCOV

1655



  TSKEY*  srcEndTsCol = (TSKEY*)pSrcEndTsCol->pData;




×








UNCOV

1656



  int64_t ver = pSrcBlock->info.version - 1;




×







1657













UNCOV

1658



  if (pInfo->partitionSup.needCalc && srcStartTsCol[0] != srcEndTsCol[0]) {





×







1659



    uint64_t     srcUid = srcUidData[0];




×







1660



    TSKEY        startTs = srcStartTsCol[0];




×







1661



    TSKEY        endTs = srcEndTsCol[0];




×







1662



    SSDataBlock* pPreRes = readPreVersionData(pInfo->pTableScanOp, srcUid, startTs, endTs, ver);




×







1663



    printDataBlock(pPreRes, "pre res", GET_TASKID(pTaskInfo));




×







1664



    blockDataCleanup(pSrcBlock);




×







1665



    int32_t code = blockDataEnsureCapacity(pSrcBlock, pPreRes->info.rows);




×







1666



    if (code != TSDB_CODE_SUCCESS) {





×







1667



      return code;




×







1668



    }










1669












1670



    SColumnInfoData* pTsCol = (SColumnInfoData*)taosArrayGet(pPreRes->pDataBlock, pInfo->primaryTsIndex);




×







1671



    rows = pPreRes->info.rows;




×







1672












1673



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





×







1674



      uint64_t groupId = calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pPreRes, i);




×







1675



      appendOneRowToStreamSpecialBlock(pSrcBlock, ((TSKEY*)pTsCol->pData) + i, ((TSKEY*)pTsCol->pData) + i, &srcUid,




×







1676



                                       &groupId, NULL);










1677



    }










1678



    printDataBlock(pSrcBlock, "new delete", GET_TASKID(pTaskInfo));




×







1679



  }











UNCOV

1680



  uint64_t* srcGp = (uint64_t*)pSrcGpCol->pData;




×








UNCOV

1681



  srcStartTsCol = (TSKEY*)pSrcStartTsCol->pData;




×








UNCOV

1682



  srcEndTsCol = (TSKEY*)pSrcEndTsCol->pData;




×








UNCOV

1683



  srcUidData = (uint64_t*)pSrcUidCol->pData;




×







1684













UNCOV

1685



  int32_t code = blockDataEnsureCapacity(pDestBlock, rows);




×








UNCOV

1686



  if (code != TSDB_CODE_SUCCESS) {





×







1687



    return code;




×







1688



  }










1689













UNCOV

1690



  SColumnInfoData* pStartTsCol = taosArrayGet(pDestBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

1691



  SColumnInfoData* pEndTsCol = taosArrayGet(pDestBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

1692



  SColumnInfoData* pDeUidCol = taosArrayGet(pDestBlock->pDataBlock, UID_COLUMN_INDEX);




×








UNCOV

1693



  SColumnInfoData* pGpCol = taosArrayGet(pDestBlock->pDataBlock, GROUPID_COLUMN_INDEX);




×








UNCOV

1694



  SColumnInfoData* pCalStartTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);




×








UNCOV

1695



  SColumnInfoData* pCalEndTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);




×








UNCOV

1696



  for (int32_t i = 0; i < rows;) {





×








UNCOV

1697



    uint64_t srcUid = srcUidData[i];




×








UNCOV

1698



    uint64_t groupId = srcGp[i];




×








UNCOV

1699



    if (groupId == 0) {





×








UNCOV

1700



      groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], ver);




×







1701



    }











UNCOV

1702



    TSKEY calStartTs = srcStartTsCol[i];




×








UNCOV

1703



    colDataSetVal(pCalStartTsCol, pDestBlock->info.rows, (const char*)(&calStartTs), false);




×








UNCOV

1704



    STimeWindow win = getSlidingWindow(srcStartTsCol, srcEndTsCol, srcGp, &pInfo->interval, &pSrcBlock->info, &i,




×








UNCOV

1705



                                       pInfo->partitionSup.needCalc);




×








UNCOV

1706



    TSKEY       calEndTs = srcStartTsCol[i - 1];




×








UNCOV

1707



    colDataSetVal(pCalEndTsCol, pDestBlock->info.rows, (const char*)(&calEndTs), false);




×








UNCOV

1708



    colDataSetVal(pDeUidCol, pDestBlock->info.rows, (const char*)(&srcUid), false);




×








UNCOV

1709



    colDataSetVal(pStartTsCol, pDestBlock->info.rows, (const char*)(&win.skey), false);




×








UNCOV

1710



    colDataSetVal(pEndTsCol, pDestBlock->info.rows, (const char*)(&win.ekey), false);




×








UNCOV

1711



    colDataSetVal(pGpCol, pDestBlock->info.rows, (const char*)(&groupId), false);




×








UNCOV

1712



    pDestBlock->info.rows++;




×







1713



  }











UNCOV

1714



  return TSDB_CODE_SUCCESS;




×







1715



}










1716













UNCOV

1717



static void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {




×








UNCOV

1718



  SExprSupp*    pTbNameCalSup = &pInfo->tbnameCalSup;




×








UNCOV

1719



  blockDataCleanup(pInfo->pCreateTbRes);




×








UNCOV

1720



  if (pInfo->tbnameCalSup.numOfExprs == 0 && pInfo->tagCalSup.numOfExprs == 0) {





×








UNCOV

1721



    pBlock->info.parTbName[0] = 0;




×







1722



  } else {











UNCOV

1723



    appendCreateTableRow(pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState, &pInfo->tbnameCalSup, &pInfo->tagCalSup,




×







1724



                         pBlock->info.id.groupId, pBlock, 0, pInfo->pCreateTbRes, &pInfo->stateStore);










1725



  }











UNCOV

1726



}




×







1727












1728



static int32_t generateDeleteResultBlock(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) {




×







1729



  blockDataCleanup(pDestBlock);




×







1730



  int32_t rows = pSrcBlock->info.rows;




×







1731



  if (rows == 0) {





×







1732



    return TSDB_CODE_SUCCESS;




×







1733



  }










1734



  int32_t code = blockDataEnsureCapacity(pDestBlock, rows);




×







1735



  if (code != TSDB_CODE_SUCCESS) {





×







1736



    return code;




×







1737



  }










1738












1739



  SColumnInfoData* pSrcStartTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX);




×







1740



  SColumnInfoData* pSrcEndTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX);




×







1741



  SColumnInfoData* pSrcUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX);




×







1742



  uint64_t*        srcUidData = (uint64_t*)pSrcUidCol->pData;




×







1743



  SColumnInfoData* pSrcGpCol = taosArrayGet(pSrcBlock->pDataBlock, GROUPID_COLUMN_INDEX);




×







1744



  uint64_t*        srcGp = (uint64_t*)pSrcGpCol->pData;




×







1745



  ASSERT(pSrcStartTsCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);





×







1746



  TSKEY*  srcStartTsCol = (TSKEY*)pSrcStartTsCol->pData;




×







1747



  TSKEY*  srcEndTsCol = (TSKEY*)pSrcEndTsCol->pData;




×







1748



  int64_t ver = pSrcBlock->info.version - 1;




×







1749



  for (int32_t i = 0; i < pSrcBlock->info.rows; i++) {





×







1750



    uint64_t srcUid = srcUidData[i];




×







1751



    uint64_t groupId = srcGp[i];




×







1752



    char     tbname[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN] = {0};




×







1753



    if (groupId == 0) {





×







1754



      groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], ver);




×







1755



    }










1756



    if (pInfo->tbnameCalSup.pExprInfo) {





×







1757



      void* parTbname = NULL;




×







1758



      code = pInfo->stateStore.streamStateGetParName(pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState, groupId, &parTbname);




×







1759



      if (code != TSDB_CODE_SUCCESS) {





×







1760



        SSDataBlock* pPreRes = readPreVersionData(pInfo->pTableScanOp, srcUid, srcStartTsCol[i], srcStartTsCol[i], ver);




×







1761



        printDataBlock(pPreRes, "pre res", GET_TASKID(pInfo->pStreamScanOp->pTaskInfo));




×







1762



        calBlockTbName(pInfo, pPreRes);




×







1763



        memcpy(varDataVal(tbname), pPreRes->info.parTbName, strlen(pPreRes->info.parTbName));




×







1764



      } else {










1765



        memcpy(varDataVal(tbname), parTbname, TSDB_TABLE_NAME_LEN);




×







1766



      }










1767



      varDataSetLen(tbname, strlen(varDataVal(tbname)));




×







1768



      pInfo->stateStore.streamStateFreeVal(parTbname);




×







1769



    }










1770



    appendOneRowToStreamSpecialBlock(pDestBlock, srcStartTsCol + i, srcEndTsCol + i, srcUidData + i, &groupId,




×







1771



                                     tbname[0] == 0 ? NULL : tbname);





×







1772



  }










1773



  return TSDB_CODE_SUCCESS;




×







1774



}










1775













UNCOV

1776



static int32_t generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock, EStreamType type) {




×








UNCOV

1777



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

1778



  if (isIntervalWindow(pInfo)) {





×








UNCOV

1779



    code = generateIntervalScanRange(pInfo, pSrcBlock, pDestBlock);




×








UNCOV

1780



  } else if (isSessionWindow(pInfo) || isStateWindow(pInfo)) {





×








UNCOV

1781



    code = generateSessionScanRange(pInfo, pSrcBlock, pDestBlock);




×








UNCOV

1782



  } else if (isCountWindow(pInfo)) {





×








UNCOV

1783



    code = generateCountScanRange(pInfo, pSrcBlock, pDestBlock, type);




×







1784



  } else {










1785



    code = generateDeleteResultBlock(pInfo, pSrcBlock, pDestBlock);




×







1786



  }











UNCOV

1787



  pDestBlock->info.type = STREAM_CLEAR;




×








UNCOV

1788



  pDestBlock->info.version = pSrcBlock->info.version;




×








UNCOV

1789



  pDestBlock->info.dataLoad = 1;




×








UNCOV

1790



  blockDataUpdateTsWindow(pDestBlock, 0);




×








UNCOV

1791



  return code;




×







1792



}










1793













UNCOV

1794



void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid,




×







1795



                                      uint64_t* pGp, void* pTbName) {











UNCOV

1796



  appendAllColumnToStreamSpecialBlock(pBlock, pStartTs, pEndTs, pStartTs, pEndTs, pUid, pGp, pTbName);




×








UNCOV

1797



}




×







1798













UNCOV

1799



void appendAllColumnToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, TSKEY* pCalStartTs,




×







1800



                                         TSKEY* pCalEndTs, uint64_t* pUid, uint64_t* pGp, void* pTbName) {











UNCOV

1801



  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

1802



  SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

1803



  SColumnInfoData* pUidCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);




×








UNCOV

1804



  SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);




×








UNCOV

1805



  SColumnInfoData* pCalStartCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);




×








UNCOV

1806



  SColumnInfoData* pCalEndCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);




×








UNCOV

1807



  SColumnInfoData* pTableCol = taosArrayGet(pBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);




×








UNCOV

1808



  colDataSetVal(pStartTsCol, pBlock->info.rows, (const char*)pStartTs, false);




×








UNCOV

1809



  colDataSetVal(pEndTsCol, pBlock->info.rows, (const char*)pEndTs, false);




×








UNCOV

1810



  colDataSetVal(pUidCol, pBlock->info.rows, (const char*)pUid, false);




×








UNCOV

1811



  colDataSetVal(pGpCol, pBlock->info.rows, (const char*)pGp, false);




×








UNCOV

1812



  colDataSetVal(pCalStartCol, pBlock->info.rows, (const char*)pCalStartTs, false);




×








UNCOV

1813



  colDataSetVal(pCalEndCol, pBlock->info.rows, (const char*)pCalEndTs, false);




×








UNCOV

1814



  colDataSetVal(pTableCol, pBlock->info.rows, (const char*)pTbName, pTbName == NULL);




×








UNCOV

1815



  pBlock->info.rows++;




×








UNCOV

1816



}




×







1817













UNCOV

1818



bool checkExpiredData(SStateStore* pAPI, SUpdateInfo* pUpdateInfo, STimeWindowAggSupp* pTwSup, uint64_t tableId, TSKEY ts) {




×








UNCOV

1819



  bool isExpired = false;




×








UNCOV

1820



  bool isInc = pAPI->isIncrementalTimeStamp(pUpdateInfo, tableId, ts);




×








UNCOV

1821



  if (!isInc) {





×








UNCOV

1822



    isExpired = isOverdue(ts, pTwSup);




×







1823



  }











UNCOV

1824



  return isExpired;




×







1825



}










1826













UNCOV

1827



static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock* pBlock, bool out) {




×








UNCOV

1828



  if (out) {





×








UNCOV

1829



    blockDataCleanup(pInfo->pUpdateDataRes);




×








UNCOV

1830



    blockDataEnsureCapacity(pInfo->pUpdateDataRes, pBlock->info.rows * 2);




×







1831



  }











UNCOV

1832



  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);




×








UNCOV

1833



  ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP);





×








UNCOV

1834



  TSKEY* tsCol = (TSKEY*)pColDataInfo->pData;




×








UNCOV

1835



  bool   tableInserted = pInfo->stateStore.updateInfoIsTableInserted(pInfo->pUpdateInfo, pBlock->info.id.uid);




×








UNCOV

1836



  for (int32_t rowId = 0; rowId < pBlock->info.rows; rowId++) {





×







1837



    SResultRowInfo dumyInfo;











UNCOV

1838



    dumyInfo.cur.pageId = -1;




×








UNCOV

1839



    bool        isClosed = false;




×








UNCOV

1840



    STimeWindow win = {.skey = INT64_MIN, .ekey = INT64_MAX};




×








UNCOV

1841



    bool        overDue = isOverdue(tsCol[rowId], &pInfo->twAggSup);




×








UNCOV

1842



    if (pInfo->igExpired && overDue) {





×







1843



      continue;




×







1844



    }










1845













UNCOV

1846



    if (tableInserted && overDue) {





×








UNCOV

1847



      win = getActiveTimeWindow(NULL, &dumyInfo, tsCol[rowId], &pInfo->interval, TSDB_ORDER_ASC);




×








UNCOV

1848



      isClosed = isCloseWindow(&win, &pInfo->twAggSup);




×







1849



    }










1850



    // must check update info first.











UNCOV

1851



    bool update = pInfo->stateStore.updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.id.uid, tsCol[rowId]);




×








UNCOV

1852



    bool closedWin = isClosed && isSignleIntervalWindow(pInfo) &&





×








UNCOV

1853



                     isDeletedStreamWindow(&win, pBlock->info.id.groupId, pInfo->pState, &pInfo->twAggSup, &pInfo->stateStore);




×








UNCOV

1854



    if ((update || closedWin) && out) {





×








UNCOV

1855



      qDebug("stream update check not pass, update %d, closedWin %d", update, closedWin);





×








UNCOV

1856



      uint64_t gpId = 0;




×








UNCOV

1857



      appendOneRowToStreamSpecialBlock(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.id.uid, &gpId,




×







1858



                                       NULL);











UNCOV

1859



      if (closedWin && pInfo->partitionSup.needCalc) {





×







1860



        gpId = calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pBlock, rowId);




×







1861



        appendOneRowToStreamSpecialBlock(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.id.uid,




×







1862



                                         &gpId, NULL);










1863



      }










1864



    }










1865



  }











UNCOV

1866



  if (out && pInfo->pUpdateDataRes->info.rows > 0) {





×








UNCOV

1867



    pInfo->pUpdateDataRes->info.version = pBlock->info.version;




×








UNCOV

1868



    pInfo->pUpdateDataRes->info.dataLoad = 1;




×








UNCOV

1869



    blockDataUpdateTsWindow(pInfo->pUpdateDataRes, 0);




×








UNCOV

1870



    pInfo->pUpdateDataRes->info.type = pInfo->partitionSup.needCalc ? STREAM_DELETE_DATA : STREAM_CLEAR;





×







1871



  }











UNCOV

1872



}




×







1873













UNCOV

1874



static void doBlockDataWindowFilter(SSDataBlock* pBlock, int32_t tsIndex, STimeWindow* pWindow, const char* id) {




×








UNCOV

1875



  if (pWindow->skey != INT64_MIN || pWindow->ekey != INT64_MAX) {





×








UNCOV

1876



    bool* p = taosMemoryCalloc(pBlock->info.rows, sizeof(bool));




×








UNCOV

1877



    bool  hasUnqualified = false;




×







1878













UNCOV

1879



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




×







1880













UNCOV

1881



    if (pWindow->skey != INT64_MIN) {





×








UNCOV

1882



      qDebug("%s filter for additional history window, skey:%" PRId64, id, pWindow->skey);





×







1883













UNCOV

1884



      ASSERT(pCol->pData != NULL);





×








UNCOV

1885



      for (int32_t i = 0; i < pBlock->info.rows; ++i) {





×








UNCOV

1886



        int64_t* ts = (int64_t*)colDataGetData(pCol, i);





×








UNCOV

1887



        p[i] = (*ts >= pWindow->skey);




×







1888













UNCOV

1889



        if (!p[i]) {





×








UNCOV

1890



          hasUnqualified = true;




×







1891



        }










1892



      }











UNCOV

1893



    } else if (pWindow->ekey != INT64_MAX) {





×








UNCOV

1894



      qDebug("%s filter for additional history window, ekey:%" PRId64, id, pWindow->ekey);





×








UNCOV

1895



      for (int32_t i = 0; i < pBlock->info.rows; ++i) {





×








UNCOV

1896



        int64_t* ts = (int64_t*)colDataGetData(pCol, i);





×








UNCOV

1897



        p[i] = (*ts <= pWindow->ekey);




×







1898













UNCOV

1899



        if (!p[i]) {





×







1900



          hasUnqualified = true;




×







1901



        }










1902



      }










1903



    }










1904













UNCOV

1905



    if (hasUnqualified) {





×








UNCOV

1906



      trimDataBlock(pBlock, pBlock->info.rows, p);




×







1907



    }










1908













UNCOV

1909



    taosMemoryFree(p);




×







1910



  }











UNCOV

1911



}




×







1912












1913



// re-build the delete block, ONLY according to the split timestamp











UNCOV

1914



static void rebuildDeleteBlockData(SSDataBlock* pBlock, STimeWindow* pWindow, const char* id) {




×








UNCOV

1915



  int32_t numOfRows = pBlock->info.rows;




×








UNCOV

1916



  bool*   p = taosMemoryCalloc(numOfRows, sizeof(bool));




×








UNCOV

1917



  bool    hasUnqualified = false;




×








UNCOV

1918



  int64_t skey = pWindow->skey;




×








UNCOV

1919



  int64_t ekey = pWindow->ekey;




×







1920













UNCOV

1921



  SColumnInfoData* pSrcStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

1922



  uint64_t*        tsStartCol = (uint64_t*)pSrcStartCol->pData;




×








UNCOV

1923



  SColumnInfoData* pSrcEndCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

1924



  uint64_t*        tsEndCol = (uint64_t*)pSrcEndCol->pData;




×







1925













UNCOV

1926



  if (pWindow->skey != INT64_MIN) {





×







1927



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





×







1928



      if (tsStartCol[i] < skey) {





×







1929



        tsStartCol[i] = skey;




×







1930



      }










1931












1932



      if (tsEndCol[i] >= skey) {





×







1933



        p[i] = true;




×







1934



      } else {  // this row should be removed, since it is not in this query time window, which is [skey, INT64_MAX]










1935



        hasUnqualified = true;




×







1936



      }










1937



    }











UNCOV

1938



  } else if (pWindow->ekey != INT64_MAX) {





×







1939



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





×







1940



      if (tsEndCol[i] > ekey) {





×







1941



        tsEndCol[i] = ekey;




×







1942



      }










1943












1944



      if (tsStartCol[i] <= ekey) {





×







1945



        p[i] = true;




×







1946



      } else {










1947



        hasUnqualified = true;




×







1948



      }










1949



    }










1950



  }










1951













UNCOV

1952



  if (hasUnqualified) {





×







1953



    trimDataBlock(pBlock, pBlock->info.rows, p);




×







1954



    qDebug("%s re-build delete datablock, start key revised to:%"PRId64", rows:%"PRId64, id, skey, pBlock->info.rows);





×







1955



  } else {











UNCOV

1956



    qDebug("%s not update the delete block", id);





×







1957



  }










1958













UNCOV

1959



  taosMemoryFree(p);




×








UNCOV

1960



}




×







1961













UNCOV

1962



static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock, STimeWindow* pTimeWindow, bool filter) {




×








UNCOV

1963



  SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;




×








UNCOV

1964



  SOperatorInfo*  pOperator = pInfo->pStreamScanOp;




×








UNCOV

1965



  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

1966



  const char*     id = GET_TASKID(pTaskInfo);




×







1967













UNCOV

1968



  blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows);




×







1969













UNCOV

1970



  pBlockInfo->rows = pBlock->info.rows;




×








UNCOV

1971



  pBlockInfo->id.uid = pBlock->info.id.uid;




×








UNCOV

1972



  pBlockInfo->type = STREAM_NORMAL;




×








UNCOV

1973



  pBlockInfo->version = pBlock->info.version;




×







1974













UNCOV

1975



  STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;




×








UNCOV

1976



  pBlockInfo->id.groupId = tableListGetTableGroupId(pTableScanInfo->base.pTableListInfo, pBlock->info.id.uid);




×







1977












1978



  // todo extract method











UNCOV

1979



  for (int32_t i = 0; i < taosArrayGetSize(pInfo->matchInfo.pList); ++i) {





×








UNCOV

1980



    SColMatchItem* pColMatchInfo = taosArrayGet(pInfo->matchInfo.pList, i);




×








UNCOV

1981



    if (!pColMatchInfo->needOutput) {





×







1982



      continue;




×







1983



    }










1984













UNCOV

1985



    bool colExists = false;




×








UNCOV

1986



    for (int32_t j = 0; j < blockDataGetNumOfCols(pBlock); ++j) {





×








UNCOV

1987



      SColumnInfoData* pResCol = bdGetColumnInfoData(pBlock, j);




×








UNCOV

1988



      if (pResCol->info.colId == pColMatchInfo->colId) {





×








UNCOV

1989



        SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pColMatchInfo->dstSlotId);




×








UNCOV

1990



        colDataAssign(pDst, pResCol, pBlock->info.rows, &pInfo->pRes->info);




×








UNCOV

1991



        colExists = true;




×








UNCOV

1992



        break;




×







1993



      }










1994



    }










1995












1996



    // the required column does not exists in submit block, let's set it to be all null value











UNCOV

1997



    if (!colExists) {





×







1998



      SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pColMatchInfo->dstSlotId);




×







1999



      colDataSetNNULL(pDst, 0, pBlockInfo->rows);





×







2000



    }










2001



  }










2002












2003



  // currently only the tbname pseudo column











UNCOV

2004



  if (pInfo->numOfPseudoExpr > 0) {





×








UNCOV

2005



    int32_t code = addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, pInfo->pRes,




×








UNCOV

2006



                                          pBlockInfo->rows, pTaskInfo, &pTableScanInfo->base.metaCache);




×







2007



    // ignore the table not exists error, since this table may have been dropped during the scan procedure.











UNCOV

2008



    if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_PAR_TABLE_NOT_EXIST) {





×







2009



      blockDataFreeRes((SSDataBlock*)pBlock);




×







2010



      T_LONG_JMP(pTaskInfo->env, code);





×







2011



    }










2012












2013



    // reset the error code.











UNCOV

2014



    terrno = 0;




×







2015



  }










2016













UNCOV

2017



  if (filter) {





×








UNCOV

2018



    doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);




×







2019



  }










2020












2021



  // filter the block extracted from WAL files, according to the time window apply additional time window filter











UNCOV

2022



  doBlockDataWindowFilter(pInfo->pRes, pInfo->primaryTsIndex, pTimeWindow, id);




×








UNCOV

2023



  pInfo->pRes->info.dataLoad = 1;




×







2024













UNCOV

2025



  blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);




×








UNCOV

2026



  if (pInfo->pRes->info.rows == 0) {





×








UNCOV

2027



    return 0;




×







2028



  }










2029













UNCOV

2030



  calBlockTbName(pInfo, pInfo->pRes);




×








UNCOV

2031



  return 0;




×







2032



}










2033













UNCOV

2034



static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {




×








UNCOV

2035



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

2036



  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;




×







2037













UNCOV

2038



  SStreamScanInfo* pInfo = pOperator->info;




×








UNCOV

2039



  const char*      id = GET_TASKID(pTaskInfo);




×







2040













UNCOV

2041



  qDebug("start to exec queue scan, %s", id);





×







2042













UNCOV

2043



  if (isTaskKilled(pTaskInfo)) {





×







2044



    return NULL;




×







2045



  }










2046













UNCOV

2047



  if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_DATA) {





×








UNCOV

2048



    SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);




×








UNCOV

2049



    if (pResult && pResult->info.rows > 0) {





×








UNCOV

2050



      tqOffsetResetToData(&pTaskInfo->streamInfo.currentOffset, pResult->info.id.uid, pResult->info.window.ekey);




×








UNCOV

2051



      return pResult;




×







2052



    }










2053













UNCOV

2054



    STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;




×








UNCOV

2055



    pAPI->tsdReader.tsdReaderClose(pTSInfo->base.dataReader);




×







2056













UNCOV

2057



    pTSInfo->base.dataReader = NULL;




×








UNCOV

2058



    int64_t validVer = pTaskInfo->streamInfo.snapshotVer + 1;




×








UNCOV

2059



    qDebug("queue scan tsdb over, switch to wal ver %" PRId64 "", validVer);





×








UNCOV

2060



    if (pAPI->tqReaderFn.tqReaderSeek(pInfo->tqReader, validVer, pTaskInfo->id.str) < 0) {





×








UNCOV

2061



      return NULL;




×







2062



    }










2063













UNCOV

2064



    tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, validVer);




×







2065



  }










2066













UNCOV

2067



  if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__LOG) {





×







2068












2069



    while (1) {




×








UNCOV

2070



      bool hasResult = pAPI->tqReaderFn.tqReaderNextBlockInWal(pInfo->tqReader, id, pTaskInfo->streamInfo.sourceExcluded);




×







2071













UNCOV

2072



      SSDataBlock* pRes = pAPI->tqReaderFn.tqGetResultBlock(pInfo->tqReader);




×








UNCOV

2073



      struct SWalReader* pWalReader = pAPI->tqReaderFn.tqReaderGetWalReader(pInfo->tqReader);




×







2074












2075



      // curVersion move to next











UNCOV

2076



      tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, pWalReader->curVersion);




×







2077












2078



      // use ts to pass time when replay, because ts not used if type is log











UNCOV

2079



      pTaskInfo->streamInfo.currentOffset.ts = pAPI->tqReaderFn.tqGetResultBlockTime(pInfo->tqReader);




×







2080













UNCOV

2081



      if (hasResult) {





×








UNCOV

2082



        qDebug("doQueueScan get data from log %" PRId64 " rows, version:%" PRId64, pRes->info.rows,





×







2083



               pTaskInfo->streamInfo.currentOffset.version);











UNCOV

2084



        blockDataCleanup(pInfo->pRes);




×








UNCOV

2085



        STimeWindow defaultWindow = {.skey = INT64_MIN, .ekey = INT64_MAX};




×








UNCOV

2086



        setBlockIntoRes(pInfo, pRes, &defaultWindow, true);




×








UNCOV

2087



        qDebug("doQueueScan after filter get data from log %" PRId64 " rows, version:%" PRId64, pInfo->pRes->info.rows,





×







2088



               pTaskInfo->streamInfo.currentOffset.version);











UNCOV

2089



        if (pInfo->pRes->info.rows > 0) {





×








UNCOV

2090



          return pInfo->pRes;




×







2091



        }










2092



      } else {











UNCOV

2093



        qDebug("doQueueScan get none from log, return, version:%" PRId64, pTaskInfo->streamInfo.currentOffset.version);





×








UNCOV

2094



        return NULL;




×







2095



      }










2096



    }










2097



  } else {










2098



    qError("unexpected streamInfo prepare type: %d", pTaskInfo->streamInfo.currentOffset.type);





×







2099



    return NULL;




×







2100



  }










2101



}










2102













UNCOV

2103



static int32_t filterDelBlockByUid(SSDataBlock* pDst, const SSDataBlock* pSrc, SStreamScanInfo* pInfo) {




×








UNCOV

2104



  STqReader* pReader = pInfo->tqReader;




×








UNCOV

2105



  int32_t    rows = pSrc->info.rows;




×








UNCOV

2106



  blockDataEnsureCapacity(pDst, rows);




×







2107













UNCOV

2108



  SColumnInfoData* pSrcStartCol = taosArrayGet(pSrc->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

2109



  uint64_t*        startCol = (uint64_t*)pSrcStartCol->pData;




×








UNCOV

2110



  SColumnInfoData* pSrcEndCol = taosArrayGet(pSrc->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

2111



  uint64_t*        endCol = (uint64_t*)pSrcEndCol->pData;




×








UNCOV

2112



  SColumnInfoData* pSrcUidCol = taosArrayGet(pSrc->pDataBlock, UID_COLUMN_INDEX);




×








UNCOV

2113



  uint64_t*        uidCol = (uint64_t*)pSrcUidCol->pData;




×







2114













UNCOV

2115



  SColumnInfoData* pDstStartCol = taosArrayGet(pDst->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

2116



  SColumnInfoData* pDstEndCol = taosArrayGet(pDst->pDataBlock, END_TS_COLUMN_INDEX);




×








UNCOV

2117



  SColumnInfoData* pDstUidCol = taosArrayGet(pDst->pDataBlock, UID_COLUMN_INDEX);




×







2118













UNCOV

2119



  int32_t j = 0;




×








UNCOV

2120



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





×








UNCOV

2121



    if (pInfo->readerFn.tqReaderIsQueriedTable(pReader, uidCol[i])) {





×








UNCOV

2122



      colDataSetVal(pDstStartCol, j, (const char*)&startCol[i], false);




×








UNCOV

2123



      colDataSetVal(pDstEndCol, j, (const char*)&endCol[i], false);




×








UNCOV

2124



      colDataSetVal(pDstUidCol, j, (const char*)&uidCol[i], false);




×







2125













UNCOV

2126



      colDataSetNULL(taosArrayGet(pDst->pDataBlock, GROUPID_COLUMN_INDEX), j);




×








UNCOV

2127



      colDataSetNULL(taosArrayGet(pDst->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX), j);




×








UNCOV

2128



      colDataSetNULL(taosArrayGet(pDst->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX), j);




×








UNCOV

2129



      j++;




×







2130



    }










2131



  }










2132













UNCOV

2133



  uint32_t cap = pDst->info.capacity;




×








UNCOV

2134



  pDst->info = pSrc->info;




×








UNCOV

2135



  pDst->info.rows = j;




×








UNCOV

2136



  pDst->info.capacity = cap;




×







2137













UNCOV

2138



  return 0;




×







2139



}










2140












2141



// for partition by tag











UNCOV

2142



static void setBlockGroupIdByUid(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {




×








UNCOV

2143



  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);




×








UNCOV

2144



  TSKEY*           startTsCol = (TSKEY*)pStartTsCol->pData;




×








UNCOV

2145



  SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);




×








UNCOV

2146



  uint64_t*        gpCol = (uint64_t*)pGpCol->pData;




×








UNCOV

2147



  SColumnInfoData* pUidCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);




×








UNCOV

2148



  uint64_t*        uidCol = (uint64_t*)pUidCol->pData;




×








UNCOV

2149



  int32_t          rows = pBlock->info.rows;




×








UNCOV

2150



  if (!pInfo->partitionSup.needCalc) {





×








UNCOV

2151



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





×








UNCOV

2152



      uint64_t groupId = getGroupIdByUid(pInfo, uidCol[i]);




×








UNCOV

2153



      colDataSetVal(pGpCol, i, (const char*)&groupId, false);




×







2154



    }










2155



  }











UNCOV

2156



}




×







2157













UNCOV

2158



static void doCheckUpdate(SStreamScanInfo* pInfo, TSKEY endKey, SSDataBlock* pBlock) {




×








UNCOV

2159



  if (pInfo->pUpdateInfo) {





×








UNCOV

2160



    pInfo->pUpdateInfo->maxDataVersion = TMAX(pInfo->pUpdateInfo->maxDataVersion, pBlock->info.version);




×







2161



  }











UNCOV

2162



  if (!pInfo->igCheckUpdate && pInfo->pUpdateInfo) {





×








UNCOV

2163



    checkUpdateData(pInfo, true, pBlock, true);




×








UNCOV

2164



    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, endKey);




×








UNCOV

2165



    if (pInfo->pUpdateDataRes->info.rows > 0) {





×








UNCOV

2166



      pInfo->updateResIndex = 0;




×








UNCOV

2167



      if (pInfo->pUpdateDataRes->info.type == STREAM_CLEAR) {





×








UNCOV

2168



        pInfo->scanMode = STREAM_SCAN_FROM_UPDATERES;




×








UNCOV

2169



      } else if (pInfo->pUpdateDataRes->info.type == STREAM_INVERT) {





×







2170



        pInfo->scanMode = STREAM_SCAN_FROM_RES;




×







2171



        // return pInfo->pUpdateDataRes;











UNCOV

2172



      } else if (pInfo->pUpdateDataRes->info.type == STREAM_DELETE_DATA) {





×








UNCOV

2173



        pInfo->scanMode = STREAM_SCAN_FROM_DELETE_DATA;




×







2174



      }










2175



    }










2176



  }











UNCOV

2177



}




×







2178













UNCOV

2179



int32_t streamScanOperatorEncode(SStreamScanInfo* pInfo, void** pBuff) {




×








UNCOV

2180



 int32_t len = pInfo->stateStore.updateInfoSerialize(NULL, 0, pInfo->pUpdateInfo);




×








UNCOV

2181



 len += encodeSTimeWindowAggSupp(NULL, &pInfo->twAggSup);




×








UNCOV

2182



 *pBuff = taosMemoryCalloc(1, len);




×








UNCOV

2183



 void* buf = *pBuff;




×








UNCOV

2184



 encodeSTimeWindowAggSupp(&buf, &pInfo->twAggSup);




×








UNCOV

2185



 pInfo->stateStore.updateInfoSerialize(buf, len, pInfo->pUpdateInfo);




×








UNCOV

2186



 return len;




×







2187



}










2188













UNCOV

2189



void streamScanOperatorSaveCheckpoint(SStreamScanInfo* pInfo) {




×








UNCOV

2190



  if (!pInfo->pState) {





×







2191



    return;




×







2192



  }











UNCOV

2193



  void* pBuf = NULL;




×








UNCOV

2194



  int32_t len = streamScanOperatorEncode(pInfo, &pBuf);




×








UNCOV

2195



  pInfo->stateStore.streamStateSaveInfo(pInfo->pState, STREAM_SCAN_OP_CHECKPOINT_NAME, strlen(STREAM_SCAN_OP_CHECKPOINT_NAME), pBuf, len);




×








UNCOV

2196



  taosMemoryFree(pBuf);




×







2197



}










2198












2199



// other properties are recovered from the execution plan










2200



void streamScanOperatorDecode(void* pBuff, int32_t len, SStreamScanInfo* pInfo) {




7✔





2201



  if (!pBuff || len == 0) {





7!





2202



    return;




×







2203



  }










2204



  void* buf = pBuff;




7✔





2205



  buf = decodeSTimeWindowAggSupp(buf, &pInfo->twAggSup);




7✔





2206



  int32_t tlen = len - encodeSTimeWindowAggSupp(NULL, &pInfo->twAggSup);




7✔





2207



  if (tlen == 0) {





7✔





2208



    return;




1✔





2209



  }










2210












2211



  void* pUpInfo = taosMemoryCalloc(1, sizeof(SUpdateInfo));




6✔





2212



  int32_t code = pInfo->stateStore.updateInfoDeserialize(buf, tlen, pUpInfo);




6✔





2213



  if (code == TSDB_CODE_SUCCESS) {





6!





2214



    pInfo->stateStore.updateInfoDestroy(pInfo->pUpdateInfo);




6✔





2215



    pInfo->pUpdateInfo = pUpInfo;




6✔





2216



  } else {










2217



    taosMemoryFree(pUpInfo);




×







2218



  }










2219



}











UNCOV

2220



static bool hasScanRange(SStreamScanInfo* pInfo) {




×








UNCOV

2221



  SStreamAggSupporter* pSup = pInfo->windowSup.pStreamAggSup;




×








UNCOV

2222



  return pSup && pSup->pScanBlock->info.rows > 0 && (isStateWindow(pInfo) || isCountWindow(pInfo));





×







2223



}










2224













UNCOV

2225



static bool isStreamWindow(SStreamScanInfo* pInfo) {




×








UNCOV

2226



  return isIntervalWindow(pInfo) || isSessionWindow(pInfo) || isStateWindow(pInfo) || isCountWindow(pInfo);





×







2227



}










2228













UNCOV

2229



static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {




×







2230



  // NOTE: this operator does never check if current status is done or not











UNCOV

2231



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

2232



  const char*    id = GET_TASKID(pTaskInfo);




×







2233













UNCOV

2234



  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;




×








UNCOV

2235



  SStreamScanInfo* pInfo = pOperator->info;




×








UNCOV

2236



  SStreamTaskInfo* pStreamInfo = &pTaskInfo->streamInfo;




×







2237













UNCOV

2238



  qDebug("stream scan started, %s", id);





×







2239













UNCOV

2240



  if (pStreamInfo->recoverStep == STREAM_RECOVER_STEP__PREPARE1 || pStreamInfo->recoverStep == STREAM_RECOVER_STEP__PREPARE2) {





×








UNCOV

2241



    STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;




×








UNCOV

2242



    memcpy(&pTSInfo->base.cond, &pStreamInfo->tableCond, sizeof(SQueryTableDataCond));




×







2243













UNCOV

2244



    if (pStreamInfo->recoverStep == STREAM_RECOVER_STEP__PREPARE1) {





×








UNCOV

2245



      pTSInfo->base.cond.startVersion = pStreamInfo->fillHistoryVer.minVer;




×








UNCOV

2246



      pTSInfo->base.cond.endVersion = pStreamInfo->fillHistoryVer.maxVer;




×







2247













UNCOV

2248



      pTSInfo->base.cond.twindows = pStreamInfo->fillHistoryWindow;




×








UNCOV

2249



      qDebug("stream scan step1, verRange:%" PRId64 "-%" PRId64 " window:%"PRId64"-%"PRId64", %s", pTSInfo->base.cond.startVersion,





×







2250



             pTSInfo->base.cond.endVersion, pTSInfo->base.cond.twindows.skey, pTSInfo->base.cond.twindows.ekey, id);











UNCOV

2251



      pStreamInfo->recoverStep = STREAM_RECOVER_STEP__SCAN1;




×








UNCOV

2252



      pStreamInfo->recoverScanFinished = false;




×







2253



    } else {











UNCOV

2254



      pTSInfo->base.cond.startVersion = pStreamInfo->fillHistoryVer.minVer;




×








UNCOV

2255



      pTSInfo->base.cond.endVersion = pStreamInfo->fillHistoryVer.maxVer;




×








UNCOV

2256



      pTSInfo->base.cond.twindows = pStreamInfo->fillHistoryWindow;




×








UNCOV

2257



      qDebug("stream scan step2 (scan wal), verRange:%" PRId64 " - %" PRId64 ", window:%" PRId64 "-%" PRId64 ", %s",





×







2258



             pTSInfo->base.cond.startVersion, pTSInfo->base.cond.endVersion, pTSInfo->base.cond.twindows.skey,










2259



             pTSInfo->base.cond.twindows.ekey, id);











UNCOV

2260



      pStreamInfo->recoverStep = STREAM_RECOVER_STEP__NONE;




×







2261



    }










2262













UNCOV

2263



    pAPI->tsdReader.tsdReaderClose(pTSInfo->base.dataReader);




×







2264













UNCOV

2265



    pTSInfo->base.dataReader = NULL;




×








UNCOV

2266



    pInfo->pTableScanOp->status = OP_OPENED;




×







2267













UNCOV

2268



    pTSInfo->scanTimes = 0;




×








UNCOV

2269



    pTSInfo->currentGroupId = -1;




×







2270



  }










2271













UNCOV

2272



  if (pStreamInfo->recoverStep == STREAM_RECOVER_STEP__SCAN1) {





×








UNCOV

2273



    if (isTaskKilled(pTaskInfo)) {





×







2274



      qInfo("===stream===stream scan is killed. task id:%s, code %s", id, tstrerror(pTaskInfo->code));





×







2275



      return NULL;




×







2276



    }










2277













UNCOV

2278



    switch (pInfo->scanMode) {





×







2279



      case STREAM_SCAN_FROM_RES: {




×







2280



        pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;




×







2281



        printSpecDataBlock(pInfo->pRecoverRes, getStreamOpName(pOperator->operatorType), "recover", GET_TASKID(pTaskInfo));




×







2282



        return pInfo->pRecoverRes;




×







2283



      } break;











UNCOV

2284



      default:




×








UNCOV

2285



        break;




×







2286



    }










2287













UNCOV

2288



    pInfo->pRecoverRes = doTableScan(pInfo->pTableScanOp);




×








UNCOV

2289



    if (pInfo->pRecoverRes != NULL) {





×








UNCOV

2290



      calBlockTbName(pInfo, pInfo->pRecoverRes);




×








UNCOV

2291



      if (!pInfo->igCheckUpdate && pInfo->pUpdateInfo) {





×








UNCOV

2292



        TSKEY maxTs = pAPI->stateStore.updateInfoFillBlockData(pInfo->pUpdateInfo, pInfo->pRecoverRes, pInfo->primaryTsIndex);




×








UNCOV

2293



        pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);




×







2294



      }











UNCOV

2295



      if (pInfo->pCreateTbRes->info.rows > 0) {





×







2296



        pInfo->scanMode = STREAM_SCAN_FROM_RES;




×







2297



        printSpecDataBlock(pInfo->pCreateTbRes, getStreamOpName(pOperator->operatorType), "recover", GET_TASKID(pTaskInfo));




×







2298



        return pInfo->pCreateTbRes;




×







2299



      }










2300













UNCOV

2301



      qDebug("stream recover scan get block, rows %" PRId64, pInfo->pRecoverRes->info.rows);





×








UNCOV

2302



      printSpecDataBlock(pInfo->pRecoverRes, getStreamOpName(pOperator->operatorType), "recover", GET_TASKID(pTaskInfo));




×








UNCOV

2303



      return pInfo->pRecoverRes;




×







2304



    }











UNCOV

2305



    pStreamInfo->recoverStep = STREAM_RECOVER_STEP__NONE;




×








UNCOV

2306



    STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;




×








UNCOV

2307



    pAPI->tsdReader.tsdReaderClose(pTSInfo->base.dataReader);




×







2308













UNCOV

2309



    pTSInfo->base.dataReader = NULL;




×







2310













UNCOV

2311



    pTSInfo->base.cond.startVersion = -1;




×








UNCOV

2312



    pTSInfo->base.cond.endVersion = -1;




×







2313













UNCOV

2314



    pStreamInfo->recoverScanFinished = true;




×








UNCOV

2315



    return NULL;




×







2316



  }










2317













UNCOV

2318



  size_t total = taosArrayGetSize(pInfo->pBlockLists);




×







2319



// TODO: refactor











UNCOV

2320



FETCH_NEXT_BLOCK:




×








UNCOV

2321



  if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {





×








UNCOV

2322



    if (pInfo->validBlockIndex >= total) {





×







2323



      doClearBufferedBlocks(pInfo);











UNCOV

2324



      return NULL;




×







2325



    }










2326













UNCOV

2327



    int32_t  current = pInfo->validBlockIndex++;




×








UNCOV

2328



    qDebug("process %d/%d input data blocks, %s", current, (int32_t) total, id);





×







2329













UNCOV

2330



    SPackedData* pPacked = taosArrayGet(pInfo->pBlockLists, current);




×








UNCOV

2331



    SSDataBlock* pBlock = pPacked->pDataBlock;




×








UNCOV

2332



    if (pBlock->info.parTbName[0]) {





×








UNCOV

2333



      pAPI->stateStore.streamStatePutParName(pStreamInfo->pState, pBlock->info.id.groupId, pBlock->info.parTbName);




×







2334



    }










2335












2336



    // TODO move into scan











UNCOV

2337



    pBlock->info.calWin.skey = INT64_MIN;




×








UNCOV

2338



    pBlock->info.calWin.ekey = INT64_MAX;




×








UNCOV

2339



    pBlock->info.dataLoad = 1;




×








UNCOV

2340



    if (pInfo->pUpdateInfo) {





×








UNCOV

2341



      pInfo->pUpdateInfo->maxDataVersion = TMAX(pInfo->pUpdateInfo->maxDataVersion, pBlock->info.version);




×







2342



    }










2343













UNCOV

2344



    blockDataUpdateTsWindow(pBlock, 0);




×








UNCOV

2345



    switch (pBlock->info.type) {





×








UNCOV

2346



      case STREAM_NORMAL:




×







2347



      case STREAM_GET_ALL:











UNCOV

2348



        return pBlock;




×








UNCOV

2349



      case STREAM_RETRIEVE: {




×








UNCOV

2350



        pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;




×








UNCOV

2351



        pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RETRIEVE;




×








UNCOV

2352



        copyDataBlock(pInfo->pUpdateRes, pBlock);




×








UNCOV

2353



        pInfo->updateResIndex = 0;




×








UNCOV

2354



        prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);




×








UNCOV

2355



        pAPI->stateStore.updateInfoAddCloseWindowSBF(pInfo->pUpdateInfo);




×








UNCOV

2356



      } break;




×








UNCOV

2357



      case STREAM_DELETE_DATA: {




×








UNCOV

2358



        printSpecDataBlock(pBlock, getStreamOpName(pOperator->operatorType), "delete recv", GET_TASKID(pTaskInfo));




×








UNCOV

2359



        SSDataBlock* pDelBlock = NULL;




×








UNCOV

2360



        if (pInfo->tqReader) {





×








UNCOV

2361



          pDelBlock = createSpecialDataBlock(STREAM_DELETE_DATA);




×








UNCOV

2362



          filterDelBlockByUid(pDelBlock, pBlock, pInfo);




×







2363



        } else {










2364



          pDelBlock = pBlock;




×







2365



        }










2366













UNCOV

2367



        setBlockGroupIdByUid(pInfo, pDelBlock);




×








UNCOV

2368



        rebuildDeleteBlockData(pDelBlock, &pStreamInfo->fillHistoryWindow, id);




×








UNCOV

2369



        printSpecDataBlock(pDelBlock, getStreamOpName(pOperator->operatorType), "delete recv filtered", GET_TASKID(pTaskInfo));




×








UNCOV

2370



        if (pDelBlock->info.rows == 0) {





×








UNCOV

2371



          if (pInfo->tqReader) {





×








UNCOV

2372



            blockDataDestroy(pDelBlock);




×







2373



          }











UNCOV

2374



          goto FETCH_NEXT_BLOCK;




×







2375



        }










2376













UNCOV

2377



        if (!isStreamWindow(pInfo)) {





×







2378



          generateDeleteResultBlock(pInfo, pDelBlock, pInfo->pDeleteDataRes);




×







2379



          pInfo->pDeleteDataRes->info.type = STREAM_DELETE_RESULT;




×







2380



          printSpecDataBlock(pDelBlock, getStreamOpName(pOperator->operatorType), "delete result", GET_TASKID(pTaskInfo));




×







2381



          blockDataDestroy(pDelBlock);




×







2382












2383



          if (pInfo->pDeleteDataRes->info.rows > 0) {





×







2384



            return pInfo->pDeleteDataRes;




×







2385



          } else {










2386



            goto FETCH_NEXT_BLOCK;




×







2387



          }










2388



        } else {











UNCOV

2389



          pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;




×








UNCOV

2390



          pInfo->updateResIndex = 0;




×








UNCOV

2391



          generateScanRange(pInfo, pDelBlock, pInfo->pUpdateRes, STREAM_DELETE_DATA);




×








UNCOV

2392



          prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);




×








UNCOV

2393



          copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes);




×








UNCOV

2394



          pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA;




×








UNCOV

2395



          printSpecDataBlock(pDelBlock, getStreamOpName(pOperator->operatorType), "delete result", GET_TASKID(pTaskInfo));




×








UNCOV

2396



          if (pInfo->tqReader) {





×








UNCOV

2397



            blockDataDestroy(pDelBlock);




×







2398



          }











UNCOV

2399



          if (pInfo->pDeleteDataRes->info.rows > 0) {





×








UNCOV

2400



            pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;




×








UNCOV

2401



            return pInfo->pDeleteDataRes;




×







2402



          } else {











UNCOV

2403



            goto FETCH_NEXT_BLOCK;




×







2404



          }










2405



        }










2406



      } break;










2407



      case STREAM_CHECKPOINT: {




×







2408



        qError("stream check point error. msg type: STREAM_INPUT__DATA_BLOCK");





×







2409



      } break;




×








UNCOV

2410



      default:




×








UNCOV

2411



        break;




×







2412



    }











UNCOV

2413



    printDataBlock(pBlock, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));




×








UNCOV

2414



    return pBlock;




×








UNCOV

2415



  } else if (pInfo->blockType == STREAM_INPUT__DATA_SUBMIT) {





×








UNCOV

2416



    qDebug("stream scan mode:%d, %s", pInfo->scanMode, id);





×








UNCOV

2417



    switch (pInfo->scanMode) {





×








UNCOV

2418



      case STREAM_SCAN_FROM_RES: {




×








UNCOV

2419



        pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;




×








UNCOV

2420



        doCheckUpdate(pInfo, pInfo->pRes->info.window.ekey, pInfo->pRes);




×








UNCOV

2421



        doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);




×








UNCOV

2422



        pInfo->pRes->info.dataLoad = 1;




×








UNCOV

2423



        blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);




×








UNCOV

2424



        if (pInfo->pRes->info.rows > 0) {





×








UNCOV

2425



          return pInfo->pRes;




×







2426



        }










2427



      } break;




×








UNCOV

2428



      case STREAM_SCAN_FROM_DELETE_DATA: {




×








UNCOV

2429



        generateScanRange(pInfo, pInfo->pUpdateDataRes, pInfo->pUpdateRes, STREAM_DELETE_DATA);




×








UNCOV

2430



        prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);




×








UNCOV

2431



        pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;




×








UNCOV

2432



        copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes);




×








UNCOV

2433



        pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA;




×








UNCOV

2434



        return pInfo->pDeleteDataRes;




×







2435



      } break;











UNCOV

2436



      case STREAM_SCAN_FROM_UPDATERES: {




×








UNCOV

2437



        generateScanRange(pInfo, pInfo->pUpdateDataRes, pInfo->pUpdateRes, STREAM_CLEAR);




×








UNCOV

2438



        prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);




×








UNCOV

2439



        pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;




×








UNCOV

2440



        return pInfo->pUpdateRes;




×







2441



      } break;











UNCOV

2442



      case STREAM_SCAN_FROM_DATAREADER_RANGE:




×







2443



      case STREAM_SCAN_FROM_DATAREADER_RETRIEVE: {











UNCOV

2444



        SSDataBlock* pSDB = doRangeScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);




×








UNCOV

2445



        if (pSDB) {





×








UNCOV

2446



          STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;




×








UNCOV

2447



          pSDB->info.type = pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE ? STREAM_NORMAL : STREAM_PULL_DATA;





×








UNCOV

2448



          if (!pInfo->igCheckUpdate && pInfo->pUpdateInfo) {





×








UNCOV

2449



            checkUpdateData(pInfo, true, pSDB, false);




×







2450



          }











UNCOV

2451



          printSpecDataBlock(pSDB, getStreamOpName(pOperator->operatorType), "update", GET_TASKID(pTaskInfo));




×








UNCOV

2452



          calBlockTbName(pInfo, pSDB);




×








UNCOV

2453



          return pSDB;




×







2454



        }











UNCOV

2455



        blockDataCleanup(pInfo->pUpdateDataRes);




×








UNCOV

2456



        pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;




×








UNCOV

2457



      } break;




×








UNCOV

2458



      default:




×








UNCOV

2459



        break;




×







2460



    }










2461













UNCOV

2462



    if (hasScanRange(pInfo)) {





×








UNCOV

2463



      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;




×








UNCOV

2464



      pInfo->updateResIndex = 0;




×








UNCOV

2465



      SStreamAggSupporter* pSup = pInfo->windowSup.pStreamAggSup;




×








UNCOV

2466



      copyDataBlock(pInfo->pUpdateRes, pSup->pScanBlock);




×








UNCOV

2467



      blockDataCleanup(pSup->pScanBlock);




×








UNCOV

2468



      prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);




×








UNCOV

2469



      pInfo->pUpdateRes->info.type = STREAM_DELETE_DATA;




×








UNCOV

2470



      printSpecDataBlock(pInfo->pUpdateRes, getStreamOpName(pOperator->operatorType), "rebuild", GET_TASKID(pTaskInfo));




×








UNCOV

2471



      return pInfo->pUpdateRes;




×







2472



    }










2473













UNCOV

2474



    SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;




×








UNCOV

2475



    int32_t         totalBlocks = taosArrayGetSize(pInfo->pBlockLists);




×







2476













UNCOV

2477



  NEXT_SUBMIT_BLK:




×







2478



    while (1) {











UNCOV

2479



      if (pInfo->readerFn.tqReaderCurrentBlockConsumed(pInfo->tqReader)) {





×








UNCOV

2480



        if (pInfo->validBlockIndex >= totalBlocks) {





×








UNCOV

2481



          pAPI->stateStore.updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);




×







2482



          doClearBufferedBlocks(pInfo);










2483













UNCOV

2484



          qDebug("stream scan return empty, all %d submit blocks consumed, %s", totalBlocks, id);





×








UNCOV

2485



          return NULL;




×







2486



        }










2487













UNCOV

2488



        int32_t      current = pInfo->validBlockIndex++;




×








UNCOV

2489



        SPackedData* pSubmit = taosArrayGet(pInfo->pBlockLists, current);




×







2490













UNCOV

2491



        qDebug("set %d/%d as the input submit block, %s", current + 1, totalBlocks, id);





×








UNCOV

2492



        if (pAPI->tqReaderFn.tqReaderSetSubmitMsg(pInfo->tqReader, pSubmit->msgStr, pSubmit->msgLen, pSubmit->ver) < 0) {





×







2493



          qError("submit msg messed up when initializing stream submit block %p, current %d/%d, %s", pSubmit, current, totalBlocks, id);





×







2494



          continue;




×







2495



        }










2496



      }










2497













UNCOV

2498



      blockDataCleanup(pInfo->pRes);




×







2499













UNCOV

2500



      while (pAPI->tqReaderFn.tqNextBlockImpl(pInfo->tqReader, id)) {





×








UNCOV

2501



        SSDataBlock* pRes = NULL;




×







2502













UNCOV

2503



        int32_t code = pAPI->tqReaderFn.tqRetrieveBlock(pInfo->tqReader, &pRes, id);




×








UNCOV

2504



        qDebug("retrieve data from submit completed code:%s rows:%" PRId64 " %s", tstrerror(code), pRes->info.rows, id);





×







2505













UNCOV

2506



        if (code != TSDB_CODE_SUCCESS || pRes->info.rows == 0) {





×







2507



          qDebug("retrieve data failed, try next block in submit block, %s", id);





×








UNCOV

2508



          continue;




×







2509



        }










2510













UNCOV

2511



        setBlockIntoRes(pInfo, pRes, &pStreamInfo->fillHistoryWindow, false);




×








UNCOV

2512



        if (pInfo->pRes->info.rows == 0) {





×








UNCOV

2513



          continue;




×







2514



        }










2515













UNCOV

2516



        if (pInfo->pCreateTbRes->info.rows > 0) {





×








UNCOV

2517



          pInfo->scanMode = STREAM_SCAN_FROM_RES;




×








UNCOV

2518



          qDebug("create table res exists, rows:%"PRId64" return from stream scan, %s", pInfo->pCreateTbRes->info.rows, id);





×








UNCOV

2519



          return pInfo->pCreateTbRes;




×







2520



        }










2521













UNCOV

2522



        doCheckUpdate(pInfo, pBlockInfo->window.ekey, pInfo->pRes);




×








UNCOV

2523



        doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);




×








UNCOV

2524



        blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);




×







2525













UNCOV

2526



        int64_t numOfUpdateRes = pInfo->pUpdateDataRes->info.rows;




×








UNCOV

2527



        qDebug("%s %" PRId64 " rows in datablock, update res:%" PRId64, id, pBlockInfo->rows, numOfUpdateRes);





×








UNCOV

2528



        if (pBlockInfo->rows > 0 || numOfUpdateRes > 0) {





×







2529



          break;










2530



        }










2531



      }










2532













UNCOV

2533



      if (pBlockInfo->rows > 0 || pInfo->pUpdateDataRes->info.rows > 0) {





×







2534



        break;










2535



      } else {











UNCOV

2536



        continue;




×







2537



      }










2538



    }










2539












2540



    // record the scan action.











UNCOV

2541



    pInfo->numOfExec++;




×








UNCOV

2542



    pOperator->resultInfo.totalRows += pBlockInfo->rows;




×







2543













UNCOV

2544



    qDebug("stream scan completed, and return source rows:%" PRId64", %s", pBlockInfo->rows, id);





×








UNCOV

2545



    if (pBlockInfo->rows > 0) {





×








UNCOV

2546



      return pInfo->pRes;




×







2547



    }










2548













UNCOV

2549



    if (pInfo->pUpdateDataRes->info.rows > 0) {





×








UNCOV

2550



      goto FETCH_NEXT_BLOCK;




×







2551



    }










2552












2553



    goto NEXT_SUBMIT_BLK;




×








UNCOV

2554



  } else if (pInfo->blockType == STREAM_INPUT__CHECKPOINT) {





×








UNCOV

2555



    if (pInfo->validBlockIndex >= total) {





×







2556



      doClearBufferedBlocks(pInfo);











UNCOV

2557



      return NULL;




×







2558



    }










2559













UNCOV

2560



    int32_t  current = pInfo->validBlockIndex++;




×








UNCOV

2561



    qDebug("process %d/%d input data blocks, %s", current, (int32_t) total, id);





×







2562













UNCOV

2563



    SPackedData* pData = taosArrayGet(pInfo->pBlockLists, current);




×








UNCOV

2564



    SSDataBlock* pBlock = taosArrayGet(pData->pDataBlock, 0);




×







2565













UNCOV

2566



    if (pBlock->info.type == STREAM_CHECKPOINT) {





×








UNCOV

2567



      streamScanOperatorSaveCheckpoint(pInfo);




×







2568



    }










2569



    // printDataBlock(pBlock, "stream scan ck");











UNCOV

2570



    return pInfo->pCheckpointRes;




×







2571



  }










2572












2573



  return NULL;




×







2574



}










2575












2576



static SArray* extractTableIdList(const STableListInfo* pTableListInfo) {




12✔





2577



  SArray* tableIdList = taosArrayInit(4, sizeof(uint64_t));




12✔





2578












2579



  // Transfer the Array of STableKeyInfo into uid list.










2580



  size_t size = tableListGetSize(pTableListInfo);




12✔





2581



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





6,012✔





2582



    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);




6,000✔





2583



    taosArrayPush(tableIdList, &pkeyInfo->uid);




6,000✔





2584



  }










2585












2586



  return tableIdList;




12✔





2587



}










2588












2589



static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {




×







2590



  // NOTE: this operator does never check if current status is done or not










2591



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×







2592



  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;




×







2593












2594



  SStreamRawScanInfo* pInfo = pOperator->info;




×







2595



  int32_t             code = TSDB_CODE_SUCCESS;




×







2596



  pTaskInfo->streamInfo.metaRsp.metaRspLen = 0;  // use metaRspLen !=0 to judge if data is meta




×







2597



  pTaskInfo->streamInfo.metaRsp.metaRsp = NULL;




×







2598












2599



  qDebug("tmqsnap doRawScan called");





×







2600



  if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_DATA) {





×







2601



    bool hasNext = false;




×







2602



    if (pInfo->dataReader && pInfo->sContext->withMeta != ONLY_META) {





×







2603



      code = pAPI->tsdReader.tsdNextDataBlock(pInfo->dataReader, &hasNext);




×







2604



      if (code) {





×







2605



        pAPI->tsdReader.tsdReaderReleaseDataBlock(pInfo->dataReader);




×







2606



        T_LONG_JMP(pTaskInfo->env, code);





×







2607



      }










2608



    }










2609












2610



    if (pInfo->dataReader && hasNext) {





×







2611



      if (isTaskKilled(pTaskInfo)) {





×







2612



        pAPI->tsdReader.tsdReaderReleaseDataBlock(pInfo->dataReader);




×







2613



        T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);





×







2614



      }










2615












2616



      SSDataBlock* pBlock = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pInfo->dataReader, NULL);




×







2617



      if (pBlock == NULL) {





×







2618



        T_LONG_JMP(pTaskInfo->env, terrno);





×







2619



      }










2620












2621



      qDebug("tmqsnap doRawScan get data uid:%" PRId64 "", pBlock->info.id.uid);





×







2622



      tqOffsetResetToData(&pTaskInfo->streamInfo.currentOffset, pBlock->info.id.uid, pBlock->info.window.ekey);




×







2623



      return pBlock;




×







2624



    }










2625












2626



    SMetaTableInfo mtInfo = pAPI->snapshotFn.getMetaTableInfoFromSnapshot(pInfo->sContext);




×







2627



    STqOffsetVal   offset = {0};




×







2628



    if (mtInfo.uid == 0 || pInfo->sContext->withMeta == ONLY_META) {  // read snapshot done, change to get data from wal





×







2629



      qDebug("tmqsnap read snapshot done, change to get data from wal");





×







2630



      tqOffsetResetToLog(&offset, pInfo->sContext->snapVersion + 1);




×







2631



    } else {










2632



      tqOffsetResetToData(&offset, mtInfo.uid, INT64_MIN);




×







2633



      qDebug("tmqsnap change get data uid:%" PRId64 "", mtInfo.uid);





×







2634



    }










2635



    qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);




×







2636



    tDeleteSchemaWrapper(mtInfo.schema);





×







2637



    return NULL;




×







2638



  } else if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_META) {





×







2639



    SSnapContext* sContext = pInfo->sContext;




×







2640



    void*         data = NULL;




×







2641



    int32_t       dataLen = 0;




×







2642



    int16_t       type = 0;




×







2643



    int64_t       uid = 0;




×







2644



    if (pAPI->snapshotFn.getTableInfoFromSnapshot(sContext, &data, &dataLen, &type, &uid) < 0) {





×







2645



      qError("tmqsnap getTableInfoFromSnapshot error");





×







2646



      taosMemoryFreeClear(data);





×







2647



      return NULL;




×







2648



    }










2649












2650



    if (!sContext->queryMeta) {  // change to get data next poll request





×







2651



      STqOffsetVal offset = {0};




×







2652



      tqOffsetResetToData(&offset, 0, INT64_MIN);










2653



      qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);




×







2654



    } else {










2655



      tqOffsetResetToMeta(&pTaskInfo->streamInfo.currentOffset, uid);




×







2656



      pTaskInfo->streamInfo.metaRsp.resMsgType = type;




×







2657



      pTaskInfo->streamInfo.metaRsp.metaRspLen = dataLen;




×







2658



      pTaskInfo->streamInfo.metaRsp.metaRsp = data;




×







2659



    }










2660












2661



    return NULL;




×







2662



  }










2663



  return NULL;




×







2664



}










2665













UNCOV

2666



static void destroyRawScanOperatorInfo(void* param) {




×








UNCOV

2667



  SStreamRawScanInfo* pRawScan = (SStreamRawScanInfo*)param;




×








UNCOV

2668



  pRawScan->pAPI->tsdReader.tsdReaderClose(pRawScan->dataReader);




×








UNCOV

2669



  pRawScan->pAPI->snapshotFn.destroySnapshot(pRawScan->sContext);




×








UNCOV

2670



  tableListDestroy(pRawScan->pTableListInfo);




×








UNCOV

2671



  taosMemoryFree(pRawScan);




×








UNCOV

2672



}




×







2673












2674



// for subscribing db or stb (not including column),










2675



// if this scan is used, meta data can be return










2676



// and schemas are decided when scanning











UNCOV

2677



SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pTaskInfo) {




×







2678



  // create operator










2679



  // create tb reader










2680



  // create meta reader










2681



  // create tq reader










2682













UNCOV

2683



  int32_t code = TSDB_CODE_SUCCESS;




×







2684













UNCOV

2685



  SStreamRawScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamRawScanInfo));




×








UNCOV

2686



  SOperatorInfo*      pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));




×








UNCOV

2687



  if (pInfo == NULL || pOperator == NULL) {





×







2688



    code = TSDB_CODE_OUT_OF_MEMORY;




×







2689



    goto _end;




×







2690



  }










2691













UNCOV

2692



  pInfo->pTableListInfo = tableListCreate();




×








UNCOV

2693



  pInfo->vnode = pHandle->vnode;




×








UNCOV

2694



  pInfo->pAPI = &pTaskInfo->storageAPI;




×







2695













UNCOV

2696



  pInfo->sContext = pHandle->sContext;




×








UNCOV

2697



  setOperatorInfo(pOperator, "RawScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, false, OP_NOT_OPENED, pInfo,




×







2698



                  pTaskInfo);










2699













UNCOV

2700



  pOperator->fpSet = createOperatorFpSet(NULL, doRawScan, NULL, destroyRawScanOperatorInfo, optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);




×








UNCOV

2701



  return pOperator;




×







2702












2703



_end:




×







2704



  taosMemoryFree(pInfo);




×







2705



  taosMemoryFree(pOperator);




×







2706



  pTaskInfo->code = code;




×







2707



  return NULL;




×







2708



}










2709












2710



static void destroyStreamScanOperatorInfo(void* param) {




13✔





2711



  SStreamScanInfo* pStreamScan = (SStreamScanInfo*)param;




13✔





2712












2713



  if (pStreamScan->pTableScanOp && pStreamScan->pTableScanOp->info) {





13!





2714



    destroyOperator(pStreamScan->pTableScanOp);




12✔





2715



  }










2716












2717



  if (pStreamScan->tqReader) {





13✔





2718



    pStreamScan->readerFn.tqReaderClose(pStreamScan->tqReader);




12✔





2719



  }










2720



  if (pStreamScan->matchInfo.pList) {





13!





2721



    taosArrayDestroy(pStreamScan->matchInfo.pList);




13✔





2722



  }










2723



  if (pStreamScan->pPseudoExpr) {





13!






UNCOV

2724



    destroyExprInfo(pStreamScan->pPseudoExpr, pStreamScan->numOfPseudoExpr);




×








UNCOV

2725



    taosMemoryFree(pStreamScan->pPseudoExpr);




×







2726



  }










2727












2728



  cleanupExprSupp(&pStreamScan->tbnameCalSup);




13✔





2729



  cleanupExprSupp(&pStreamScan->tagCalSup);




13✔





2730












2731



  pStreamScan->stateStore.updateInfoDestroy(pStreamScan->pUpdateInfo);




13✔





2732



  blockDataDestroy(pStreamScan->pRes);




13✔





2733



  blockDataDestroy(pStreamScan->pUpdateRes);




13✔





2734



  blockDataDestroy(pStreamScan->pDeleteDataRes);




13✔





2735



  blockDataDestroy(pStreamScan->pUpdateDataRes);




13✔





2736



  blockDataDestroy(pStreamScan->pCreateTbRes);




13✔





2737



  taosArrayDestroy(pStreamScan->pBlockLists);




13✔





2738



  blockDataDestroy(pStreamScan->pCheckpointRes);




13✔





2739












2740



  taosMemoryFree(pStreamScan);




13✔





2741



}




13✔





2742













UNCOV

2743



void streamScanReleaseState(SOperatorInfo* pOperator) {




×








UNCOV

2744



  SStreamScanInfo* pInfo = pOperator->info;




×








UNCOV

2745



  if (!pInfo->pState) {





×







2746



    return;




×







2747



  }











UNCOV

2748



  if (!pInfo->pUpdateInfo) {





×








UNCOV

2749



    return;




×







2750



  }











UNCOV

2751



  int32_t len = pInfo->stateStore.updateInfoSerialize(NULL, 0, pInfo->pUpdateInfo);




×








UNCOV

2752



  void* pBuff = taosMemoryCalloc(1, len);




×








UNCOV

2753



  pInfo->stateStore.updateInfoSerialize(pBuff, len, pInfo->pUpdateInfo);




×








UNCOV

2754



  pInfo->stateStore.streamStateSaveInfo(pInfo->pState, STREAM_SCAN_OP_STATE_NAME, strlen(STREAM_SCAN_OP_STATE_NAME), pBuff, len);




×








UNCOV

2755



  taosMemoryFree(pBuff);




×







2756



}










2757













UNCOV

2758



void streamScanReloadState(SOperatorInfo* pOperator) {




×








UNCOV

2759



  SStreamScanInfo* pInfo = pOperator->info;




×








UNCOV

2760



  if (!pInfo->pState) {





×







2761



    return;




×







2762



  }











UNCOV

2763



  void*   pBuff = NULL;




×








UNCOV

2764



  int32_t len = 0;




×








UNCOV

2765



  pInfo->stateStore.streamStateGetInfo(pInfo->pState, STREAM_SCAN_OP_STATE_NAME, strlen(STREAM_SCAN_OP_STATE_NAME), &pBuff, &len);




×








UNCOV

2766



  SUpdateInfo* pUpInfo = taosMemoryCalloc(1, sizeof(SUpdateInfo));




×








UNCOV

2767



  int32_t      code = pInfo->stateStore.updateInfoDeserialize(pBuff, len, pUpInfo);




×








UNCOV

2768



  taosMemoryFree(pBuff);




×








UNCOV

2769



  if (code == TSDB_CODE_SUCCESS && pInfo->pUpdateInfo) {





×








UNCOV

2770



    if (pInfo->pUpdateInfo->minTS < 0) {





×








UNCOV

2771



      pInfo->stateStore.updateInfoDestroy(pInfo->pUpdateInfo);




×








UNCOV

2772



      pInfo->pUpdateInfo = pUpInfo;




×







2773



    } else {











UNCOV

2774



      pInfo->stateStore.windowSBfDelete(pInfo->pUpdateInfo, 1);




×








UNCOV

2775



      pInfo->stateStore.windowSBfAdd(pInfo->pUpdateInfo, 1);




×








UNCOV

2776



      ASSERT(pInfo->pUpdateInfo->minTS > pUpInfo->minTS);





×








UNCOV

2777



      pInfo->pUpdateInfo->maxDataVersion = TMAX(pInfo->pUpdateInfo->maxDataVersion, pUpInfo->maxDataVersion);




×








UNCOV

2778



      SHashObj* curMap = pInfo->pUpdateInfo->pMap;




×








UNCOV

2779



      void *pIte = taosHashIterate(curMap, NULL);




×








UNCOV

2780



      while (pIte != NULL) {





×








UNCOV

2781



        size_t keySize = 0;




×








UNCOV

2782



        int64_t* pUid = taosHashGetKey(pIte, &keySize);




×








UNCOV

2783



        taosHashPut(pUpInfo->pMap, pUid, sizeof(int64_t), pIte, sizeof(TSKEY));




×








UNCOV

2784



        pIte = taosHashIterate(curMap, pIte);




×







2785



      }











UNCOV

2786



      taosHashCleanup(curMap);




×








UNCOV

2787



      pInfo->pUpdateInfo->pMap = pUpInfo->pMap;




×








UNCOV

2788



      pUpInfo->pMap = NULL;




×








UNCOV

2789



      pInfo->stateStore.updateInfoDestroy(pUpInfo);




×







2790



    }










2791



  } else {











UNCOV

2792



    pInfo->stateStore.updateInfoDestroy(pUpInfo);




×







2793



  }










2794



}










2795












2796



SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond,




13✔





2797



                                            STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo) {










2798



  SArray*          pColIds = NULL;




13✔





2799



  SStreamScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamScanInfo));




13✔





2800



  SOperatorInfo*   pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));




13✔





2801



  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;




13✔





2802



  const char* idstr = pTaskInfo->id.str;




13✔





2803












2804



  if (pInfo == NULL || pOperator == NULL) {





13!





2805



    terrno = TSDB_CODE_OUT_OF_MEMORY;




×







2806



    tableListDestroy(pTableListInfo);




×







2807



    goto _error;




×







2808



  }










2809












2810



  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;




13✔





2811



  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;




13✔





2812












2813



  pInfo->pTagCond = pTagCond;




13✔





2814



  pInfo->pGroupTags = pTableScanNode->pGroupTags;




13✔





2815












2816



  int32_t numOfCols = 0;




13✔





2817



  int32_t code =










2818



      extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);




13✔





2819



  if (code != TSDB_CODE_SUCCESS) {





13!





2820



    tableListDestroy(pTableListInfo);




×







2821



    goto _error;




×







2822



  }










2823












2824



  int32_t numOfOutput = taosArrayGetSize(pInfo->matchInfo.pList);




13✔





2825



  pColIds = taosArrayInit(numOfOutput, sizeof(int16_t));




13✔





2826



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





40✔





2827



    SColMatchItem* id = taosArrayGet(pInfo->matchInfo.pList, i);




27✔





2828












2829



    int16_t colId = id->colId;




27✔





2830



    taosArrayPush(pColIds, &colId);










2831



    if (id->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {





27✔





2832



      pInfo->primaryTsIndex = id->dstSlotId;




12✔





2833



    }










2834



  }










2835












2836



  if (pTableScanNode->pSubtable != NULL) {





13!






UNCOV

2837



    SExprInfo* pSubTableExpr = taosMemoryCalloc(1, sizeof(SExprInfo));




×








UNCOV

2838



    if (pSubTableExpr == NULL) {





×







2839



      terrno = TSDB_CODE_OUT_OF_MEMORY;




×







2840



      tableListDestroy(pTableListInfo);




×







2841



      goto _error;




×







2842



    }










2843













UNCOV

2844



    pInfo->tbnameCalSup.pExprInfo = pSubTableExpr;




×








UNCOV

2845



    createExprFromOneNode(pSubTableExpr, pTableScanNode->pSubtable, 0);




×








UNCOV

2846



    if (initExprSupp(&pInfo->tbnameCalSup, pSubTableExpr, 1, &pTaskInfo->storageAPI.functionStore) != 0) {





×







2847



      tableListDestroy(pTableListInfo);




×







2848



      goto _error;




×







2849



    }










2850



  }










2851












2852



  if (pTableScanNode->pTags != NULL) {





13!





2853



    int32_t    numOfTags;











UNCOV

2854



    SExprInfo* pTagExpr = createExpr(pTableScanNode->pTags, &numOfTags);




×








UNCOV

2855



    if (pTagExpr == NULL) {





×







2856



      terrno = TSDB_CODE_OUT_OF_MEMORY;




×







2857



      tableListDestroy(pTableListInfo);




×







2858



      goto _error;




×







2859



    }











UNCOV

2860



    if (initExprSupp(&pInfo->tagCalSup, pTagExpr, numOfTags, &pTaskInfo->storageAPI.functionStore) != 0) {





×







2861



      terrno = TSDB_CODE_OUT_OF_MEMORY;




×







2862



      tableListDestroy(pTableListInfo);




×







2863



      goto _error;




×







2864



    }










2865



  }










2866












2867



  pInfo->pBlockLists = taosArrayInit(4, sizeof(SPackedData));




13✔





2868



  if (pInfo->pBlockLists == NULL) {





13!






UNCOV

2869



    terrno = TSDB_CODE_OUT_OF_MEMORY;




×







2870



    tableListDestroy(pTableListInfo);




×







2871



    goto _error;




×







2872



  }










2873












2874



  if (pHandle->vnode) {





13✔





2875



    SOperatorInfo*  pTableScanOp = createTableScanOperatorInfo(pTableScanNode, pHandle, pTableListInfo, pTaskInfo);




12✔





2876



    if (pTableScanOp == NULL) {





12!





2877



      qError("createTableScanOperatorInfo error, errorcode: %d", pTaskInfo->code);





×







2878



      goto _error;




×







2879



    }










2880



    STableScanInfo* pTSInfo = (STableScanInfo*)pTableScanOp->info;




12✔





2881



    if (pHandle->version > 0) {





12!






UNCOV

2882



      pTSInfo->base.cond.endVersion = pHandle->version;




×







2883



    }










2884












2885



    STableKeyInfo* pList = NULL;




12✔





2886



    int32_t        num = 0;




12✔





2887



    tableListGetGroupList(pTableListInfo, 0, &pList, &num);




12✔





2888












2889



    if (pHandle->initTableReader) {





12!






UNCOV

2890



      pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;




×








UNCOV

2891



      pTSInfo->base.dataReader = NULL;




×







2892



    }










2893












2894



    if (pHandle->initTqReader) {





12!





2895



      ASSERT(pHandle->tqReader == NULL);





12!





2896



      pInfo->tqReader = pAPI->tqReaderFn.tqReaderOpen(pHandle->vnode);




12✔





2897



      ASSERT(pInfo->tqReader);





12!





2898



    } else {










2899



      ASSERT(pHandle->tqReader);





×







2900



      pInfo->tqReader = pHandle->tqReader;




×







2901



    }










2902












2903



    pInfo->pUpdateInfo = NULL;




12✔





2904



    pInfo->pTableScanOp = pTableScanOp;




12✔





2905



    if (pInfo->pTableScanOp->pTaskInfo->streamInfo.pState) {





12!





2906



      pAPI->stateStore.streamStateSetNumber(pInfo->pTableScanOp->pTaskInfo->streamInfo.pState, -1);




12✔





2907



    }










2908












2909



    pInfo->readHandle = *pHandle;




12✔





2910



    pTaskInfo->streamInfo.snapshotVer = pHandle->version;




12✔





2911



    pInfo->pCreateTbRes = buildCreateTableBlock(&pInfo->tbnameCalSup, &pInfo->tagCalSup);




12✔





2912



    blockDataEnsureCapacity(pInfo->pCreateTbRes, 8);




12✔





2913












2914



    // set the extract column id to streamHandle










2915



    pAPI->tqReaderFn.tqReaderSetColIdList(pInfo->tqReader, pColIds);




12✔





2916



    SArray* tableIdList = extractTableIdList(((STableScanInfo*)(pInfo->pTableScanOp->info))->base.pTableListInfo);




12✔





2917



    code = pAPI->tqReaderFn.tqReaderSetQueryTableList(pInfo->tqReader, tableIdList, idstr);




12✔





2918



    if (code != 0) {





12!





2919



      taosArrayDestroy(tableIdList);




×







2920



      goto _error;




×







2921



    }










2922












2923



    taosArrayDestroy(tableIdList);




12✔





2924



    memcpy(&pTaskInfo->streamInfo.tableCond, &pTSInfo->base.cond, sizeof(SQueryTableDataCond));




12✔





2925



  } else {










2926



    taosArrayDestroy(pColIds);




1✔





2927



    tableListDestroy(pTableListInfo);




1✔





2928



    pColIds = NULL;




1✔





2929



  }










2930












2931



  // create the pseduo columns info










2932



  if (pTableScanNode->scan.pScanPseudoCols != NULL) {





13!






UNCOV

2933



    pInfo->pPseudoExpr = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pInfo->numOfPseudoExpr);




×







2934



  }










2935












2936



  code = filterInitFromNode((SNode*)pScanPhyNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);




13✔





2937



  if (code != TSDB_CODE_SUCCESS) {





13!





2938



    goto _error;




×







2939



  }










2940












2941



  pInfo->pRes = createDataBlockFromDescNode(pDescNode);




13✔





2942



  pInfo->pUpdateRes = createSpecialDataBlock(STREAM_CLEAR);




13✔





2943



  pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;




13✔





2944



  pInfo->windowSup = (SWindowSupporter){.pStreamAggSup = NULL, .gap = -1, .parentType = QUERY_NODE_PHYSICAL_PLAN};




13✔





2945



  pInfo->groupId = 0;




13✔





2946



  pInfo->pStreamScanOp = pOperator;




13✔





2947



  pInfo->deleteDataIndex = 0;




13✔





2948



  pInfo->pDeleteDataRes = createSpecialDataBlock(STREAM_DELETE_DATA);




13✔





2949



  pInfo->updateWin = (STimeWindow){.skey = INT64_MAX, .ekey = INT64_MAX};




13✔





2950



  pInfo->pUpdateDataRes = createSpecialDataBlock(STREAM_CLEAR);




13✔





2951



  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;




13✔





2952



  pInfo->partitionSup.needCalc = false;




13✔





2953



  pInfo->igCheckUpdate = pTableScanNode->igCheckUpdate;




13✔





2954



  pInfo->igExpired = pTableScanNode->igExpired;




13✔





2955



  pInfo->twAggSup.maxTs = INT64_MIN;




13✔





2956



  pInfo->pState = pTaskInfo->streamInfo.pState;




13✔





2957



  pInfo->stateStore = pTaskInfo->storageAPI.stateStore;




13✔





2958



  pInfo->readerFn = pTaskInfo->storageAPI.tqReaderFn;




13✔





2959



  pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT);




13✔





2960












2961



  // for stream










2962



  if (pTaskInfo->streamInfo.pState) {





13!





2963



    void*   buff = NULL;




13✔





2964



    int32_t len = 0;




13✔





2965



    int32_t res = pAPI->stateStore.streamStateGetInfo(pTaskInfo->streamInfo.pState, STREAM_SCAN_OP_CHECKPOINT_NAME, strlen(STREAM_SCAN_OP_CHECKPOINT_NAME), &buff, &len);




13✔





2966



    if (res == TSDB_CODE_SUCCESS) {





13✔





2967



      streamScanOperatorDecode(buff, len, pInfo);




7✔





2968



      taosMemoryFree(buff);




7✔





2969



    }










2970



  }










2971












2972



  setOperatorInfo(pOperator, STREAM_SCAN_OP_NAME, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, false, OP_NOT_OPENED, pInfo,




13✔





2973



                  pTaskInfo);










2974



  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);




13✔





2975












2976



  __optr_fn_t nextFn = (pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM) ? doStreamScan : doQueueScan;





13!





2977



  pOperator->fpSet =










2978



      createOperatorFpSet(optrDummyOpenFn, nextFn, NULL, destroyStreamScanOperatorInfo, optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);




13✔





2979



  setOperatorStreamStateFn(pOperator, streamScanReleaseState, streamScanReloadState);




13✔





2980












2981



  return pOperator;




13✔





2982












2983



_error:




×







2984



  if (pColIds != NULL) {





×







2985



    taosArrayDestroy(pColIds);




×







2986



  }










2987












2988



  if (pInfo != NULL) {





×







2989



    destroyStreamScanOperatorInfo(pInfo);




×







2990



  }










2991












2992



  taosMemoryFreeClear(pOperator);





×







2993



  return NULL;




×







2994



}










2995













UNCOV

2996



static void doTagScanOneTable(SOperatorInfo* pOperator, const SSDataBlock* pRes, int32_t count, SMetaReader* mr, SStorageAPI* pAPI) {




×








UNCOV

2997



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

2998



  STagScanInfo* pInfo = pOperator->info;




×








UNCOV

2999



  SExprInfo*    pExprInfo = &pOperator->exprSupp.pExprInfo[0];




×







3000













UNCOV

3001



  STableKeyInfo* item = tableListGetInfo(pInfo->pTableListInfo, pInfo->curPos);




×








UNCOV

3002



  int32_t        code = pAPI->metaReaderFn.getTableEntryByUid(mr, item->uid);




×








UNCOV

3003



  tDecoderClear(&(*mr).coder);




×








UNCOV

3004



  if (code != TSDB_CODE_SUCCESS) {





×







3005



    qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", item->uid, tstrerror(terrno),





×







3006



           GET_TASKID(pTaskInfo));










3007



    pAPI->metaReaderFn.clearReader(mr);




×







3008



    T_LONG_JMP(pTaskInfo->env, terrno);





×







3009



  }










3010












3011



  char str[512];











UNCOV

3012



  for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {





×








UNCOV

3013



    SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pExprInfo[j].base.resSchema.slotId);




×







3014












3015



    // refactor later











UNCOV

3016



    if (FUNCTION_TYPE_TBNAME == pExprInfo[j].pExpr->_function.functionType) {





×








UNCOV

3017



      STR_TO_VARSTR(str, (*mr).me.name);




×








UNCOV

3018



      colDataSetVal(pDst, (count), str, false);




×








UNCOV

3019



    } else if (FUNCTION_TYPE_TBUID == pExprInfo[j].pExpr->_function.functionType) {





×







3020



      colDataSetVal(pDst, (count), (char*)&(*mr).me.uid, false);    




×








UNCOV

3021



    } else if (FUNCTION_TYPE_VGID == pExprInfo[j].pExpr->_function.functionType) {





×







3022



      colDataSetVal(pDst, (count), (char*)&pTaskInfo->id.vgId, false);    




×







3023



    } else {  // it is a tag value











UNCOV

3024



      STagVal val = {0};




×








UNCOV

3025



      val.cid = pExprInfo[j].base.pParam[0].pCol->colId;




×








UNCOV

3026



      const char* p = pAPI->metaFn.extractTagVal((*mr).me.ctbEntry.pTags, pDst->info.type, &val);




×







3027













UNCOV

3028



      char* data = NULL;




×








UNCOV

3029



      if (pDst->info.type != TSDB_DATA_TYPE_JSON && p != NULL) {





×








UNCOV

3030



        data = tTagValToData((const STagVal*)p, false);




×







3031



      } else {










3032



        data = (char*)p;




×







3033



      }











UNCOV

3034



      colDataSetVal(pDst, (count), data,




×








UNCOV

3035



                    (data == NULL) || (pDst->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data)));





×







3036













UNCOV

3037



      if (pDst->info.type != TSDB_DATA_TYPE_JSON && p != NULL && IS_VAR_DATA_TYPE(((const STagVal*)p)->type) &&





×







3038



          data != NULL) {











UNCOV

3039



        taosMemoryFree(data);




×







3040



      }










3041



    }










3042



  }











UNCOV

3043



}




×







3044













UNCOV

3045



static void tagScanFreeUidTag(void* p) {




×








UNCOV

3046



  STUidTagInfo* pInfo = p;




×








UNCOV

3047



  if (pInfo->pTagVal != NULL) {





×








UNCOV

3048



    taosMemoryFree(pInfo->pTagVal);




×







3049



  }











UNCOV

3050



}




×







3051













UNCOV

3052



static int32_t tagScanCreateResultData(SDataType* pType, int32_t numOfRows, SScalarParam* pParam) {




×








UNCOV

3053



  SColumnInfoData* pColumnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));




×








UNCOV

3054



  if (pColumnData == NULL) {





×







3055



    terrno = TSDB_CODE_OUT_OF_MEMORY;




×







3056



    return terrno;




×







3057



  }










3058













UNCOV

3059



  pColumnData->info.type = pType->type;




×








UNCOV

3060



  pColumnData->info.bytes = pType->bytes;




×








UNCOV

3061



  pColumnData->info.scale = pType->scale;




×








UNCOV

3062



  pColumnData->info.precision = pType->precision;




×







3063













UNCOV

3064



  int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows, true);




×








UNCOV

3065



  if (code != TSDB_CODE_SUCCESS) {





×







3066



    terrno = code;




×







3067



    taosMemoryFree(pColumnData);




×







3068



    return terrno;




×







3069



  }










3070













UNCOV

3071



  pParam->columnData = pColumnData;




×








UNCOV

3072



  pParam->colAlloced = true;




×








UNCOV

3073



  return TSDB_CODE_SUCCESS;




×







3074



}










3075













UNCOV

3076



static EDealRes tagScanRewriteTagColumn(SNode** pNode, void* pContext) {




×








UNCOV

3077



  SColumnNode* pSColumnNode = NULL;




×








UNCOV

3078



  if (QUERY_NODE_COLUMN == nodeType((*pNode))) {





×








UNCOV

3079



    pSColumnNode = *(SColumnNode**)pNode;




×








UNCOV

3080



  } else if (QUERY_NODE_FUNCTION == nodeType((*pNode))) {





×







3081



    SFunctionNode* pFuncNode = *(SFunctionNode**)(pNode);




×







3082



    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {





×







3083



      pSColumnNode = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);




×







3084



      if (NULL == pSColumnNode) {





×







3085



        return DEAL_RES_ERROR;




×







3086



      }










3087



      pSColumnNode->colId = -1;




×







3088



      pSColumnNode->colType = COLUMN_TYPE_TBNAME;




×







3089



      pSColumnNode->node.resType.type = TSDB_DATA_TYPE_VARCHAR;




×







3090



      pSColumnNode->node.resType.bytes = TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE;




×







3091



      nodesDestroyNode(*pNode);




×







3092



      *pNode = (SNode*)pSColumnNode;




×







3093



    } else {










3094



      return DEAL_RES_CONTINUE;




×







3095



    }










3096



  } else {











UNCOV

3097



    return DEAL_RES_CONTINUE;




×







3098



  }










3099













UNCOV

3100



  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;




×








UNCOV

3101



  void*            data = taosHashGet(pCtx->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId));




×








UNCOV

3102



  if (!data) {





×








UNCOV

3103



    taosHashPut(pCtx->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId), pNode, sizeof((*pNode)));




×








UNCOV

3104



    pSColumnNode->slotId = pCtx->index++;




×








UNCOV

3105



    SColumnInfo cInfo = {.colId = pSColumnNode->colId,




×








UNCOV

3106



                         .type = pSColumnNode->node.resType.type,




×








UNCOV

3107



                         .bytes = pSColumnNode->node.resType.bytes};




×








UNCOV

3108



    taosArrayPush(pCtx->cInfoList, &cInfo);




×







3109



  } else {










3110



    SColumnNode* col = *(SColumnNode**)data;




×







3111



    pSColumnNode->slotId = col->slotId;




×







3112



  }










3113













UNCOV

3114



  return DEAL_RES_CONTINUE;




×







3115



}










3116












3117













UNCOV

3118



static int32_t tagScanFilterByTagCond(SArray* aUidTags, SNode* pTagCond, SArray* aFilterIdxs, void* pVnode, SStorageAPI* pAPI, STagScanInfo* pInfo) {




×








UNCOV

3119



  int32_t code = 0;




×








UNCOV

3120



  int32_t numOfTables = taosArrayGetSize(aUidTags);




×







3121













UNCOV

3122



  SSDataBlock* pResBlock = createTagValBlockForFilter(pInfo->filterCtx.cInfoList, numOfTables, aUidTags, pVnode, pAPI);




×







3123













UNCOV

3124



  SArray* pBlockList = taosArrayInit(1, POINTER_BYTES);




×







3125



  taosArrayPush(pBlockList, &pResBlock);











UNCOV

3126



  SDataType type = {.type = TSDB_DATA_TYPE_BOOL, .bytes = sizeof(bool)};




×







3127













UNCOV

3128



  SScalarParam output = {0};




×








UNCOV

3129



  code = tagScanCreateResultData(&type, numOfTables, &output);




×








UNCOV

3130



  if (TSDB_CODE_SUCCESS != code) {





×







3131



    return code;




×







3132



  }










3133













UNCOV

3134



  code = scalarCalculate(pTagCond, pBlockList, &output);




×








UNCOV

3135



  if (TSDB_CODE_SUCCESS != code) {





×







3136



    return code;




×







3137



  }










3138













UNCOV

3139



  bool* result = (bool*)output.columnData->pData;




×








UNCOV

3140



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





×








UNCOV

3141



    if (result[i]) {





×







3142



      taosArrayPush(aFilterIdxs, &i);










3143



    }










3144



  }










3145













UNCOV

3146



  colDataDestroy(output.columnData);




×








UNCOV

3147



  taosMemoryFreeClear(output.columnData);





×







3148













UNCOV

3149



  blockDataDestroy(pResBlock);




×








UNCOV

3150



  taosArrayDestroy(pBlockList);




×







3151













UNCOV

3152



  return TSDB_CODE_SUCCESS;




×







3153



}










3154













UNCOV

3155



static void tagScanFillOneCellWithTag(SOperatorInfo* pOperator, const STUidTagInfo* pUidTagInfo, SExprInfo* pExprInfo, SColumnInfoData* pColInfo, int rowIndex, const SStorageAPI* pAPI, void* pVnode) {




×








UNCOV

3156



  if (QUERY_NODE_FUNCTION == pExprInfo->pExpr->nodeType) {





×








UNCOV

3157



    if (FUNCTION_TYPE_TBNAME == pExprInfo->pExpr->_function.functionType) {  // tbname





×







3158



      char str[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};




×







3159



      STR_TO_VARSTR(str, "ctbidx");




×







3160












3161



      colDataSetVal(pColInfo, rowIndex, str, false);




×








UNCOV

3162



    } else if (FUNCTION_TYPE_TBUID == pExprInfo->pExpr->_function.functionType) {





×








UNCOV

3163



      colDataSetVal(pColInfo, rowIndex, (char*)&pUidTagInfo->uid, false);    




×








UNCOV

3164



    } else if (FUNCTION_TYPE_VGID == pExprInfo->pExpr->_function.functionType) {





×








UNCOV

3165



      colDataSetVal(pColInfo, rowIndex, (char*)&pOperator->pTaskInfo->id.vgId, false);   




×







3166



    } 










3167



  } else {











UNCOV

3168



    STagVal tagVal = {0};




×








UNCOV

3169



    tagVal.cid = pExprInfo->base.pParam[0].pCol->colId;




×








UNCOV

3170



    if (pUidTagInfo->pTagVal == NULL) {





×







3171



      colDataSetNULL(pColInfo, rowIndex);





×







3172



    } else {











UNCOV

3173



      const char* p = pAPI->metaFn.extractTagVal(pUidTagInfo->pTagVal, pColInfo->info.type, &tagVal);




×







3174













UNCOV

3175



      if (p == NULL || (pColInfo->info.type == TSDB_DATA_TYPE_JSON && ((STag*)p)->nTag == 0)) {





×







3176



        colDataSetNULL(pColInfo, rowIndex);





×








UNCOV

3177



      } else if (pColInfo->info.type == TSDB_DATA_TYPE_JSON) {





×







3178



        colDataSetVal(pColInfo, rowIndex, p, false);




×








UNCOV

3179



      } else if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {





×








UNCOV

3180



        char* tmp = taosMemoryMalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1);




×








UNCOV

3181



        varDataSetLen(tmp, tagVal.nData);




×








UNCOV

3182



        memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData);




×








UNCOV

3183



        colDataSetVal(pColInfo, rowIndex, tmp, false);




×








UNCOV

3184



        taosMemoryFree(tmp);




×







3185



      } else {











UNCOV

3186



        colDataSetVal(pColInfo, rowIndex, (const char*)&tagVal.i64, false);




×







3187



      }










3188



    }










3189



  }











UNCOV

3190



}




×







3191













UNCOV

3192



static int32_t tagScanFillResultBlock(SOperatorInfo* pOperator, SSDataBlock* pRes, SArray* aUidTags, SArray* aFilterIdxs, bool ignoreFilterIdx,




×







3193



                                      SStorageAPI* pAPI) {











UNCOV

3194



  STagScanInfo* pInfo = pOperator->info;




×








UNCOV

3195



  SExprInfo*    pExprInfo = &pOperator->exprSupp.pExprInfo[0];




×








UNCOV

3196



  if (!ignoreFilterIdx) {





×








UNCOV

3197



    size_t szTables = taosArrayGetSize(aFilterIdxs);




×








UNCOV

3198



    for (int i = 0; i < szTables; ++i) {





×








UNCOV

3199



      int32_t idx = *(int32_t*)taosArrayGet(aFilterIdxs, i);




×








UNCOV

3200



      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);




×








UNCOV

3201



      for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {





×








UNCOV

3202



        SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pExprInfo[j].base.resSchema.slotId);




×








UNCOV

3203



        tagScanFillOneCellWithTag(pOperator, pUidTagInfo, &pExprInfo[j], pDst, i, pAPI, pInfo->readHandle.vnode);




×







3204



      }










3205



    }










3206



  } else {











UNCOV

3207



    size_t szTables = taosArrayGetSize(aUidTags);




×








UNCOV

3208



    for (int i = 0; i < szTables; ++i) {





×








UNCOV

3209



      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);




×








UNCOV

3210



      for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {





×








UNCOV

3211



        SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pExprInfo[j].base.resSchema.slotId);




×








UNCOV

3212



        tagScanFillOneCellWithTag(pOperator, pUidTagInfo, &pExprInfo[j], pDst, i, pAPI, pInfo->readHandle.vnode);




×







3213



      }










3214



    }










3215



  }











UNCOV

3216



  return 0;




×







3217



}










3218













UNCOV

3219



static SSDataBlock* doTagScanFromCtbIdx(SOperatorInfo* pOperator) {




×








UNCOV

3220



  if (pOperator->status == OP_EXEC_DONE) {





×








UNCOV

3221



    return NULL;




×







3222



  }











UNCOV

3223



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

3224



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×







3225













UNCOV

3226



  STagScanInfo* pInfo = pOperator->info;




×








UNCOV

3227



  SSDataBlock*  pRes = pInfo->pRes;




×








UNCOV

3228



  blockDataCleanup(pRes);




×







3229













UNCOV

3230



  if (pInfo->pCtbCursor == NULL) {





×








UNCOV

3231



    pInfo->pCtbCursor = pAPI->metaFn.openCtbCursor(pInfo->readHandle.vnode, pInfo->suid, 1);




×







3232



  } else {










3233



    pAPI->metaFn.resumeCtbCursor(pInfo->pCtbCursor, 0);




×







3234



  }










3235













UNCOV

3236



  SArray* aUidTags = pInfo->aUidTags;




×








UNCOV

3237



  SArray* aFilterIdxs = pInfo->aFilterIdxs;




×








UNCOV

3238



  int32_t count = 0;




×








UNCOV

3239



  bool ctbCursorFinished = false;




×








UNCOV

3240



  while (1) {




×








UNCOV

3241



    taosArrayClearEx(aUidTags, tagScanFreeUidTag);




×








UNCOV

3242



    taosArrayClear(aFilterIdxs);




×







3243













UNCOV

3244



    int32_t numTables = 0;




×








UNCOV

3245



    while (numTables < pOperator->resultInfo.capacity) {





×








UNCOV

3246



      SMCtbCursor* pCur = pInfo->pCtbCursor;




×








UNCOV

3247



      tb_uid_t     uid = pAPI->metaFn.ctbCursorNext(pInfo->pCtbCursor);




×








UNCOV

3248



      if (uid == 0) {





×








UNCOV

3249



        ctbCursorFinished = true;




×








UNCOV

3250



        break;




×







3251



      }











UNCOV

3252



      STUidTagInfo info = {.uid = uid, .pTagVal = pCur->pVal};




×








UNCOV

3253



      info.pTagVal = taosMemoryMalloc(pCur->vLen);




×








UNCOV

3254



      memcpy(info.pTagVal, pCur->pVal, pCur->vLen);




×







3255



      taosArrayPush(aUidTags, &info);











UNCOV

3256



      ++numTables;




×







3257



    }










3258













UNCOV

3259



    if (numTables == 0) {





×








UNCOV

3260



      break;




×







3261



    }











UNCOV

3262



    bool ignoreFilterIdx = true;




×








UNCOV

3263



    if (pInfo->pTagCond != NULL) {





×








UNCOV

3264



      ignoreFilterIdx = false;




×








UNCOV

3265



      int32_t code = tagScanFilterByTagCond(aUidTags, pInfo->pTagCond, aFilterIdxs, pInfo->readHandle.vnode, pAPI, pInfo);




×








UNCOV

3266



      if (TSDB_CODE_SUCCESS != code) {





×







3267



        pOperator->pTaskInfo->code = code;




×







3268



        T_LONG_JMP(pOperator->pTaskInfo->env, code);





×







3269



      }










3270



    } else {











UNCOV

3271



      ignoreFilterIdx = true;




×







3272



    }










3273













UNCOV

3274



    tagScanFillResultBlock(pOperator, pRes, aUidTags, aFilterIdxs, ignoreFilterIdx, pAPI);




×







3275



    











UNCOV

3276



    count = ignoreFilterIdx ? taosArrayGetSize(aUidTags): taosArrayGetSize(aFilterIdxs);





×







3277













UNCOV

3278



    if (count != 0) {





×








UNCOV

3279



      break;




×







3280



    }










3281



  }










3282













UNCOV

3283



  if (count > 0) {





×








UNCOV

3284



    pAPI->metaFn.pauseCtbCursor(pInfo->pCtbCursor);




×







3285



  }











UNCOV

3286



  if (count == 0 || ctbCursorFinished) {





×








UNCOV

3287



    pAPI->metaFn.closeCtbCursor(pInfo->pCtbCursor);




×








UNCOV

3288



    pInfo->pCtbCursor = NULL;




×








UNCOV

3289



    setOperatorCompleted(pOperator);




×







3290



  }











UNCOV

3291



  pRes->info.rows = count;




×







3292













UNCOV

3293



  bool bLimitReached = applyLimitOffset(&pInfo->limitInfo, pRes, pTaskInfo);




×








UNCOV

3294



  if (bLimitReached) {





×







3295



    setOperatorCompleted(pOperator);




×







3296



  }











UNCOV

3297



  pOperator->resultInfo.totalRows += pRes->info.rows;




×








UNCOV

3298



  return (pRes->info.rows == 0) ? NULL : pInfo->pRes;





×







3299



}










3300













UNCOV

3301



static SSDataBlock* doTagScanFromMetaEntry(SOperatorInfo* pOperator) {




×








UNCOV

3302



  if (pOperator->status == OP_EXEC_DONE) {





×








UNCOV

3303



    return NULL;




×







3304



  }










3305













UNCOV

3306



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

3307



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×







3308













UNCOV

3309



  STagScanInfo* pInfo = pOperator->info;




×








UNCOV

3310



  SExprInfo*    pExprInfo = &pOperator->exprSupp.pExprInfo[0];




×








UNCOV

3311



  SSDataBlock*  pRes = pInfo->pRes;




×








UNCOV

3312



  blockDataCleanup(pRes);




×







3313













UNCOV

3314



  int32_t size = tableListGetSize(pInfo->pTableListInfo);




×








UNCOV

3315



  if (size == 0) {





×








UNCOV

3316



    setTaskStatus(pTaskInfo, TASK_COMPLETED);




×








UNCOV

3317



    return NULL;




×







3318



  }










3319













UNCOV

3320



  char        str[512] = {0};




×








UNCOV

3321



  int32_t     count = 0;




×








UNCOV

3322



  SMetaReader mr = {0};




×








UNCOV

3323



  pAPI->metaReaderFn.initReader(&mr, pInfo->readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);




×







3324













UNCOV

3325



  while (pInfo->curPos < size && count < pOperator->resultInfo.capacity) {





×








UNCOV

3326



    doTagScanOneTable(pOperator, pRes, count, &mr, &pTaskInfo->storageAPI);




×








UNCOV

3327



    ++count;




×








UNCOV

3328



    if (++pInfo->curPos >= size) {





×








UNCOV

3329



      setOperatorCompleted(pOperator);




×







3330



    }










3331



  }











UNCOV

3332



  pRes->info.rows = count;




×







3333













UNCOV

3334



  pAPI->metaReaderFn.clearReader(&mr);




×








UNCOV

3335



  bool bLimitReached = applyLimitOffset(&pInfo->limitInfo, pRes, pTaskInfo);




×








UNCOV

3336



  if (bLimitReached) {





×







3337



    setOperatorCompleted(pOperator);




×







3338



  }










3339



  // qDebug("QInfo:0x%"PRIx64" create tag values results completed, rows:%d", GET_TASKID(pRuntimeEnv), count);











UNCOV

3340



  if (pOperator->status == OP_EXEC_DONE) {





×








UNCOV

3341



    setTaskStatus(pTaskInfo, TASK_COMPLETED);




×







3342



  }










3343













UNCOV

3344



  pOperator->resultInfo.totalRows += pRes->info.rows;




×







3345













UNCOV

3346



  return (pRes->info.rows == 0) ? NULL : pInfo->pRes;





×







3347



}










3348













UNCOV

3349



static void destroyTagScanOperatorInfo(void* param) {




×








UNCOV

3350



  STagScanInfo* pInfo = (STagScanInfo*)param;




×








UNCOV

3351



  if (pInfo->pCtbCursor != NULL) {





×







3352



    pInfo->pStorageAPI->metaFn.closeCtbCursor(pInfo->pCtbCursor);




×







3353



  }











UNCOV

3354



  taosHashCleanup(pInfo->filterCtx.colHash);




×








UNCOV

3355



  taosArrayDestroy(pInfo->filterCtx.cInfoList);




×








UNCOV

3356



  taosArrayDestroy(pInfo->aFilterIdxs);




×








UNCOV

3357



  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);




×







3358













UNCOV

3359



  pInfo->pRes = blockDataDestroy(pInfo->pRes);




×








UNCOV

3360



  taosArrayDestroy(pInfo->matchInfo.pList);




×








UNCOV

3361



  pInfo->pTableListInfo = tableListDestroy(pInfo->pTableListInfo);




×








UNCOV

3362



  taosMemoryFreeClear(param);





×








UNCOV

3363



}




×







3364













UNCOV

3365



SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,




×







3366



                                         STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond, SExecTaskInfo* pTaskInfo) {











UNCOV

3367



  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;




×








UNCOV

3368



  STagScanInfo*  pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo));




×








UNCOV

3369



  SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));




×








UNCOV

3370



  if (pInfo == NULL || pOperator == NULL) {





×







3371



    goto _error;




×







3372



  }










3373













UNCOV

3374



  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;




×







3375













UNCOV

3376



  int32_t    numOfExprs = 0;




×








UNCOV

3377



  SExprInfo* pExprInfo = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &numOfExprs);




×








UNCOV

3378



  int32_t    code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs, &pTaskInfo->storageAPI.functionStore);




×








UNCOV

3379



  if (code != TSDB_CODE_SUCCESS) {





×







3380



    goto _error;




×







3381



  }










3382













UNCOV

3383



  int32_t num = 0;




×








UNCOV

3384



  code = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);




×








UNCOV

3385



  if (code != TSDB_CODE_SUCCESS) {





×







3386



    goto _error;




×







3387



  }










3388













UNCOV

3389



  pInfo->pTagCond = pTagCond;




×








UNCOV

3390



  pInfo->pTagIndexCond = pTagIndexCond;




×








UNCOV

3391



  pInfo->suid = pPhyNode->suid;




×








UNCOV

3392



  pInfo->pStorageAPI = &pTaskInfo->storageAPI;




×







3393













UNCOV

3394



  pInfo->pTableListInfo = pTableListInfo;




×








UNCOV

3395



  pInfo->pRes = createDataBlockFromDescNode(pDescNode);




×








UNCOV

3396



  pInfo->readHandle = *pReadHandle;




×








UNCOV

3397



  pInfo->curPos = 0;




×







3398













UNCOV

3399



  initLimitInfo(pPhyNode->node.pLimit, pPhyNode->node.pSlimit, &pInfo->limitInfo);




×








UNCOV

3400



  setOperatorInfo(pOperator, "TagScanOperator", QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN, false, OP_NOT_OPENED, pInfo,




×







3401



                  pTaskInfo);











UNCOV

3402



  initResultSizeInfo(&pOperator->resultInfo, 4096);




×








UNCOV

3403



  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);




×







3404













UNCOV

3405



  if (pTagScanNode->onlyMetaCtbIdx) {





×








UNCOV

3406



    pInfo->aUidTags = taosArrayInit(pOperator->resultInfo.capacity, sizeof(STUidTagInfo));




×








UNCOV

3407



    pInfo->aFilterIdxs = taosArrayInit(pOperator->resultInfo.capacity, sizeof(int32_t));




×








UNCOV

3408



    pInfo->filterCtx.colHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);




×








UNCOV

3409



    pInfo->filterCtx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));




×








UNCOV

3410



    if (pInfo->pTagCond != NULL) {





×








UNCOV

3411



      nodesRewriteExprPostOrder(&pTagCond, tagScanRewriteTagColumn, (void*)&pInfo->filterCtx);




×







3412



    }










3413



  }











UNCOV

3414



  __optr_fn_t tagScanNextFn = (pTagScanNode->onlyMetaCtbIdx) ? doTagScanFromCtbIdx : doTagScanFromMetaEntry;





×







3415



  pOperator->fpSet =











UNCOV

3416



      createOperatorFpSet(optrDummyOpenFn, tagScanNextFn, NULL, destroyTagScanOperatorInfo, optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);




×







3417













UNCOV

3418



  return pOperator;




×







3419












3420



_error:




×







3421



  taosMemoryFree(pInfo);




×







3422



  taosMemoryFree(pOperator);




×







3423



  terrno = TSDB_CODE_OUT_OF_MEMORY;




×







3424



  return NULL;




×







3425



}










3426












3427



// table merge scan operator










3428












3429



// table merge scan operator










3430












3431



static int32_t subTblRowCompareFn(const void* pLeft, const void* pRight, void* param) {




×







3432



  int32_t left = *(int32_t*)pLeft;




×







3433



  int32_t right = *(int32_t*)pRight;




×







3434



  STmsSubTablesMergeInfo* pInfo = (STmsSubTablesMergeInfo*)param;




×







3435












3436



  int32_t leftIdx = pInfo->aInputs[left].rowIdx;




×







3437



  int32_t rightIdx = pInfo->aInputs[right].rowIdx;




×







3438












3439



  if (leftIdx == -1) {





×







3440



    return 1;




×







3441



  } else if (rightIdx == -1) {





×







3442



    return -1;




×







3443



  }










3444












3445



  int64_t leftTs = pInfo->aInputs[left].aTs[leftIdx];




×







3446



  int64_t rightTs = pInfo->aInputs[right].aTs[rightIdx];




×







3447



  int32_t ret = leftTs>rightTs ? 1 : ((leftTs < rightTs) ? -1 : 0);





×







3448



  if (pInfo->pOrderInfo->order == TSDB_ORDER_DESC) {





×







3449



    ret = -1 * ret;




×







3450



  }










3451



  return ret;




×







3452



}










3453












3454



int32_t dumpQueryTableCond(const SQueryTableDataCond* src, SQueryTableDataCond* dst) {




×







3455



  memcpy((void*)dst, (void*)src, sizeof(SQueryTableDataCond));




×







3456



  dst->colList = taosMemoryCalloc(src->numOfCols, sizeof(SColumnInfo));




×







3457



  for (int i = 0; i < src->numOfCols; i++) {





×







3458



    dst->colList[i] = src->colList[i];




×







3459



  }










3460



  return 0;




×







3461



}










3462












3463



static int32_t fetchNextSubTableBlockFromReader(SOperatorInfo* pOperator, STmsSubTableInput* pInput, bool* pSubTableHasBlock) {




×







3464



  int32_t code = 0;




×







3465












3466



  STableMergeScanInfo* pInfo = pOperator->info;




×







3467



  SReadHandle* pHandle = &pInfo->base.readHandle;




×







3468



  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;




×







3469



  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;




×







3470



  const SStorageAPI* pAPI= &pTaskInfo->storageAPI;




×







3471












3472



  blockDataCleanup(pInput->pReaderBlock);




×







3473



  if (!pInput->bInMemReader) {





×







3474



    code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, pInput->pKeyInfo, 1, pInput->pReaderBlock,




×







3475



                                        (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);




×







3476



    if (code != 0) {





×







3477



      T_LONG_JMP(pTaskInfo->env, code);





×







3478



    }










3479



  }










3480












3481



  pInfo->base.dataReader = pInput->pReader;




×







3482












3483



  while (true) {




×







3484



    bool hasNext = false;




×







3485



    int32_t code = pAPI->tsdReader.tsdNextDataBlock(pInfo->base.dataReader, &hasNext);




×







3486



    if (code != 0) {





×







3487



      pAPI->tsdReader.tsdReaderReleaseDataBlock(pInfo->base.dataReader);




×







3488



      pInfo->base.dataReader = NULL;




×







3489



      T_LONG_JMP(pTaskInfo->env, code);





×







3490



    }










3491



    if (!hasNext || isTaskKilled(pTaskInfo)) {





×







3492



      if (isTaskKilled(pTaskInfo)) {





×







3493



        pAPI->tsdReader.tsdReaderReleaseDataBlock(pInfo->base.dataReader);




×







3494



        pInfo->base.dataReader = NULL;




×







3495



        T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);





×







3496



      }










3497



      *pSubTableHasBlock = false;




×







3498



      break;




×







3499



    }










3500












3501



   if (pInput->tblCond.order == TSDB_ORDER_ASC) {





×







3502



      pInput->tblCond.twindows.skey = pInput->pReaderBlock->info.window.ekey + 1;




×







3503



    } else {










3504



      pInput->tblCond.twindows.ekey = pInput->pReaderBlock->info.window.skey - 1;




×







3505



    }










3506












3507



    uint32_t status = 0;




×







3508



    code = loadDataBlock(pOperator, &pInfo->base, pInput->pReaderBlock, &status);




×







3509



    if (code != 0) {





×







3510



      pInfo->base.dataReader = NULL;




×







3511



      T_LONG_JMP(pTaskInfo->env, code);





×







3512



    }










3513



    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {





×







3514



      *pSubTableHasBlock = false;




×







3515



      break;




×







3516



    }










3517



    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pInput->pReaderBlock->info.rows == 0) {





×







3518



      continue;




×







3519



    }










3520












3521



    *pSubTableHasBlock = true;




×







3522



    break;




×







3523



  }










3524












3525



  if (*pSubTableHasBlock) {





×







3526



    pInput->pReaderBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pInput->pReaderBlock->info.id.uid);




×







3527



    pOperator->resultInfo.totalRows += pInput->pReaderBlock->info.rows;




×







3528



  }










3529



  if (!pInput->bInMemReader || !*pSubTableHasBlock) {





×







3530



    pAPI->tsdReader.tsdReaderClose(pInput->pReader);




×







3531



    pInput->pReader = NULL;




×







3532



  }










3533












3534



  pInfo->base.dataReader = NULL;




×







3535



  return TSDB_CODE_SUCCESS;




×







3536



}










3537












3538



static void setGroupStartEndIndex(STableMergeScanInfo* pInfo) {




×







3539



  pInfo->bGroupProcessed = false;




×







3540












3541



  size_t  numOfTables = tableListGetSize(pInfo->base.pTableListInfo);




×







3542



  int32_t i = pInfo->tableStartIndex + 1;




×







3543



  for (; i < numOfTables; ++i) {





×







3544



    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);




×







3545



    if (tableKeyInfo->groupId != pInfo->groupId) {





×







3546



      break;




×







3547



    }










3548



  }










3549



  pInfo->tableEndIndex = i - 1;




×







3550



}




×







3551












3552



static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {




×







3553



  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {





×







3554



    STmsSubTableInput * pInput = pSubTblsInfo->aInputs + i;




×







3555



    if (pInput->rowIdx == -1) {





×







3556



      continue;




×







3557



    }










3558



    if (pInput->type == SUB_TABLE_MEM_BLOCK) {





×







3559



      pInput->rowIdx = 0;




×







3560



      pInput->pageIdx = -1;




×







3561



    }










3562



    SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;





×







3563



    SColumnInfoData* col = taosArrayGet(pInputBlock->pDataBlock, pSubTblsInfo->pOrderInfo->slotId);




×







3564



    pInput->aTs = (int64_t*)col->pData;




×







3565



  }










3566



  tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, subTblRowCompareFn);




×







3567



  return  TSDB_CODE_SUCCESS;




×







3568



}










3569












3570



static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {




×







3571



  setGroupStartEndIndex(pInfo);




×







3572



  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));




×







3573



  if (pSubTblsInfo == NULL) {





×







3574



    return TSDB_CODE_OUT_OF_MEMORY;




×







3575



  }










3576



  pSubTblsInfo->pOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);




×







3577



  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;




×







3578



  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));




×







3579



  if (pSubTblsInfo->aInputs == NULL) {





×







3580



    taosMemoryFree(pSubTblsInfo);




×







3581



    return TSDB_CODE_OUT_OF_MEMORY;




×







3582



  }










3583



  int32_t bufPageSize = pInfo->bufPageSize;




×







3584



  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;




×







3585



  int32_t code =










3586



      createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);




×







3587



  if (code != TSDB_CODE_SUCCESS) {





×







3588



    taosMemoryFree(pSubTblsInfo->aInputs);




×







3589



    taosMemoryFree(pSubTblsInfo);




×







3590



    return code;




×







3591



  }










3592



  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;




×







3593



  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;




×







3594












3595



  pInfo->pSubTablesMergeInfo = pSubTblsInfo;




×







3596



  return TSDB_CODE_SUCCESS;




×







3597



} 










3598












3599



static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {




×







3600



  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;




×







3601



  SReadHandle* pHandle = &pInfo->base.readHandle;




×







3602



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×







3603












3604



  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;




×







3605












3606



  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {





×







3607



    STmsSubTableInput* pInput =  pSubTblsInfo->aInputs + i;




×







3608



    pInput->type = SUB_TABLE_MEM_BLOCK;




×







3609



    dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);




×







3610



    pInput->pReaderBlock = createOneDataBlock(pInfo->pResBlock, false);




×







3611



    pInput->pPageBlock = createOneDataBlock(pInfo->pResBlock, false);




×







3612



    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);




×







3613



    pInput->pKeyInfo = keyInfo;




×







3614












3615



    if (isTaskKilled(pTaskInfo)) {





×







3616



      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);





×







3617



    }










3618












3619



    if (i + 1 < pSubTblsInfo->numInMemReaders) {





×







3620



      pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, keyInfo, 1, pInput->pReaderBlock,




×







3621



                                  (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);




×







3622



      pInput->bInMemReader = true;




×







3623



    } else {










3624



      pInput->pReader = NULL;




×







3625



      pInput->bInMemReader = false;




×







3626



    }










3627



    bool hasNext = true;




×







3628



    fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);




×







3629



    if (!hasNext) {





×







3630



      pInput->rowIdx = -1;




×







3631



      ++pSubTblsInfo->numSubTablesCompleted;




×







3632



      continue;




×







3633



    } else {










3634



      pInput->rowIdx = 0;




×







3635



      pInput->pageIdx = -1;




×







3636



    }










3637



  }  










3638



  return TSDB_CODE_SUCCESS;




×







3639



}










3640












3641



static int32_t adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {




×







3642



  STableMergeScanInfo* pInfo = pOperatorInfo->info;




×







3643



  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);




×







3644



  bool hasNext = true;




×







3645



  fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);




×







3646



  if (!hasNext) {





×







3647



    pInput->rowIdx = -1;




×







3648



    ++pSubTblsInfo->numSubTablesCompleted;




×







3649



  } else {










3650



    pInput->rowIdx = 0;     




×







3651



  }










3652



  










3653



  return TSDB_CODE_SUCCESS;




×







3654



}










3655












3656



static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {




×







3657



  STableMergeScanInfo* pInfo = pOperatorInfo->info;




×







3658



  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);




×







3659












3660



  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;





×







3661



  if (pInput->rowIdx < pInputBlock->info.rows - 1) {





×







3662



    ++pInput->rowIdx;




×







3663



  } else if (pInput->rowIdx == pInputBlock->info.rows -1 ) {





×







3664



    if (pInput->type == SUB_TABLE_MEM_BLOCK) {





×







3665



      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);




×







3666



    }










3667



    if (pInput->rowIdx != -1) {





×







3668



      SColumnInfoData* col = taosArrayGet(pInputBlock->pDataBlock, pSubTblsInfo->pOrderInfo->slotId);




×







3669



      pInput->aTs = (int64_t*)col->pData;




×







3670



    }










3671



  }










3672












3673



  tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));




×







3674












3675



  return TSDB_CODE_SUCCESS;




×







3676



}










3677












3678



static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {




×







3679



  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);




×







3680



  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;





×







3681












3682



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





×







3683



    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);




×







3684












3685



    SColumnInfoData* pSrcColInfo = taosArrayGet(pInputBlock->pDataBlock, i);




×







3686



    bool             isNull = colDataIsNull(pSrcColInfo, pInputBlock->info.rows, pInput->rowIdx, NULL);





×







3687












3688



    if (isNull) {





×







3689



      colDataSetVal(pColInfo, pBlock->info.rows, NULL, true);




×







3690



    } else {










3691



      if (pSrcColInfo->pData != NULL) {





×







3692



        char* pData = colDataGetData(pSrcColInfo, pInput->rowIdx);





×







3693



        colDataSetVal(pColInfo, pBlock->info.rows, pData, false);




×







3694



      }










3695



    }










3696



  }










3697



  pBlock->info.dataLoad = 1;




×







3698



  pBlock->info.scanFlag = pInputBlock->info.scanFlag;




×







3699



  pBlock->info.rows += 1;




×







3700



  return TSDB_CODE_SUCCESS;




×







3701



}










3702












3703



static SSDataBlock* getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pResBlock, int32_t capacity) {




×







3704



  STableMergeScanInfo* pInfo = pOperator->info;




×







3705



  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;




×







3706



  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;




×







3707












3708



  blockDataCleanup(pResBlock);




×







3709



  bool finished = false;




×







3710



  while (true) {




×







3711



    while (1) {










3712



      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {





×







3713



        finished = true;




×







3714



        break;




×







3715



      }










3716












3717



      appendChosenRowToDataBlock(pSubTblsInfo, pResBlock);




×







3718



      adjustSubTableForNextRow(pOperator, pSubTblsInfo);




×







3719



      










3720



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





×







3721



        break;




×







3722



      }










3723



    }










3724












3725



    if (isTaskKilled(pTaskInfo)) {





×







3726



      T_LONG_JMP(pOperator->pTaskInfo->env, pTaskInfo->code);





×







3727



    }










3728












3729



    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);




×







3730



    if (finished || limitReached || pResBlock->info.rows > 0) {





×







3731



      break;










3732



    }










3733



  }










3734



  return (pResBlock->info.rows > 0) ? pResBlock : NULL;





×







3735



}










3736












3737



static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {




×







3738



  STableMergeScanInfo* pInfo = pOperator->info;




×







3739












3740



  initSubTablesMergeInfo(pInfo);




×







3741












3742



  initSubTableInputs(pOperator, pInfo);




×







3743



  










3744



  openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);




×







3745












3746



  return TSDB_CODE_SUCCESS;




×







3747



}










3748













UNCOV

3749



static int32_t stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {




×








UNCOV

3750



  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;




×








UNCOV

3751



  if (pSubTblsInfo != NULL) {





×







3752



    tMergeTreeDestroy(&pSubTblsInfo->pTree);




×







3753












3754



    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {





×







3755



      STmsSubTableInput* pInput =  pSubTblsInfo->aInputs + i;




×







3756



      taosMemoryFree(pInput->tblCond.colList);




×







3757



      blockDataDestroy(pInput->pReaderBlock);




×







3758



      blockDataDestroy(pInput->pPageBlock);




×







3759



      taosArrayDestroy(pInput->aBlockPages);




×







3760



      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);




×







3761



      pInput->pReader = NULL;




×







3762



    }










3763












3764



    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);




×







3765



    taosMemoryFree(pSubTblsInfo->aInputs);




×







3766












3767



    taosMemoryFree(pSubTblsInfo);




×







3768



    pInfo->pSubTablesMergeInfo = NULL;




×







3769












3770



    taosMemoryTrim(0);




×







3771



  }











UNCOV

3772



  return TSDB_CODE_SUCCESS;




×







3773



}










3774












3775



SSDataBlock* doTableMergeScanParaSubTables(SOperatorInfo* pOperator) {




×







3776



  if (pOperator->status == OP_EXEC_DONE) {





×







3777



    return NULL;




×







3778



  }










3779












3780



  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;




×







3781



  STableMergeScanInfo* pInfo = pOperator->info;




×







3782












3783



  int32_t code = pOperator->fpSet._openFn(pOperator);




×







3784



  if (code != TSDB_CODE_SUCCESS) {





×







3785



    T_LONG_JMP(pTaskInfo->env, code);





×







3786



  }










3787












3788



  size_t tableListSize = tableListGetSize(pInfo->base.pTableListInfo);




×







3789



  if (!pInfo->hasGroupId) {





×







3790



    pInfo->hasGroupId = true;




×







3791












3792



    if (tableListSize == 0) {





×







3793



      setOperatorCompleted(pOperator);




×







3794



      return NULL;




×







3795



    }










3796



    pInfo->tableStartIndex = 0;




×







3797



    pInfo->groupId = ((STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex))->groupId;




×







3798



    startSubTablesTableMergeScan(pOperator);




×







3799



  }










3800












3801



  SSDataBlock* pBlock = NULL;




×







3802



  while (pInfo->tableStartIndex < tableListSize) {





×







3803



    if (isTaskKilled(pTaskInfo)) {





×







3804



      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);





×







3805



    }










3806












3807



    pBlock = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity);




×







3808



    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {





×







3809



      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);




×







3810



      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);




×







3811



    }










3812



    if (pBlock != NULL) {





×







3813



      pBlock->info.id.groupId = pInfo->groupId;




×







3814



      pOperator->resultInfo.totalRows += pBlock->info.rows;




×







3815



      pInfo->bGroupProcessed = true;




×







3816



      return pBlock;




×







3817



    } else {










3818



      // Data of this group are all dumped, let's try the next group










3819



      stopSubTablesTableMergeScan(pInfo);




×







3820



      if (pInfo->tableEndIndex >= tableListSize - 1) {





×







3821



        setOperatorCompleted(pOperator);




×







3822



        break;




×







3823



      }










3824












3825



      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;




×







3826



      pInfo->groupId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex)->groupId;




×







3827



      startSubTablesTableMergeScan(pOperator);




×







3828



      resetLimitInfoForNextGroup(&pInfo->limitInfo);




×







3829



    }










3830



  }










3831












3832



  return pBlock;




×







3833



}










3834












3835













UNCOV

3836



static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeScanInfo) {




×








UNCOV

3837



  STableMergeScanInfo* pInfo = pTableMergeScanInfo;




×








UNCOV

3838



  if (pInfo->mSkipTables == NULL) {





×








UNCOV

3839



      pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,




×







3840



                                        taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);










3841



  }











UNCOV

3842



  int bSkip = 1;




×








UNCOV

3843



  if (pInfo->mSkipTables != NULL) {





×








UNCOV

3844



    taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));




×







3845



  }  











UNCOV

3846



}




×







3847













UNCOV

3848



static void doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {




×








UNCOV

3849



  STableMergeScanInfo*            pInfo = pOperator->info;




×








UNCOV

3850



  SExecTaskInfo*                  pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

3851



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×







3852













UNCOV

3853



  SSDataBlock*                    pBlock = pInfo->pReaderBlock;




×








UNCOV

3854



  int32_t                         code = 0;




×








UNCOV

3855



  bool hasNext = false;




×








UNCOV

3856



  STsdbReader* reader = pInfo->base.dataReader;




×







3857













UNCOV

3858



  code = pAPI->tsdReader.tsdNextDataBlock(reader, &hasNext);




×








UNCOV

3859



  if (code != 0) {





×







3860



    pAPI->tsdReader.tsdReaderReleaseDataBlock(reader);




×







3861



    qError("table merge scan fetch next data block error code: %d, %s", code, GET_TASKID(pTaskInfo));





×







3862



    T_LONG_JMP(pTaskInfo->env, code);





×







3863



  }










3864













UNCOV

3865



  if (!hasNext || isTaskKilled(pTaskInfo)) {





×








UNCOV

3866



    if (isTaskKilled(pTaskInfo)) {





×







3867



      qInfo("table merge scan fetch next data block found task killed. %s", GET_TASKID(pTaskInfo));





×







3868



      pAPI->tsdReader.tsdReaderReleaseDataBlock(reader);




×







3869



    }











UNCOV

3870



    *pFinished = true;




×








UNCOV

3871



    return;




×







3872



  }










3873













UNCOV

3874



  uint32_t status = 0;




×








UNCOV

3875



  code = loadDataBlock(pOperator, &pInfo->base, pBlock, &status);




×







3876













UNCOV

3877



  if (code != TSDB_CODE_SUCCESS) {





×







3878



    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));





×







3879



    T_LONG_JMP(pTaskInfo->env, code);





×







3880



  }










3881













UNCOV

3882



  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {





×







3883



    *pFinished = true;




×







3884



    return;




×







3885



  }










3886












3887



  // current block is filter out according to filter condition, continue load the next block











UNCOV

3888



  if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {





×








UNCOV

3889



    *pSkipped = true;




×








UNCOV

3890



    return;




×







3891



  }











UNCOV

3892



  return;




×







3893



}










3894













UNCOV

3895



static SSDataBlock* getBlockForTableMergeScan(void* param) {




×








UNCOV

3896



  STableMergeScanSortSourceParam* source = param;




×








UNCOV

3897



  SOperatorInfo*                  pOperator = source->pOperator;




×








UNCOV

3898



  STableMergeScanInfo*            pInfo = pOperator->info;




×








UNCOV

3899



  SExecTaskInfo*                  pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

3900



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×







3901













UNCOV

3902



  SSDataBlock*                    pBlock = NULL;




×








UNCOV

3903



  int32_t                         code = 0;




×







3904













UNCOV

3905



  int64_t      st = taosGetTimestampUs();




×








UNCOV

3906



  bool         hasNext = false;




×







3907













UNCOV

3908



  STsdbReader* reader = pInfo->base.dataReader;




×







3909



  while (true) {











UNCOV

3910



    if (pInfo->rtnNextDurationBlocks) {





×








UNCOV

3911



      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d", 





×







3912



              GET_TASKID(pTaskInfo), pInfo->nextDurationBlocksIdx, pInfo->numNextDurationBlocks);











UNCOV

3913



      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {





×








UNCOV

3914



        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];




×








UNCOV

3915



        ++pInfo->nextDurationBlocksIdx;




×







3916



      } else {











UNCOV

3917



        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {





×








UNCOV

3918



          blockDataDestroy(pInfo->nextDurationBlocks[i]);




×








UNCOV

3919



          pInfo->nextDurationBlocks[i] = NULL;




×







3920



        }











UNCOV

3921



        pInfo->rtnNextDurationBlocks = false;




×








UNCOV

3922



        pInfo->nextDurationBlocksIdx = 0;




×








UNCOV

3923



        pInfo->numNextDurationBlocks = 0;




×








UNCOV

3924



        continue;




×







3925



      }










3926



    } else {










3927













UNCOV

3928



      bool bFinished = false;




×








UNCOV

3929



      bool bSkipped = false;




×








UNCOV

3930



      doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);




×








UNCOV

3931



      pBlock = pInfo->pReaderBlock;




×








UNCOV

3932



      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d", 





×







3933



              GET_TASKID(pTaskInfo), bFinished, bSkipped, pInfo->bNextDurationBlockEvent, pInfo->bNewFilesetEvent);











UNCOV

3934



      if (bFinished) {





×








UNCOV

3935



        pInfo->bNewFilesetEvent = false;




×








UNCOV

3936



        break;




×







3937



      }










3938













UNCOV

3939



      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {





×








UNCOV

3940



        if (!bSkipped) {





×








UNCOV

3941



          pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks] = createOneDataBlock(pBlock, true);




×








UNCOV

3942



          ++pInfo->numNextDurationBlocks;




×








UNCOV

3943



          if (pInfo->numNextDurationBlocks > 2) {





×







3944



            qError("%s table merge scan prefetch %d next duration blocks. end early.", GET_TASKID(pTaskInfo), pInfo->numNextDurationBlocks);





×







3945



            pInfo->bNewFilesetEvent = false;




×







3946



            break;




×







3947



          }










3948



        }











UNCOV

3949



        if (pInfo->bNewFilesetEvent) {





×








UNCOV

3950



          pInfo->rtnNextDurationBlocks = true;




×








UNCOV

3951



          return NULL;




×







3952



        }










3953



        if (pInfo->bNextDurationBlockEvent) {





×







3954



          pInfo->bNextDurationBlockEvent = false;




×








UNCOV

3955



          continue;




×







3956



        }










3957



      }











UNCOV

3958



      if (bSkipped) continue;





×







3959



    }











UNCOV

3960



    pBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pBlock->info.id.uid);




×







3961













UNCOV

3962



    pOperator->resultInfo.totalRows += pBlock->info.rows;




×







3963













UNCOV

3964



    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;




×







3965



    











UNCOV

3966



    return pBlock;




×







3967



  }










3968













UNCOV

3969



  return NULL;




×







3970



}










3971












3972













UNCOV

3973



SArray* generateSortByTsInfo(SArray* colMatchInfo, int32_t order) {




×








UNCOV

3974



  int32_t tsTargetSlotId = 0;




×








UNCOV

3975



  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {





×








UNCOV

3976



    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);




×








UNCOV

3977



    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {





×








UNCOV

3978



      tsTargetSlotId = colInfo->dstSlotId;




×







3979



    }










3980



  }










3981













UNCOV

3982



  SArray*         pList = taosArrayInit(1, sizeof(SBlockOrderInfo));




×








UNCOV

3983



  SBlockOrderInfo bi = {0};




×








UNCOV

3984



  bi.order = order;




×








UNCOV

3985



  bi.slotId = tsTargetSlotId;




×








UNCOV

3986



  bi.nullFirst = NULL_ORDER_FIRST;




×







3987












3988



  taosArrayPush(pList, &bi);










3989













UNCOV

3990



  return pList;




×







3991



}










3992













UNCOV

3993



void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {




×








UNCOV

3994



  STableMergeScanInfo* pTmsInfo = param;




×








UNCOV

3995



  if (type == TSD_READER_NOTIFY_DURATION_START) {





×








UNCOV

3996



    pTmsInfo->bNewFilesetEvent = true;




×







3997



  } else if (type == TSD_READER_NOTIFY_NEXT_DURATION_BLOCK) {





×







3998



    pTmsInfo->bNextDurationBlockEvent = true;




×







3999



  }











UNCOV

4000



  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);





×







4001













UNCOV

4002



  return;




×







4003



}










4004













UNCOV

4005



int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {




×








UNCOV

4006



  STableMergeScanInfo* pInfo = pOperator->info;




×








UNCOV

4007



  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

4008



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

4009



  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;




×







4010













UNCOV

4011



  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));





×








UNCOV

4012



  pInfo->bNewFilesetEvent = false;




×








UNCOV

4013



  pInfo->bNextDurationBlockEvent = false;




×







4014













UNCOV

4015



  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;




×








UNCOV

4016



  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;




×







4017













UNCOV

4018



  pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,




×








UNCOV

4019



                                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0);




×








UNCOV

4020



  if (pInfo->bSortRowId && numOfTable != 1) {





×








UNCOV

4021



    int32_t memSize = 512 * 1024 * 1024;




×








UNCOV

4022



    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);




×








UNCOV

4023



    if (code != TSDB_CODE_SUCCESS) {





×







4024



      return code;




×







4025



    }










4026



  }











UNCOV

4027



  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);




×








UNCOV

4028



  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pInfo);




×








UNCOV

4029



  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);




×







4030













UNCOV

4031



  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);




×







4032













UNCOV

4033



  STableMergeScanSortSourceParam *param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));




×








UNCOV

4034



  param->pOperator = pOperator;




×







4035













UNCOV

4036



  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));




×








UNCOV

4037



  ps->param = param;




×








UNCOV

4038



  ps->onlyRef = false;




×








UNCOV

4039



  tsortAddSource(pInfo->pSortHandle, ps);




×







4040













UNCOV

4041



  if (numOfTable == 1) {





×








UNCOV

4042



    tsortSetSingleTableMerge(pInfo->pSortHandle);




×







4043



  } else {











UNCOV

4044



    code = tsortOpen(pInfo->pSortHandle);




×







4045



  }











UNCOV

4046



  return code;




×







4047



}










4048













UNCOV

4049



void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {




×








UNCOV

4050



  STableMergeScanInfo* pInfo = pOperator->info;




×








UNCOV

4051



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

4052



  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));





×







4053













UNCOV

4054



  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);




×








UNCOV

4055



  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;




×








UNCOV

4056



  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;




×








UNCOV

4057



  pInfo->sortExecInfo.loops += sortExecInfo.loops;




×








UNCOV

4058



  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;




×








UNCOV

4059



  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;




×







4060













UNCOV

4061



  tsortDestroySortHandle(pInfo->pSortHandle);




×








UNCOV

4062



  pInfo->pSortHandle = NULL;




×







4063













UNCOV

4064



}




×







4065













UNCOV

4066



int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {




×








UNCOV

4067



  STableMergeScanInfo* pInfo = pOperator->info;




×








UNCOV

4068



  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

4069



  SReadHandle* pHandle = &pInfo->base.readHandle;




×








UNCOV

4070



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×








UNCOV

4071



  qDebug("%s table merge scan start group %"PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);





×







4072












4073



  {











UNCOV

4074



    size_t  numOfTables = tableListGetSize(pInfo->base.pTableListInfo);




×








UNCOV

4075



    int32_t i = pInfo->tableStartIndex + 1;




×








UNCOV

4076



    for (; i < numOfTables; ++i) {





×








UNCOV

4077



      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);




×








UNCOV

4078



      if (tableKeyInfo->groupId != pInfo->groupId) {





×








UNCOV

4079



        break;




×







4080



      }










4081



    }











UNCOV

4082



    pInfo->tableEndIndex = i - 1;




×







4083



  }











UNCOV

4084



  pInfo->bGroupProcessed = false;




×








UNCOV

4085



  int32_t tableStartIdx = pInfo->tableStartIndex;




×








UNCOV

4086



  int32_t tableEndIdx = pInfo->tableEndIndex;




×







4087













UNCOV

4088



  int32_t numOfTable = tableEndIdx - tableStartIdx + 1;




×








UNCOV

4089



  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);




×








UNCOV

4090



  pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,




×








UNCOV

4091



                                (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->mSkipTables);




×








UNCOV

4092



  if (pInfo->filesetDelimited) {





×








UNCOV

4093



    pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);




×







4094



  }











UNCOV

4095



  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);




×







4096













UNCOV

4097



  int32_t code = startDurationForGroupTableMergeScan(pOperator);




×







4098













UNCOV

4099



  if (code != TSDB_CODE_SUCCESS) {





×







4100



    T_LONG_JMP(pTaskInfo->env, terrno);





×







4101



  }










4102













UNCOV

4103



  return TSDB_CODE_SUCCESS;




×







4104



}










4105













UNCOV

4106



int32_t stopGroupTableMergeScan(SOperatorInfo* pOperator) {




×








UNCOV

4107



  STableMergeScanInfo* pInfo = pOperator->info;




×








UNCOV

4108



  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

4109



  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;




×







4110













UNCOV

4111



  stopDurationForGroupTableMergeScan(pOperator);




×







4112













UNCOV

4113



  if (pInfo->base.dataReader != NULL) {





×








UNCOV

4114



    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);




×








UNCOV

4115



    pInfo->base.dataReader = NULL;




×







4116



  }











UNCOV

4117



  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {





×







4118



    if (pInfo->nextDurationBlocks[i]) {





×







4119



      blockDataDestroy(pInfo->nextDurationBlocks[i]);




×







4120



      pInfo->nextDurationBlocks[i] = NULL;




×







4121



    }










4122



    pInfo->numNextDurationBlocks = 0;




×







4123



    pInfo->nextDurationBlocksIdx = 0;




×







4124



  }











UNCOV

4125



  resetLimitInfoForNextGroup(&pInfo->limitInfo);




×








UNCOV

4126



  taosHashCleanup(pInfo->mSkipTables);




×








UNCOV

4127



  pInfo->mSkipTables = NULL;




×








UNCOV

4128



  qDebug("%s table merge scan stop group %"PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);





×







4129













UNCOV

4130



  return TSDB_CODE_SUCCESS;




×







4131



}










4132












4133



// all data produced by this function only belongs to one group










4134



// slimit/soffset does not need to be concerned here, since this function only deal with data within one group.











UNCOV

4135



SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,




×







4136



                                              SOperatorInfo* pOperator) {











UNCOV

4137



  STableMergeScanInfo* pInfo = pOperator->info;




×








UNCOV

4138



  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;




×







4139













UNCOV

4140



  blockDataCleanup(pResBlock);




×








UNCOV

4141



  STupleHandle* pTupleHandle = NULL;




×







4142



  while (1) {




×







4143



    while (1) {











UNCOV

4144



      pTupleHandle = tsortNextTuple(pHandle);




×








UNCOV

4145



      if (pTupleHandle == NULL) {





×








UNCOV

4146



        break;




×







4147



      }











UNCOV

4148



      tsortAppendTupleToBlock(pInfo->pSortHandle, pResBlock, pTupleHandle);




×








UNCOV

4149



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





×








UNCOV

4150



        break;




×







4151



      }










4152



    }










4153













UNCOV

4154



    if (tsortIsClosed(pHandle)) {





×







4155



      terrno = TSDB_CODE_TSC_QUERY_CANCELLED;




×







4156



      T_LONG_JMP(pOperator->pTaskInfo->env, terrno);





×







4157



    }










4158



    











UNCOV

4159



    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);




×








UNCOV

4160



    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,





×







4161



          pInfo->limitInfo.numOfOutputRows);











UNCOV

4162



    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {





×







4163



      break;










4164



    }  










4165



  }











UNCOV

4166



  return (pResBlock->info.rows > 0) ? pResBlock : NULL;





×







4167



}










4168













UNCOV

4169



SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {




×








UNCOV

4170



  if (pOperator->status == OP_EXEC_DONE) {





×








UNCOV

4171



    return NULL;




×







4172



  }










4173













UNCOV

4174



  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

4175



  STableMergeScanInfo* pInfo = pOperator->info;




×







4176













UNCOV

4177



  int32_t code = pOperator->fpSet._openFn(pOperator);




×








UNCOV

4178



  if (code != TSDB_CODE_SUCCESS) {





×







4179



    T_LONG_JMP(pTaskInfo->env, code);





×







4180



  }










4181













UNCOV

4182



  size_t tableListSize = tableListGetSize(pInfo->base.pTableListInfo);




×








UNCOV

4183



  if (!pInfo->hasGroupId) {





×








UNCOV

4184



    pInfo->hasGroupId = true;




×







4185













UNCOV

4186



    if (tableListSize == 0) {





×








UNCOV

4187



      setOperatorCompleted(pOperator);




×








UNCOV

4188



      return NULL;




×







4189



    }











UNCOV

4190



    pInfo->tableStartIndex = 0;




×








UNCOV

4191



    pInfo->groupId = ((STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex))->groupId;




×








UNCOV

4192



    startGroupTableMergeScan(pOperator);




×







4193



  }










4194













UNCOV

4195



  SSDataBlock* pBlock = NULL;




×








UNCOV

4196



  while (pInfo->tableStartIndex < tableListSize) {





×








UNCOV

4197



    if (isTaskKilled(pTaskInfo)) {





×







4198



      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);





×







4199



    }










4200













UNCOV

4201



    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,




×







4202



                                              pOperator);











UNCOV

4203



    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {





×








UNCOV

4204



      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);




×








UNCOV

4205



      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);




×







4206



    }











UNCOV

4207



    if (pBlock != NULL) {





×








UNCOV

4208



      pBlock->info.id.groupId = pInfo->groupId;




×








UNCOV

4209



      pOperator->resultInfo.totalRows += pBlock->info.rows;




×








UNCOV

4210



      pInfo->bGroupProcessed = true;




×








UNCOV

4211



      return pBlock;




×







4212



    } else {











UNCOV

4213



      if (pInfo->bNewFilesetEvent) {





×








UNCOV

4214



        stopDurationForGroupTableMergeScan(pOperator);




×








UNCOV

4215



        code = startDurationForGroupTableMergeScan(pOperator);




×








UNCOV

4216



        if (code != TSDB_CODE_SUCCESS) {





×







4217



          T_LONG_JMP(pTaskInfo->env, terrno);





×







4218



        }










4219



      } else {










4220



        // Data of this group are all dumped, let's try the next group











UNCOV

4221



        stopGroupTableMergeScan(pOperator);




×








UNCOV

4222



        if (pInfo->tableEndIndex >= tableListSize - 1) {





×








UNCOV

4223



          setOperatorCompleted(pOperator);




×








UNCOV

4224



          break;




×







4225



        }










4226













UNCOV

4227



        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;




×








UNCOV

4228



        pInfo->groupId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex)->groupId;




×








UNCOV

4229



        startGroupTableMergeScan(pOperator);




×








UNCOV

4230



        resetLimitInfoForNextGroup(&pInfo->limitInfo);




×







4231



      }










4232



    }










4233



  }










4234













UNCOV

4235



  return pBlock;




×







4236



}










4237













UNCOV

4238



void destroyTableMergeScanOperatorInfo(void* param) {




×








UNCOV

4239



  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;




×







4240












4241



  // start one reader variable











UNCOV

4242



  pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);




×








UNCOV

4243



  pTableScanInfo->base.dataReader = NULL;




×







4244













UNCOV

4245



  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {





×







4246



    if (pTableScanInfo->nextDurationBlocks[i] != NULL) {





×







4247



      blockDataDestroy(pTableScanInfo->nextDurationBlocks[i]);




×







4248



      pTableScanInfo->nextDurationBlocks[i] = NULL;




×







4249



    }










4250



  }










4251













UNCOV

4252



  tsortDestroySortHandle(pTableScanInfo->pSortHandle);




×








UNCOV

4253



  pTableScanInfo->pSortHandle = NULL;




×








UNCOV

4254



  taosHashCleanup(pTableScanInfo->mSkipTables);




×








UNCOV

4255



  pTableScanInfo->mSkipTables = NULL;




×








UNCOV

4256



  pTableScanInfo->pSortInputBlock = blockDataDestroy(pTableScanInfo->pSortInputBlock);




×







4257



  // end one reader variable










4258













UNCOV

4259



  cleanupQueryTableDataCond(&pTableScanInfo->base.cond);




×








UNCOV

4260



  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);




×







4261













UNCOV

4262



  pTableScanInfo->pResBlock = blockDataDestroy(pTableScanInfo->pResBlock);




×








UNCOV

4263



  pTableScanInfo->pReaderBlock = blockDataDestroy(pTableScanInfo->pReaderBlock);




×








UNCOV

4264



  taosArrayDestroy(pTableScanInfo->pSortInfo);




×







4265













UNCOV

4266



  stopSubTablesTableMergeScan(pTableScanInfo);




×







4267













UNCOV

4268



  taosMemoryFreeClear(param);





×








UNCOV

4269



}




×







4270













UNCOV

4271



int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {




×








UNCOV

4272



  ASSERT(pOptr != NULL);





×







4273



  // TODO: merge these two info into one struct











UNCOV

4274



  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));




×








UNCOV

4275



  STableMergeScanInfo*     pInfo = pOptr->info;




×








UNCOV

4276



  execInfo->blockRecorder = pInfo->base.readRecorder;




×








UNCOV

4277



  execInfo->sortExecInfo = pInfo->sortExecInfo;




×







4278













UNCOV

4279



  *pOptrExplain = execInfo;




×








UNCOV

4280



  *len = sizeof(STableMergeScanExecInfo);




×







4281













UNCOV

4282



  return TSDB_CODE_SUCCESS;




×







4283



}










4284













UNCOV

4285



SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,




×







4286



                                                STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo) {











UNCOV

4287



  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));




×








UNCOV

4288



  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));




×








UNCOV

4289



  if (pInfo == NULL || pOperator == NULL) {





×







4290



    goto _error;




×







4291



  }










4292













UNCOV

4293



  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;




×







4294













UNCOV

4295



  int32_t numOfCols = 0;




×








UNCOV

4296



  int32_t code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,




×







4297



                                     &pInfo->base.matchInfo);











UNCOV

4298



  if (code != TSDB_CODE_SUCCESS) {





×







4299



    goto _error;




×







4300



  }










4301













UNCOV

4302



  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle);




×








UNCOV

4303



  if (code != TSDB_CODE_SUCCESS) {





×







4304



    taosArrayDestroy(pInfo->base.matchInfo.pList);




×







4305



    goto _error;




×







4306



  }










4307













UNCOV

4308



  if (pTableScanNode->scan.pScanPseudoCols != NULL) {





×








UNCOV

4309



    SExprSupp* pSup = &pInfo->base.pseudoSup;




×








UNCOV

4310



    pSup->pExprInfo = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->numOfExprs);




×








UNCOV

4311



    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset, &pTaskInfo->storageAPI.functionStore);




×







4312



  }










4313













UNCOV

4314



  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};




×







4315













UNCOV

4316



  pInfo->base.metaCache.pTableMetaEntryCache = taosLRUCacheInit(1024 * 128, -1, .5);




×








UNCOV

4317



  if (pInfo->base.metaCache.pTableMetaEntryCache == NULL) {





×







4318



    code = terrno;




×







4319



    goto _error;




×







4320



  }










4321













UNCOV

4322



  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;




×








UNCOV

4323



  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;




×








UNCOV

4324



  pInfo->base.scanFlag = MAIN_SCAN;




×








UNCOV

4325



  pInfo->base.readHandle = *readHandle;




×







4326













UNCOV

4327



  pInfo->readIdx = -1;




×







4328













UNCOV

4329



  pInfo->base.limitInfo.limit.limit = -1;




×








UNCOV

4330



  pInfo->base.limitInfo.slimit.limit = -1;




×








UNCOV

4331



  pInfo->base.pTableListInfo = pTableListInfo;




×







4332













UNCOV

4333



  pInfo->sample.sampleRatio = pTableScanNode->ratio;




×








UNCOV

4334



  pInfo->sample.seed = taosGetTimestampSec();




×







4335













UNCOV

4336



  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);




×








UNCOV

4337



  if (code != TSDB_CODE_SUCCESS) {





×







4338



    goto _error;




×







4339



  }










4340













UNCOV

4341



  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);




×







4342













UNCOV

4343



  pInfo->mergeLimit = -1;




×








UNCOV

4344



  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;





×








UNCOV

4345



  if (hasLimit) {





×








UNCOV

4346



    pInfo->mergeLimit = pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset;




×








UNCOV

4347



    pInfo->mSkipTables = NULL;




×







4348



  }










4349













UNCOV

4350



  initResultSizeInfo(&pOperator->resultInfo, 1024);




×








UNCOV

4351



  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);




×








UNCOV

4352



  blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);




×








UNCOV

4353



  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {





×








UNCOV

4354



    pInfo->bSortRowId = true;




×







4355



  } else {











UNCOV

4356



    pInfo->bSortRowId = false;




×







4357



  }










4358













UNCOV

4359



  pInfo->pSortInfo = generateSortByTsInfo(pInfo->base.matchInfo.pList, pInfo->base.cond.order);




×








UNCOV

4360



  pInfo->pReaderBlock = createOneDataBlock(pInfo->pResBlock, false);




×







4361













UNCOV

4362



  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;





×







4363













UNCOV

4364



  int32_t  rowSize = pInfo->pResBlock->info.rowSize;




×








UNCOV

4365



  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);




×







4366



  











UNCOV

4367



  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);




×







4368












4369



  //start one reader variable











UNCOV

4370



  pInfo->pSortInputBlock = createOneDataBlock(pInfo->pResBlock, false);




×







4371













UNCOV

4372



  if (!tsExperimental) {





×







4373



    pInfo->filesetDelimited = false;




×







4374



  } else {











UNCOV

4375



    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;




×







4376



  }










4377



  // end one reader variable











UNCOV

4378



  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,




×







4379



                  pInfo, pTaskInfo);











UNCOV

4380



  pOperator->exprSupp.numOfExprs = numOfCols;




×







4381













UNCOV

4382



  pOperator->fpSet = createOperatorFpSet(




×








UNCOV

4383



      optrDummyOpenFn, pTableScanNode->paraTablesSort ? doTableMergeScanParaSubTables : doTableMergeScan, NULL,





×







4384



      destroyTableMergeScanOperatorInfo, optrDefaultBufFn, getTableMergeScanExplainExecInfo, optrDefaultGetNextExtFn,










4385



      NULL);











UNCOV

4386



  pOperator->cost.openCost = 0;




×








UNCOV

4387



  return pOperator;




×







4388












4389



_error:




×







4390



  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;




×







4391



  taosMemoryFree(pInfo);




×







4392



  taosMemoryFree(pOperator);




×







4393



  return NULL;




×







4394



}










4395












4396



// ====================================================================================================================










4397



// TableCountScanOperator










4398



static SSDataBlock* doTableCountScan(SOperatorInfo* pOperator);










4399



static void         destoryTableCountScanOperator(void* param);










4400



static void         buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,










4401



                                                   SSDataBlock* pRes, char* dbName, tb_uid_t stbUid, SStorageAPI* pAPI);










4402



static void         buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,










4403



                                                   SSDataBlock* pRes, char* dbName, SStorageAPI* pAPI);










4404



static void         buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,










4405



                                              STableCountScanSupp* pSupp, SSDataBlock* pRes, char* dbName);










4406



static void         buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,










4407



                                                STableCountScanSupp* pSupp, SSDataBlock* pRes, int32_t vgId, char* dbName);










4408



static SSDataBlock* buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,










4409



                                           STableCountScanSupp* pSupp, SSDataBlock* pRes);










4410



static void         buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,










4411



                                                STableCountScanSupp* pSupp, SSDataBlock* pRes, size_t infodbTableNum,










4412



                                                size_t perfdbTableNum);










4413



static void         buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,










4414



                                               size_t infodbTableNum, size_t perfdbTableNum);










4415



static const char*  GROUP_TAG_DB_NAME = "db_name";










4416



static const char*  GROUP_TAG_STABLE_NAME = "stable_name";










4417













UNCOV

4418



int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {




×








UNCOV

4419



  if (scanCols != NULL) {





×








UNCOV

4420



    SNode* pNode = NULL;




×








UNCOV

4421



    FOREACH(pNode, scanCols) {





×








UNCOV

4422



      if (nodeType(pNode) != QUERY_NODE_TARGET) {





×







4423



        return TSDB_CODE_QRY_SYS_ERROR;




×







4424



      }











UNCOV

4425



      STargetNode* targetNode = (STargetNode*)pNode;




×








UNCOV

4426



      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {





×







4427



        return TSDB_CODE_QRY_SYS_ERROR;




×







4428



      }











UNCOV

4429



      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);




×








UNCOV

4430



      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {





×








UNCOV

4431



        supp->dbNameSlotId = targetNode->slotId;




×








UNCOV

4432



      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {





×








UNCOV

4433



        supp->stbNameSlotId = targetNode->slotId;




×







4434



      }










4435



    }










4436



  }











UNCOV

4437



  return TSDB_CODE_SUCCESS;




×







4438



}










4439













UNCOV

4440



int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {




×








UNCOV

4441



  if (pseudoCols != NULL) {





×








UNCOV

4442



    SNode* pNode = NULL;




×








UNCOV

4443



    FOREACH(pNode, pseudoCols) {





×








UNCOV

4444



      if (nodeType(pNode) != QUERY_NODE_TARGET) {





×







4445



        return TSDB_CODE_QRY_SYS_ERROR;




×







4446



      }











UNCOV

4447



      STargetNode* targetNode = (STargetNode*)pNode;




×








UNCOV

4448



      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {





×







4449



        return TSDB_CODE_QRY_SYS_ERROR;




×







4450



      }











UNCOV

4451



      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);




×








UNCOV

4452



      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {





×








UNCOV

4453



        supp->tbCountSlotId = targetNode->slotId;




×







4454



      }










4455



    }










4456



  }











UNCOV

4457



  return TSDB_CODE_SUCCESS;




×







4458



}










4459













UNCOV

4460



int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {




×








UNCOV

4461



  if (groupTags != NULL) {





×








UNCOV

4462



    SNode* pNode = NULL;




×








UNCOV

4463



    FOREACH(pNode, groupTags) {





×








UNCOV

4464



      if (nodeType(pNode) != QUERY_NODE_COLUMN) {





×







4465



        return TSDB_CODE_QRY_SYS_ERROR;




×







4466



      }











UNCOV

4467



      SColumnNode* colNode = (SColumnNode*)pNode;




×








UNCOV

4468



      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {





×








UNCOV

4469



        supp->groupByDbName = true;




×







4470



      }











UNCOV

4471



      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {





×








UNCOV

4472



        supp->groupByStbName = true;




×







4473



      }










4474



    }










4475



  } else {











UNCOV

4476



    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);




×








UNCOV

4477



    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);




×







4478



  }











UNCOV

4479



  return TSDB_CODE_SUCCESS;




×







4480



}










4481













UNCOV

4482



int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,




×







4483



                              STableCountScanSupp* supp, SExecTaskInfo* taskInfo) {











UNCOV

4484



  int32_t code = 0;




×








UNCOV

4485



  code = tblCountScanGetInputs(groupTags, tableName, supp);




×








UNCOV

4486



  if (code != TSDB_CODE_SUCCESS) {





×







4487



    qError("%s get table count scan supp. get inputs error", GET_TASKID(taskInfo));





×







4488



    return code;




×







4489



  }










4490













UNCOV

4491



  supp->dbNameSlotId = -1;




×








UNCOV

4492



  supp->stbNameSlotId = -1;




×








UNCOV

4493



  supp->tbCountSlotId = -1;




×







4494













UNCOV

4495



  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);




×








UNCOV

4496



  if (code != TSDB_CODE_SUCCESS) {





×







4497



    qError("%s get table count scan supp. get group tags slot id error", GET_TASKID(taskInfo));





×







4498



    return code;




×







4499



  }










4500













UNCOV

4501



  code = tblCountScanGetCountSlotId(pseudoCols, supp);




×








UNCOV

4502



  if (code != TSDB_CODE_SUCCESS) {





×







4503



    qError("%s get table count scan supp. get count error", GET_TASKID(taskInfo));





×







4504



    return code;




×







4505



  }











UNCOV

4506



  return code;




×







4507



}










4508













UNCOV

4509



SOperatorInfo* createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,




×







4510



                                                SExecTaskInfo* pTaskInfo) {











UNCOV

4511



  int32_t code = TSDB_CODE_SUCCESS;




×







4512













UNCOV

4513



  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;




×








UNCOV

4514



  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));




×








UNCOV

4515



  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));




×







4516













UNCOV

4517



  if (!pInfo || !pOperator) {





×







4518



    goto _error;




×







4519



  }










4520













UNCOV

4521



  pInfo->readHandle = *readHandle;




×







4522













UNCOV

4523



  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;




×








UNCOV

4524



  initResultSizeInfo(&pOperator->resultInfo, 1);




×








UNCOV

4525



  pInfo->pRes = createDataBlockFromDescNode(pDescNode);




×








UNCOV

4526



  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);




×







4527













UNCOV

4528



  getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,




×







4529



                        pTblCountScanNode->scan.pScanCols, pTblCountScanNode->scan.pScanPseudoCols, &pInfo->supp,










4530



                        pTaskInfo);










4531













UNCOV

4532



  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,




×







4533



                  pInfo, pTaskInfo);











UNCOV

4534



  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScan, NULL, destoryTableCountScanOperator,




×







4535



                                         optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);











UNCOV

4536



  return pOperator;




×







4537












4538



_error:




×







4539



  if (pInfo != NULL) {





×







4540



    destoryTableCountScanOperator(pInfo);




×







4541



  }










4542



  taosMemoryFreeClear(pOperator);





×







4543



  pTaskInfo->code = code;




×







4544



  return NULL;




×







4545



}










4546













UNCOV

4547



void fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,




×







4548



                                 SSDataBlock* pRes) {











UNCOV

4549



  if (pSupp->dbNameSlotId != -1) {





×








UNCOV

4550



    ASSERT(strlen(dbName));





×








UNCOV

4551



    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);




×







4552













UNCOV

4553



    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};




×








UNCOV

4554



    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);




×







4555













UNCOV

4556



    varDataSetLen(varDbName, strlen(dbName));




×








UNCOV

4557



    colDataSetVal(colInfoData, 0, varDbName, false);




×







4558



  }










4559













UNCOV

4560



  if (pSupp->stbNameSlotId != -1) {





×








UNCOV

4561



    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);




×








UNCOV

4562



    if (strlen(stbName) != 0) {





×








UNCOV

4563



      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};




×








UNCOV

4564



      strncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);




×








UNCOV

4565



      varDataSetLen(varStbName, strlen(stbName));




×








UNCOV

4566



      colDataSetVal(colInfoData, 0, varStbName, false);




×







4567



    } else {










4568



      colDataSetNULL(colInfoData, 0);










4569



    }










4570



  }










4571













UNCOV

4572



  if (pSupp->tbCountSlotId != -1) {





×








UNCOV

4573



    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);




×








UNCOV

4574



    colDataSetVal(colInfoData, 0, (char*)&count, false);




×







4575



  }











UNCOV

4576



  pRes->info.rows = 1;




×








UNCOV

4577



}




×







4578













UNCOV

4579



static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {




×








UNCOV

4580



  STableCountScanSupp* pSupp = &pInfo->supp;




×








UNCOV

4581



  SSDataBlock*         pRes = pInfo->pRes;




×







4582












4583



  size_t infodbTableNum;











UNCOV

4584



  getInfosDbMeta(NULL, &infodbTableNum);




×







4585



  size_t perfdbTableNum;











UNCOV

4586



  getPerfDbMeta(NULL, &perfdbTableNum);




×







4587













UNCOV

4588



  if (pSupp->groupByDbName || pSupp->groupByStbName) {





×








UNCOV

4589



    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);




×








UNCOV

4590



    return (pRes->info.rows > 0) ? pRes : NULL;





×







4591



  } else {











UNCOV

4592



    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);




×








UNCOV

4593



    return (pRes->info.rows > 0) ? pRes : NULL;





×







4594



  }










4595



}










4596













UNCOV

4597



static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,




×







4598



                                       size_t infodbTableNum, size_t perfdbTableNum) {











UNCOV

4599



  if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {





×







4600



    fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);




×








UNCOV

4601



  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {





×







4602



    fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);




×








UNCOV

4603



  } else if (strlen(pSupp->dbNameFilter) == 0) {





×








UNCOV

4604



    fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);




×







4605



  }











UNCOV

4606



  setOperatorCompleted(pOperator);




×








UNCOV

4607



}




×







4608













UNCOV

4609



static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,




×







4610



                                        STableCountScanSupp* pSupp, SSDataBlock* pRes, size_t infodbTableNum,










4611



                                        size_t perfdbTableNum) {











UNCOV

4612



  if (pInfo->currGrpIdx == 0) {





×








UNCOV

4613



    uint64_t groupId = 0;




×








UNCOV

4614



    if (pSupp->groupByDbName) {





×








UNCOV

4615



      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));




×







4616



    } else {











UNCOV

4617



      groupId = calcGroupId("", 0);




×







4618



    }










4619













UNCOV

4620



    pRes->info.id.groupId = groupId;




×








UNCOV

4621



    fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);




×








UNCOV

4622



  } else if (pInfo->currGrpIdx == 1) {





×








UNCOV

4623



    uint64_t groupId = 0;




×








UNCOV

4624



    if (pSupp->groupByDbName) {





×








UNCOV

4625



      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));




×







4626



    } else {











UNCOV

4627



      groupId = calcGroupId("", 0);




×







4628



    }










4629













UNCOV

4630



    pRes->info.id.groupId = groupId;




×








UNCOV

4631



    fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);




×







4632



  } else {











UNCOV

4633



    setOperatorCompleted(pOperator);




×







4634



  }











UNCOV

4635



  pInfo->currGrpIdx++;




×








UNCOV

4636



}




×







4637













UNCOV

4638



static SSDataBlock* doTableCountScan(SOperatorInfo* pOperator) {




×








UNCOV

4639



  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

4640



  STableCountScanOperatorInfo* pInfo = pOperator->info;




×








UNCOV

4641



  STableCountScanSupp*         pSupp = &pInfo->supp;




×








UNCOV

4642



  SSDataBlock*                 pRes = pInfo->pRes;




×








UNCOV

4643



  blockDataCleanup(pRes);




×







4644













UNCOV

4645



  if (pOperator->status == OP_EXEC_DONE) {





×








UNCOV

4646



    return NULL;




×







4647



  }











UNCOV

4648



  if (pInfo->readHandle.mnd != NULL) {





×








UNCOV

4649



    return buildSysDbTableCount(pOperator, pInfo);




×







4650



  }










4651













UNCOV

4652



  return buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);




×







4653



}










4654













UNCOV

4655



static SSDataBlock* buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,




×







4656



                                           STableCountScanSupp* pSupp, SSDataBlock* pRes) {











UNCOV

4657



  const char* db = NULL;




×








UNCOV

4658



  int32_t     vgId = 0;




×








UNCOV

4659



  char        dbName[TSDB_DB_NAME_LEN] = {0};




×








UNCOV

4660



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

4661



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×







4662












4663



  // get dbname











UNCOV

4664



  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);




×








UNCOV

4665



  SName sn = {0};




×








UNCOV

4666



  tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);




×








UNCOV

4667



  tNameGetDbName(&sn, dbName);




×







4668













UNCOV

4669



  if (pSupp->groupByDbName || pSupp->groupByStbName) {





×








UNCOV

4670



    buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);




×







4671



  } else {











UNCOV

4672



    buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);




×







4673



  }











UNCOV

4674



  return pRes->info.rows > 0 ? pRes : NULL;





×







4675



}










4676













UNCOV

4677



static void buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,




×







4678



                                        STableCountScanSupp* pSupp, SSDataBlock* pRes, int32_t vgId, char* dbName) {











UNCOV

4679



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

4680



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×







4681













UNCOV

4682



  if (pSupp->groupByStbName) {





×








UNCOV

4683



    if (pInfo->stbUidList == NULL) {





×








UNCOV

4684



      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));




×








UNCOV

4685



      if (pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList) < 0) {





×







4686



        qError("vgId:%d, failed to get stb id list error: %s", vgId, terrstr());





×







4687



      }










4688



    }











UNCOV

4689



    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {





×








UNCOV

4690



      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);




×








UNCOV

4691



      buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);




×







4692













UNCOV

4693



      pInfo->currGrpIdx++;




×








UNCOV

4694



    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {





×








UNCOV

4695



      buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);




×







4696













UNCOV

4697



      pInfo->currGrpIdx++;




×







4698



    } else {











UNCOV

4699



      setOperatorCompleted(pOperator);




×







4700



    }










4701



  } else {











UNCOV

4702



    uint64_t groupId = calcGroupId(dbName, strlen(dbName));




×








UNCOV

4703



    pRes->info.id.groupId = groupId;




×







4704













UNCOV

4705



    int64_t dbTableCount = 0;




×








UNCOV

4706



    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);




×








UNCOV

4707



    fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);




×








UNCOV

4708



    setOperatorCompleted(pOperator);




×







4709



  }











UNCOV

4710



}




×







4711













UNCOV

4712



static void buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,




×







4713



                                      STableCountScanSupp* pSupp, SSDataBlock* pRes, char* dbName) {











UNCOV

4714



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×








UNCOV

4715



  SStorageAPI* pAPI = &pTaskInfo->storageAPI;




×







4716













UNCOV

4717



  if (strlen(pSupp->dbNameFilter) != 0) {





×








UNCOV

4718



    if (strlen(pSupp->stbNameFilter) != 0) {





×








UNCOV

4719



      uint64_t uid = 0;




×








UNCOV

4720



      pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid);




×







4721













UNCOV

4722



      int64_t numOfChildTables = 0;




×








UNCOV

4723



      pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL);




×







4724













UNCOV

4725



      fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes);




×







4726



    } else {










4727



      int64_t tbNumVnode = 0;




×







4728



      pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);




×







4729



      fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);




×







4730



    }










4731



  } else {











UNCOV

4732



    int64_t tbNumVnode = 0;




×








UNCOV

4733



    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);




×








UNCOV

4734



    fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);




×







4735



  }










4736













UNCOV

4737



  setOperatorCompleted(pOperator);




×








UNCOV

4738



}




×







4739













UNCOV

4740



static void buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,




×







4741



                                           SSDataBlock* pRes, char* dbName, SStorageAPI* pAPI) {











UNCOV

4742



  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};




×








UNCOV

4743



  if (pSupp->groupByDbName) {





×








UNCOV

4744



    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");




×







4745



  }










4746













UNCOV

4747



  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));




×








UNCOV

4748



  pRes->info.id.groupId = groupId;




×







4749













UNCOV

4750



  int64_t numOfTables = 0;




×








UNCOV

4751



  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);




×







4752













UNCOV

4753



  if (numOfTables != 0) {





×








UNCOV

4754



    fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);




×







4755



  }











UNCOV

4756



}




×







4757













UNCOV

4758



static void buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,




×







4759



                                           SSDataBlock* pRes, char* dbName, tb_uid_t stbUid, SStorageAPI* pAPI) {











UNCOV

4760



  char stbName[TSDB_TABLE_NAME_LEN] = {0};




×








UNCOV

4761



  pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);




×







4762













UNCOV

4763



  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};




×








UNCOV

4764



  if (pSupp->groupByDbName) {





×








UNCOV

4765



    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));




×







4766



  } else {











UNCOV

4767



    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));




×







4768



  }










4769













UNCOV

4770



  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));




×








UNCOV

4771



  pRes->info.id.groupId = groupId;




×







4772













UNCOV

4773



  int64_t ctbNum = 0;




×








UNCOV

4774



  int32_t code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL);




×








UNCOV

4775



  fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);




×








UNCOV

4776



}




×







4777













UNCOV

4778



static void destoryTableCountScanOperator(void* param) {




×








UNCOV

4779



  STableCountScanOperatorInfo* pTableCountScanInfo = param;




×








UNCOV

4780



  blockDataDestroy(pTableCountScanInfo->pRes);




×







4781













UNCOV

4782



  taosArrayDestroy(pTableCountScanInfo->stbUidList);




×








UNCOV

4783



  taosMemoryFreeClear(param);





×








UNCOV

4784



}




×





















    

  • 

    •