taosdata/TDengine | Build 4987 | source/libs/executor/src/scanoperator.c | Coveralls

79

bool processBlockWithProbability(const SSampleExecInfo* pInfo) {

687,228,645✔

88

  return true;

687,228,645✔

92

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

69,379✔

93

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

69,379✔

96

69,379✔

98

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

697,083,095✔

99

  int32_t     code = TSDB_CODE_SUCCESS;

697,083,095✔

100

  STimeWindow w = {0};

697,083,095✔

103

  if (pInterval->interval == 0) {

697,206,566✔

104

    *overlap = false;

658,815,228✔

105

    return code;

658,816,520✔

108

  if (order == TSDB_ORDER_ASC) {

38,430,981✔

109

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

37,166,791✔

110

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

37,171,229✔

115

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

37,166,302✔

116

      *overlap = true;

32,357,062✔

117

      return code;

32,357,001✔

121

      getNextTimeWindow(pInterval, &w, order);

4,809,467✔

122

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

4,814,724✔

123

        break;

4,531,719✔

126

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

282,085✔

130

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

282,085✔

131

        *overlap = true;

282,085✔

132

        return code;

282,085✔

136

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

1,264,190✔

137

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

1,263,553✔

142

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

1,262,890✔

143

      *overlap = true;

1,089,521✔

144

      return code;

1,089,521✔

148

      getNextTimeWindow(pInterval, &w, order);

175,322✔

149

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

175,322✔

150

        break;

172,628✔

153

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

2,694✔

157

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

2,694✔

158

        *overlap = true;

2,694✔

159

        return code;

2,694✔

164

  *overlap = false;

4,704,347✔

165

  return code;

4,703,085✔

169

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

196,989,747✔

170

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

196,989,747✔

174

  int64_t buf[2] = {0};

197,006,359✔

175

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

196,989,230✔

177

  STableScanInfo* pTableScanInfo = pOperator->info;

196,992,863✔

179

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

196,971,687✔

182

  if (p1 == NULL) {

197,030,012✔

183

    return NULL;

51,324,357✔

186

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

145,705,655✔

187

  if (NULL == *pPage) {

145,677,956✔

191

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

145,678,048✔

194

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

13,236,842✔

195

  if (NULL == pTableScanInfo->pIgnoreTables) {

13,236,842✔

196

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

1,502,693✔

197

    pTableScanInfo->pIgnoreTables =

1,502,693✔

198

        taosHashInit(tableNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);

1,502,092✔

199

    if (NULL == pTableScanInfo->pIgnoreTables) {

1,502,693✔

204

  int32_t tempRes = taosHashPut(pTableScanInfo->pIgnoreTables, &uid, sizeof(uid), &pTableScanInfo->scanTimes,

13,236,842✔

206

  if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {

13,236,410✔

210

  return TSDB_CODE_SUCCESS;

13,236,410✔

213

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

782,971,062✔

214

  STableScanInfo* pTableScanInfo = pOperator->info;

782,971,062✔

215

  int32_t         code = TSDB_CODE_SUCCESS;

783,161,422✔

217

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

783,161,422✔

218

    return TSDB_CODE_SUCCESS;

586,100,035✔

221

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

196,980,860✔

223

  SFilePage*  pPage = NULL;

196,994,039✔

224

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

196,981,702✔

226

  if (pRow == NULL) {

197,001,323✔

227

    return TSDB_CODE_SUCCESS;

51,310,705✔

230

  bool notLoadBlock = true;

145,690,618✔

231

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

2,147,483,647✔

232

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

2,147,483,647✔

234

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

2,147,483,647✔

236

    EFuncDataRequired reqStatus = fmFuncDynDataRequired(functionId, pEntry, pBlockInfo);

2,147,483,647✔

237

    if (reqStatus != FUNC_DATA_REQUIRED_NOT_LOAD && !pSup1->pCtx[i].skipDynDataCheck) {

2,147,483,647✔

238

      notLoadBlock = false;

132,463,978✔

239

      break;

132,463,978✔

244

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

145,308,480✔

246

  if (notLoadBlock) {

145,693,570✔

247

    *status = FUNC_DATA_REQUIRED_NOT_LOAD;

13,236,382✔

248

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

13,236,382✔

251

  return code;

145,656,070✔

254

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

9,415,007✔

256

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

9,415,007✔

261

  return filterRangeExecute(pFilterInfo, pColsAgg, numOfCols, numOfRows, keep);

9,415,859✔

264

static int32_t doLoadBlockSMA(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo,

192,523,470✔

266

  SStorageAPI* pAPI = &pTaskInfo->storageAPI;

192,523,470✔

267

  bool         allColumnsHaveAgg = true;

192,632,960✔

268

  bool         hasNullSMA = false;

192,674,418✔

269

  if (pLoad != NULL) {

192,658,003✔

270

    *pLoad = false;

192,664,125✔

273

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

192,654,629✔

275

  if (code != TSDB_CODE_SUCCESS) {

192,568,226✔

279

  if (!allColumnsHaveAgg || hasNullSMA) {

192,568,226✔

280

    *pLoad = false;

160,756,102✔

282

    *pLoad = true;

31,812,124✔

285

  return code;

192,556,263✔

288

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

766,244,435✔

290

  int32_t    code = 0;

766,244,435✔

291

  SExprSupp* pSup = &pTableScanInfo->pseudoSup;

766,244,435✔

292

  if (pSup->numOfExprs > 0) {

766,525,730✔

293

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

328,985,905✔

296

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

328,837,133✔

301

  return code;

766,362,926✔

304

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

1,229,120,609✔

305

  SLimit*     pLimit = &pLimitInfo->limit;

1,229,120,609✔

306

  const char* id = GET_TASKID(pTaskInfo);

1,229,159,570✔

308

  if (pLimitInfo->remainOffset > 0) {

1,229,196,725✔

309

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

5,585,361✔

310

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

2,818,095✔

311

      blockDataEmpty(pBlock);

2,818,095✔

312

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

2,818,095✔

313

      return false;

2,818,095✔

315

      int32_t code = blockDataTrimFirstRows(pBlock, pLimitInfo->remainOffset);

2,767,266✔

316

      if (code != TSDB_CODE_SUCCESS) {

2,767,266✔

321

      pLimitInfo->remainOffset = 0;

2,767,266✔

325

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

1,226,421,690✔

327

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

28,202,793✔

328

    blockDataKeepFirstNRows(pBlock, keep);

28,204,110✔

330

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

28,204,699✔

331

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

28,203,063✔

332

    return true;

28,204,948✔

335

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

1,197,991,461✔

336

  return false;

1,198,242,446✔

339

static bool skipSetTagColumnData(SOperatorInfo* pOperator) {

813,662,253✔

340

  return (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) &&

1,486,388,992✔

341

         ((STableScanInfo*)pOperator->info)->ignoreTag;

672,498,588✔

344

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

830,642,254✔

346

  int32_t        code = TSDB_CODE_SUCCESS;

830,642,254✔

347

  int32_t        lino = 0;

830,642,254✔

348

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

830,642,254✔

349

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

830,838,733✔

350

  bool           loadSMA = false;

830,831,630✔

352

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

830,831,630✔

354

  pCost->totalBlocks += 1;

830,797,160✔

355

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

830,766,691✔

356

  *status = pTableScanInfo->dataBlockLoadFlag;

830,646,769✔

358

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

830,732,607✔

359

    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

133,435,938✔

361

    bool overlap = false;

697,228,193✔

363

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

697,267,582✔

364

    if (ret != TSDB_CODE_SUCCESS) {

697,000,739✔

367

    if (overlap) {

697,000,739✔

368

      (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

33,728,189✔

372

  SDataBlockInfo* pBlockInfo = &pBlock->info;

830,570,660✔

373

  taosMemoryFreeClear(pBlock->pBlockAgg);

830,671,545✔

375

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {

830,679,998✔

382

  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

830,431,320✔

383

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

21,677,902✔

386

    if (!skipSetTagColumnData(pOperator)) {

21,683,381✔

387

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

21,686,501✔

389

    pCost->skipBlocks += 1;

21,684,868✔

390

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

21,684,825✔

391

    return code;

21,684,857✔

392

  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {

808,735,230✔

393

    pCost->loadBlockStatis += 1;

59,215,251✔

394

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

59,210,432✔

395

    bool success = true;

59,210,432✔

396

    code = doLoadBlockSMA(pTableScanInfo, pBlock, pTaskInfo, &success);

59,221,021✔

397

    if (code) {

59,119,045✔

403

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

59,104,151✔

404

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

22,348,720✔

406

      if (!skipSetTagColumnData(pOperator)) {

22,349,425✔

407

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

5,373,702✔

409

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

22,407,111✔

410

      return code;

22,363,577✔

412

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

36,755,431✔

413

      *status = FUNC_DATA_REQUIRED_DATA_LOAD;

36,775,931✔

417

  if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {

786,153,349✔

424

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

786,263,414✔

425

    bool success = true;

133,427,720✔

426

    code = doLoadBlockSMA(pTableScanInfo, pBlock, pTaskInfo, &success);

133,438,131✔

427

    if (code) {

133,404,834✔

430

      QUERY_CHECK_CODE(code, lino, _end);

839✔

433

    if (success) {

133,403,464✔

434

      size_t size = taosArrayGetSize(pBlock->pDataBlock);

9,413,871✔

435

      bool   keep = false;

9,414,439✔

436

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

9,414,013✔

437

      if (code) {

9,415,433✔

443

      if (!keep) {

9,414,865✔

444

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

3,431,562✔

446

        pCost->filterOutBlocks += 1;

3,431,562✔

447

        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

3,431,562✔

448

        taosMemoryFreeClear(pBlock->pBlockAgg);

3,431,562✔

450

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

3,431,562✔

451

        return TSDB_CODE_SUCCESS;

3,431,562✔

457

  taosMemoryFreeClear(pBlock->pBlockAgg);

783,020,755✔

460

  code = doDynamicPruneDataBlock(pOperator, pBlockInfo, status);

782,851,846✔

461

  if (code) {

783,096,012✔

466

  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

783,096,012✔

467

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

13,236,410✔

469

    pCost->skipBlocks += 1;

13,236,410✔

470

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

13,236,842✔

472

    STableScanInfo* p1 = pOperator->info;

13,236,842✔

473

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

13,236,842✔

474

      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;

3,967,673✔

476

      *status = FUNC_DATA_REQUIRED_FILTEROUT;

9,269,169✔

478

    return TSDB_CODE_SUCCESS;

13,236,842✔

481

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

769,676,480✔

482

  pCost->loadBlocks += 1;

769,925,626✔

484

  SSDataBlock* p = NULL;

769,761,957✔

485

  code = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pTableScanInfo->dataReader, &p);

769,700,746✔

486

  if (p == NULL || code != TSDB_CODE_SUCCESS || p != pBlock) {

769,872,132✔

490

  if (!skipSetTagColumnData(pOperator)) {

769,872,132✔

491

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

690,798,018✔

492

    QUERY_CHECK_CODE(code, lino, _end);

690,715,533✔

496

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

769,740,236✔

498

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

769,930,131✔

499

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

129,980,218✔

500

    QUERY_CHECK_CODE(code, lino, _end);

129,965,875✔

502

    int64_t st = taosGetTimestampUs();

129,033,450✔

503

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

129,032,583✔

504

    pTableScanInfo->readRecorder.filterTime += el;

129,032,583✔

506

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

129,040,042✔

507

      pCost->filterOutBlocks += 1;

35,621,470✔

508

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

35,614,783✔

511

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

93,385,443✔

515

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

768,967,428✔

516

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

768,985,428✔

517

    setOperatorCompleted(pOperator);

3,741,204✔

520

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

768,985,054✔

522

_end:

769,930,208✔

523

  if (code != TSDB_CODE_SUCCESS) {

769,928,722✔

524

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

945,872✔

526

  return code;

769,710,551✔

529

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

69,379✔

530

  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

69,379✔

532

  switchCtxOrder(pCtx, numOfOutput);

69,379✔

533

  pTableScanInfo->cond.order = TSDB_ORDER_DESC;

69,379✔

536

69,379✔

599

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

329,306,787✔

601

  int32_t          code = TSDB_CODE_SUCCESS;

329,306,787✔

602

  int32_t          lino = 0;

329,306,787✔

603

  bool             freeReader = false;

329,306,787✔

604

  LRUHandle*       h = NULL;

329,306,787✔

605

  STableCachedVal  val = {0};

329,306,787✔

606

  SMetaReader      mr = {0};

329,306,787✔

607

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

329,301,252✔

608

  int32_t          insertRet = TAOS_LRU_STATUS_OK;

329,355,061✔

609

  STableCachedVal* pVal = NULL;

329,355,061✔

612

  if (numOfExpr <= 0) {

329,442,009✔

613

    return TSDB_CODE_SUCCESS;

254,087✔

619

  int32_t backupRows = pBlock->info.rows;

329,187,922✔

620

  pBlock->info.rows = rows;

329,186,830✔

626

  if (pCache == NULL || pCache->pTableMetaEntryCache == NULL) {

329,131,790✔

627

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

329,138,350✔

628

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

329,219,041✔

629

    if (code != TSDB_CODE_SUCCESS) {

328,984,720✔

644

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

328,984,720✔

646

    val.pName = mr.me.name;

329,117,167✔

647

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

329,117,167✔

649

    freeReader = true;

329,117,167✔

691

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

1,259,456,623✔

692

    const SExprInfo* pExpr1 = &pExpr[j];

930,292,708✔

693

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

930,237,429✔

695

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

930,475,158✔

696

    QUERY_CHECK_NULL(pColInfoData, code, lino, _end, terrno);

930,430,312✔

697

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

930,430,312✔

699

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

930,456,108✔

702

    if (fmIsScanPseudoColumnFunc(functionId)) {

930,524,209✔

703

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

132,588,078✔

704

      if (fType == FUNCTION_TYPE_TBNAME) {

132,590,634✔

705

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

132,590,634✔

706

        QUERY_CHECK_CODE(code, lino, _end);

132,566,496✔

715

      STagVal tagVal = {0};

797,661,098✔

716

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

797,722,380✔

717

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

797,891,108✔

719

      char* data = NULL;

797,873,490✔

720

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

797,873,490✔

721

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

792,547,581✔

723

        data = (char*)p;

5,292,723✔

726

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

797,729,113✔

727

      if (isNullVal) {

797,750,333✔

728

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

4,789,363✔

729

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

792,960,970✔

730

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

792,567,326✔

731

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

792,653,438✔

732

          char* tmp = taosMemoryCalloc(1, varDataLen(data) + 1);

283,173,391✔

733

          if (tmp != NULL) {

283,203,705✔

734

            memcpy(tmp, varDataVal(data), varDataLen(data));

283,203,705✔

735

            qDebug("get tag value:%s, cid:%d, table name:%s, uid%" PRId64, tmp, tagVal.cid, val.pName,

283,187,042✔

737

            taosMemoryFree(tmp);

283,244,943✔

739

          taosMemoryFree(data);

283,261,041✔

741

        QUERY_CHECK_CODE(code, lino, _end);

792,578,221✔

743

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

3,128,300✔

744

          code = colDataSetVal(pColInfoData, i, data, false);

2,623,215✔

745

          QUERY_CHECK_CODE(code, lino, _end);

2,624,758✔

752

  pBlock->info.rows = backupRows;

329,163,915✔

754

_end:

329,231,281✔

756

  if (NULL != pVal) {

329,204,449✔

765

  if (freeReader) {

329,204,449✔

766

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

329,108,203✔

768

  if (code != TSDB_CODE_SUCCESS) {

329,204,507✔

771

  return code;

329,151,237✔

774

int32_t setTbNameColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId,

132,571,760✔

776

  int32_t                     code = TSDB_CODE_SUCCESS;

132,571,760✔

777

  int32_t                     lino = 0;

132,571,760✔

778

  struct SScalarFuncExecFuncs fpSet = {0};

132,571,760✔

779

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

132,574,395✔

780

  QUERY_CHECK_CODE(code, lino, _end);

132,581,181✔

782

  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;

132,581,181✔

783

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

132,581,181✔

784

  STR_TO_VARSTR(buf, name)

132,579,026✔

786

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

132,571,240✔

788

  code = colInfoDataEnsureCapacity(&infoData, 1, false);

132,573,403✔

789

  QUERY_CHECK_CODE(code, lino, _end);

132,579,820✔

791

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

132,579,820✔

792

  QUERY_CHECK_CODE(code, lino, _end);

132,567,581✔

794

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

132,567,581✔

795

  SScalarParam param = {.columnData = pColInfoData};

132,572,869✔

797

  if (fpSet.process != NULL) {

132,575,092✔

798

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

132,575,092✔

799

    QUERY_CHECK_CODE(code, lino, _end);

132,580,744✔

804

  colDataDestroy(&infoData);

132,580,744✔

806

_end:

132,575,377✔

807

  if (code != TSDB_CODE_SUCCESS) {

132,579,148✔

810

  return code;

132,579,148✔

879

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

257,285,778✔

880

  int32_t code = TSDB_CODE_SUCCESS;

257,285,778✔

881

  int32_t lino = 0;

257,285,778✔

882

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

257,285,778✔

883

  QUERY_CHECK_CODE(code, lino, _end);

257,123,020✔

885

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

257,123,020✔

886

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

257,213,690✔

887

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

257,221,597✔

888

  taosMemoryFreeClear(pInfo->pResBlock->pBlockAgg);

257,199,067✔

890

  if (!pInfo->needCountEmptyTable) {

257,234,826✔

891

    pInfo->countState = TABLE_COUNT_STATE_END;

237,796,699✔

893

    pInfo->countState = TABLE_COUNT_STATE_SCAN;

19,408,768✔

896

_end:

257,221,204✔

897

  if (code != TSDB_CODE_SUCCESS) {

257,221,204✔

900

  return code;

257,135,752✔

903

void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {

686,506,505✔

904

  if (pInfo->countState == TABLE_COUNT_STATE_END) {

686,506,505✔

905

    return;

575,607,094✔

907

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

111,002,830✔

908

    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;

225,748✔

910

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

110,781,573✔

911

    if (code != TSDB_CODE_SUCCESS) {

110,781,771✔

912

      qDebug("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

44,029,015✔

917

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

48,621,395✔

919

  blockDataEmpty(pBlock);

48,621,395✔

920

  pBlock->info.rows = 1;

48,624,645✔

921

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

48,624,645✔

922

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

48,624,645✔

923

  pBlock->info.blankFill = true;

48,623,055✔

926

  int32_t col_num = blockDataGetNumOfCols(pBlock);

48,619,345✔

927

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

201,054,111✔

928

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

152,431,126✔

933

  terrno = doSetTagColumnData(pBase, pBlock, pTaskInfo, 1);

48,622,985✔

935

  return pBlock;

48,619,875✔

938

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

46,718,049✔

939

  STableScanInfo* pTableScanInfo = pOperator->info;

46,718,049✔

941

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

46,719,639✔

943

  pOperator->resultInfo.totalRows++;

46,712,819✔

944

  return pBlock;

46,720,169✔

951

static int32_t prepareTableScan(SOperatorInfo* pOperator) {

916,822,512✔

952

  int32_t code = TSDB_CODE_SUCCESS;

916,822,512✔

953

  int32_t lino = 0;

916,822,512✔

955

  qDebug("%s, prepareTableScan start", GET_TASKID(pOperator->pTaskInfo));

916,822,512✔

956

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

916,875,665✔

957

  if (NULL != pGetParam && NULL != pGetParam->value &&

917,033,909✔

958

      ((STableScanOperatorParam*)pGetParam->value)->paramType ==

36,042,021✔

959

         NOTIFY_TYPE_SCAN_PARAM &&

3,087,596✔

960

      ((STableScanOperatorParam*)pGetParam->value)->notifyToProcess) {

3,087,596✔

961

    STableScanOperatorParam* pNotify =

3,087,596✔

964

    STableScanInfo* pTableScanInfo = pOperator->info;

3,087,596✔

965

    SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

3,087,596✔

966

    SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

3,087,596✔

967

    code = pAPI->tsdReader.tsdReaderStepDone(pTableScanInfo->base.dataReader,

3,087,596✔

969

    QUERY_CHECK_CODE(code, lino, _end);

3,087,596✔

971

    if (!pGetParam->reUse) {

3,087,596✔

972

      freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

98,800✔

978

      pGetParam->reUse = false;

2,988,796✔

983

    pOperator->pOperatorGetParam = NULL;

3,087,596✔

986

  qDebug("%s, prepareTableScan end", GET_TASKID(pOperator->pTaskInfo));

916,992,399✔

987

_end:

268,044,744✔

988

  if (code != TSDB_CODE_SUCCESS) {

916,984,676✔

992

  return code;

916,972,708✔

995

static int32_t doTableScanImplNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

877,899,657✔

996

  int32_t         code = TSDB_CODE_SUCCESS;

877,899,657✔

997

  int32_t         lino = 0;

877,899,657✔

998

  STableScanInfo* pTableScanInfo = pOperator->info;

877,899,657✔

999

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

877,982,749✔

1000

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

877,885,402✔

1001

  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;

877,963,175✔

1002

  bool            hasNext = false;

877,924,983✔

1003

  int64_t         st = taosGetTimestampUs();

877,904,234✔

1005

  QRY_PARAM_CHECK(ppRes);

877,904,234✔

1006

  pBlock->info.dataLoad = false;

877,934,656✔

1008

  while (true) {

42,613,576✔

1009

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

920,432,957✔

1010

    if (code != TSDB_CODE_SUCCESS) {

920,274,063✔

1015

    if (!hasNext) {

920,188,115✔

1016

      break;

233,026,014✔

1019

    if (isTaskKilled(pTaskInfo)) {

687,162,101✔

1020

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

30,034✔

1021

      code = pTaskInfo->code;

30,034✔

1022

      QUERY_CHECK_CODE(code, lino, _end);

30,034✔

1025

    if (pOperator->status == OP_EXEC_DONE) {

687,522,326✔

1026

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

30,800✔

1027

      break;

30,800✔

1031

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

687,322,931✔

1032

    if (!processThisBlock) {

687,418,660✔

1036

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

687,418,660✔

1037

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

687,515,913✔

1040

    uint32_t status = 0;

687,577,299✔

1041

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

687,588,432✔

1042

    QUERY_CHECK_CODE(code, lino, _end);

687,326,805✔

1044

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

686,397,373✔

1045

      break;

3,967,072✔

1049

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

682,430,301✔

1050

      continue;

42,615,735✔

1053

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

639,843,463✔

1054

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

639,927,800✔

1056

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

640,030,837✔

1057

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

640,064,114✔

1059

    (*ppRes) = pBlock;

640,029,499✔

1060

    return code;

639,924,153✔

1063

_end:

237,992,754✔

1064

  if (code != TSDB_CODE_SUCCESS) {

237,991,411✔

1065

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

967,799✔

1066

    pTaskInfo->code = code;

967,799✔

1067

    T_LONG_JMP(pTaskInfo->env, code);

959,466✔

1069

  return code;

237,023,612✔

1072

static int32_t doGroupedTableScan(SOperatorInfo* pOperator, SSDataBlock** pBlock) {

927,519,965✔

1073

  int32_t         code = TSDB_CODE_SUCCESS;

927,519,965✔

1074

  int32_t         lino = 0;

927,519,965✔

1075

  STableScanInfo* pTableScanInfo = pOperator->info;

927,519,965✔

1076

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

927,612,600✔

1077

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

927,533,788✔

1079

  QRY_PARAM_CHECK(pBlock);

927,498,117✔

1082

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

927,591,722✔

1083

    return code;

3,012,349✔

1087

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

1,144,917,872✔

1088

    SSDataBlock* p = NULL;

785,907,754✔

1089

    code = doTableScanImplNext(pOperator, &p);

785,863,002✔

1090

    QUERY_CHECK_CODE(code, lino, _end);

784,734,524✔

1092

    if (p != NULL) {

784,734,524✔

1093

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

564,279,332✔

1094

      *pBlock = p;

564,276,157✔

1095

      return code;

564,354,293✔

1098

    pTableScanInfo->scanTimes += 1;

220,455,192✔

1099

    taosHashClear(pTableScanInfo->pIgnoreTables);

220,452,650✔

1101

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

220,455,598✔

1102

      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

85,921✔

1103

      pTableScanInfo->base.scanFlag = MAIN_SCAN;

85,921✔

1104

      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

85,921✔

1105

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

85,921✔

1108

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

85,921✔

1109

      QUERY_CHECK_CODE(code, lino, _end);

70,159✔

1113

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

359,125,776✔

1114

  if (pTableScanInfo->scanTimes < total) {

359,121,353✔

1115

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

92,084,684✔

1116

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

69,379✔

1117

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

69,379✔

1118

      QUERY_CHECK_CODE(code, lino, _end);

69,379✔

1119

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

69,379✔

1122

    while (pTableScanInfo->scanTimes < total) {

108,638,062✔

1123

      SSDataBlock* p = NULL;

92,088,803✔

1124

      code = doTableScanImplNext(pOperator, &p);

92,079,366✔

1125

      QUERY_CHECK_CODE(code, lino, _end);

92,045,833✔

1127

      if (p != NULL) {

92,045,833✔

1128

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

75,483,293✔

1129

        *pBlock = p;

75,518,308✔

1130

        return code;

75,511,633✔

1133

      pTableScanInfo->scanTimes += 1;

16,562,540✔

1134

      taosHashClear(pTableScanInfo->pIgnoreTables);

16,564,567✔

1136

      if (pTableScanInfo->scanTimes < total) {

16,563,019✔

1147

  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {

283,566,186✔

1148

    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;

65,477,490✔

1149

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

65,477,569✔

1150

      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {

110,120✔

1163

      int32_t numOfTables = 0;

65,363,502✔

1164

      code = tableListGetSize(pTableListInfo, &numOfTables);

65,361,991✔

1165

      QUERY_CHECK_CODE(code, lino, _end);

65,368,337✔

1167

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

65,368,337✔

1169

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

63,591,291✔

1170

        if (pIte != NULL) {

63,588,380✔

1171

          size_t        keySize = 0;

46,719,639✔

1172

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

46,721,759✔

1173

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

46,715,399✔

1174

          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);

46,714,339✔

1175

          markGroupProcessed(pTableScanInfo, *pGroupId);

46,719,109✔

1176

          *pBlock = getBlockForEmptyTable(pOperator, &info);

46,717,519✔

1178

          return code;

46,718,119✔

1182

    pTableScanInfo->countState = TABLE_COUNT_STATE_END;

18,755,573✔

1185

_end:

218,118,468✔

1186

  if (code != TSDB_CODE_SUCCESS) {

236,855,571✔

1191

  return code;

236,855,571✔

1194

static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {

991,057✔

1195

  STableScanInfo*          pInfo = pOperator->info;

991,057✔

1196

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

991,057✔

1197

  int32_t                  code = 0;

991,057✔

1198

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

991,057✔

1199

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

991,057✔

1200

  int32_t                  num = taosArrayGetSize(pParam->pUidList);

991,057✔

1201

  if (num <= 0) {

991,057✔

1206

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

991,057✔

1210

  if (pParam->tableSeq) {

991,057✔

1211

    pListInfo->oneTableForEachGroup = true;

733,347✔

1212

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

733,347✔

1218

    pListInfo->oneTableForEachGroup = false;

257,710✔

1219

    pListInfo->numOfOuputGroups = 1;

257,710✔

1221

  if (pParam->window.skey == INT64_MAX && pParam->window.ekey == INT64_MIN) {

991,057✔

1224

    pInfo->base.cond.twindows.skey = pParam->window.skey;

728,715✔

1225

    pInfo->base.cond.twindows.ekey = pParam->window.ekey;

728,715✔

1228

  STableKeyInfo info = {.groupId = 0};

991,057✔

1229

  int32_t       tableIdx = 0;

991,057✔

1230

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

9,344,244✔

1231

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

8,353,187✔

1232

    if (!pUid) {

8,353,187✔

1237

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

8,353,187✔

1238

      if (TSDB_CODE_DUP_KEY == terrno) {

7,001,182✔

1239

        continue;

7,001,182✔

1244

    info.uid = *pUid;

1,352,005✔

1245

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

1,352,005✔

1246

    if (p == NULL) {

1,352,005✔

1250

    tableIdx++;

1,352,005✔

1251

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

1,352,005✔

1254

  return code;

991,057✔

1257

static int32_t doInitReader(STableScanInfo* pInfo, SExecTaskInfo* pTaskInfo, SStorageAPI* pAPI, int32_t* pNum,

239,070,641✔

1259

  const char* idStr = GET_TASKID(pTaskInfo);

239,070,641✔

1260

  int32_t     code = initNextGroupScan(pInfo, pList, pNum);

239,106,832✔

1261

  if (code) {

238,876,014✔

1266

  if (pInfo->base.dataReader != NULL) {

238,876,014✔

1271

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

477,886,374✔

1272

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

239,035,139✔

1273

  if (code) {

238,983,069✔

1274

    qError("%s failed to open tsdbReader, code:%s at line:%d", idStr, tstrerror(code), __LINE__);

2,350✔

1277

  return code;

238,896,064✔

1280

int compareColIdPair(const void* elem1, const void* elem2) {

345,131,533✔

1281

  SColIdPair* node1 = (SColIdPair*)elem1;

345,131,533✔

1282

  SColIdPair* node2 = (SColIdPair*)elem2;

345,131,533✔

1284

  if (node1->orgColId == node2->orgColId) {

345,131,533✔

1285

    return 0;

113,712✔

1288

  return node1->orgColId < node2->orgColId ? -1 : 1;

345,166,934✔

1291

int compareColIdSlotIdPair(const void* elem1, const void* elem2) {

454,730,391✔

1292

  SColIdSlotIdPair* node1 = (SColIdSlotIdPair*)elem1;

454,730,391✔

1293

  SColIdSlotIdPair* node2 = (SColIdSlotIdPair*)elem2;

454,730,391✔

1295

  if (node1->orgColId == node2->orgColId) {

454,730,391✔

1296

    return 0;

173,712✔

1299

  return node1->orgColId < node2->orgColId ? -1 : 1;

454,712,402✔

1313

static SColMatchItem* findMatchItemByColId(SArray* pList, col_id_t colId) {

101,694,064✔

1314

  if (pList == NULL) {

101,694,064✔

1318

  int32_t left = 0;

101,694,064✔

1319

  int32_t right = taosArrayGetSize(pList) - 1;

101,694,064✔

1320

  while (left <= right) {

313,500,764✔

1321

    int32_t mid = left + (right - left) / 2;

313,500,764✔

1322

    SColMatchItem* pItem = taosArrayGet(pList, mid);

313,500,764✔

1323

    if (pItem->colId == colId) {

313,500,764✔

1325

      while (mid > 0) {

101,694,064✔

1326

        SColMatchItem* pPrev = taosArrayGet(pList, mid - 1);

75,891,964✔

1327

        if (pPrev->colId != colId) {

75,891,964✔

1328

          break;

75,891,964✔

1332

      return taosArrayGet(pList, mid);

101,694,064✔

1335

    if (pItem->colId < colId) {

211,806,700✔

1336

      left = mid + 1;

99,821,548✔

1338

      right = mid - 1;

111,985,152✔

1345

static bool isNewScanParam(STableScanOperatorParam* pParam) {

44,160,560✔

1346

  return pParam->isNewParam;

44,160,560✔

1349

static int32_t createVTableScanInfoFromBatchParam(SOperatorInfo* pOperator) {

88,132,437✔

1350

  int32_t                  code = 0;

88,132,437✔

1351

  int32_t                  lino = 0;

88,132,437✔

1352

  STableScanInfo*          pInfo = pOperator->info;

88,132,437✔

1353

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

88,183,099✔

1354

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

88,171,598✔

1355

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

88,197,058✔

1356

  SMetaReader              orgTable = {0};

88,185,510✔

1357

  SMetaReader              superTable = {0};

88,217,029✔

1358

  SSchemaWrapper*          schema = NULL;

88,213,181✔

1359

  SArray*                  pColArray = NULL;

88,213,181✔

1360

  SArray*                  pBlockColArray = NULL;

88,213,181✔

1361

  int32_t                  num = 0;

88,213,181✔

1362

  STableKeyInfo*           pList = NULL;

88,213,593✔

1366

  cleanupQueryTableDataCond(&pInfo->base.cond);

88,178,114✔

1368

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

88,189,052✔

1369

  if (pGetParam &&

88,194,709✔

1370

      *(ETableScanGetParamType*)pGetParam->value == DYN_TYPE_SCAN_PARAM) {

6,156,372✔

1371

    STableScanOperatorParam* pParam =

6,149,932✔

1373

    pInfo->pBatchColMap = pParam->pBatchTbInfo;

6,155,422✔

1374

    pParam->pBatchTbInfo = NULL;

6,141,918✔

1375

    pInfo->cachedTimeWindow = pParam->window;

6,131,110✔

1376

    pInfo->currentBatchIdx = 0;

6,144,926✔

1377

    pInfo->lastBatchIdx = -1;

6,136,687✔

1378

    pInfo->cachedTagList = pParam->pTagList;

6,130,578✔

1379

    pParam->pTagList = NULL;

6,134,073✔

1380

    pInfo->cachedGroupId = pParam->groupid;

6,140,939✔

1382

    qDebug("vgId:%d dynamic vtable scan with batch param, total %zd tables to scan, groupid:%" PRIu64 ", %s",

6,143,122✔

1384

    taosHashClear(pInfo->pIgnoreTables);

6,143,122✔

1386

    if (pInfo->base.pdInfo.pExprSup) {

6,150,763✔

1387

      SExprSupp* pSup = pInfo->base.pdInfo.pExprSup;

5,654,295✔

1388

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

70,499,465✔

1389

        SqlFunctionCtx ctx = pSup->pCtx[i];

64,915,405✔

1390

        ctx.skipDynDataCheck = false;

64,847,721✔

1395

  pOrgTbInfo = taosArrayGet(pInfo->pBatchColMap, pInfo->currentBatchIdx);

88,118,819✔

1396

  QUERY_CHECK_NULL(pOrgTbInfo, code, lino, _return, terrno);

88,228,744✔

1398

  if (pInfo->currentBatchIdx == pInfo->lastBatchIdx) {

88,228,744✔

1399

    isNewTable = false;

77,794,836✔

1401

    isNewTable = true;

10,429,788✔

1402

    pInfo->lastBatchIdx = pInfo->currentBatchIdx;

10,429,788✔

1405

  if (isNewTable) {

88,191,625✔

1406

    pAPI->metaReaderFn.initReader(&orgTable, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

10,429,995✔

1407

    code = pAPI->metaReaderFn.getTableEntryByName(&orgTable, strstr(pOrgTbInfo->tbName, ".") + 1);

10,437,268✔

1408

    pAPI->metaReaderFn.readerReleaseLock(&orgTable);

10,410,132✔

1409

    qDebug("dynamic vtable scan for origin table:%s, %s", pOrgTbInfo->tbName, GET_TASKID(pTaskInfo));

10,422,112✔

1410

    QUERY_CHECK_CODE(code, lino, _return);

10,386,923✔

1411

    switch (orgTable.me.type) {

10,386,923✔

1412

      case TSDB_CHILD_TABLE:

591,196✔

1413

        pAPI->metaReaderFn.initReader(&superTable, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

591,196✔

1414

        code = pAPI->metaReaderFn.getTableEntryByUid(&superTable, orgTable.me.ctbEntry.suid);

591,196✔

1415

        pAPI->metaReaderFn.readerReleaseLock(&superTable);

591,196✔

1416

        QUERY_CHECK_CODE(code, lino, _return);

591,196✔

1417

        schema = &superTable.me.stbEntry.schemaRow;

591,196✔

1418

        break;

591,196✔

1419

      case TSDB_NORMAL_TABLE:

9,781,589✔

1420

        schema = &orgTable.me.ntbEntry.schemaRow;

9,781,589✔

1421

        break;

9,781,589✔

1422

      default:

14,138✔

1423

        qError("invalid table type:%d", orgTable.me.type);

14,138✔

1428

    pListInfo->oneTableForEachGroup = true;

10,372,785✔

1429

    taosHashClear(pListInfo->map);

10,416,845✔

1430

    taosArrayClear(pListInfo->pTableList);

10,443,538✔

1432

    uint64_t      pUid = orgTable.me.uid;

10,440,631✔

1433

    STableKeyInfo info = {.groupId = pInfo->cachedGroupId, .uid = pUid};

10,440,631✔

1434

    int32_t       tableIdx = 0;

10,440,631✔

1435

    code = taosHashPut(pListInfo->map, &pUid, sizeof(uint64_t), &tableIdx, sizeof(int32_t));

10,440,989✔

1436

    QUERY_CHECK_CODE(code, lino, _return);

10,443,008✔

1437

    QUERY_CHECK_NULL(taosArrayPush(pListInfo->pTableList, &info), code, lino, _return, terrno);

20,877,603✔

1438

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

10,434,595✔

1440

    pColArray = taosArrayInit(schema->nCols, sizeof(SColIdPair));

10,434,595✔

1441

    QUERY_CHECK_NULL(pColArray, code, lino, _return, terrno);

10,421,031✔

1442

    pBlockColArray = taosArrayInit(schema->nCols, sizeof(SColIdSlotIdPair));

10,421,031✔

1443

    QUERY_CHECK_NULL(pBlockColArray, code, lino, _return, terrno);

10,420,313✔

1446

    SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID, .orgColId = PRIMARYKEY_TIMESTAMP_COL_ID};

10,420,313✔

1447

    QUERY_CHECK_NULL(taosArrayPush(pColArray, &tsPair), code, lino, _return, terrno);

10,428,658✔

1449

    for (int32_t i = 0; i < taosArrayGetSize(pOrgTbInfo->colMap); ++i) {

38,751,462✔

1450

      SColIdNameKV* kv = taosArrayGet(pOrgTbInfo->colMap, i);

28,319,882✔

1451

      for (int32_t j = 0; j < schema->nCols; j++) {

221,591,115✔

1452

        if (strcmp(kv->colName, schema->pSchema[j].name) == 0) {

221,314,818✔

1453

          SColIdPair pPair = {.vtbColId = kv->colId, .orgColId = (col_id_t)(schema->pSchema[j].colId)};

28,338,276✔

1454

          QUERY_CHECK_NULL(taosArrayPush(pColArray, &pPair), code, lino, _return, terrno);

28,318,106✔

1455

          qDebug("dynamic vtable scan map col:%s, orgColId:%d, vtbColId:%d, %s", kv->colName, pPair.orgColId,

28,318,106✔

1457

          break;

28,311,126✔

1462

    for (int32_t i = 0; i < taosArrayGetSize(pColArray); i++) {

49,208,920✔

1463

      SColIdPair* pPair = (SColIdPair*)taosArrayGet(pColArray, i);

38,756,573✔

1464

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

226,384,006✔

1465

        SColMatchItem* pItem = taosArrayGet(pInfo->base.matchInfo.pList, j);

226,032,847✔

1466

        if (pItem->colId == pPair->vtbColId) {

226,124,886✔

1467

          SColIdSlotIdPair colIdSlotIdPair = {.orgColId = pPair->orgColId, .vtbSlotId = pItem->dstSlotId};

38,776,877✔

1468

          QUERY_CHECK_NULL(taosArrayPush(pBlockColArray, &colIdSlotIdPair), code, lino, _return, terrno);

38,786,429✔

1469

          qDebug("dynamic vtable scan block col map orgColId:%d, vtbSlotId:%d, %s", colIdSlotIdPair.orgColId,

38,786,429✔

1471

          break;

38,761,056✔

1476

    taosArraySort(pColArray, compareColIdPair);

10,440,620✔

1477

    taosArraySort(pBlockColArray, compareColIdSlotIdPair);

10,429,256✔

1478

    if (pInfo->pBlockColMap) {

10,434,347✔

1479

      taosArrayDestroy(pInfo->pBlockColMap);

7,410,258✔

1480

      pInfo->pBlockColMap = NULL;

7,404,822✔

1482

    pInfo->pBlockColMap = taosArrayDup(pBlockColArray, NULL);

10,422,219✔

1483

    QUERY_CHECK_NULL(pInfo->pBlockColMap, code, lino, _return, terrno)

10,429,352✔

1486

    taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

10,407,116✔

1487

    taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

10,410,012✔

1489

    if (pInfo->base.pdInfo.pExprSup) {

10,431,363✔

1490

      SExprSupp* pSup = pInfo->base.pdInfo.pExprSup;

8,823,116✔

1491

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

108,747,429✔

1492

        SqlFunctionCtx* ctx = &pSup->pCtx[i];

99,957,734✔

1493

        bool           needScan = false;

99,957,694✔

1494

        for (int32_t j = 0; j < (fmIsImplicitTsFunc(ctx->functionId) ? ctx->numOfParams - 1 : ctx->numOfParams); j++) {

201,352,120✔

1495

          SFunctParam pParam = ctx->param[j];

101,431,642✔

1496

          for (int32_t k = 0; k < taosArrayGetSize(pColArray); k++) {

487,072,537✔

1497

            SColIdPair* pPair = (SColIdPair*)taosArrayGet(pColArray, k);

409,360,902✔

1498

            if (pParam.pCol && pParam.pCol->colId == pPair->vtbColId) {

410,429,811✔

1499

              needScan |= true;

25,225,407✔

1500

              break;

25,225,407✔

1504

        qDebug("dynamic vtable scan expr funcId:%d, needScan:%d, %s", ctx->functionId, needScan, GET_TASKID(pTaskInfo));

99,972,607✔

1505

        if (!needScan) {

100,017,575✔

1506

          ctx->skipDynDataCheck = true;

74,804,135✔

1508

          ctx->skipDynDataCheck = false;

25,213,440✔

1515

    pColArray = pInfo->lastColArray;

77,761,630✔

1516

    pBlockColArray = pInfo->lastBlockColArray;

77,773,419✔

1519

  code = initQueryTableDataCondWithColArray(&pInfo->base.cond, &pInfo->base.orgCond, &pInfo->base.readHandle, pColArray);

88,223,713✔

1520

  QUERY_CHECK_CODE(code, lino, _return);

88,233,479✔

1522

  if (pInfo->pResBlock) {

88,233,479✔

1523

    blockDataDestroy(pInfo->pResBlock);

82,074,340✔

1524

    pInfo->pResBlock = NULL;

82,045,420✔

1527

  qDebug("dynamic vtable scan isNewTable:%d, order:%d, lastTimeWindow:[%" PRId64 ", %" PRId64 "], "

88,205,720✔

1535

  if (isNewTable) {

88,133,767✔

1536

    if (pInfo->cachedTimeWindow.skey == INT64_MAX && pInfo->cachedTimeWindow.ekey == INT64_MIN) {

10,407,442✔

1537

      pInfo->base.cond.twindows.skey = pInfo->base.orgCond.twindows.skey;

8,947,965✔

1538

      pInfo->base.cond.twindows.ekey = pInfo->base.orgCond.twindows.ekey;

8,934,076✔

1540

      pInfo->base.cond.twindows.skey = pInfo->cachedTimeWindow.skey;

1,467,046✔

1541

      pInfo->base.cond.twindows.ekey = pInfo->cachedTimeWindow.ekey;

1,461,558✔

1544

    if (pInfo->base.orgCond.order == TSDB_ORDER_DESC)

77,726,325✔

1545

      pInfo->base.cond.twindows.ekey = pInfo->lastTimeWindow.skey - 1;

1,960,165✔

1547

      pInfo->base.cond.twindows.skey = pInfo->lastTimeWindow.ekey + 1;

75,777,997✔

1551

  pInfo->base.cond.suid = orgTable.me.type == TSDB_CHILD_TABLE ? superTable.me.uid : 0;

88,164,102✔

1552

  pInfo->currentGroupId = 0;

88,185,270✔

1553

  pInfo->ignoreTag = true;

88,172,362✔

1555

  code = createOneDataBlockWithColArray(pInfo->pOrgBlock, pBlockColArray, &pInfo->pResBlock);

88,167,641✔

1556

  QUERY_CHECK_CODE(code, lino, _return);

88,196,924✔

1558

  if (isNewTable) {

88,196,924✔

1559

    if (pInfo->base.dataReader != NULL) {

10,428,839✔

1560

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

7,416,376✔

1562

    pInfo->base.dataReader = NULL;

10,403,995✔

1564

    taosRLockLatch(&pTaskInfo->lock);

10,409,446✔

1565

    code = doInitReader(pInfo, pTaskInfo, pAPI, &num, &pList);

10,441,966✔

1566

    taosRUnLockLatch(&pTaskInfo->lock);

10,374,207✔

1567

    QUERY_CHECK_CODE(code, lino, _return);

10,414,653✔

1569

    pInfo->newReader = true;

10,414,653✔

1570

    if (pInfo->lastColArray) {

10,410,809✔

1571

      taosArrayDestroy(pInfo->lastColArray);

4,274,429✔

1572

      pInfo->lastColArray = NULL;

4,273,311✔

1574

    if (pInfo->lastBlockColArray) {

10,404,887✔

1575

      taosArrayDestroy(pInfo->lastBlockColArray);

4,273,369✔

1576

      pInfo->lastBlockColArray = NULL;

4,272,953✔

1578

    pInfo->lastColArray = pColArray;

10,349,713✔

1579

    pInfo->lastBlockColArray = pBlockColArray;

10,356,254✔

1581

    code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

77,768,085✔

1582

    QUERY_CHECK_CODE(code, lino, _return);

77,790,199✔

1584

    pAPI->tsdReader.tsdReaderSetDatablock(pInfo->base.dataReader, pInfo->pResBlock);

77,790,199✔

1585

    pInfo->newReader = false;

77,794,191✔

1586

    pInfo->scanTimes = 0;

77,794,549✔

1589

  pOperator->status = OP_OPENED;

88,139,001✔

1591

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

88,158,180✔

1595

  pInfo->currentGroupId = -1;

88,182,352✔

1596

_return:

88,179,406✔

1597

  if (code) {

88,143,711✔

1600

  pAPI->metaReaderFn.clearReader(&superTable);

88,143,711✔

1601

  pAPI->metaReaderFn.clearReader(&orgTable);

88,164,216✔

1602

  return code;

88,214,462✔

1605

static int32_t createVTableScanInfoFromParam(SOperatorInfo* pOperator) {

25,802,100✔

1606

  int32_t                  code = 0;

25,802,100✔

1607

  int32_t                  lino = 0;

25,802,100✔

1608

  STableScanInfo*          pInfo = pOperator->info;

25,802,100✔

1609

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

25,802,100✔

1610

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

25,802,100✔

1611

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

25,802,100✔

1612

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

25,802,100✔

1613

  SMetaReader              orgTable = {0};

25,802,100✔

1614

  SMetaReader              superTable = {0};

25,802,100✔

1615

  SSchemaWrapper*          schema = NULL;

25,802,100✔

1616

  SArray*                  pColArray = NULL;

25,802,100✔

1617

  SArray*                  pBlockColArray = NULL;

25,802,100✔

1618

  int32_t                  num = 0;

25,802,100✔

1619

  STableKeyInfo*           pList = NULL;

25,802,100✔

1621

  SArray*                  pMatchList = NULL;

25,802,100✔

1623

  cleanupQueryTableDataCond(&pInfo->base.cond);

25,802,100✔

1625

  pOrgTbInfo = pParam->pOrgTbInfo;

25,802,100✔

1627

  QUERY_CHECK_NULL(pOrgTbInfo, code, lino, _return, terrno);

25,802,100✔

1629

  pAPI->metaReaderFn.initReader(&orgTable, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

25,802,100✔

1630

  code = pAPI->metaReaderFn.getTableEntryByName(&orgTable, strstr(pOrgTbInfo->tbName, ".") + 1);

25,802,100✔

1631

  pAPI->metaReaderFn.readerReleaseLock(&orgTable);

25,802,100✔

1632

  qDebug("dynamic vtable scan for origin table:%s, %s", pOrgTbInfo->tbName, GET_TASKID(pTaskInfo));

25,802,100✔

1633

  QUERY_CHECK_CODE(code, lino, _return);

25,802,100✔

1634

  switch (orgTable.me.type) {

25,802,100✔

1635

    case TSDB_CHILD_TABLE:

7,175,260✔

1636

      pAPI->metaReaderFn.initReader(&superTable, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

7,175,260✔

1637

      code = pAPI->metaReaderFn.getTableEntryByUid(&superTable, orgTable.me.ctbEntry.suid);

7,175,260✔

1638

      pAPI->metaReaderFn.readerReleaseLock(&superTable);

7,175,260✔

1639

      QUERY_CHECK_CODE(code, lino, _return);

7,175,260✔

1640

      schema = &superTable.me.stbEntry.schemaRow;

7,175,260✔

1641

      break;

7,175,260✔

1642

    case TSDB_NORMAL_TABLE:

18,626,840✔

1643

      schema = &orgTable.me.ntbEntry.schemaRow;

18,626,840✔

1644

      break;

18,626,840✔

1651

  pListInfo->oneTableForEachGroup = true;

25,802,100✔

1652

  taosHashClear(pListInfo->map);

25,802,100✔

1653

  taosArrayClear(pListInfo->pTableList);

25,802,100✔

1655

  uint64_t      pUid = orgTable.me.uid;

25,802,100✔

1656

  STableKeyInfo info = {.groupId = 0, .uid = pUid};

25,802,100✔

1657

  int32_t       tableIdx = 0;

25,802,100✔

1658

  code = taosHashPut(pListInfo->map, &pUid, sizeof(uint64_t), &tableIdx, sizeof(int32_t));

25,802,100✔

1659

  QUERY_CHECK_CODE(code, lino, _return);

25,802,100✔

1660

  QUERY_CHECK_NULL(taosArrayPush(pListInfo->pTableList, &info), code, lino, _return, terrno);

51,604,200✔

1661

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

25,802,100✔

1663

  pColArray = taosArrayInit(schema->nCols, sizeof(SColIdPair));

25,802,100✔

1664

  QUERY_CHECK_NULL(pColArray, code, lino, _return, terrno);

25,802,100✔

1665

  pBlockColArray = taosArrayInit(schema->nCols, sizeof(SColIdSlotIdPair));

25,802,100✔

1666

  QUERY_CHECK_NULL(pBlockColArray, code, lino, _return, terrno);

25,802,100✔

1667

  if (pInfo->pBlockColMap) {

25,802,100✔

1668

    taosArrayDestroy(pInfo->pBlockColMap);

23,107,251✔

1669

    pInfo->pBlockColMap = NULL;

23,107,251✔

1671

  pInfo->pBlockColMap = taosArrayInit(schema->nCols, sizeof(SColIdSlotIdPair));

25,802,100✔

1672

  QUERY_CHECK_NULL(pBlockColArray, code, lino, _return, terrno);

25,802,100✔

1675

  SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID,

25,802,100✔

1679

  QUERY_CHECK_NULL(taosArrayPush(pColArray, &tsPair), code, lino, _return, terrno);

25,802,100✔

1681

  for (int32_t i = 0; i < taosArrayGetSize(pOrgTbInfo->colMap); ++i) {

101,694,064✔

1682

    SColIdNameKV* kv = taosArrayGet(pOrgTbInfo->colMap, i);

75,891,964✔

1683

    for (int32_t j = 0; j < schema->nCols; j++) {

639,059,684✔

1684

      if (strcmp(kv->colName, schema->pSchema[j].name) == 0) {

639,059,684✔

1685

        SColIdPair pPair = {.vtbColId = kv->colId,

75,891,964✔

1686

                            .orgColId = (col_id_t)(schema->pSchema[j].colId),

75,891,964✔

1687

                            .type.type = schema->pSchema[j].type,

75,891,964✔

1688

                            .type.bytes = schema->pSchema[j].bytes};

75,891,964✔

1689

        QUERY_CHECK_NULL(taosArrayPush(pColArray, &pPair), code, lino, _return, terrno);

75,891,964✔

1690

        break;

75,891,964✔

1695

  if (pInfo->base.matchInfo.pList) {

25,802,100✔

1696

    if (!pInfo->base.matchInfo.colIdOrdered) {

25,802,100✔

1701

      pMatchList = pInfo->base.matchInfo.pList;

25,802,100✔

1706

  for (int32_t i = 0; i < taosArrayGetSize(pColArray); i++) {

127,495,503✔

1707

    SColIdPair* pPair = (SColIdPair*)taosArrayGet(pColArray, i);

101,694,064✔

1708

    SColMatchItem* pItem = findMatchItemByColId(pMatchList, pPair->vtbColId);

101,694,064✔

1709

    if (pItem) {

101,694,064✔

1710

      if (pItem->dataType.type != pPair->type.type || pItem->dataType.bytes != pPair->type.bytes) {

101,694,064✔

1711

        qError("column type not match for vtable colId:%d, org colId:%d, org table name:%s", pPair->vtbColId,

661✔

1713

        code = TSDB_CODE_VTABLE_COLUMN_TYPE_MISMATCH;

661✔

1714

        goto _return;

661✔

1716

      SColIdSlotIdPair colIdSlotIdPair = {.orgColId = pPair->orgColId, .vtbSlotId = pItem->dstSlotId};

101,693,403✔

1717

      QUERY_CHECK_NULL(taosArrayPush(pBlockColArray, &colIdSlotIdPair), code, lino, _return, terrno);

101,693,403✔

1718

      if (i > 0) {

101,693,403✔

1719

        QUERY_CHECK_NULL(taosArrayPush(pInfo->pBlockColMap, &colIdSlotIdPair), code, lino, _return, terrno);

151,782,606✔

1724

  taosArraySort(pColArray, compareColIdPair);

25,801,439✔

1725

  taosArraySort(pBlockColArray, compareColIdSlotIdPair);

25,801,439✔

1726

  taosArraySort(pInfo->pBlockColMap, compareColIdSlotIdPair);

25,801,439✔

1728

  taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

25,801,439✔

1729

  taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

25,801,439✔

1731

  code = initQueryTableDataCondWithColArray(&pInfo->base.cond, &pInfo->base.orgCond, &pInfo->base.readHandle, pColArray);

25,801,439✔

1732

  QUERY_CHECK_CODE(code, lino, _return);

25,801,439✔

1734

  if (pInfo->pResBlock) {

25,801,439✔

1735

    blockDataDestroy(pInfo->pResBlock);

18,361,485✔

1736

    pInfo->pResBlock = NULL;

18,361,485✔

1739

  if (isNewScanParam(pParam)) {

25,801,439✔

1740

    if (pParam->window.skey == INT64_MAX && pParam->window.ekey == INT64_MIN) {

7,449,858✔

1741

      pInfo->base.cond.twindows.skey = pInfo->base.orgCond.twindows.skey;

7,449,858✔

1742

      pInfo->base.cond.twindows.ekey = pInfo->base.orgCond.twindows.ekey;

7,449,858✔

1748

    pInfo->base.cond.twindows.skey = pParam->window.ekey + 1;

18,351,581✔

1750

  pInfo->base.cond.suid = orgTable.me.type == TSDB_CHILD_TABLE ? superTable.me.uid : 0;

25,801,439✔

1751

  pInfo->currentGroupId = 0;

25,801,439✔

1752

  pInfo->ignoreTag = true;

25,801,439✔

1754

  code = createOneDataBlockWithColArray(pInfo->pOrgBlock, pBlockColArray, &pInfo->pResBlock);

25,801,439✔

1755

  QUERY_CHECK_CODE(code, lino, _return);

25,801,439✔

1757

  void **reader = taosHashGet(pInfo->readerCache, &pUid, sizeof(uint64_t));

25,801,439✔

1758

  if (reader) {

25,801,439✔

1759

    if (isNewScanParam(pParam)) {

18,359,121✔

1760

      pAPI->tsdReader.tsdReaderClose(*reader);

7,540✔

1761

      pInfo->base.dataReader = NULL;

7,540✔

1763

      code = taosHashRemove(pInfo->readerCache, &pUid, sizeof(uint64_t));

7,540✔

1764

      QUERY_CHECK_CODE(code, lino, _return);

7,540✔

1766

      taosRLockLatch(&pTaskInfo->lock);

7,540✔

1767

      code = doInitReader(pInfo, pTaskInfo, pAPI, &num, &pList);

7,540✔

1768

      taosRUnLockLatch(&pTaskInfo->lock);

7,540✔

1769

      QUERY_CHECK_CODE(code, lino, _return);

7,540✔

1771

      code = taosHashPut(pInfo->readerCache, &pUid, sizeof(uint64_t), &pInfo->base.dataReader, POINTER_BYTES);

7,540✔

1772

      QUERY_CHECK_CODE(code, lino, _return);

7,540✔

1773

      pInfo->newReader = true;

7,540✔

1775

      pInfo->base.dataReader = *reader;

18,351,581✔

1776

      code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

18,351,581✔

1777

      QUERY_CHECK_CODE(code, lino, _return);

18,351,581✔

1779

      pAPI->tsdReader.tsdReaderSetDatablock(pInfo->base.dataReader, pInfo->pResBlock);

18,351,581✔

1780

      pInfo->newReader = false;

18,351,581✔

1781

      pInfo->scanTimes = 0;

18,351,581✔

1784

    pInfo->base.dataReader = NULL;

7,442,318✔

1786

    taosRLockLatch(&pTaskInfo->lock);

7,442,318✔

1787

    code = doInitReader(pInfo, pTaskInfo, pAPI, &num, &pList);

7,442,318✔

1788

    taosRUnLockLatch(&pTaskInfo->lock);

7,442,318✔

1789

    QUERY_CHECK_CODE(code, lino, _return);

7,442,318✔

1791

    code = taosHashPut(pInfo->readerCache, &pUid, sizeof(uint64_t), &pInfo->base.dataReader, POINTER_BYTES);

7,442,318✔

1792

    QUERY_CHECK_CODE(code, lino, _return);

7,442,318✔

1793

    pInfo->newReader = true;

7,442,318✔

1795

  pOperator->status = OP_OPENED;

25,801,439✔

1797

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

25,801,439✔

1801

  pInfo->currentGroupId = -1;

25,801,439✔

1802

_return:

25,802,100✔

1803

  if (code) {

25,802,100✔

1804

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

661✔

1806

  if (!pInfo->base.matchInfo.colIdOrdered) {

25,802,100✔

1809

  taosArrayDestroy(pColArray);

25,802,100✔

1810

  taosArrayDestroy(pBlockColArray);

25,802,100✔

1811

  pAPI->metaReaderFn.clearReader(&superTable);

25,802,100✔

1812

  pAPI->metaReaderFn.clearReader(&orgTable);

25,802,100✔

1813

  return code;

25,802,100✔

1816

static int32_t startNextGroupScan(SOperatorInfo* pOperator, SSDataBlock** pResult) {

233,884,133✔

1817

  int32_t         code = TSDB_CODE_SUCCESS;

233,884,133✔

1818

  int32_t         lino = 0;

233,884,133✔

1819

  STableScanInfo* pInfo = pOperator->info;

233,884,133✔

1820

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

233,890,258✔

1821

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

233,888,825✔

1822

  int32_t         numOfTables = 0;

233,872,590✔

1824

  QRY_PARAM_CHECK(pResult);

233,828,591✔

1826

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

233,830,604✔

1827

  QUERY_CHECK_CODE(code, lino, _end);

233,895,799✔

1829

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

233,895,799✔

1830

    setOperatorCompleted(pOperator);

215,655,471✔

1831

    if (pOperator->dynamicTask) {

215,651,880✔

1832

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

989,901✔

1833

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

989,510✔

1835

    return code;

215,650,088✔

1839

  pOperator->status = OP_OPENED;

18,150,359✔

1840

  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);

18,207,245✔

1842

  int32_t        num = 0;

18,216,003✔

1843

  STableKeyInfo* pList = NULL;

18,216,788✔

1844

  code = initNextGroupScan(pInfo, &pList, &num);

18,213,900✔

1845

  QUERY_CHECK_CODE(code, lino, _end);

18,229,830✔

1847

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

18,229,830✔

1848

  QUERY_CHECK_CODE(code, lino, _end);

18,208,582✔

1850

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

18,208,582✔

1851

  QUERY_CHECK_CODE(code, lino, _end);

18,232,665✔

1852

  pInfo->scanTimes = 0;

18,232,665✔

1854

  code = doGroupedTableScan(pOperator, pResult);

18,230,119✔

1855

  QUERY_CHECK_CODE(code, lino, _end);

18,258,744✔

1857

  if (*pResult != NULL) {

18,258,744✔

1858

    if (pOperator->dynamicTask) {

14,900,608✔

1859

      (*pResult)->info.id.groupId = (*pResult)->info.id.uid;

14,642,138✔

1861

    return code;

14,900,608✔

1864

_end:

3,358,720✔

1865

  if (code != TSDB_CODE_SUCCESS) {

3,356,306✔

1869

  return code;

3,356,258✔

1872

static int32_t groupSeqTableScan(SOperatorInfo* pOperator, SSDataBlock** pResBlock) {

800,891,521✔

1873

  int32_t         code = TSDB_CODE_SUCCESS;

800,891,521✔

1874

  int32_t         lino = 0;

800,891,521✔

1875

  STableScanInfo* pInfo = pOperator->info;

800,891,521✔

1876

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

800,931,012✔

1877

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

800,871,905✔

1878

  int32_t         num = 0;

800,869,352✔

1879

  STableKeyInfo*  pList = NULL;

800,876,805✔

1880

  SSDataBlock*    pResult = NULL;

800,836,474✔

1881

  const char*     idStr = GET_TASKID(pTaskInfo);

800,874,391✔

1883

  QRY_PARAM_CHECK(pResBlock);

800,855,475✔

1885

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

800,860,848✔

1886

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

221,141,288✔

1891

    taosRLockLatch(&pTaskInfo->lock);

221,194,047✔

1892

    code = doInitReader(pInfo, pTaskInfo, pAPI, &num, &pList);

221,247,562✔

1893

    taosRUnLockLatch(&pTaskInfo->lock);

221,149,271✔

1894

    QUERY_CHECK_CODE(code, lino, _end);

221,209,001✔

1896

    if (pInfo->filesetDelimited) {

221,206,651✔

1897

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

259,251✔

1900

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

221,206,346✔

1905

  code = doGroupedTableScan(pOperator, &pResult);

800,896,626✔

1906

  QUERY_CHECK_CODE(code, lino, _end);

799,692,276✔

1908

  if (pResult != NULL) {

799,692,276✔

1909

    if (pOperator->dynamicTask) {

583,966,151✔

1910

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

2,106,592✔

1913

    *pResBlock = pResult;

583,966,378✔

1914

    return code;

583,776,521✔

1918

    code = startNextGroupScan(pOperator, &pResult);

215,807,706✔

1919

    QUERY_CHECK_CODE(code, lino, _end);

216,016,198✔

1921

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

216,016,198✔

1922

      *pResBlock = pResult;

215,925,637✔

1923

      return code;

215,916,057✔

1927

_end:

2,350✔

1928

  if (code != TSDB_CODE_SUCCESS) {

2,350✔

1929

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

2,350✔

1930

    pTaskInfo->code = code;

2,350✔

1933

  return code;

2,350✔

1936

static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {

25,801,439✔

1937

  return (pWindow == NULL) || (pWindow->skey > pWindow->ekey);

25,801,439✔

1940

static int32_t doStbJoinDynamicTableScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

991,057✔

1941

  int32_t                  code = TSDB_CODE_SUCCESS;

991,057✔

1942

  int32_t                  lino = 0;

991,057✔

1943

  STableScanInfo*          pInfo = pOperator->info;

991,057✔

1944

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

991,057✔

1945

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

991,057✔

1946

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

991,057✔

1948

  code = createTableListInfoFromParam(pOperator);

991,057✔

1949

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

991,057✔

1950

  pOperator->pOperatorGetParam = NULL;

991,057✔

1951

  QUERY_CHECK_CODE(code, lino, _end);

991,057✔

1953

  if (pOperator->status == OP_EXEC_DONE) {

991,057✔

1954

    pInfo->currentGroupId = -1;

23,961✔

1955

    pOperator->status = OP_OPENED;

23,961✔

1956

    SSDataBlock* result = NULL;

23,961✔

1959

      code = startNextGroupScan(pOperator, &result);

23,961✔

1960

      QUERY_CHECK_CODE(code, lino, _end);

23,961✔

1962

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

23,961✔

1963

        (*ppRes) = result;

23,961✔

1964

        return code;

23,961✔

1968

_end:

967,096✔

1969

  if (code) {

967,096✔

1972

  return code;

967,096✔

1975

static int32_t blockSetVstbSlotId(const SSDataBlock* pSrcBlock,SArray* pColMap) {

18,361,037✔

1976

  int32_t      code = TSDB_CODE_SUCCESS;

18,361,037✔

1977

  int32_t      lino = 0;

18,361,037✔

1978

  bool         found = false;

18,361,037✔

1980

  QUERY_CHECK_NULL(pSrcBlock, code, lino, _return, TSDB_CODE_INVALID_PARA);

18,361,037✔

1981

  QUERY_CHECK_NULL(pColMap, code, lino, _return, TSDB_CODE_INVALID_PARA);

18,361,037✔

1983

  for (int32_t i = 0; i < taosArrayGetSize(pSrcBlock->pDataBlock); i++) {

90,304,249✔

1984

    SColumnInfoData* pSrcCol = taosArrayGet(pSrcBlock->pDataBlock, i);

71,943,212✔

1985

    QUERY_CHECK_NULL(pSrcCol, code, lino, _return, terrno);

71,943,212✔

1986

    found = false;

71,943,212✔

1987

    for (int32_t j = 0; j < taosArrayGetSize(pColMap); j++) {

204,574,048✔

1988

      SColIdSlotIdPair* pColPair = taosArrayGet(pColMap, j);

186,216,131✔

1989

      QUERY_CHECK_NULL(pColPair, code, lino, _return, terrno);

186,216,131✔

1990

      if (pSrcCol->info.colId == pColPair->orgColId) {

186,216,131✔

1991

        pSrcCol->info.slotId = (int16_t)pColPair->vtbSlotId;

53,585,295✔

1992

        found = true;

53,585,295✔

1993

        break;

53,585,295✔

1996

    if (!found) {

71,943,212✔

1997

      if (pSrcCol->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

18,357,917✔

1998

        pSrcCol->info.slotId = -1;

18,357,917✔

2007

  return code;

18,361,037✔

2013

static int32_t doVstbSingleDynamicTableScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

25,802,100✔

2014

  int32_t                  code = TSDB_CODE_SUCCESS;

25,802,100✔

2015

  int32_t                  lino = 0;

25,802,100✔

2016

  STableScanInfo*          pInfo = pOperator->info;

25,802,100✔

2017

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

25,802,100✔

2018

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

25,802,100✔

2019

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

25,802,100✔

2021

  QRY_PARAM_CHECK(ppRes);

25,802,100✔

2022

  qTrace("%s call", __FUNCTION__);

25,802,100✔

2024

  code = createVTableScanInfoFromParam(pOperator);

25,802,100✔

2025

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

25,802,100✔

2026

  pOperator->pOperatorGetParam = NULL;

25,802,100✔

2028

  QUERY_CHECK_CODE(code, lino, _end);

25,802,100✔

2030

  SSDataBlock* result = NULL;

25,801,439✔

2031

  if (isEmptyQueryTimeWindow(&pInfo->base.cond.twindows) && pInfo->base.cond.type == TIMEWINDOW_RANGE_CONTAINED) {

25,801,439✔

2032

    (*ppRes) = result;

20,278✔

2033

    return code;

20,278✔

2036

  if (pInfo->newReader) {

25,781,161✔

2037

    code = startNextGroupScan(pOperator, &result);

7,449,858✔

2038

    QUERY_CHECK_CODE(code, lino, _end);

7,449,858✔

2040

    code = doGroupedTableScan(pOperator, &result);

18,331,303✔

2041

    QUERY_CHECK_CODE(code, lino, _end);

18,331,303✔

2044

  if (result) {

25,781,161✔

2045

    SSDataBlock* res = NULL;

18,361,037✔

2046

    pAPI->tsdReader.tsdReaderSetDatablock(pInfo->base.dataReader, NULL);

18,361,037✔

2047

    pInfo->lastTimeWindow.skey = result->info.window.skey;

18,361,037✔

2048

    pInfo->lastTimeWindow.ekey = result->info.window.ekey;

18,361,037✔

2050

    code = blockSetVstbSlotId(result, pInfo->pBlockColMap);

18,361,037✔

2051

    QUERY_CHECK_CODE(code, lino, _end);

18,361,037✔

2057

    (*ppRes) = result;

18,361,037✔

2059

    STableKeyInfo *keyInfo = taosArrayGet(pInfo->base.pTableListInfo->pTableList, 0);

7,420,124✔

2060

    QUERY_CHECK_NULL(keyInfo, code, lino, _end, terrno)

7,420,124✔

2062

    blockDataDestroy(pInfo->pResBlock);

7,420,124✔

2063

    pInfo->pResBlock = NULL;

7,420,124✔

2065

    void** reader = taosHashGet(pInfo->readerCache, &keyInfo->uid, sizeof(uint64_t));

7,420,124✔

2066

    if (reader) {

7,420,124✔

2067

      if (*reader == pInfo->base.dataReader) {

7,420,124✔

2068

        pAPI->tsdReader.tsdReaderSetDatablock(pInfo->base.dataReader, NULL);

7,420,124✔

2069

        pInfo->base.dataReader = NULL;

7,420,124✔

2071

      if (pAPI->tsdReader.tsdReaderClose) {

7,420,124✔

2072

        pAPI->tsdReader.tsdReaderClose(*reader);

7,420,124✔

2076

    code = taosHashRemove(pInfo->readerCache, &keyInfo->uid, sizeof(uint64_t));

7,420,124✔

2077

    QUERY_CHECK_CODE(code, lino, _end);

7,420,124✔

2079

_end:

25,781,822✔

2080

  if (code) {

25,781,822✔

2081

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

661✔

2083

  return code;

25,781,822✔

2086

static int32_t setTagValFromTagList(SOperatorInfo* pOperator, SSDataBlock* pRes) {

17,163,521✔

2087

  int32_t                  code = TSDB_CODE_SUCCESS;

17,163,521✔

2088

  int32_t                  lino = 0;

17,163,521✔

2089

  STableScanInfo*          pInfo = pOperator->info;

17,163,521✔

2090

  int32_t                  index = 0;

17,163,521✔

2091

  char*                    tagVal = NULL;

17,163,521✔

2093

  for (int32_t i = taosArrayGetSize(pRes->pDataBlock) - taosArrayGetSize(pInfo->cachedTagList); i < taosArrayGetSize(pRes->pDataBlock); i++) {

122,449,489✔

2094

    SColumnInfoData* pTagCol = taosArrayGet(pRes->pDataBlock, i);

105,287,194✔

2095

    STagVal*         pTagVal = taosArrayGet(pInfo->cachedTagList, index);

105,285,980✔

2097

    QUERY_CHECK_NULL(pTagVal, code, lino, _end, terrno);

105,282,939✔

2098

    QUERY_CHECK_NULL(pTagCol, code, lino, _end, terrno);

105,282,939✔

2100

    for(int32_t j = 0; j < pRes->info.rows; j++) {

2,147,483,647✔

2101

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

2,147,483,647✔

2102

        tagVal = taosMemoryMalloc(pTagVal->nData + VARSTR_HEADER_SIZE + 1);

2,147,483,647✔

2103

        QUERY_CHECK_NULL(tagVal, code, lino, _end, terrno);

2,147,483,647✔

2105

        varDataSetLen(tagVal, pTagVal->nData);

2,147,483,647✔

2106

        memcpy(tagVal + VARSTR_HEADER_SIZE, pTagVal->pData, pTagVal->nData);

2,147,483,647✔

2107

        code = colDataSetVal(pTagCol, j, tagVal, false);

2,147,483,647✔

2108

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2109

        taosMemoryFreeClear(tagVal);

2,147,483,647✔

2111

        code = colDataSetVal(pTagCol, j, (const char*)&pTagVal->i64, false);

2,147,483,647✔

2112

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2115

    index++;

105,285,968✔

2117

  return code;

17,163,521✔

2124

static void clearVstbBatchDynamicTableScanInfo(STableScanInfo* pInfo) {

6,165,644✔

2125

  if (pInfo->pBatchColMap) {

6,165,644✔

2126

    for (int32_t i = 0; i < taosArrayGetSize(pInfo->pBatchColMap); i++) {

16,608,709✔

2127

      SOrgTbInfo* pOrgTbInfo = (SOrgTbInfo*)taosArrayGet(pInfo->pBatchColMap, i);

10,439,830✔

2128

      taosArrayDestroy(pOrgTbInfo->colMap);

10,440,674✔

2130

    taosArrayDestroy(pInfo->pBatchColMap);

6,167,677✔

2131

    pInfo->pBatchColMap = NULL;

6,166,603✔

2133

  if (pInfo->cachedTagList) {

6,166,360✔

2134

    for (int32_t i = 0; i < taosArrayGetSize(pInfo->cachedTagList); i++) {

18,601,154✔

2135

      STagVal* pTagVal = (STagVal*)taosArrayGet(pInfo->cachedTagList, i);

15,820,826✔

2136

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

15,821,900✔

2137

        taosMemoryFreeClear(pTagVal->pData);

6,946,168✔

2140

    taosArrayDestroy(pInfo->cachedTagList);

2,778,896✔

2141

    pInfo->cachedTagList = NULL;

2,779,612✔

2143

  if (pInfo->lastColArray) {

6,167,677✔

2144

    taosArrayDestroy(pInfo->lastColArray);

6,168,035✔

2145

    pInfo->lastColArray = NULL;

6,168,035✔

2147

  if (pInfo->lastBlockColArray) {

6,166,961✔

2148

    taosArrayDestroy(pInfo->lastBlockColArray);

6,168,393✔

2149

    pInfo->lastBlockColArray = NULL;

6,166,245✔

2152

  blockDataDestroy(pInfo->pResBlock);

6,164,928✔

2153

  pInfo->pResBlock = NULL;

6,169,109✔

2154

6,168,751✔

2156

static int32_t doVstbBatchDynamicTableScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

83,850,804✔

2157

  int32_t                  code = TSDB_CODE_SUCCESS;

83,850,804✔

2158

  int32_t                  lino = 0;

83,850,804✔

2159

  STableScanInfo*          pInfo = pOperator->info;

83,850,804✔

2160

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

83,904,649✔

2161

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

83,900,229✔

2162

  SSDataBlock*             result = NULL;

83,897,125✔

2164

  QRY_PARAM_CHECK(ppRes);

83,900,220✔

2165

  qTrace("%s call", __FUNCTION__);

83,907,020✔

2167

  while (!result) {

165,814,990✔

2168

    code = createVTableScanInfoFromBatchParam(pOperator);

88,137,558✔

2169

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

88,183,958✔

2170

    pOperator->pOperatorGetParam = NULL;

88,153,750✔

2171

    QUERY_CHECK_CODE(code, lino, _end);

88,171,556✔

2173

    if (pInfo->newReader) {

88,171,556✔

2174

      code = startNextGroupScan(pOperator, &result);

10,379,976✔

2175

      QUERY_CHECK_CODE(code, lino, _end);

10,439,074✔

2177

      code = doGroupedTableScan(pOperator, &result);

77,779,069✔

2178

      QUERY_CHECK_CODE(code, lino, _end);

77,678,636✔

2180

    if (!result) {

88,117,710✔

2181

      qDebug("%s, finished batch scan for current table, switch to next table, idx:%d,  %s",

10,436,245✔

2183

      pInfo->currentBatchIdx++;

10,436,245✔

2184

      if (pInfo->currentBatchIdx >= taosArrayGetSize(pInfo->pBatchColMap)) {

10,433,004✔

2186

        break;

6,165,887✔

2191

  if (result) {

83,843,319✔

2192

    SSDataBlock* res = NULL;

77,676,716✔

2193

    pAPI->tsdReader.tsdReaderSetDatablock(pInfo->base.dataReader, NULL);

77,680,310✔

2194

    pInfo->lastTimeWindow.skey = result->info.window.skey;

77,771,154✔

2195

    pInfo->lastTimeWindow.ekey = result->info.window.ekey;

77,765,617✔

2197

    code = createOneDataBlockWithTwoBlock(result, pInfo->pOrgBlock, pInfo->pBlockColMap, &res);

77,763,790✔

2198

    QUERY_CHECK_CODE(code, lino, _end);

77,788,237✔

2200

    if (pInfo->cachedTagList) {

77,788,237✔

2201

      code = setTagValFromTagList(pOperator, res);

17,163,521✔

2202

      QUERY_CHECK_CODE(code, lino, _end);

17,163,521✔

2205

    res->info.id.groupId = pInfo->cachedGroupId;

77,794,271✔

2206

    pInfo->pResBlock = res;

77,797,853✔

2207

    blockDataDestroy(result);

77,796,109✔

2208

    (*ppRes) = res;

77,756,937✔

2210

    printDataBlock(res, __func__, pTaskInfo->id.str, pTaskInfo->id.queryId);

77,784,523✔

2212

    clearVstbBatchDynamicTableScanInfo(pInfo);

6,166,603✔

2213

    setOperatorCompleted(pOperator);

6,168,393✔

2216

_end:

83,949,891✔

2217

  return code;

83,947,821✔

2220

static int32_t doDynamicTableScanNext(SOperatorInfo* pOperator, ETableScanDynType type, SSDataBlock** ppRes) {

110,688,158✔

2221

  int32_t                  code = TSDB_CODE_SUCCESS;

110,688,158✔

2222

  int32_t                  lino = 0;

110,688,158✔

2223

  STableScanInfo*          pInfo = pOperator->info;

110,688,158✔

2224

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

110,733,837✔

2225

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

110,741,021✔

2227

  QRY_PARAM_CHECK(ppRes);

110,728,188✔

2228

  qTrace("%s call", __FUNCTION__);

110,722,568✔

2230

  pOperator->dynamicTask = true;

110,722,568✔

2232

  switch (type) {

110,675,641✔

2233

    case DYN_TYPE_VSTB_WIN_SCAN:

991,057✔

2235

      code = doStbJoinDynamicTableScanNext(pOperator, ppRes);

991,057✔

2236

      QUERY_CHECK_CODE(code, lino, _end);

991,057✔

2237

      break;

991,057✔

2239

    case DYN_TYPE_VSTB_SINGLE_SCAN: {

25,802,100✔

2240

      code = doVstbSingleDynamicTableScanNext(pOperator, ppRes);

25,802,100✔

2241

      QUERY_CHECK_CODE(code, lino, _end);

25,802,100✔

2242

      break;

25,801,439✔

2244

    case DYN_TYPE_VSTB_BATCH_SCAN:

83,882,484✔

2245

      code = doVstbBatchDynamicTableScanNext(pOperator, ppRes);

83,882,484✔

2246

      QUERY_CHECK_CODE(code, lino, _end);

83,928,979✔

2247

      break;

83,928,979✔

2254

_end:

110,722,136✔

2255

  return code;

110,722,136✔

2258

int32_t doTableScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

916,901,682✔

2259

  int32_t         code = TSDB_CODE_SUCCESS;

916,901,682✔

2260

  int32_t         lino = 0;

916,901,682✔

2261

  STableScanInfo* pInfo = pOperator->info;

916,901,682✔

2262

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

917,042,154✔

2263

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

917,009,644✔

2264

  QRY_PARAM_CHECK(ppRes);

917,027,665✔

2265

  qTrace("%s call", __FUNCTION__);

916,958,882✔

2267

  code = pOperator->fpSet._openFn(pOperator);

916,960,034✔

2268

  QUERY_CHECK_CODE(code, lino, _end);

917,010,213✔

2270

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

917,010,213✔

2271

  if (pGetParam || pInfo->pBatchColMap) {

916,984,414✔

2272

    ETableScanDynType type = pGetParam ?

110,709,248✔

2273

      ((STableScanOperatorParam*)pGetParam->value)->dynType :

110,708,075✔

2276

    code = doDynamicTableScanNext(pOperator, type, ppRes);

110,709,248✔

2277

    QUERY_CHECK_CODE(code, lino, _end);

110,717,547✔

2279

    if (*ppRes != NULL || (type != DYN_TYPE_STB_JOIN && type != DYN_TYPE_VSTB_WIN_SCAN)) {

110,716,886✔

2280

      return code;

109,780,505✔

2285

  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {

807,205,263✔

2286

    int32_t       numOfTables = 0;

6,395,141✔

2287

    STableKeyInfo tInfo = {0};

6,395,141✔

2288

    pInfo->countState = TABLE_COUNT_STATE_END;

6,395,473✔

2290

    while (1) {

5,973,348✔

2291

      SSDataBlock* result = NULL;

12,368,489✔

2292

      code = doGroupedTableScan(pOperator, &result);

12,367,174✔

2293

      QUERY_CHECK_CODE(code, lino, _end);

12,359,895✔

2295

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

12,359,895✔

2296

        (*ppRes) = result;

6,343,042✔

2297

        return code;

6,349,364✔

2301

      pInfo->currentTable++;

6,017,147✔

2303

      taosRLockLatch(&pTaskInfo->lock);

6,015,179✔

2304

      numOfTables = 0;

6,021,744✔

2305

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

6,021,744✔

2306

      if (code != TSDB_CODE_SUCCESS) {

6,018,765✔

2311

      if (pInfo->currentTable >= numOfTables) {

6,018,765✔

2312

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

45,776✔

2313

        taosRUnLockLatch(&pTaskInfo->lock);

45,776✔

2314

        (*ppRes) = NULL;

45,776✔

2315

        return code;

45,776✔

2318

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

5,966,746✔

2319

      if (!tmp) {

5,968,698✔

2325

      tInfo = *tmp;

5,968,698✔

2326

      taosRUnLockLatch(&pTaskInfo->lock);

5,966,061✔

2328

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

5,974,619✔

2329

      QUERY_CHECK_CODE(code, lino, _end);

5,971,675✔

2330

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

5,971,675✔

2333

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

5,981,245✔

2334

      QUERY_CHECK_CODE(code, lino, _end);

5,965,764✔

2335

      pInfo->scanTimes = 0;

5,965,764✔

2338

    code = groupSeqTableScan(pOperator, ppRes);

800,831,606✔

2339

    QUERY_CHECK_CODE(code, lino, _end);

799,702,404✔

2342

_end:

799,700,054✔

2343

  if (code != TSDB_CODE_SUCCESS) {

799,679,460✔

2344

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

3,011✔

2345

    pTaskInfo->code = code;

3,011✔

2346

    T_LONG_JMP(pTaskInfo->env, code);

3,011✔

2349

  return code;

799,676,449✔

2352

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

1,942,458✔

2353

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

1,942,458✔

2354

  if (!pRecorder) {

1,942,458✔

2357

  STableScanInfo* pTableScanInfo = pOptr->info;

1,942,458✔

2358

  *pRecorder = pTableScanInfo->base.readRecorder;

1,942,458✔

2359

  *pOptrExplain = pRecorder;

1,942,458✔

2360

  *len = sizeof(SFileBlockLoadRecorder);

1,942,458✔

2361

  return 0;

1,942,458✔

2364

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

258,444,427✔

2365

  cleanupQueryTableDataCond(&pBase->cond);

258,444,427✔

2366

  cleanupQueryTableDataCond(&pBase->orgCond);

258,414,706✔

2368

  if (pAPI->tsdReaderClose) {

258,432,783✔

2369

    pAPI->tsdReaderClose(pBase->dataReader);

258,436,771✔

2371

  pBase->dataReader = NULL;

258,410,283✔

2373

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

258,411,021✔

2374

    taosArrayDestroy(pBase->matchInfo.pList);

258,420,594✔

2377

  tableListDestroy(pBase->pTableListInfo);

258,389,458✔

2378

  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);

258,436,737✔

2379

  cleanupExprSupp(&pBase->pseudoSup);

258,399,448✔

2380

258,440,247✔

2382

static void cleanReaderForVTable(STableScanInfo* pInfo){

6,829,759✔

2383

  if (pInfo != NULL && pInfo->base.readerAPI.tsdReaderClose) {

6,829,759✔

2384

    void *pIter = taosHashIterate(pInfo->readerCache, NULL);

6,829,229✔

2385

    while (pIter != NULL) {

6,851,595✔

2386

      void **reader = pIter;

22,194✔

2387

      if (*reader) {

22,194✔

2388

        if (*reader == pInfo->base.dataReader) {

22,194✔

2389

          pInfo->base.dataReader = NULL;

19,830✔

2391

        pInfo->base.readerAPI.tsdReaderClose(*reader);

22,194✔

2393

      pIter = taosHashIterate(pInfo->readerCache, pIter);

22,194✔

2396

6,830,289✔

2398

static void destroyTableScanOperatorInfo(void* param) {

233,126,603✔

2399

  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;

233,126,603✔

2400

  blockDataDestroy(pTableScanInfo->pResBlock);

233,126,603✔

2401

  blockDataDestroy(pTableScanInfo->pOrgBlock);

233,119,754✔

2402

  taosHashCleanup(pTableScanInfo->pIgnoreTables);

233,120,790✔

2403

  if (pTableScanInfo->virtualStableScan && pTableScanInfo->readerCache) {

233,101,872✔

2404

    cleanReaderForVTable(pTableScanInfo);

6,827,846✔

2405

    taosHashCleanup(pTableScanInfo->readerCache);

6,828,828✔

2407

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

233,102,025✔

2408

  taosArrayDestroy(pTableScanInfo->pBlockColMap);

233,109,156✔

2409

  taosMemoryFreeClear(param);

233,117,770✔

2410

233,104,476✔

2412

static void resetClolumnReserve(SSDataBlock* pBlock, int32_t dataRequireFlag) {

233,036,121✔

2413

  if (pBlock && dataRequireFlag == FUNC_DATA_REQUIRED_NOT_LOAD) {

233,036,121✔

2414

    int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

7,899,724✔

2415

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

19,945,587✔

2416

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

12,058,824✔

2417

      if (pCol) {

12,054,320✔

2418

        pCol->info.noData = true;

12,053,184✔

2422

233,023,160✔

2424

static int32_t resetTableScanOperatorState(SOperatorInfo* pOper) {

461,645✔

2425

  int32_t         code = TSDB_CODE_SUCCESS;

461,645✔

2426

  STableScanInfo*   pInfo = pOper->info;

461,645✔

2427

  pOper->status = OP_NOT_OPENED;

461,645✔

2429

  pInfo->scanTimes = 0;

461,645✔

2430

  pInfo->currentGroupId = -1;

461,645✔

2431

  pInfo->tableEndIndex = -1;

461,645✔

2432

  pInfo->tableStartIndex = 0;

461,645✔

2433

  pInfo->currentTable = 0;

461,645✔

2434

  pInfo->scanMode = 0;

461,645✔

2435

  pInfo->countState = 0;

461,645✔

2436

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

461,645✔

2438

  if (!pInfo->virtualStableScan) {

461,645✔

2444

    if (pInfo->base.readerAPI.tsdReaderClose) {

278,991✔

2445

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

278,991✔

2447

    pInfo->base.dataReader = NULL;

278,991✔

2449

  tableListDestroy(pInfo->base.pTableListInfo);

461,645✔

2450

  pInfo->base.pTableListInfo = tableListCreate();

461,645✔

2451

  if (!pInfo->base.pTableListInfo) {

461,645✔

2455

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

461,251✔

2457

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

461,645✔

2458

  if (!pTableScanNode->scan.node.dynamicOp) {

461,645✔

2459

    code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

554,970✔

2461

                                    pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

554,970✔

2462

    if (code) {

277,911✔

2468

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

461,645✔

2469

  cleanupQueryTableDataCond(&pInfo->base.cond);

461,645✔

2470

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

461,645✔

2471

  if (code) {

461,645✔

2475

  if (pTableScanNode->scan.node.dynamicOp && pTableScanNode->scan.virtualStableScan) {

461,645✔

2476

    cleanupQueryTableDataCond(&pInfo->base.orgCond);

182,654✔

2477

    memcpy(&pInfo->base.orgCond, &pInfo->base.cond, sizeof(SQueryTableDataCond));

182,654✔

2478

    memset(&pInfo->base.cond, 0, sizeof(SQueryTableDataCond));

182,654✔

2481

  pOper->resultInfo.totalRows = 0;

461,645✔

2482

  blockDataEmpty(pInfo->pResBlock);

461,645✔

2483

  blockDataEmpty(pInfo->pOrgBlock);

461,645✔

2484

  taosHashClear(pInfo->pIgnoreTables);

461,645✔

2485

  taosArrayDestroy(pInfo->pBlockColMap);

461,645✔

2486

  pInfo->pBlockColMap = NULL;

461,645✔

2487

  return code;

461,645✔

2490

int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

233,121,929✔

2493

  QRY_PARAM_CHECK(pOptrInfo);

233,121,929✔

2495

  int32_t         code = TSDB_CODE_SUCCESS;

233,129,909✔

2496

  int32_t         lino = 0;

233,129,909✔

2497

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

233,129,909✔

2498

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

233,118,488✔

2499

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

233,091,046✔

2504

  pOperator->pPhyNode = pTableScanNode;

233,091,597✔

2505

  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;

233,095,916✔

2506

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

233,096,980✔

2508

  int32_t numOfCols = 0;

233,101,123✔

2510

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

233,104,427✔

2511

  QUERY_CHECK_CODE(code, lino, _error);

233,040,050✔

2513

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

233,040,050✔

2514

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

233,025,137✔

2515

  QUERY_CHECK_CODE(code, lino, _error);

232,908,454✔

2517

  if (pScanNode->pScanPseudoCols != NULL) {

232,908,454✔

2518

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

55,079,723✔

2519

    pSup->pExprInfo = NULL;

55,078,366✔

2520

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

55,072,759✔

2521

    QUERY_CHECK_CODE(code, lino, _error);

55,074,777✔

2523

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

55,074,777✔

2525

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

55,054,406✔

2528

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

232,860,505✔

2529

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

232,946,155✔

2531

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

232,902,700✔

2532

  pInfo->base.readHandle = *readHandle;

232,854,117✔

2533

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

232,884,559✔

2535

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

232,869,804✔

2536

  pInfo->sample.seed = taosGetTimestampSec();

232,906,286✔

2538

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

233,018,292✔

2539

  initResultSizeInfo(&pOperator->resultInfo, 4096);

232,933,119✔

2540

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

232,940,146✔

2541

  resetClolumnReserve(pInfo->pResBlock, pInfo->base.dataBlockLoadFlag);

233,044,278✔

2542

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

232,948,245✔

2544

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

232,898,263✔

2545

  QUERY_CHECK_CODE(code, lino, _error);

232,910,854✔

2547

  pInfo->virtualStableScan = pScanNode->virtualStableScan;

232,910,854✔

2548

  if (pScanNode->node.dynamicOp && pScanNode->virtualStableScan) {

232,891,422✔

2549

    TSWAP(pInfo->pOrgBlock, pInfo->pResBlock);

6,823,452✔

2550

    pInfo->pResBlock = NULL;

6,819,878✔

2551

    memcpy(&pInfo->base.orgCond, &pInfo->base.cond, sizeof(SQueryTableDataCond));

6,822,033✔

2552

    memset(&pInfo->base.cond, 0, sizeof(SQueryTableDataCond));

6,822,676✔

2555

  if (pInfo->virtualStableScan) {

232,873,359✔

2556

    pInfo->readerCache = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);

6,823,567✔

2557

    QUERY_CHECK_NULL(pInfo->readerCache, code, lino, _error, terrno);

6,830,475✔

2560

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

232,857,204✔

2561

                            pTaskInfo->pStreamRuntimeInfo);

232,912,456✔

2562

  QUERY_CHECK_CODE(code, lino, _error);

232,917,371✔

2564

  pInfo->currentGroupId = -1;

232,893,945✔

2566

  pInfo->tableEndIndex = -1;

232,914,583✔

2567

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

232,910,228✔

2568

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

232,920,155✔

2570

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

232,839,822✔

2572

  pOperator->exprSupp.numOfExprs = numOfCols;

232,827,775✔

2574

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

232,930,766✔

2575

  pInfo->ignoreTag = false;

232,918,808✔

2577

  pInfo->lastBatchIdx = -1;

232,871,194✔

2578

  pInfo->currentBatchIdx = 0;

232,862,112✔

2579

  pInfo->base.pTableListInfo = pTableListInfo;

232,864,270✔

2580

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

232,849,409✔

2590

  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

232,849,409✔

2592

  pOperator->fpSet = createOperatorFpSet(prepareTableScan, doTableScanNext,

232,942,102✔

2598

  setOperatorResetStateFn(pOperator, resetTableScanOperatorState);

232,912,449✔

2600

  pOperator->cost.openCost = 0;

232,821,079✔

2601

  *pOptrInfo = pOperator;

232,823,616✔

2602

  return TSDB_CODE_SUCCESS;

232,804,344✔

2604

_error:

23,426✔

2605

  if (pInfo != NULL) {

23,426✔

2606

    pInfo->base.pTableListInfo = NULL;  // this attribute will be destroy outside of this function

23,426✔

2607

    destroyTableScanOperatorInfo(pInfo);

23,426✔

2610

  if (pOperator != NULL) {

23,426✔

2611

    pOperator->info = NULL;

23,426✔

2612

    destroyOperator(pOperator);

23,426✔

2614

  pTaskInfo->code = code;

23,426✔

2615

  return code;

23,426✔

2652

static int32_t doBlockDataPrimaryKeyFilter(SSDataBlock* pBlock, STqOffsetVal* offset) {

14,319✔

2653

  int32_t code = TSDB_CODE_SUCCESS;

14,319✔

2654

  int32_t lino = 0;

14,319✔

2655

  if (pBlock->info.window.skey != offset->ts || offset->primaryKey.type == 0) {

14,319✔

2656

    return code;

1,733✔

2658

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

12,586✔

2659

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

12,586✔

2660

  bool hasUnqualified = false;

12,586✔

2662

  SColumnInfoData* pColTs = taosArrayGet(pBlock->pDataBlock, 0);

12,586✔

2663

  SColumnInfoData* pColPk = taosArrayGet(pBlock->pDataBlock, 1);

12,586✔

2665

  qDebug("doBlockDataWindowFilter primary key, ts:%" PRId64 " %" PRId64, offset->ts,

12,586✔

2667

  QUERY_CHECK_CONDITION((pColPk->info.type == offset->primaryKey.type), code, lino, _end,

12,586✔

2670

  __compar_fn_t func = getComparFunc(pColPk->info.type, 0);

12,586✔

2671

  QUERY_CHECK_NULL(func, code, lino, _end, terrno);

12,586✔

2672

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

45,212,812✔

2673

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

45,200,226✔

2674

    void*    data = colDataGetData(pColPk, i);

45,200,226✔

2675

    if (IS_VAR_DATA_TYPE(pColPk->info.type)) {

45,202,666✔

2676

      if (IS_STR_DATA_BLOB(pColPk->info.type)) {

2,440✔

2679

      void* tmq = taosMemoryMalloc(offset->primaryKey.nData + VARSTR_HEADER_SIZE);

2,440✔

2680

      QUERY_CHECK_NULL(tmq, code, lino, _end, terrno);

2,440✔

2681

      memcpy(varDataVal(tmq), offset->primaryKey.pData, offset->primaryKey.nData);

2,440✔

2682

      varDataLen(tmq) = offset->primaryKey.nData;

2,440✔

2683

      p[i] = (*ts > offset->ts) || (func(data, tmq) > 0);

2,440✔

2684

      taosMemoryFree(tmq);

2,440✔

2686

      p[i] = (*ts > offset->ts) || (func(data, VALUE_GET_DATUM(&offset->primaryKey, pColPk->info.type)) > 0);

45,197,786✔

2689

    if (!p[i]) {

45,200,226✔

2690

      hasUnqualified = true;

16,081,506✔

2694

  if (hasUnqualified) {

12,586✔

2695

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

4,231✔

2696

    QUERY_CHECK_CODE(code, lino, _end);

4,231✔

2699

_end:

12,586✔

2700

  taosMemoryFree(p);

12,586✔

2701

  if (code != TSDB_CODE_SUCCESS) {

12,586✔

2704

  return code;

12,586✔

2718

static int32_t processBlock(STmqQueryScanInfo* pInfo) {

19,853,947✔

2719

  int32_t         code = TSDB_CODE_SUCCESS;

19,853,947✔

2720

  int32_t         lino = 0;

19,853,947✔

2721

  SOperatorInfo*  pOperator = pInfo->pTmqScanOp;

19,853,947✔

2723

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

19,853,947✔

2724

  QUERY_CHECK_CODE(code, lino, _end);

19,853,947✔

2726

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

19,853,947✔

2727

  QUERY_CHECK_CODE(code, lino, _end);

19,853,747✔

2729

_end:

19,853,747✔

2730

  if (code != TSDB_CODE_SUCCESS) {

19,853,747✔

2733

  return code;

19,853,947✔

2736

static int32_t processPrimaryKey(SSDataBlock* pBlock, bool hasPrimaryKey, STqOffsetVal* offset) {

7,133,274✔

2737

  int32_t code = TSDB_CODE_SUCCESS;

7,133,274✔

2738

  int32_t lino = 0;

7,133,274✔

2739

  SValue  val = {0};

7,133,274✔

2740

  if (hasPrimaryKey) {

7,133,586✔

2741

    code = doBlockDataPrimaryKeyFilter(pBlock, offset);

14,319✔

2742

    if (code != TSDB_CODE_SUCCESS) {

14,319✔

2746

    SColumnInfoData* pColPk = taosArrayGet(pBlock->pDataBlock, 1);

14,319✔

2748

    if (pBlock->info.rows < 1) {

14,319✔

2749

      return code;

3,926✔

2751

    void* tmp = colDataGetData(pColPk, pBlock->info.rows - 1);

10,393✔

2752

    val.type = pColPk->info.type;

10,393✔

2753

    if (IS_VAR_DATA_TYPE(pColPk->info.type)) {

10,393✔

2754

      if (IS_STR_DATA_BLOB(pColPk->info.type)) {

662✔

2757

      val.pData = taosMemoryMalloc(varDataLen(tmp));

662✔

2758

      QUERY_CHECK_NULL(val.pData, code, lino, _end, terrno);

662✔

2759

      val.nData = varDataLen(tmp);

662✔

2760

      memcpy(val.pData, varDataVal(tmp), varDataLen(tmp));

662✔

2762

      valueSetDatum(&val, pColPk->info.type, tmp, pColPk->info.bytes);

9,731✔

2765

  tqOffsetResetToData(offset, pBlock->info.id.uid, pBlock->info.window.ekey, val);

7,129,660✔

2767

_end:

7,129,660✔

2768

  if (code != TSDB_CODE_SUCCESS) {

7,129,972✔

2771

  return code;

7,129,660✔

2774

static int32_t doQueueScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

20,048,014✔

2775

  int32_t        code = TSDB_CODE_SUCCESS;

20,048,014✔

2776

  int32_t        lino = 0;

20,048,014✔

2777

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

20,048,014✔

2778

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

20,048,979✔

2780

  STmqQueryScanInfo* pInfo = pOperator->info;

20,048,659✔

2781

  const char*      id = GET_TASKID(pTaskInfo);

20,047,894✔

2783

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

20,048,059✔

2785

  if (isTaskKilled(pTaskInfo)) {

20,050,041✔

2790

  blockDataCleanup(pInfo->pRes);

20,049,675✔

2791

  int64_t st = taosGetTimestampMs();

20,049,475✔

2793

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

20,049,475✔

2794

    while (1) {

6,063,987✔

2795

      SSDataBlock* pResult = NULL;

6,393,959✔

2796

      code = doTableScanNext(pInfo->pTableScanOp, &pResult);

6,393,493✔

2797

      QUERY_CHECK_CODE(code, lino, _end);

6,395,140✔

2799

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

6,395,140✔

2800

        bool hasPrimaryKey = pAPI->tqReaderFn.tqGetTablePrimaryKey(pInfo->tqReader);

6,349,364✔

2801

        code = processPrimaryKey(pResult, hasPrimaryKey, &pTaskInfo->tmqInfo.currentOffset);

6,349,389✔

2802

        QUERY_CHECK_CODE(code, lino, _end);

6,349,364✔

2803

        qDebug("tmqsnap doQueueScanNext get data uid:%" PRId64, pResult->info.id.uid);

6,349,364✔

2804

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

6,349,364✔

2805

          code = blockDataMerge(pInfo->pRes, pResult);

6,345,771✔

2806

          QUERY_CHECK_CODE(code, lino, _end);

6,344,449✔

2812

      if (pInfo->pRes->info.rows >= pTaskInfo->tmqInfo.minPollRows){

6,348,375✔

2813

        (*ppRes) = pInfo->pRes;

280,114✔

2814

        return code;

280,114✔

2816

      int64_t elapsed = taosGetTimestampMs() - st;

6,067,929✔

2817

      if (elapsed > pTaskInfo->tmqInfo.timeout || elapsed < 0) {

6,067,929✔

2818

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

3,610✔

2821

        (*ppRes) = pInfo->pRes;

3,942✔

2822

        return code;

3,942✔

2826

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

45,776✔

2827

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

45,776✔

2829

    pTSInfo->base.dataReader = NULL;

45,776✔

2830

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

45,776✔

2831

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

45,776✔

2832

    code = pAPI->tqReaderFn.tqReaderSeek(pInfo->tqReader, validVer, pTaskInfo->id.str);

45,776✔

2833

    if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) {

45,776✔

2834

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

13,175✔

2835

        (*ppRes) = NULL;

5,268✔

2837

        (*ppRes) = pInfo->pRes;

7,907✔

2839

      return 0;

13,175✔

2841

    QUERY_CHECK_CODE(code, lino, _end);

32,601✔

2843

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

32,601✔

2846

  while (1) {

101,503✔

2847

    code = pAPI->tqReaderFn.tqReaderNextBlockInWal(pInfo->tqReader, pInfo->pRes,

59,559,771✔

2851

                                                              pTaskInfo->tmqInfo.sourceExcluded,

19,853,239✔

2854

                                                              pTaskInfo->tmqInfo.enableReplay);

19,853,547✔

2855

    if (code != 0 && code != TSDB_CODE_TMQ_FETCH_TIMEOUT && code != TSDB_CODE_WAL_LOG_NOT_EXIST) {

19,853,584✔

2858

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

19,853,584✔

2861

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

19,853,747✔

2864

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

19,853,947✔

2866

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

19,853,384✔

2868

    int32_t ret = processBlock(pInfo);

19,853,584✔

2869

    if (ret != 0) {

19,853,947✔

2873

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

19,853,947✔

2876

    if (code != 0) {

19,853,584✔

2877

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

19,237,255✔

2878

        (*ppRes) = NULL;

9,161,827✔

2880

        (*ppRes) = pInfo->pRes;

10,075,791✔

2882

      return 0;

19,237,618✔

2885

    if (pInfo->pRes->info.rows >= pTaskInfo->tmqInfo.minPollRows ||

616,329✔

2886

        (pInfo->pRes->info.rows > 0 && pTaskInfo->tmqInfo.enableReplay)) {

106,831✔

2887

      (*ppRes) = pInfo->pRes;

514,826✔

2888

      return 0;

514,826✔

2902

int32_t extractTableIdList(const STableListInfo* pTableListInfo, SArray** ppArrayRes) {

346,218✔

2903

  int32_t code = TSDB_CODE_SUCCESS;

346,218✔

2904

  int32_t lino = 0;

346,218✔

2905

  SArray* tableIdList = taosArrayInit(4, sizeof(uint64_t));

346,218✔

2906

  QUERY_CHECK_NULL(tableIdList, code, lino, _end, terrno);

347,421✔

2909

  int32_t size = 0;

347,421✔

2910

  code = tableListGetSize(pTableListInfo, &size);

347,421✔

2911

  QUERY_CHECK_CODE(code, lino, _end);

347,421✔

2912

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

10,035,552✔

2913

    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);

9,683,577✔

2914

    QUERY_CHECK_NULL(pkeyInfo, code, lino, _end, terrno);

9,675,241✔

2915

    void* tmp = taosArrayPush(tableIdList, &pkeyInfo->uid);

9,675,241✔

2916

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

9,688,131✔

2919

  (*ppArrayRes) = tableIdList;

351,975✔

2921

_end:

347,100✔

2922

  if (code != TSDB_CODE_SUCCESS) {

347,421✔

2925

  return code;

347,421✔

2928

static int32_t doRawScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

1,581,560✔

2929

  int32_t        code = TSDB_CODE_SUCCESS;

1,581,560✔

2930

  int32_t        lino = 0;

1,581,560✔

2931

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,581,560✔

2932

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,581,560✔

2934

  STmqRawScanInfo* pInfo = pOperator->info;

1,581,560✔

2935

  pTaskInfo->tmqInfo.btMetaRsp.batchMetaReq = NULL;  // use batchMetaReq != NULL to judge if data is meta

1,581,560✔

2936

  pTaskInfo->tmqInfo.btMetaRsp.batchMetaLen = NULL;

1,581,560✔

2938

  qDebug("tmqsnap doRawScan called");

1,581,560✔

2939

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

1,581,560✔

2940

    bool hasNext = false;

1,574,950✔

2941

    if (pInfo->dataReader && pInfo->sContext->withMeta != ONLY_META) {

1,574,950✔

2942

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

1,573,677✔

2943

      if (code != TSDB_CODE_SUCCESS) {

1,572,429✔

2949

    if (pInfo->dataReader && hasNext) {

1,573,702✔

2950

      if (isTaskKilled(pTaskInfo)) {

784,846✔

2955

      SSDataBlock* pBlock = NULL;

784,846✔

2956

      code = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pInfo->dataReader, &pBlock);

784,846✔

2957

      QUERY_CHECK_CODE(code, lino, _end);

784,222✔

2959

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

784,222✔

2960

        bool hasPrimaryKey = pAPI->snapshotFn.taosXGetTablePrimaryKey(pInfo->sContext);

784,222✔

2961

        code = processPrimaryKey(pBlock, hasPrimaryKey, &pTaskInfo->tmqInfo.currentOffset);

784,534✔

2962

        QUERY_CHECK_CODE(code, lino, _end);

784,222✔

2963

        qDebug("tmqsnap doRawScan get data uid:%" PRId64, pBlock->info.id.uid);

784,222✔

2964

        (*ppRes) = pBlock;

784,846✔

2965

        return code;

784,846✔

2969

    SMetaTableInfo mtInfo = {0};

788,856✔

2970

    code = pAPI->snapshotFn.getMetaTableInfoFromSnapshot(pInfo->sContext, &mtInfo);

790,104✔

2971

    QUERY_CHECK_CODE(code, lino, _end);

788,544✔

2972

    if (code != 0) {

788,544✔

2976

    STqOffsetVal offset = {0};

788,544✔

2977

    if (mtInfo.uid == 0 || pInfo->sContext->withMeta == ONLY_META) {  // read snapshot done, change to get data from wal

788,856✔

2978

      qDebug("tmqsnap read snapshot done, change to get data from wal");

11,029✔

2979

      tqOffsetResetToLog(&offset, pInfo->sContext->snapVersion + 1);

11,029✔

2981

      SValue val = {0};

778,763✔

2982

      tqOffsetResetToData(&offset, mtInfo.uid, INT64_MIN, val);

778,763✔

2983

      qDebug("tmqsnap change get data uid:%" PRId64, mtInfo.uid);

777,767✔

2986

    code = qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);

790,104✔

2987

    QUERY_CHECK_CODE(code, lino, _end);

790,104✔

2988

    (*ppRes) = NULL;

790,104✔

2989

    return code;

790,104✔

2990

  } else if (pTaskInfo->tmqInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_META) {

6,610✔

2991

    SSnapContext* sContext = pInfo->sContext;

6,610✔

2992

    int64_t st = taosGetTimestampMs();

6,610✔

2993

    for (int32_t i = 0; i < pTaskInfo->tmqInfo.minPollRows; i++) {

51,354✔

2994

      void*   data = NULL;

50,640✔

2995

      int32_t dataLen = 0;

50,640✔

2996

      int16_t type = 0;

50,640✔

2997

      int64_t uid = 0;

50,352✔

2998

      if (pAPI->snapshotFn.getTableInfoFromSnapshot(sContext, &data, &dataLen, &type, &uid) < 0) {

50,640✔

3004

      if (!sContext->queryMeta) {  // change to get data next poll request

50,640✔

3005

        STqOffsetVal offset = {0};

5,896✔

3006

        SValue       val = {0};

5,896✔

3008

        code = qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);

5,896✔

3009

        QUERY_CHECK_CODE(code, lino, _end);

5,896✔

3010

        break;

5,896✔

3012

        tqOffsetResetToMeta(&pTaskInfo->tmqInfo.currentOffset, uid);

44,744✔

3013

        SMqMetaRsp tmpMetaRsp = {0};

44,744✔

3014

        tmpMetaRsp.resMsgType = type;

44,744✔

3015

        tmpMetaRsp.metaRspLen = dataLen;

44,744✔

3016

        tmpMetaRsp.metaRsp = data;

44,744✔

3017

        if (!pTaskInfo->tmqInfo.btMetaRsp.batchMetaReq) {

44,744✔

3018

          pTaskInfo->tmqInfo.btMetaRsp.batchMetaReq = taosArrayInit(4, POINTER_BYTES);

5,320✔

3019

          QUERY_CHECK_NULL(pTaskInfo->tmqInfo.btMetaRsp.batchMetaReq, code, lino, _end, terrno);

5,320✔

3021

          pTaskInfo->tmqInfo.btMetaRsp.batchMetaLen = taosArrayInit(4, sizeof(int32_t));

5,320✔

3022

          QUERY_CHECK_NULL(pTaskInfo->tmqInfo.btMetaRsp.batchMetaLen, code, lino, _end, terrno);

5,320✔

3024

        int32_t  tempRes = TSDB_CODE_SUCCESS;

44,744✔

3025

        uint32_t len = 0;

44,744✔

3026

        tEncodeSize(tEncodeMqMetaRsp, &tmpMetaRsp, len, tempRes);

44,744✔

3027

        if (TSDB_CODE_SUCCESS != tempRes) {

44,744✔

3033

        int32_t tLen = sizeof(SMqRspHead) + len;

44,744✔

3034

        void*   tBuf = taosMemoryCalloc(1, tLen);

44,744✔

3035

        QUERY_CHECK_NULL(tBuf, code, lino, _end, terrno);

44,744✔

3037

        void*    metaBuff = POINTER_SHIFT(tBuf, sizeof(SMqRspHead));

44,744✔

3038

        SEncoder encoder = {0};

44,744✔

3039

        tEncoderInit(&encoder, metaBuff, len);

44,744✔

3040

        int32_t tempLen = tEncodeMqMetaRsp(&encoder, &tmpMetaRsp);

44,744✔

3041

        if (tempLen < 0) {

44,744✔

3048

        taosMemoryFreeClear(data);

44,744✔

3049

        void* tmp = taosArrayPush(pTaskInfo->tmqInfo.btMetaRsp.batchMetaReq, &tBuf);

44,744✔

3050

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

44,744✔

3052

        tmp = taosArrayPush(pTaskInfo->tmqInfo.btMetaRsp.batchMetaLen, &tLen);

44,744✔

3053

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

44,744✔

3055

        int64_t elapsed = taosGetTimestampMs() - st;

44,744✔

3056

        if (elapsed > pTaskInfo->tmqInfo.timeout || elapsed < 0) {

44,744✔

3062

    (*ppRes) = NULL;

6,610✔

3063

    return code;

6,610✔

3077

static void destroyTmqRawScanOperatorInfo(void* param) {

128,902✔

3078

  STmqRawScanInfo* pRawScan = (STmqRawScanInfo*)param;

128,902✔

3079

  pRawScan->pAPI->tsdReader.tsdReaderClose(pRawScan->dataReader);

128,902✔

3080

  pRawScan->pAPI->snapshotFn.destroySnapshot(pRawScan->sContext);

128,902✔

3081

  tableListDestroy(pRawScan->pTableListInfo);

128,902✔

3082

  taosMemoryFree(pRawScan);

128,902✔

3083

128,902✔

3088

int32_t createTmqRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {

128,257✔

3094

  QRY_PARAM_CHECK(pOptrInfo);

128,257✔

3095

  int32_t code = TSDB_CODE_SUCCESS;

128,902✔

3096

  int32_t lino = 0;

128,902✔

3098

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

128,902✔

3099

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

128,614✔

3100

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

128,614✔

3106

  pInfo->pTableListInfo = tableListCreate();

128,614✔

3107

  QUERY_CHECK_NULL(pInfo->pTableListInfo, code, lino, _end, terrno);

128,902✔

3108

  pInfo->vnode = pHandle->vnode;

128,902✔

3109

  pInfo->pAPI = &pTaskInfo->storageAPI;

128,902✔

3111

  pInfo->sContext = pHandle->sContext;

128,614✔

3112

  pHandle->sContext = NULL;  // transfer the ownership

128,614✔

3113

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

128,902✔

3116

  pOperator->fpSet = createOperatorFpSet(NULL, doRawScanNext, NULL, destroyTmqRawScanOperatorInfo, optrDefaultBufFn,

128,614✔

3118

  *pOptrInfo = pOperator;

128,614✔

3119

  return code;

128,614✔

3131

void destroyTmqScanOperatorInfo(void* param) {

347,421✔

3132

  if (param == NULL) {

347,421✔

3136

  STmqQueryScanInfo* pTmqScan = (STmqQueryScanInfo*)param;

347,421✔

3137

  if (pTmqScan->pTableScanOp && pTmqScan->pTableScanOp->info) {

347,421✔

3138

    destroyOperator(pTmqScan->pTableScanOp);

347,421✔

3141

  if (pTmqScan->tqReader != NULL && pTmqScan->readerFn.tqReaderClose != NULL) {

347,129✔

3142

    pTmqScan->readerFn.tqReaderClose(pTmqScan->tqReader);

347,421✔

3145

  if (pTmqScan->pTableListInfo) {

347,129✔

3149

  if (pTmqScan->pPseudoExpr) {

347,421✔

3150

    destroyExprInfo(pTmqScan->pPseudoExpr, pTmqScan->numOfPseudoExpr);

202,510✔

3151

    taosMemoryFree(pTmqScan->pPseudoExpr);

202,218✔

3154

  taosHashCleanup(pTmqScan->pCol2SlotId);

346,805✔

3155

  pTmqScan->pCol2SlotId = NULL;

346,821✔

3157

  cleanupExprSupp(&pTmqScan->tagCalSup);

346,821✔

3159

  blockDataDestroy(pTmqScan->pRes);

347,421✔

3161

  taosMemoryFree(pTmqScan);

347,421✔

3181

static void getTsSlotId(SNodeList* pNodeList, int32_t* pTsSlotId) {

347,124✔

3182

  size_t  numOfCols = LIST_LENGTH(pNodeList);

347,124✔

3183

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

2,187,959✔

3184

    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);

1,841,109✔

3185

    if (pNode && nodeType(pNode->pExpr) == QUERY_NODE_COLUMN && ((SColumnNode*)pNode->pExpr)->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

1,841,109✔

3186

      *pTsSlotId = pNode->slotId;

347,421✔

3189

346,850✔

3191

static int32_t buildCol2SlotId(SNodeList* pNodeList, SHashObj* pCol2SlotId) {

346,805✔

3192

  size_t  numOfCols = LIST_LENGTH(pNodeList);

346,805✔

3193

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

2,187,914✔

3194

    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);

1,840,493✔

3195

    if (pNode == NULL) {

1,841,109✔

3198

    if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN) {

1,841,109✔

3199

      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;

1,840,493✔

3200

      int32_t code = taosHashPut(pCol2SlotId, (const void*)&pColNode->colId, sizeof(pColNode->colId), (const void*)&pNode->slotId, sizeof(pNode->slotId));

1,840,493✔

3201

      if (code != TSDB_CODE_SUCCESS) {

1,841,087✔

3206

  return TSDB_CODE_SUCCESS;

347,421✔

3209

int32_t createTmqScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,

346,968✔

3212

  QRY_PARAM_CHECK(pOptrInfo);

346,968✔

3214

  int32_t          code = TSDB_CODE_SUCCESS;

347,421✔

3215

  int32_t          lino = 0;

347,421✔

3216

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

347,421✔

3217

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

347,421✔

3218

  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;

347,124✔

3219

  const char*      idstr = pTaskInfo->id.str;

347,124✔

3221

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

347,124✔

3226

  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;

347,124✔

3227

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;

347,124✔

3229

  pInfo->pTagCond = pTagCond;

347,124✔

3230

  pInfo->pGroupTags = pTableScanNode->pGroupTags;

347,124✔

3232

  pInfo->primaryTsIndex = -1;

347,421✔

3233

  getTsSlotId(pScanPhyNode->pScanCols, &pInfo->primaryTsIndex);

347,421✔

3234

  pInfo->pCol2SlotId = taosHashInit(LIST_LENGTH(pScanPhyNode->pScanCols), taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_ENTRY_LOCK);

347,421✔

3235

  QUERY_CHECK_NULL(pInfo->pCol2SlotId, code, lino, _error, terrno);

347,102✔

3236

  code = buildCol2SlotId(pScanPhyNode->pScanCols, pInfo->pCol2SlotId);

346,805✔

3237

  QUERY_CHECK_CODE(code, lino, _error);

347,421✔

3239

  SDataBlockDescNode* pNode = (SDataBlockDescNode*)pDescNode;

347,421✔

3240

  if (LIST_LENGTH(pNode->pSlots) != LIST_LENGTH(pScanPhyNode->pScanCols) + LIST_LENGTH(pScanPhyNode->pScanPseudoCols)) {

347,421✔

3246

  if (pTableScanNode->pTags != NULL) {

347,124✔

3258

  SOperatorInfo* pTableScanOp = NULL;

347,124✔

3259

  code = createTableScanOperatorInfo(pTableScanNode, pHandle, pTableListInfo, pTaskInfo, &pTableScanOp);

347,124✔

3260

  if (pTableScanOp == NULL || code != 0) {

346,850✔

3261

    qError("createTableScanOperatorInfo error, code:%d", pTaskInfo->code);

615✔

3265

  STableScanInfo* pTSInfo = (STableScanInfo*)pTableScanOp->info;

346,235✔

3266

  if (pHandle->version > 0) {

346,850✔

3267

    pTSInfo->base.cond.endVersion = pHandle->version;

346,828✔

3270

  STableKeyInfo* pList = NULL;

344,947✔

3271

  int32_t        num = 0;

345,318✔

3272

  code = tableListGetGroupList(pTableListInfo, 0, &pList, &num);

345,615✔

3273

  QUERY_CHECK_CODE(code, lino, _error);

345,947✔

3275

  if (pHandle->initTableReader) {

345,947✔

3276

    pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;

345,943✔

3277

    pTSInfo->base.dataReader = NULL;

346,239✔

3280

  if (pHandle->initTqReader) {

345,618✔

3281

    pInfo->tqReader = pAPI->tqReaderFn.tqReaderOpen(pHandle->vnode);

346,266✔

3282

    QUERY_CHECK_NULL(pInfo->tqReader, code, lino, _error, terrno);

347,147✔

3288

  pInfo->pTableScanOp = pTableScanOp;

347,421✔

3290

  pInfo->readHandle = *pHandle;

346,539✔

3291

  pTaskInfo->tmqInfo.snapshotVer = pHandle->version;

347,109✔

3293

  SArray* tableIdList = NULL;

346,813✔

3294

  code = extractTableIdList(((STableScanInfo*)(pInfo->pTableScanOp->info))->base.pTableListInfo, &tableIdList);

347,421✔

3295

  QUERY_CHECK_CODE(code, lino, _error);

346,826✔

3296

  code = pAPI->tqReaderFn.tqReaderSetQueryTableList(pInfo->tqReader, tableIdList, idstr);

346,826✔

3297

  QUERY_CHECK_CODE(code, lino, _error);

347,421✔

3298

  taosArrayDestroy(tableIdList);

347,421✔

3299

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

347,421✔

3302

  code = createExprInfo(pScanPhyNode->pScanPseudoCols, NULL, &pInfo->pPseudoExpr, &pInfo->numOfPseudoExpr);

347,421✔

3303

  QUERY_CHECK_CODE(code, lino, _error);

347,421✔

3305

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

347,421✔

3306

  QUERY_CHECK_CODE(code, lino, _error);

347,421✔

3308

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

347,421✔

3309

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

347,421✔

3311

  pInfo->pTmqScanOp = pOperator;

347,124✔

3312

  pInfo->readerFn = pTaskInfo->storageAPI.tqReaderFn;

347,124✔

3314

  setOperatorInfo(pOperator, STREAM_SCAN_OP_NAME, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, false, OP_NOT_OPENED, pInfo,

347,124✔

3316

  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);

347,124✔

3318

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doQueueScanNext, NULL, destroyTmqScanOperatorInfo, optrDefaultBufFn,

347,124✔

3321

  *pOptrInfo = pOperator;

346,803✔

3322

  return code;

346,803✔

3341

static int32_t doTagScanOneTable(SOperatorInfo* pOperator, SSDataBlock* pRes, SMetaReader* mr, SStorageAPI* pAPI) {

11,523,467✔

3342

  int32_t        code = TSDB_CODE_SUCCESS;

11,523,467✔

3343

  int32_t        lino = 0;

11,523,467✔

3344

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

11,523,467✔

3345

  STagScanInfo*  pInfo = pOperator->info;

11,525,558✔

3346

  SExprInfo*     pExprInfo = &pOperator->exprSupp.pExprInfo[0];

11,525,322✔

3347

  int32_t        count = pRes->info.rows;

11,525,465✔

3349

  STableKeyInfo* item = tableListGetInfo(pInfo->pTableListInfo, pInfo->curPos);

11,521,175✔

3350

  if (!item) {

11,525,362✔

3357

  code = pAPI->metaReaderFn.getTableEntryByUid(mr, item->uid);

11,525,362✔

3358

  tDecoderClear(&(*mr).coder);

11,528,874✔

3359

  if (code != TSDB_CODE_SUCCESS) {

11,527,015✔

3365

  char str[512];

11,526,911✔

3366

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

52,734,032✔

3367

    SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pExprInfo[j].base.resSchema.slotId);

41,206,232✔

3370

    if (FUNCTION_TYPE_TBNAME == pExprInfo[j].pExpr->_function.functionType) {

41,208,785✔

3371

      STR_TO_VARSTR(str, (*mr).me.name);

4,850,336✔

3372

      code = colDataSetVal(pDst, (count), str, false);

4,847,672✔

3373

      QUERY_CHECK_CODE(code, lino, _end);

4,849,806✔

3374

    } else if (FUNCTION_TYPE_TBUID == pExprInfo[j].pExpr->_function.functionType) {

36,361,196✔

3375

      code = colDataSetVal(pDst, (count), (char*)&(*mr).me.uid, false);

7,345,442✔

3376

      QUERY_CHECK_CODE(code, lino, _end);

7,347,032✔

3377

    } else if (FUNCTION_TYPE_VGID == pExprInfo[j].pExpr->_function.functionType) {

29,014,969✔

3378

      code = colDataSetVal(pDst, (count), (char*)&pTaskInfo->id.vgId, false);

15,868✔

3379

      QUERY_CHECK_CODE(code, lino, _end);

15,868✔

3381

      STagVal val = {0};

28,997,070✔

3382

      val.cid = pExprInfo[j].base.pParam[0].pCol->colId;

28,996,093✔

3383

      const char* p = pAPI->metaFn.extractTagVal((*mr).me.ctbEntry.pTags, pDst->info.type, &val);

28,999,556✔

3385

      char* data = NULL;

28,993,593✔

3386

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

28,993,593✔

3387

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

28,994,621✔

3389

        data = (char*)p;

1,440✔

3392

      code = colDataSetVal(pDst, (count), data,

28,998,156✔

3393

                           (data == NULL) || (pDst->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data)));

28,997,268✔

3394

      QUERY_CHECK_CODE(code, lino, _end);

29,000,584✔

3396

      if ((pDst->info.type != TSDB_DATA_TYPE_JSON) && (p != NULL) && IS_VAR_DATA_TYPE(((const STagVal*)p)->type) &&

29,000,584✔

3398

        taosMemoryFree(data);

10,490,412✔

3403

_end:

11,534,514✔

3404

  if (code != TSDB_CODE_SUCCESS) {

11,528,079✔

3408

    pRes->info.rows++;

11,528,079✔

3411

  return code;

11,527,377✔

3414

static void tagScanFreeUidTag(void* p) {

7,419,415✔

3415

  STUidTagInfo* pInfo = p;

7,419,415✔

3416

  if (pInfo->pTagVal != NULL) {

7,419,415✔

3417

    taosMemoryFree(pInfo->pTagVal);

7,424,611✔

3419

7,417,307✔

3421

static int32_t tagScanCreateResultData(SDataType* pType, int32_t numOfRows, SScalarParam* pParam) {

1,681,082✔

3422

  SColumnInfoData* pColumnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));

1,681,082✔

3423

  if (pColumnData == NULL) {

1,681,830✔

3427

  pColumnData->info.type = pType->type;

1,681,830✔

3428

  pColumnData->info.bytes = pType->bytes;

1,681,594✔

3429

  pColumnData->info.scale = pType->scale;

1,681,594✔

3430

  pColumnData->info.precision = pType->precision;

1,681,600✔

3432

  int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows, true);

1,681,076✔

3433

  if (code != TSDB_CODE_SUCCESS) {

1,680,199✔

3439

  pParam->columnData = pColumnData;

1,680,199✔

3440

  pParam->colAlloced = true;

1,681,082✔

3441

  return TSDB_CODE_SUCCESS;

1,681,226✔

3444

static EDealRes tagScanRewriteTagColumn(SNode** pNode, void* pContext) {

10,327,387✔

3445

  int32_t                code = TSDB_CODE_SUCCESS;

10,327,387✔

3446

  int32_t                lino = 0;

10,327,387✔

3447

  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;

10,327,387✔

3448

  SColumnNode*           pSColumnNode = NULL;

10,327,387✔

3449

  if (QUERY_NODE_COLUMN == nodeType((*pNode))) {

10,328,114✔

3450

    pSColumnNode = *(SColumnNode**)pNode;

2,901,490✔

3451

  } else if (QUERY_NODE_FUNCTION == nodeType((*pNode))) {

7,428,771✔

3452

    SFunctionNode* pFuncNode = *(SFunctionNode**)(pNode);

540✔

3453

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

540✔

3466

      return DEAL_RES_CONTINUE;

540✔

3469

    return DEAL_RES_CONTINUE;

7,427,722✔

3472

  void* data = taosHashGet(pCtx->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId));

2,901,864✔

3473

  if (!data) {

2,900,624✔

3474

    code = taosHashPut(pCtx->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId), pNode, sizeof((*pNode)));

1,684,189✔

3475

    if (code == TSDB_CODE_DUP_KEY) {

1,684,926✔

3478

    QUERY_CHECK_CODE(code, lino, _end);

1,684,926✔

3479

    pSColumnNode->slotId = pCtx->index++;

1,684,926✔

3480

    SColumnInfo cInfo = {.colId = pSColumnNode->colId,

1,684,690✔

3481

                         .type = pSColumnNode->node.resType.type,

1,684,184✔

3482

                         .bytes = pSColumnNode->node.resType.bytes};

1,684,552✔

3483

    void*       tmp = taosArrayPush(pCtx->cInfoList, &cInfo);

1,683,810✔

3484

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

1,685,435✔

3486

    SColumnNode* col = *(SColumnNode**)data;

1,216,435✔

3487

    pSColumnNode->slotId = col->slotId;

1,216,435✔

3490

_end:

2,902,747✔

3491

  if (code != TSDB_CODE_SUCCESS) {

2,900,844✔

3495

  return DEAL_RES_CONTINUE;

2,902,750✔

3498

static int32_t tagScanFilterByTagCond(SArray* aUidTags, SNode* pTagCond, SArray* aFilterIdxs, void* pVnode,

1,681,457✔

3500

  int32_t code = TSDB_CODE_SUCCESS;

1,681,457✔

3501

  int32_t lino = 0;

1,681,457✔

3502

  int32_t numOfTables = taosArrayGetSize(aUidTags);

1,681,457✔

3503

  SArray* pBlockList = NULL;

1,681,831✔

3505

  SSDataBlock* pResBlock = createTagValBlockForFilter(pInfo->filterCtx.cInfoList, numOfTables, aUidTags, pVnode, pAPI);

1,681,831✔

3506

  QUERY_CHECK_NULL(pResBlock, code, lino, _end, terrno);

1,681,457✔

3508

  pBlockList = taosArrayInit(1, POINTER_BYTES);

1,681,457✔

3509

  QUERY_CHECK_NULL(pBlockList, code, lino, _end, terrno);

1,681,456✔

3511

  void* tmp = taosArrayPush(pBlockList, &pResBlock);

1,682,342✔

3512

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

1,682,342✔

3514

  SDataType type = {.type = TSDB_DATA_TYPE_BOOL, .bytes = sizeof(bool)};

1,682,342✔

3516

  SScalarParam output = {0};

1,681,830✔

3517

  code = tagScanCreateResultData(&type, numOfTables, &output);

1,681,830✔

3518

  QUERY_CHECK_CODE(code, lino, _end);

1,681,600✔

3520

  gTaskScalarExtra.pStreamInfo = (void*)pStreamRuntimeInfo;

1,681,600✔

3521

  gTaskScalarExtra.pStreamRange = NULL;

1,681,600✔

3522

  code = scalarCalculate(pTagCond, pBlockList, &output, &gTaskScalarExtra);

1,680,349✔

3523

  QUERY_CHECK_CODE(code, lino, _end);

1,680,358✔

3525

  bool* result = (bool*)output.columnData->pData;

1,680,358✔

3526

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

4,201,140✔

3527

    if (result[i]) {

2,519,723✔

3528

      void* tmp = taosArrayPush(aFilterIdxs, &i);

2,360,859✔

3529

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

2,360,859✔

3533

  colDataDestroy(output.columnData);

1,681,100✔

3534

  taosMemoryFreeClear(output.columnData);

1,678,327✔

3536

_end:

1,678,706✔

3537

  blockDataDestroy(pResBlock);

1,678,618✔

3538

  taosArrayDestroy(pBlockList);

1,676,701✔

3539

  if (code != TSDB_CODE_SUCCESS) {

1,675,212✔

3542

  return code;

1,675,212✔

3545

static int32_t tagScanFillOneCellWithTag(SOperatorInfo* pOperator, const STUidTagInfo* pUidTagInfo,

30,183,820✔

3548

  int32_t code = TSDB_CODE_SUCCESS;

30,183,820✔

3549

  int32_t lino = 0;

30,183,820✔

3550

  if (QUERY_NODE_FUNCTION == pExprInfo->pExpr->nodeType) {

30,183,820✔

3551

    if (FUNCTION_TYPE_TBNAME == pExprInfo->pExpr->_function.functionType) {  // tbname

14,413,036✔

3557

    } else if (FUNCTION_TYPE_TBUID == pExprInfo->pExpr->_function.functionType) {

14,402,042✔

3558

      code = colDataSetVal(pColInfo, rowIndex, (char*)&pUidTagInfo->uid, false);

7,199,512✔

3559

      QUERY_CHECK_CODE(code, lino, _end);

7,206,696✔

3560

    } else if (FUNCTION_TYPE_VGID == pExprInfo->pExpr->_function.functionType) {

7,206,702✔

3561

      code = colDataSetVal(pColInfo, rowIndex, (char*)&pOperator->pTaskInfo->id.vgId, false);

7,204,544✔

3562

      QUERY_CHECK_CODE(code, lino, _end);

7,207,070✔

3565

    STagVal tagVal = {0};

15,770,433✔

3566

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

15,770,359✔

3567

    if (pUidTagInfo->pTagVal == NULL) {

15,774,602✔

3570

      const char* p = pAPI->metaFn.extractTagVal(pUidTagInfo->pTagVal, pColInfo->info.type, &tagVal);

15,771,753✔

3572

      if (p == NULL || (pColInfo->info.type == TSDB_DATA_TYPE_JSON && ((STag*)p)->nTag == 0)) {

15,772,601✔

3573

        colDataSetNULL(pColInfo, rowIndex);

7,424✔

3574

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

15,764,435✔

3577

      } else if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {

19,624,320✔

3578

        if (IS_STR_DATA_BLOB(pColInfo->info.type)) {

3,873,805✔

3581

        char* tmp = taosMemoryMalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1);

3,868,434✔

3582

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

3,863,730✔

3584

        varDataSetLen(tmp, tagVal.nData);

3,863,730✔

3585

        memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData);

3,866,116✔

3586

        code = colDataSetVal(pColInfo, rowIndex, tmp, false);

3,865,627✔

3587

        taosMemoryFree(tmp);

3,870,983✔

3588

        QUERY_CHECK_CODE(code, lino, _end);

3,870,614✔

3590

        code = colDataSetVal(pColInfo, rowIndex, (const char*)&tagVal.i64, false);

11,881,527✔

3591

        QUERY_CHECK_CODE(code, lino, _end);

11,894,888✔

3596

_end:

30,186,992✔

3597

  if (code != TSDB_CODE_SUCCESS) {

30,186,992✔

3600

  return code;

30,174,732✔

3603

static int32_t tagScanFillResultBlock(SOperatorInfo* pOperator, SSDataBlock* pRes, SArray* aUidTags,

4,745,075✔

3605

  int32_t       code = TSDB_CODE_SUCCESS;

4,745,075✔

3606

  int32_t       lino = 0;

4,745,075✔

3607

  STagScanInfo* pInfo = pOperator->info;

4,745,075✔

3608

  SExprInfo*    pExprInfo = &pOperator->exprSupp.pExprInfo[0];

4,746,717✔

3609

  if (!ignoreFilterIdx) {

4,746,123✔

3610

    size_t szTables = taosArrayGetSize(aFilterIdxs);

1,676,893✔

3611

    for (int i = 0; i < szTables; ++i) {

4,039,773✔

3612

      int32_t       idx = *(int32_t*)taosArrayGet(aFilterIdxs, i);

2,356,337✔

3613

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);

2,357,741✔

3614

      QUERY_CHECK_NULL(pUidTagInfo, code, lino, _end, terrno);

2,359,007✔

3615

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

13,443,048✔

3616

        SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pExprInfo[j].base.resSchema.slotId);

11,080,083✔

3617

        QUERY_CHECK_NULL(pDst, code, lino, _end, terrno);

11,082,440✔

3618

        code = tagScanFillOneCellWithTag(pOperator, pUidTagInfo, &pExprInfo[j], pDst, i, pAPI, pInfo->readHandle.vnode);

11,082,440✔

3619

        QUERY_CHECK_CODE(code, lino, _end);

11,084,041✔

3623

    size_t szTables = taosArrayGetSize(aUidTags);

3,069,230✔

3624

    for (int i = 0; i < szTables; ++i) {

7,976,901✔

3625

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);

4,907,297✔

3626

      QUERY_CHECK_NULL(pUidTagInfo, code, lino, _end, terrno);

4,907,671✔

3627

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

23,997,960✔

3628

        SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pExprInfo[j].base.resSchema.slotId);

19,091,971✔

3629

        QUERY_CHECK_NULL(pDst, code, lino, _end, terrno);

19,092,739✔

3630

        code = tagScanFillOneCellWithTag(pOperator, pUidTagInfo, &pExprInfo[j], pDst, i, pAPI, pInfo->readHandle.vnode);

19,092,739✔

3631

        QUERY_CHECK_CODE(code, lino, _end);

19,090,289✔

3636

_end:

3,069,604✔

3637

  if (code != TSDB_CODE_SUCCESS) {

4,753,040✔

3640

  return code;

4,749,318✔

3643

static int32_t doTagScanFromCtbIdxNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

9,453,532✔

3644

  int32_t        code = TSDB_CODE_SUCCESS;

9,453,532✔

3645

  int32_t        lino = 0;

9,453,532✔

3646

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,453,532✔

3647

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

9,456,927✔

3649

  STagScanInfo* pInfo = pOperator->info;

9,454,930✔

3650

  SSDataBlock*  pRes = pInfo->pRes;

9,454,709✔

3652

  QRY_PARAM_CHECK(ppRes);

9,449,289✔

3654

  if (pOperator->status == OP_EXEC_DONE) {

9,450,997✔

3655

    return TSDB_CODE_SUCCESS;

4,632,522✔

3657

  blockDataCleanup(pRes);

4,817,401✔

3659

  if (pInfo->pCtbCursor == NULL) {

4,822,775✔

3660

    pInfo->pCtbCursor = pAPI->metaFn.openCtbCursor(pInfo->readHandle.vnode, pInfo->suid, 1);

4,823,287✔

3661

    QUERY_CHECK_NULL(pInfo->pCtbCursor, code, lino, _end, terrno);

4,824,187✔

3667

  SArray* aUidTags = pInfo->aUidTags;

4,823,071✔

3668

  SArray* aFilterIdxs = pInfo->aFilterIdxs;

4,825,825✔

3669

  int32_t count = 0;

4,822,061✔

3670

  bool    ctbCursorFinished = false;

4,822,061✔

3671

  while (1) {

104,737✔

3672

    taosArrayClearEx(aUidTags, tagScanFreeUidTag);

4,926,798✔

3673

    taosArrayClear(aFilterIdxs);

4,931,118✔

3675

    int32_t numTables = 0;

4,929,215✔

3676

    while (numTables < pOperator->resultInfo.capacity) {

12,363,600✔

3677

      SMCtbCursor* pCur = pInfo->pCtbCursor;

12,353,466✔

3678

      tb_uid_t     uid = pAPI->metaFn.ctbCursorNext(pInfo->pCtbCursor);

12,355,500✔

3679

      if (uid == 0) {

12,354,171✔

3680

        ctbCursorFinished = true;

4,932,346✔

3681

        break;

4,932,346✔

3683

      STUidTagInfo info = {.uid = uid, .pTagVal = pCur->pVal};

7,421,825✔

3684

      info.pTagVal = taosMemoryMalloc(pCur->vLen);

7,423,005✔

3685

      QUERY_CHECK_NULL(info.pTagVal, code, lino, _end, terrno);

7,420,224✔

3687

      memcpy(info.pTagVal, pCur->pVal, pCur->vLen);

7,420,224✔

3688

      void* tmp = taosArrayPush(aUidTags, &info);

7,427,093✔

3689

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

7,427,093✔

3690

      ++numTables;

7,427,093✔

3693

    if (numTables == 0) {

4,931,852✔

3694

      break;

180,280✔

3696

    bool ignoreFilterIdx = true;

4,751,572✔

3697

    if (pInfo->pTagCond != NULL) {

4,751,572✔

3698

      ignoreFilterIdx = false;

1,682,342✔

3699

      code = tagScanFilterByTagCond(aUidTags, pInfo->pTagCond, aFilterIdxs, pInfo->readHandle.vnode, pAPI, pInfo, pTaskInfo->pStreamRuntimeInfo);

1,682,342✔

3700

      QUERY_CHECK_CODE(code, lino, _end);

1,677,918✔

3702

      ignoreFilterIdx = true;

3,069,230✔

3705

    code = tagScanFillResultBlock(pOperator, pRes, aUidTags, aFilterIdxs, ignoreFilterIdx, pAPI);

4,747,148✔

3706

    QUERY_CHECK_CODE(code, lino, _end);

4,749,318✔

3708

    count = ignoreFilterIdx ? taosArrayGetSize(aUidTags) : taosArrayGetSize(aFilterIdxs);

4,749,318✔

3710

    if (count != 0) {

4,749,021✔

3711

      break;

4,644,284✔

3715

  if (count > 0) {

4,824,564✔

3716

    pAPI->metaFn.pauseCtbCursor(pInfo->pCtbCursor);

4,643,398✔

3718

  if (count == 0 || ctbCursorFinished) {

4,824,708✔

3719

    pAPI->metaFn.closeCtbCursor(pInfo->pCtbCursor);

4,825,226✔

3720

    pInfo->pCtbCursor = NULL;

4,819,942✔

3721

    setOperatorCompleted(pOperator);

4,821,328✔

3724

  pRes->info.rows = count;

4,819,437✔

3725

  bool bLimitReached = applyLimitOffset(&pInfo->limitInfo, pRes, pTaskInfo);

4,820,081✔

3726

  if (bLimitReached) {

4,819,985✔

3730

  pOperator->resultInfo.totalRows += pRes->info.rows;

4,819,985✔

3731

  (*ppRes) = (pRes->info.rows == 0) ? NULL : pInfo->pRes;

4,817,420✔

3733

_end:

4,814,045✔

3734

  if (code != TSDB_CODE_SUCCESS) {

4,814,045✔

3740

  return code;

4,814,045✔

3743

static int32_t createTagScanTableListInfoFromParam(SOperatorInfo* pOperator) {

3,648,648✔

3744

  STagScanInfo*          pInfo = pOperator->info;

3,648,648✔

3745

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

3,648,648✔

3746

  int32_t                code = 0;

3,648,648✔

3747

  STableListInfo*        pListInfo = pInfo->pTableListInfo;

3,648,648✔

3748

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

3,648,648✔

3749

  tb_uid_t               pUid = pParam->vcUid;

3,648,648✔

3755

  STableKeyInfo info = {.groupId = 0};

3,648,648✔

3757

  int32_t tableIdx = 0;

3,648,648✔

3758

  taosHashClear(pListInfo->map);

3,648,648✔

3759

  taosArrayClear(pListInfo->pTableList);

3,648,648✔

3761

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

3,648,648✔

3768

  info.uid = pUid;

3,648,648✔

3769

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

3,648,648✔

3770

  if (p == NULL) {

3,648,648✔

3774

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

3,648,648✔

3776

  return code;

3,648,648✔

3779

static int32_t doTagScanFromMetaEntryNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

7,830,176✔

3780

  int32_t       code = TSDB_CODE_SUCCESS;

7,830,176✔

3781

  int32_t       lino = 0;

7,830,176✔

3782

  STagScanInfo* pInfo = pOperator->info;

7,830,176✔

3783

  if (pOperator->pOperatorGetParam) {

7,830,742✔

3784

    pOperator->resultInfo.totalRows = 0;

3,648,648✔

3785

    pOperator->dynamicTask = true;

3,648,648✔

3786

    pInfo->curPos = 0;

3,648,648✔

3787

    code = createTagScanTableListInfoFromParam(pOperator);

3,648,648✔

3788

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

3,648,648✔

3789

    pOperator->pOperatorGetParam = NULL;

3,648,648✔

3790

    QUERY_CHECK_CODE(code, lino, _end);

3,648,648✔

3792

    if (pOperator->status == OP_EXEC_DONE) {

3,648,648✔

3793

      pOperator->status = OP_OPENED;

1,969,948✔

3797

  if (pOperator->status == OP_EXEC_DONE) {

7,827,562✔

3798

    (*ppRes) = NULL;

2,077,155✔

3799

    return code;

2,077,155✔

3802

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

5,751,431✔

3803

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

5,752,885✔

3805

  SExprInfo*   pExprInfo = &pOperator->exprSupp.pExprInfo[0];

5,753,458✔

3806

  SSDataBlock* pRes = pInfo->pRes;

5,754,611✔

3807

  blockDataCleanup(pRes);

5,754,876✔

3809

  int32_t size = 0;

5,755,936✔

3810

  code = tableListGetSize(pInfo->pTableListInfo, &size);

5,755,936✔

3811

  QUERY_CHECK_CODE(code, lino, _end);

5,752,978✔

3813

  if (size == 0) {

5,752,978✔

3814

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

19,721✔

3815

    (*ppRes) = NULL;

19,721✔

3816

    return code;

19,721✔

3819

  SMetaReader mr = {0};

5,733,257✔

3820

  pAPI->metaReaderFn.initReader(&mr, pInfo->readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

5,733,508✔

3821

  pRes->info.rows = 0;

5,735,606✔

3823

  while (pInfo->curPos < size && pRes->info.rows < pOperator->resultInfo.capacity) {

17,261,114✔

3824

    code = doTagScanOneTable(pOperator, pRes, &mr, &pTaskInfo->storageAPI);

11,525,322✔

3825

    if (code != TSDB_CODE_OUT_OF_MEMORY && code != TSDB_CODE_QRY_REACH_QMEM_THRESHOLD &&

11,526,661✔

3828

      code = TSDB_CODE_SUCCESS;

11,527,721✔

3830

    QUERY_CHECK_CODE(code, lino, _end);

11,526,661✔

3832

    if (++pInfo->curPos >= size) {

11,526,661✔

3833

      setOperatorCompleted(pOperator);

5,736,215✔

3837

  pAPI->metaReaderFn.clearReader(&mr);

5,736,000✔

3838

  bool bLimitReached = applyLimitOffset(&pInfo->limitInfo, pRes, pTaskInfo);

5,735,155✔

3839

  if (bLimitReached) {

5,735,685✔

3844

  if (pOperator->status == OP_EXEC_DONE) {

5,735,685✔

3845

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

5,735,685✔

3848

  pOperator->resultInfo.totalRows += pRes->info.rows;

5,735,155✔

3850

  (*ppRes) = (pRes->info.rows == 0) ? NULL : pInfo->pRes;

5,735,685✔

3851

_end:

5,735,685✔

3852

  if (code != TSDB_CODE_SUCCESS) {

5,735,685✔

3857

  return code;

5,735,685✔

3860

static void destroyTagScanOperatorInfo(void* param) {

8,950,577✔

3861

  STagScanInfo* pInfo = (STagScanInfo*)param;

8,950,577✔

3862

  if (pInfo->pCtbCursor != NULL && pInfo->pStorageAPI != NULL) {

8,950,577✔

3865

  taosHashCleanup(pInfo->filterCtx.colHash);

8,950,041✔

3866

  taosArrayDestroy(pInfo->filterCtx.cInfoList);

8,944,791✔

3867

  taosArrayDestroy(pInfo->aFilterIdxs);

8,941,732✔

3868

  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);

8,944,211✔

3870

  blockDataDestroy(pInfo->pRes);

8,952,706✔

3871

  taosArrayDestroy(pInfo->matchInfo.pList);

8,950,831✔

3872

  tableListDestroy(pInfo->pTableListInfo);

8,948,315✔

3874

  pInfo->pRes = NULL;

8,951,817✔

3875

  pInfo->pTableListInfo = NULL;

8,950,423✔

3876

  taosMemoryFreeClear(param);

8,951,816✔

3877

8,955,587✔

3879

static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {

5,202✔

3880

  int32_t         code = TSDB_CODE_SUCCESS;

5,202✔

3881

  STagScanInfo*   pInfo = pOper->info;

5,202✔

3883

  pOper->status = OP_NOT_OPENED;

5,202✔

3884

  STagScanPhysiNode* pTagScanNode = (STagScanPhysiNode*)pOper->pPhyNode;

5,202✔

3885

  if (pTagScanNode->onlyMetaCtbIdx) {

5,202✔

3886

      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

2,160✔

3887

      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,160✔

3888

      pAPI->metaFn.closeCtbCursor(pInfo->pCtbCursor);

2,160✔

3889

      pInfo->pCtbCursor = NULL;

2,160✔

3891

      pInfo->curPos = 0;

3,042✔

3894

  return code;

5,202✔

3897

int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,

8,951,484✔

3900

  QRY_PARAM_CHECK(pOptrInfo);

8,951,484✔

3902

  int32_t         code = TSDB_CODE_SUCCESS;

8,955,473✔

3903

  int32_t         lino = 0;

8,955,473✔

3904

  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;

8,955,473✔

3905

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

8,955,473✔

3906

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

8,945,094✔

3907

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

8,937,818✔

3911

  pOperator->pPhyNode = pTagScanNode;

8,938,176✔

3912

  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

8,937,344✔

3914

  int32_t    numOfExprs = 0;

8,941,569✔

3915

  SExprInfo* pExprInfo = NULL;

8,944,321✔

3917

  code = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &pExprInfo, &numOfExprs);

8,943,443✔

3918

  QUERY_CHECK_CODE(code, lino, _error);

8,948,853✔

3920

  code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs, &pTaskInfo->storageAPI.functionStore);

8,948,853✔

3921

  QUERY_CHECK_CODE(code, lino, _error);

8,935,778✔

3923

  int32_t num = 0;

8,935,778✔

3924

  code = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);

8,931,886✔

3925

  QUERY_CHECK_CODE(code, lino, _error);

8,937,111✔

3927

  pInfo->pTagCond = pTagCond;

8,937,111✔

3928

  pInfo->pTagIndexCond = pTagIndexCond;

8,939,454✔

3929

  pInfo->suid = pPhyNode->suid;

8,940,250✔

3930

  pInfo->pStorageAPI = &pTaskInfo->storageAPI;

8,927,872✔

3932

  pInfo->pTableListInfo = pTableListInfo;

8,938,285✔

3933

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

8,929,845✔

3934

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

8,947,312✔

3936

  pInfo->readHandle = *pReadHandle;

8,925,486✔

3937

  pInfo->curPos = 0;

8,941,216✔

3939

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

8,925,621✔

3940

  setOperatorInfo(pOperator, "TagScanOperator", QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN, false, OP_NOT_OPENED, pInfo,

8,931,211✔

3942

  initResultSizeInfo(&pOperator->resultInfo, 4096);

8,929,342✔

3943

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

8,934,810✔

3944

  QUERY_CHECK_CODE(code, lino, _error);

8,954,355✔

3946

  if (pTagScanNode->onlyMetaCtbIdx) {

8,954,355✔

3947

    pInfo->aUidTags = taosArrayInit(pOperator->resultInfo.capacity, sizeof(STUidTagInfo));

4,830,393✔

3948

    QUERY_CHECK_NULL(pInfo->aUidTags, code, lino, _error, terrno);

4,833,187✔

3950

    pInfo->aFilterIdxs = taosArrayInit(pOperator->resultInfo.capacity, sizeof(int32_t));

4,832,803✔

3951

    QUERY_CHECK_NULL(pInfo->aFilterIdxs, code, lino, _error, terrno);

4,832,695✔

3953

    pInfo->filterCtx.colHash =

4,830,833✔

3954

        taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);

4,830,544✔

3955

    QUERY_CHECK_NULL(pInfo->filterCtx.colHash, code, lino, _error, terrno);

4,831,348✔

3957

    pInfo->filterCtx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));

4,829,810✔

3958

    QUERY_CHECK_NULL(pInfo->filterCtx.cInfoList, code, lino, _error, terrno);

4,833,337✔

3960

    if (pInfo->pTagCond != NULL) {

4,829,276✔

3961

      nodesRewriteExprPostOrder(&pTagCond, tagScanRewriteTagColumn, (void*)&pInfo->filterCtx);

1,685,947✔

3964

  __optr_fn_t tagScanNextFn = (pTagScanNode->onlyMetaCtbIdx) ? doTagScanFromCtbIdxNext : doTagScanFromMetaEntryNext;

8,954,002✔

3965

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, tagScanNextFn, NULL, destroyTagScanOperatorInfo,

8,952,494✔

3967

  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);

8,951,305✔

3969

  *pOptrInfo = pOperator;

8,949,386✔

3970

  return code;

8,948,936✔

3987

static int32_t subTblRowCompareTsFn(const void* pLeft, const void* pRight, void* param) {

2,147,483,647✔

3988

  int32_t                 left = *(int32_t*)pLeft;

2,147,483,647✔

3989

  int32_t                 right = *(int32_t*)pRight;

2,147,483,647✔

3990

  STmsSubTablesMergeInfo* pInfo = (STmsSubTablesMergeInfo*)param;

2,147,483,647✔

3992

  int32_t leftIdx = pInfo->aInputs[left].rowIdx;

2,147,483,647✔

3993

  int32_t rightIdx = pInfo->aInputs[right].rowIdx;

2,147,483,647✔

3995

  if (leftIdx == -1) {

2,147,483,647✔

3996

    return 1;

1,162,282,644✔

3997

  } else if (rightIdx == -1) {

1,437,504,184✔

3998

    return -1;

446,452✔

4001

  int64_t leftTs = pInfo->aInputs[left].aTs[leftIdx];

1,437,057,732✔

4002

  int64_t rightTs = pInfo->aInputs[right].aTs[rightIdx];

1,442,179,260✔

4003

  int32_t ret = leftTs > rightTs ? 1 : ((leftTs < rightTs) ? -1 : 0);

1,442,522,492✔

4004

  if (pInfo->pTsOrderInfo->order == TSDB_ORDER_DESC) {

1,442,522,492✔

4007

  return ret;

1,442,273,948✔

4037

int32_t dumpQueryTableCond(const SQueryTableDataCond* src, SQueryTableDataCond* dst) {

3,408,690✔

4038

  int32_t code = TSDB_CODE_SUCCESS;

3,408,690✔

4039

  int32_t lino = 0;

3,408,690✔

4040

  memcpy((void*)dst, (void*)src, sizeof(SQueryTableDataCond));

3,408,690✔

4041

  dst->colList = NULL;

3,408,690✔

4042

  dst->pSlotList = NULL;

3,409,402✔

4044

  if (src->numOfCols > 0) {

3,405,758✔

4045

    dst->colList = taosMemoryCalloc(src->numOfCols, sizeof(SColumnInfo));

3,374,430✔

4046

    QUERY_CHECK_NULL(dst->colList, code, lino, _end, terrno);

3,394,634✔

4047

    memcpy(dst->colList, src->colList, src->numOfCols * sizeof(SColumnInfo));

3,385,760✔

4049

    if (src->pSlotList != NULL) {

3,403,110✔

4050

      dst->pSlotList = taosMemoryMalloc(src->numOfCols * sizeof(int32_t));

3,402,024✔

4051

      QUERY_CHECK_NULL(dst->pSlotList, code, lino, _end, terrno);

3,400,148✔

4052

      memcpy(dst->pSlotList, src->pSlotList, src->numOfCols * sizeof(int32_t));

3,388,686✔

4057

  if (code != TSDB_CODE_SUCCESS) {

3,396,884✔

4062

  return code;

3,384,270✔

4065

static int32_t fetchNextSubTableBlockFromReader(SOperatorInfo* pOperator, STmsSubTableInput* pInput,

5,391,854✔

4067

  int32_t code = 0;

5,391,854✔

4069

  STableMergeScanInfo*    pInfo = pOperator->info;

5,391,854✔

4070

  SReadHandle*            pHandle = &pInfo->base.readHandle;

5,391,854✔

4071

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

5,397,410✔

4072

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

5,392,602✔

4073

  const SStorageAPI*      pAPI = &pTaskInfo->storageAPI;

5,392,234✔

4075

  blockDataCleanup(pInput->pReaderBlock);

5,391,094✔

4076

  if (!pInput->bInMemReader) {

5,396,336✔

4077

    code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, pInput->pKeyInfo, 1, pInput->pReaderBlock,

5,271,072✔

4078

                                         (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

2,636,462✔

4079

    if (code != 0) {

2,634,768✔

4084

  pInfo->base.dataReader = pInput->pReader;

5,395,384✔

4086

  while (true) {

5,610✔

4087

    bool hasNext = false;

5,400,210✔

4088

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

5,399,836✔

4089

    if (code != 0) {

5,405,270✔

4095

    if (!hasNext || isTaskKilled(pTaskInfo)) {

5,405,270✔

4096

      if (isTaskKilled(pTaskInfo)) {

3,355,606✔

4102

      *pSubTableHasBlock = false;

3,355,974✔

4103

      break;

3,354,870✔

4106

    if (pInput->tblCond.order == TSDB_ORDER_ASC) {

2,049,664✔

4107

      pInput->tblCond.twindows.skey = pInput->pReaderBlock->info.window.ekey + 1;

2,022,736✔

4109

      pInput->tblCond.twindows.ekey = pInput->pReaderBlock->info.window.skey - 1;

26,928✔

4112

    uint32_t status = 0;

2,049,664✔

4113

    code = loadDataBlock(pOperator, &pInfo->base, pInput->pReaderBlock, &status);

2,049,664✔

4114

    if (code != 0) {

2,048,916✔

4119

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

2,048,916✔

4123

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

2,048,916✔

4124

      continue;

5,610✔

4127

    *pSubTableHasBlock = true;

2,043,680✔

4128

    break;

2,042,564✔

4131

  if (*pSubTableHasBlock) {

5,401,138✔

4132

    pInput->pReaderBlock->info.id.groupId =

4,086,986✔

4133

        tableListGetTableGroupId(pInfo->base.pTableListInfo, pInput->pReaderBlock->info.id.uid);

4,085,870✔

4134

    pOperator->resultInfo.totalRows += pInput->pReaderBlock->info.rows;

2,044,054✔

4136

  if (!pInput->bInMemReader || !*pSubTableHasBlock) {

5,401,506✔

4137

    pAPI->tsdReader.tsdReaderClose(pInput->pReader);

3,932,184✔

4138

    pInput->pReader = NULL;

3,931,454✔

4141

  pInfo->base.dataReader = NULL;

5,398,924✔

4142

  return TSDB_CODE_SUCCESS;

5,398,924✔

4145

static int32_t setGroupStartEndIndex(STableMergeScanInfo* pInfo) {

2,038,438✔

4146

  pInfo->bGroupProcessed = false;

2,038,438✔

4148

  int32_t numOfTables = 0;

2,038,064✔

4149

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

2,018,114✔

4150

  if (code != TSDB_CODE_SUCCESS) {

2,012,956✔

4155

  int32_t i = pInfo->tableStartIndex + 1;

2,012,956✔

4156

  for (; i < numOfTables; ++i) {

3,304,184✔

4157

    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

1,350,752✔

4158

    if (!tableKeyInfo) {

1,348,134✔

4162

    if (tableKeyInfo->groupId != pInfo->groupId) {

1,348,134✔

4163

      break;

59,120✔

4166

  pInfo->tableEndIndex = i - 1;

2,012,552✔

4167

  return TSDB_CODE_SUCCESS;

2,054,274✔

4170

static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {

2,061,718✔

4171

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,414,748✔

4172

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,355,612✔

4173

    if (pInput->rowIdx == -1) {

3,355,244✔

4174

      continue;

1,611,222✔

4177

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,743,654✔

4178

      pInput->rowIdx = 0;

1,742,906✔

4179

      pInput->pageIdx = -1;

1,743,648✔

4182

    pInput->pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,743,274✔

4183

    SColumnInfoData* col = taosArrayGet(pInput->pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

1,743,274✔

4184

    if (!col) {

1,744,022✔

4188

    pInput->aTs = (int64_t*)col->pData;

1,744,022✔

4191

  __merge_compare_fn_t mergeCompareFn = (!pSubTblsInfo->pPkOrderInfo) ? subTblRowCompareTsFn : subTblRowCompareTsPkFn;

2,060,632✔

4192

  return tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, mergeCompareFn);

2,062,840✔

4195

static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {

2,051,028✔

4196

  int32_t code = setGroupStartEndIndex(pInfo);

2,051,028✔

4197

  if (code != TSDB_CODE_SUCCESS) {

2,049,080✔

4201

  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));

2,049,080✔

4202

  if (pSubTblsInfo == NULL) {

2,034,728✔

4205

  pSubTblsInfo->pTsOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);

2,034,728✔

4206

  if (!pSubTblsInfo->pTsOrderInfo) {

2,045,454✔

4210

  if (taosArrayGetSize(pInfo->pSortInfo) == 2) {

2,036,586✔

4217

    pSubTblsInfo->pPkOrderInfo = NULL;

2,029,528✔

4219

  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

2,043,988✔

4220

  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));

2,026,638✔

4221

  if (pSubTblsInfo->aInputs == NULL) {

2,026,252✔

4225

  int32_t bufPageSize = pInfo->bufPageSize;

2,032,900✔

4226

  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;

2,027,356✔

4227

  code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);

2,029,232✔

4228

  if (code != TSDB_CODE_SUCCESS) {

2,037,328✔

4233

  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;

2,037,328✔

4234

  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;

2,041,424✔

4236

  pInfo->pSubTablesMergeInfo = pSubTblsInfo;

2,033,304✔

4237

  return TSDB_CODE_SUCCESS;

2,019,990✔

4250

static int32_t setSubTableCondSuidByUid(STableMergeScanInfo* pInfo, SStorageAPI* pAPI, uint64_t uid,

3,341,930✔

4252

  int32_t     code = TSDB_CODE_SUCCESS;

3,341,930✔

4253

  int32_t     lino = 0;

3,341,930✔

4254

  SMetaReader tableReader = {0};

3,341,930✔

4256

  QUERY_CHECK_NULL(pInfo, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,346,364✔

4257

  QUERY_CHECK_NULL(pAPI, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,346,364✔

4258

  QUERY_CHECK_NULL(pCond, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,346,364✔

4260

  pAPI->metaReaderFn.initReader(&tableReader, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

3,346,364✔

4261

  code = pAPI->metaReaderFn.getTableEntryByUid(&tableReader, uid);

3,352,276✔

4262

  pAPI->metaReaderFn.readerReleaseLock(&tableReader);

3,343,378✔

4263

  QUERY_CHECK_CODE(code, lino, _end);

3,346,732✔

4265

  if (tableReader.me.type == TSDB_CHILD_TABLE) {

3,346,732✔

4266

    pCond->suid = tableReader.me.ctbEntry.suid;

3,241,252✔

4267

  } else if (tableReader.me.type == TSDB_NORMAL_TABLE) {

105,480✔

4268

    pCond->suid = 0;

105,480✔

4274

_end:

3,348,204✔

4275

  if (code != TSDB_CODE_SUCCESS) {

3,341,550✔

4278

  pAPI->metaReaderFn.clearReader(&tableReader);

3,341,550✔

4279

  return code;

3,348,204✔

4282

static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {

2,052,138✔

4283

  int32_t        code = TSDB_CODE_SUCCESS;

2,052,138✔

4284

  int32_t        lino = 0;

2,052,138✔

4285

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

2,052,138✔

4286

  SReadHandle*   pHandle = &pInfo->base.readHandle;

2,052,144✔

4287

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,018,144✔

4289

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

2,033,678✔

4291

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,371,904✔

4292

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,345,990✔

4293

    pInput->type = SUB_TABLE_MEM_BLOCK;

3,331,216✔

4295

    code = dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);

3,328,272✔

4296

    QUERY_CHECK_CODE(code, lino, _end);

3,324,876✔

4298

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pReaderBlock);

3,324,876✔

4299

    QUERY_CHECK_CODE(code, lino, _end);

3,351,540✔

4301

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pPageBlock);

3,351,540✔

4302

    QUERY_CHECK_CODE(code, lino, _end);

3,357,078✔

4304

    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);

3,357,078✔

4305

    pInput->pKeyInfo = keyInfo;

3,354,122✔

4306

    QUERY_CHECK_NULL(pInput->pKeyInfo, code, lino, _end, terrno);

3,354,122✔

4308

    code = setSubTableCondSuidByUid(pInfo, pAPI, pInput->pKeyInfo->uid, &pInput->tblCond);

3,347,468✔

4309

    QUERY_CHECK_CODE(code, lino, _end);

3,342,292✔

4311

    if (isTaskKilled(pTaskInfo)) {

3,342,292✔

4315

    if (i + 1 < pSubTblsInfo->numInMemReaders) {

3,348,958✔

4316

      code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, keyInfo, 1, pInput->pReaderBlock,

2,582,516✔

4317

                                           (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

1,292,380✔

4318

      QUERY_CHECK_CODE(code, lino, _end);

1,289,014✔

4319

      pInput->bInMemReader = true;

1,289,014✔

4321

      pInput->pReader = NULL;

2,057,658✔

4322

      pInput->bInMemReader = false;

2,058,768✔

4324

    bool hasNext = true;

3,348,156✔

4325

    code = fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);

3,347,040✔

4326

    QUERY_CHECK_CODE(code, lino, _end);

3,354,864✔

4327

    if (!hasNext) {

3,354,864✔

4328

      pInput->rowIdx = -1;

1,611,216✔

4329

      ++pSubTblsInfo->numSubTablesCompleted;

1,611,222✔

4330

      continue;

1,612,694✔

4332

      pInput->rowIdx = 0;

1,743,648✔

4333

      pInput->pageIdx = -1;

1,743,648✔

4337

_end:

2,062,472✔

4338

  if (code != TSDB_CODE_SUCCESS) {

2,062,472✔

4341

  return code;

2,060,632✔

4344

static void adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

2,044,054✔

4345

  int32_t              code = TSDB_CODE_SUCCESS;

2,044,054✔

4346

  int32_t              lino = 0;

2,044,054✔

4347

  SExecTaskInfo*       pTaskInfo = pOperatorInfo->pTaskInfo;

2,044,054✔

4348

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

2,044,054✔

4349

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

2,044,054✔

4350

  bool                 hasNext = true;

2,044,054✔

4351

  code = fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);

2,044,054✔

4352

  QUERY_CHECK_CODE(code, lino, _end);

2,044,054✔

4354

  if (!hasNext) {

2,044,054✔

4355

    pInput->rowIdx = -1;

1,744,022✔

4356

    ++pSubTblsInfo->numSubTablesCompleted;

1,744,022✔

4358

    pInput->rowIdx = 0;

300,032✔

4361

_end:

2,044,054✔

4362

  if (code != TSDB_CODE_SUCCESS) {

2,044,054✔

4367

2,044,054✔

4369

static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

1,787,488,444✔

4370

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

1,787,488,444✔

4371

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,787,566,130✔

4373

  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,787,331,534✔

4374

  if (pInput->rowIdx < pInputBlock->info.rows - 1) {

1,787,205,198✔

4375

    ++pInput->rowIdx;

1,784,893,716✔

4376

  } else if (pInput->rowIdx == pInputBlock->info.rows - 1) {

2,044,054✔

4377

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

2,044,054✔

4378

      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);

2,044,054✔

4380

    if (pInput->rowIdx != -1) {

2,044,054✔

4381

      SColumnInfoData* col = taosArrayGet(pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

300,032✔

4382

      if (!col) {

300,032✔

4386

      pInput->pInputBlock = pInputBlock;

300,032✔

4387

      pInput->aTs = (int64_t*)col->pData;

300,032✔

4391

  return tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));

1,787,073,410✔

4394

static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {

1,788,062,694✔

4395

  int32_t            code = TSDB_CODE_SUCCESS;

1,788,062,694✔

4396

  int32_t            lino = 0;

1,788,062,694✔

4397

  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,788,062,694✔

4398

  SSDataBlock*       pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,788,433,510✔

4400

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

2,147,483,647✔

4401

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

2,147,483,647✔

4402

    QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);

2,147,483,647✔

4404

    SColumnInfoData* pSrcColInfo = taosArrayGet(pInputBlock->pDataBlock, i);

2,147,483,647✔

4405

    QUERY_CHECK_NULL(pSrcColInfo, code, lino, _end, terrno);

2,147,483,647✔

4406

    bool isNull = colDataIsNull(pSrcColInfo, pInputBlock->info.rows, pInput->rowIdx, NULL);

2,147,483,647✔

4408

    if (isNull) {

2,147,483,647✔

4409

      code = colDataSetVal(pColInfo, pBlock->info.rows, NULL, true);

99,775,840✔

4410

      QUERY_CHECK_CODE(code, lino, _end);

99,770,604✔

4412

      if (pSrcColInfo->pData != NULL) {

2,147,483,647✔

4413

        char* pData = colDataGetData(pSrcColInfo, pInput->rowIdx);

2,147,483,647✔

4414

        code = colDataSetVal(pColInfo, pBlock->info.rows, pData, false);

2,147,483,647✔

4415

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

4419

  pBlock->info.dataLoad = 1;

1,788,421,852✔

4420

  pBlock->info.scanFlag = pInputBlock->info.scanFlag;

1,788,436,512✔

4421

  pBlock->info.rows += 1;

1,788,199,068✔

4423

_end:

1,788,223,570✔

4424

  if (code != TSDB_CODE_SUCCESS) {

1,788,223,570✔

4427

  return code;

1,787,608,162✔

4430

static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,

4,797,186✔

4432

  int32_t                 code = TSDB_CODE_SUCCESS;

4,797,186✔

4433

  int32_t                 lino = 0;

4,797,186✔

4434

  STableMergeScanInfo*    pInfo = pOperator->info;

4,797,186✔

4435

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

4,798,302✔

4436

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

4,796,824✔

4437

  bool                    finished = false;

4,794,984✔

4439

  QRY_PARAM_CHECK(pResBlock);

4,794,984✔

4441

  blockDataCleanup(pBlock);

4,793,874✔

4445

      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {

1,791,998,680✔

4446

        finished = true;

3,362,902✔

4447

        break;

3,362,902✔

4450

      code = appendChosenRowToDataBlock(pSubTblsInfo, pBlock);

1,788,417,622✔

4451

      QUERY_CHECK_CODE(code, lino, _end);

1,787,660,610✔

4453

      code = adjustSubTableForNextRow(pOperator, pSubTblsInfo);

1,787,660,610✔

4454

      QUERY_CHECK_CODE(code, lino, _end);

1,788,607,104✔

4456

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

1,788,607,104✔

4457

        break;

1,436,872✔

4461

    if (isTaskKilled(pTaskInfo)) {

4,799,774✔

4465

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

4,797,934✔

4466

    if (finished || limitReached || pBlock->info.rows > 0) {

4,797,934✔

4471

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

4,797,934✔

4472

    *pResBlock = pBlock;

2,736,198✔

4475

_end:

2,061,368✔

4476

  if (code != TSDB_CODE_SUCCESS) {

4,797,566✔

4480

  return code;

4,797,934✔

4483

static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {

2,052,132✔

4484

  int32_t              code = TSDB_CODE_SUCCESS;

2,052,132✔

4485

  int32_t              lino = 0;

2,052,132✔

4486

  STableMergeScanInfo* pInfo = pOperator->info;

2,052,132✔

4488

  code = initSubTablesMergeInfo(pInfo);

2,055,070✔

4489

  QUERY_CHECK_CODE(code, lino, _end);

2,025,166✔

4491

  code = initSubTableInputs(pOperator, pInfo);

2,025,166✔

4492

  QUERY_CHECK_CODE(code, lino, _end);

2,061,730✔

4494

  code = openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);

2,061,730✔

4495

  QUERY_CHECK_CODE(code, lino, _end);

2,059,148✔

4497

_end:

2,059,148✔

4498

  if (code != TSDB_CODE_SUCCESS) {

2,059,148✔

4501

  return code;

2,061,356✔

4504

static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {

29,819,328✔

4505

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

29,819,328✔

4506

  if (pSubTblsInfo != NULL) {

29,826,289✔

4507

    tMergeTreeDestroy(&pSubTblsInfo->pTree);

2,061,368✔

4509

    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,418,820✔

4510

      STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,356,710✔

4511

      cleanupQueryTableDataCond(&pInput->tblCond);

3,354,134✔

4512

      blockDataDestroy(pInput->pReaderBlock);

3,354,140✔

4513

      blockDataDestroy(pInput->pPageBlock);

3,356,348✔

4514

      taosArrayDestroy(pInput->aBlockPages);

3,355,244✔

4515

      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);

3,355,244✔

4516

      pInput->pReader = NULL;

3,355,980✔

4519

    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);

2,062,840✔

4520

    taosMemoryFree(pSubTblsInfo->aInputs);

2,061,368✔

4522

    taosMemoryFree(pSubTblsInfo);

2,059,528✔

4523

    pInfo->pSubTablesMergeInfo = NULL;

2,060,258✔

4527

29,824,811✔

4529

void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

38,456,797✔

4530

  STableMergeScanInfo* pInfo = pOperator->info;

38,456,797✔

4531

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

38,460,883✔

4532

  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));

38,460,883✔

4534

  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

38,468,306✔

4535

  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;

38,450,455✔

4536

  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;

38,455,993✔

4537

  pInfo->sortExecInfo.loops += sortExecInfo.loops;

38,451,647✔

4538

  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;

38,446,507✔

4539

  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

38,442,856✔

4541

  tsortDestroySortHandle(pInfo->pSortHandle);

38,444,422✔

4542

  pInfo->pSortHandle = NULL;

38,454,322✔

4543

38,450,791✔

4552

static int32_t createVTableMergeScanInfoFromBatchParam(SOperatorInfo* pOperator) {

65,632✔

4553

  int32_t                  code = TSDB_CODE_SUCCESS;

65,632✔

4554

  int32_t                  lino = 0;

65,632✔

4555

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

65,632✔

4556

  STableMergeScanInfo*     pInfo = pOperator->info;

65,632✔

4557

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

65,632✔

4558

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

65,632✔

4559

  SMetaReader              orgTable = {0};

65,632✔

4560

  SOperatorParam*          pGetParam = pOperator->pOperatorGetParam;

65,632✔

4561

  STableScanOperatorParam* pParam = (STableScanOperatorParam*)pGetParam->value;

65,632✔

4563

  if (pParam->paramType != DYN_TYPE_SCAN_PARAM) {

65,632✔

4567

  if (pInfo->pSubTablesMergeInfo) {

65,632✔

4570

  if (pInfo->pSortHandle) {

65,632✔

4573

  if (pInfo->base.dataReader) {

65,632✔

4578

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

65,632✔

4584

  pInfo->numNextDurationBlocks = 0;

65,632✔

4585

  pInfo->nextDurationBlocksIdx = 0;

65,632✔

4586

  pInfo->rtnNextDurationBlocks = false;

65,632✔

4587

  pInfo->bNewFilesetEvent = false;

65,632✔

4588

  pInfo->bNextDurationBlockEvent = false;

65,632✔

4589

  taosHashCleanup(pInfo->mSkipTables);

65,632✔

4590

  pInfo->mSkipTables = NULL;

65,632✔

4592

  if (pInfo->base.orgCond.colList && pInfo->base.orgCond.numOfCols > 0) {

65,632✔

4593

    cleanupQueryTableDataCond(&pInfo->base.cond);

65,632✔

4594

    code = dumpQueryTableCond(&pInfo->base.orgCond, &pInfo->base.cond);

65,632✔

4595

    QUERY_CHECK_CODE(code, lino, _return);

65,632✔

4598

  if (pParam->window.skey == INT64_MAX && pParam->window.ekey == INT64_MIN) {

65,632✔

4599

    if (pInfo->base.orgCond.colList && pInfo->base.orgCond.numOfCols > 0) {

65,632✔

4600

      pInfo->base.cond.twindows.skey = pInfo->base.orgCond.twindows.skey;

65,632✔

4601

      pInfo->base.cond.twindows.ekey = pInfo->base.orgCond.twindows.ekey;

65,632✔

4608

  pInfo->tableStartIndex = 0;

65,632✔

4609

  pInfo->tableEndIndex = 0;

65,632✔

4610

  pInfo->hasGroupId = false;

65,632✔

4611

  pInfo->groupId = pParam->groupid;

65,632✔

4612

  pInfo->bGroupProcessed = false;

65,632✔

4613

  pOperator->status = OP_NOT_OPENED;

65,632✔

4615

  pListInfo->oneTableForEachGroup = true;

65,632✔

4616

  pListInfo->numOfOuputGroups = 1;

65,632✔

4617

  taosHashClear(pListInfo->map);

65,632✔

4618

  taosArrayClear(pListInfo->pTableList);

65,632✔

4620

  int32_t tableIdx = 0;

65,632✔

4621

  if (pParam->pBatchTbInfo) {

65,632✔

4622

    int32_t numOfBatchTables = taosArrayGetSize(pParam->pBatchTbInfo);

65,632✔

4623

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

230,884✔

4624

      SOrgTbInfo* pOrgTbInfo = taosArrayGet(pParam->pBatchTbInfo, i);

165,252✔

4625

      QUERY_CHECK_NULL(pOrgTbInfo, code, lino, _return, terrno);

165,252✔

4627

      const char* pTableName = strstr(pOrgTbInfo->tbName, ".");

165,252✔

4628

      pTableName = pTableName ? (pTableName + 1) : pOrgTbInfo->tbName;

165,252✔

4629

      pAPI->metaReaderFn.initReader(&orgTable, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

165,252✔

4630

      code = pAPI->metaReaderFn.getTableEntryByName(&orgTable, pTableName);

165,252✔

4631

      pAPI->metaReaderFn.readerReleaseLock(&orgTable);

165,252✔

4632

      QUERY_CHECK_CODE(code, lino, _return);

165,252✔

4634

      uint64_t uid = orgTable.me.uid;

165,252✔

4635

      code = taosHashPut(pListInfo->map, &uid, sizeof(uid), &tableIdx, sizeof(tableIdx));

165,252✔

4636

      if (code == TSDB_CODE_DUP_KEY) {

165,252✔

4640

      QUERY_CHECK_CODE(code, lino, _return);

165,252✔

4642

      STableKeyInfo keyInfo = {.groupId = pParam->groupid, .uid = uid};

165,252✔

4643

      QUERY_CHECK_NULL(taosArrayPush(pListInfo->pTableList, &keyInfo), code, lino, _return, terrno);

330,504✔

4644

      ++tableIdx;

165,252✔

4646

      qDebug("%s table merge dynamic batch add table name:%s uid:%" PRIu64 " group:%" PRIu64, GET_TASKID(pTaskInfo),

165,252✔

4648

      pAPI->metaReaderFn.clearReader(&orgTable);

165,252✔

4652

  qDebug("%s table merge dynamic batch param consumed, group:%" PRIu64 ", tables:%d", GET_TASKID(pTaskInfo),

65,632✔

4655

_return:

65,632✔

4656

  if (code != TSDB_CODE_SUCCESS) {

65,632✔

4659

  pAPI->metaReaderFn.clearReader(&orgTable);

65,632✔

4660

  return code;

65,632✔

4663

int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

4,762,104✔

4664

  int32_t              code = TSDB_CODE_SUCCESS;

4,762,104✔

4665

  int32_t              lino = 0;

4,762,104✔

4666

  int32_t              tableListSize = 0;

4,762,104✔

4667

  int64_t              st = taosGetTimestampUs();

4,747,704✔

4668

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

4,747,704✔

4669

  STableMergeScanInfo* pInfo = pOperator->info;

4,751,016✔

4671

  QRY_PARAM_CHECK(ppRes);

4,747,330✔

4673

  if (pOperator->status == OP_EXEC_DONE && !pOperator->pOperatorGetParam) {

4,738,830✔

4678

  if (pOperator->pOperatorGetParam) {

4,736,260✔

4679

    code = createVTableMergeScanInfoFromBatchParam(pOperator);

65,632✔

4680

    QUERY_CHECK_CODE(code, lino, _end);

65,632✔

4682

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

65,632✔

4683

    pOperator->pOperatorGetParam = NULL;

65,632✔

4684

    pOperator->dynamicTask = true;

65,632✔

4687

  code = pOperator->fpSet._openFn(pOperator);

4,736,984✔

4688

  QUERY_CHECK_CODE(code, lino, _end);

4,736,622✔

4690

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

4,736,622✔

4691

  QUERY_CHECK_CODE(code, lino, _end);

4,751,770✔

4693

  if (!pInfo->hasGroupId) {

4,751,770✔

4694

    pInfo->hasGroupId = true;

2,012,984✔

4696

    if (tableListSize == 0) {

2,001,540✔

4697

      setOperatorCompleted(pOperator);

31,042✔

4698

      (*ppRes) = NULL;

31,042✔

4699

      return code;

31,042✔

4702

    pInfo->tableStartIndex = 0;

1,970,498✔

4703

    STableKeyInfo* pTmpGpId = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,972,688✔

4704

    QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

1,975,282✔

4705

    pInfo->groupId = pTmpGpId->groupId;

1,975,282✔

4706

    code = startSubTablesTableMergeScan(pOperator);

1,984,506✔

4707

    QUERY_CHECK_CODE(code, lino, _end);

2,001,868✔

4710

  SSDataBlock* pBlock = NULL;

4,740,286✔

4711

  while (pInfo->tableStartIndex < tableListSize) {

4,796,812✔

4712

    if (isTaskKilled(pTaskInfo)) {

4,797,192✔

4716

    code = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity, &pBlock);

4,797,186✔

4717

    QUERY_CHECK_CODE(code, lino, _end);

4,798,670✔

4719

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

4,798,670✔

4726

    if (pBlock != NULL) {

4,797,566✔

4727

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

2,736,198✔

4728

      pOperator->resultInfo.totalRows += pBlock->info.rows;

2,736,198✔

4729

      pInfo->bGroupProcessed = true;

2,736,198✔

4730

      break;

2,736,198✔

4733

      stopSubTablesTableMergeScan(pInfo);

2,061,368✔

4734

      if (pInfo->tableEndIndex >= tableListSize - 1) {

2,059,522✔

4735

        setOperatorCompleted(pOperator);

2,001,512✔

4736

        break;

2,002,616✔

4739

      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

59,120✔

4740

      STableKeyInfo* pTmpGpId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

59,120✔

4741

      QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

59,120✔

4743

      pInfo->groupId = pTmpGpId->groupId;

59,120✔

4744

      code = startSubTablesTableMergeScan(pOperator);

59,120✔

4745

      QUERY_CHECK_CODE(code, lino, _end);

59,120✔

4746

      resetLimitInfoForNextGroup(&pInfo->limitInfo);

59,120✔

4750

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

4,739,550✔

4752

_end:

4,740,286✔

4753

  if (code != TSDB_CODE_SUCCESS) {

4,740,286✔

4758

    (*ppRes) = pBlock;

4,740,286✔

4761

  return code;

4,739,182✔

4764

static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {

3,658,430✔

4765

  int32_t              code = TSDB_CODE_SUCCESS;

3,658,430✔

4766

  int32_t              lino = 0;

3,658,430✔

4767

  SOperatorInfo*       pOperator = (SOperatorInfo*)pTableMergeOpInfo;

3,658,430✔

4768

  STableMergeScanInfo* pInfo = pOperator->info;

3,658,430✔

4769

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,658,430✔

4771

  if (pInfo->mSkipTables == NULL) {

3,658,430✔

4772

    pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,

258,204✔

4774

    QUERY_CHECK_NULL(pInfo->mSkipTables, code, lino, _end, terrno);

258,204✔

4776

  int bSkip = 1;

3,658,430✔

4777

  if (pInfo->mSkipTables != NULL) {

3,658,430✔

4778

    code = taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));

3,658,430✔

4779

    if (code == TSDB_CODE_DUP_KEY) {

3,658,430✔

4782

    QUERY_CHECK_CODE(code, lino, _end);

3,658,430✔

4785

_end:

3,658,430✔

4786

  if (code != TSDB_CODE_SUCCESS) {

3,658,430✔

4791

3,658,430✔

4793

static int32_t doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {

170,016,448✔

4794

  STableMergeScanInfo* pInfo = pOperator->info;

170,016,448✔

4795

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

170,039,099✔

4796

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

170,026,116✔

4797

  SSDataBlock*         pBlock = pInfo->pReaderBlock;

170,007,124✔

4798

  int32_t              code = 0;

170,005,202✔

4799

  bool                 hasNext = false;

170,005,202✔

4800

  STsdbReader*         reader = pInfo->base.dataReader;

169,994,235✔

4802

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

170,020,814✔

4803

  if (code != 0) {

170,000,091✔

4810

  if (!hasNext || isTaskKilled(pTaskInfo)) {

170,000,091✔

4811

    if (isTaskKilled(pTaskInfo)) {

28,789,896✔

4815

    *pFinished = true;

28,800,677✔

4816

    return code;

28,790,719✔

4819

  uint32_t status = 0;

141,218,158✔

4820

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

141,210,926✔

4822

  if (code != TSDB_CODE_SUCCESS) {

141,202,561✔

4823

    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));

16,440✔

4824

    pTaskInfo->code = code;

16,440✔

4825

    return code;

16,440✔

4828

  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

141,186,121✔

4834

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

141,186,121✔

4835

    *pSkipped = true;

6,559,799✔

4836

    return code;

6,544,014✔

4839

  return code;

134,618,557✔

4842

static int32_t getBlockForTableMergeScan(void* param, SSDataBlock** ppBlock) {

172,910,264✔

4843

  STableMergeScanSortSourceParam* source = param;

172,910,264✔

4845

  SOperatorInfo*       pOperator = source->pOperator;

172,910,264✔

4846

  STableMergeScanInfo* pInfo = pOperator->info;

172,916,446✔

4847

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

172,911,529✔

4848

  SSDataBlock*         pBlock = NULL;

172,923,622✔

4849

  int64_t              st = taosGetTimestampUs();

172,925,439✔

4850

  int32_t              code = TSDB_CODE_SUCCESS;

172,925,439✔

4853

    if (pInfo->rtnNextDurationBlocks) {

188,094,495✔

4854

      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d",

18,084,851✔

4857

      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {

18,090,677✔

4858

        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];

8,636,510✔

4859

        ++pInfo->nextDurationBlocksIdx;

8,636,510✔

4861

        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

18,082,056✔

4862

          blockDataDestroy(pInfo->nextDurationBlocks[i]);

8,635,410✔

4863

          pInfo->nextDurationBlocks[i] = NULL;

8,633,106✔

4866

        pInfo->rtnNextDurationBlocks = false;

9,446,646✔

4867

        pInfo->nextDurationBlocksIdx = 0;

9,447,798✔

4868

        pInfo->numNextDurationBlocks = 0;

9,442,806✔

4869

        continue;

9,442,038✔

4872

      bool bFinished = false;

170,003,316✔

4873

      bool bSkipped = false;

170,012,360✔

4875

      code = doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);

170,042,811✔

4876

      if (code != 0) {

170,000,013✔

4877

        return code;

16,440✔

4880

      pBlock = pInfo->pReaderBlock;

169,983,573✔

4881

      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d",

170,012,853✔

4883

      if (bFinished) {

170,035,051✔

4884

        pInfo->bNewFilesetEvent = false;

28,806,479✔

4885

        break;

28,805,413✔

4888

      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {

141,228,572✔

4889

        if (!bSkipped) {

9,452,731✔

4890

          code = createOneDataBlock(pBlock, true, &pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks]);

8,636,510✔

4891

          if (code) {

8,630,366✔

4896

          ++pInfo->numNextDurationBlocks;

8,630,366✔

4897

          if (pInfo->numNextDurationBlocks > 2) {

8,623,070✔

4905

        if (pInfo->bNewFilesetEvent) {

9,438,139✔

4906

          pInfo->rtnNextDurationBlocks = true;

9,445,442✔

4907

          *ppBlock = NULL;

9,443,522✔

4908

          return code;

9,437,378✔

4916

      if (bSkipped) continue;

131,778,063✔

4919

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

134,686,900✔

4921

    pOperator->resultInfo.totalRows += pBlock->info.rows;

134,688,432✔

4922

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

134,681,033✔

4923

    *ppBlock = pBlock;

134,684,134✔

4925

    return code;

134,683,083✔

4928

  *ppBlock = NULL;

28,805,931✔

4929

  return code;

28,805,439✔

4932

int32_t generateSortByTsPkInfo(SArray* colMatchInfo, int32_t order, SArray** ppSortArray) {

25,310,755✔

4933

  int32_t code = TSDB_CODE_SUCCESS;

25,310,755✔

4934

  int32_t lino = 0;

25,310,755✔

4935

  SArray* pSortInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));

25,310,755✔

4936

  QUERY_CHECK_NULL(pSortInfo, code, lino, _end, terrno);

25,274,812✔

4937

  SBlockOrderInfo biTs = {0};

25,274,812✔

4938

  SBlockOrderInfo biPk = {0};

25,275,854✔

4940

  int32_t tsTargetSlotId = 0;

25,281,340✔

4941

  int32_t pkTargetSlotId = -1;

25,281,340✔

4942

  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {

96,952,777✔

4943

    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);

71,715,969✔

4944

    QUERY_CHECK_NULL(colInfo, code, lino, _end, terrno);

71,728,332✔

4945

    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

71,728,332✔

4946

      tsTargetSlotId = colInfo->dstSlotId;

25,305,635✔

4947

      biTs.order = order;

25,285,503✔

4948

      biTs.slotId = tsTargetSlotId;

25,285,503✔

4949

      biTs.nullFirst = (order == TSDB_ORDER_ASC);

25,285,503✔

4950

      biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, order);

25,285,503✔

4953

    if (colInfo->isPk) {

71,714,646✔

4954

      pkTargetSlotId = colInfo->dstSlotId;

2,202,559✔

4955

      biPk.order = order;

2,203,213✔

4956

      biPk.slotId = pkTargetSlotId;

2,203,213✔

4957

      biPk.nullFirst = (order == TSDB_ORDER_ASC);

2,203,213✔

4958

      biPk.compFn = getKeyComparFunc(colInfo->dataType.type, order);

2,203,213✔

4962

  void* tmp = taosArrayPush(pSortInfo, &biTs);

25,325,250✔

4963

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

25,325,250✔

4964

  if (pkTargetSlotId != -1) {

25,325,250✔

4965

    tmp = taosArrayPush(pSortInfo, &biPk);

2,202,040✔

4966

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

2,202,040✔

4969

  (*ppSortArray) = pSortInfo;

25,324,686✔

4971

_end:

25,285,583✔

4972

  if (code != TSDB_CODE_SUCCESS) {

25,265,375✔

4975

  return code;

25,265,375✔

4978

void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {

30,984,615✔

4979

  STableMergeScanInfo* pTmsInfo = param;

30,984,615✔

4980

  if (type == TSD_READER_NOTIFY_DURATION_START) {

30,984,615✔

4981

    pTmsInfo->bNewFilesetEvent = true;

30,994,452✔

4985

  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);

30,986,599✔

4987

  return;

31,025,746✔

4990

int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

38,469,327✔

4991

  STableMergeScanInfo* pInfo = pOperator->info;

38,469,327✔

4992

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

38,482,420✔

4994

  int32_t code = TSDB_CODE_SUCCESS;

38,477,691✔

4995

  int32_t lino = 0;

38,477,691✔

4996

  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

38,477,691✔

4998

  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));

38,469,253✔

4999

  pInfo->bNewFilesetEvent = false;

38,513,254✔

5000

  pInfo->bNextDurationBlockEvent = false;

38,497,717✔

5002

  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;

38,494,297✔

5003

  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;

38,496,493✔

5005

  pInfo->pSortHandle = NULL;

38,493,530✔

5006

  code = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,

76,989,944✔

5007

                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0, &pInfo->pSortHandle);

38,486,025✔

5008

  if (code) {

38,459,229✔

5012

  if (pInfo->bSortRowId && numOfTable != 1) {

38,459,229✔

5013

    int32_t memSize = 512 * 1024 * 1024;

2,984,148✔

5014

    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);

2,984,148✔

5015

    QUERY_CHECK_CODE(code, lino, _end);

2,984,523✔

5018

  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);

38,463,893✔

5019

  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pOperator);

38,462,153✔

5020

  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);

38,460,468✔

5022

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);

38,442,068✔

5023

  QUERY_CHECK_CODE(code, lino, _end);

38,416,280✔

5025

  STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));

38,416,280✔

5026

  QUERY_CHECK_NULL(param, code, lino, _end, terrno);

38,428,261✔

5027

  param->pOperator = pOperator;

38,428,261✔

5029

  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));

38,431,299✔

5030

  if (ps == NULL) {

38,435,181✔

5035

  ps->param = param;

38,435,181✔

5036

  ps->onlyRef = false;

38,442,961✔

5037

  code = tsortAddSource(pInfo->pSortHandle, ps);

38,462,893✔

5038

  QUERY_CHECK_CODE(code, lino, _end);

38,466,570✔

5040

  if (numOfTable == 1) {

38,466,570✔

5041

    tsortSetSingleTableMerge(pInfo->pSortHandle);

27,620,338✔

5043

    code = tsortOpen(pInfo->pSortHandle);

10,846,232✔

5044

    QUERY_CHECK_CODE(code, lino, _end);

10,854,850✔

5047

_end:

10,846,630✔

5048

  if (code != TSDB_CODE_SUCCESS) {

38,471,209✔

5049

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

8,220✔

5051

  return code;

38,466,976✔

5054

void startGroupTableMergeScan(SOperatorInfo* pOperator) {

29,036,882✔

5055

  int32_t              code = TSDB_CODE_SUCCESS;

29,036,882✔

5056

  int32_t              lino = 0;

29,036,882✔

5057

  STableMergeScanInfo* pInfo = pOperator->info;

29,036,882✔

5058

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

29,035,324✔

5059

  SReadHandle*         pHandle = &pInfo->base.readHandle;

29,036,343✔

5060

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

29,025,087✔

5061

  qDebug("%s table merge scan start group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

29,023,197✔

5064

    int32_t numOfTables = 0;

29,031,833✔

5065

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

29,021,955✔

5066

    QUERY_CHECK_CODE(code, lino, _end);

29,027,442✔

5068

    int32_t i = pInfo->tableStartIndex + 1;

29,027,442✔

5069

    for (; i < numOfTables; ++i) {

139,518,120✔

5070

      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

116,997,559✔

5071

      QUERY_CHECK_NULL(tableKeyInfo, code, lino, _end, terrno);

116,999,699✔

5072

      if (tableKeyInfo->groupId != pInfo->groupId) {

116,999,699✔

5073

        break;

6,504,384✔

5076

    pInfo->tableEndIndex = i - 1;

29,024,945✔

5078

  pInfo->bGroupProcessed = false;

29,047,177✔

5079

  int32_t tableStartIdx = pInfo->tableStartIndex;

29,050,785✔

5080

  int32_t tableEndIdx = pInfo->tableEndIndex;

29,041,385✔

5082

  int32_t        numOfTable = tableEndIdx - tableStartIdx + 1;

29,047,311✔

5083

  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);

29,047,311✔

5084

  code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,

58,072,352✔

5085

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

29,031,194✔

5086

  QUERY_CHECK_CODE(code, lino, _end);

29,024,866✔

5087

  if (pInfo->filesetDelimited) {

29,024,866✔

5088

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

29,020,772✔

5090

  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);

29,044,035✔

5092

  code = startDurationForGroupTableMergeScan(pOperator);

28,999,980✔

5093

  QUERY_CHECK_CODE(code, lino, _end);

29,014,659✔

5095

_end:

29,006,439✔

5096

  if (code != TSDB_CODE_SUCCESS) {

29,016,109✔

5097

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

8,220✔

5098

    pTaskInfo->code = code;

8,220✔

5099

    T_LONG_JMP(pTaskInfo->env, code);

8,220✔

5101

29,007,889✔

5103

void stopGroupTableMergeScan(SOperatorInfo* pOperator) {

29,010,211✔

5104

  STableMergeScanInfo* pInfo = pOperator->info;

29,010,211✔

5105

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

29,011,247✔

5106

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

29,011,247✔

5108

  stopDurationForGroupTableMergeScan(pOperator);

29,011,247✔

5110

  if (pInfo->base.dataReader != NULL) {

29,004,854✔

5111

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

29,005,014✔

5112

    pInfo->base.dataReader = NULL;

28,975,735✔

5114

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

28,982,661✔

5115

    if (pInfo->nextDurationBlocks[i]) {

1,100✔

5116

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

1,100✔

5117

      pInfo->nextDurationBlocks[i] = NULL;

1,100✔

5119

    pInfo->numNextDurationBlocks = 0;

1,100✔

5120

    pInfo->nextDurationBlocksIdx = 0;

1,100✔

5122

  resetLimitInfoForNextGroup(&pInfo->limitInfo);

28,990,202✔

5123

  taosHashCleanup(pInfo->mSkipTables);

28,951,725✔

5124

  pInfo->mSkipTables = NULL;

28,962,564✔

5125

  qDebug("%s table merge scan stop group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,964,349✔

5126

29,007,417✔

5130

SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,

80,059,601✔

5132

  STableMergeScanInfo* pInfo = pOperator->info;

80,059,601✔

5133

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

80,077,531✔

5134

  STupleHandle*        pTupleHandle = NULL;

80,076,818✔

5136

  blockDataCleanup(pResBlock);

80,076,444✔

5139

    while (1) {

2,147,483,647✔

5140

      pTupleHandle = NULL;

2,147,483,647✔

5141

      int32_t code = tsortNextTuple(pHandle, &pTupleHandle);

2,147,483,647✔

5142

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5143

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

54✔

5144

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

54✔

5146

      if (pTupleHandle == NULL) {

2,147,483,647✔

5147

        break;

63,117,756✔

5150

      code = tsortAppendTupleToBlock(pInfo->pSortHandle, pResBlock, pTupleHandle);

2,147,483,647✔

5151

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5156

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

2,147,483,647✔

5157

        break;

16,958,823✔

5161

    if (tsortIsClosed(pHandle)) {

80,076,579✔

5166

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);

80,076,376✔

5167

    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,

80,080,541✔

5169

    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {

80,095,519✔

5173

  return (pResBlock->info.rows > 0) ? pResBlock : NULL;

80,095,519✔

5176

int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

67,371,708✔

5177

  if (pOperator->status == OP_EXEC_DONE) {

67,371,708✔

5178

    (*ppRes) = NULL;

46,240✔

5179

    return TSDB_CODE_SUCCESS;

46,240✔

5182

  int32_t              code = TSDB_CODE_SUCCESS;

67,310,950✔

5183

  int32_t              lino = 0;

67,310,950✔

5184

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

67,310,950✔

5185

  STableMergeScanInfo* pInfo = pOperator->info;

67,317,322✔

5187

  code = pOperator->fpSet._openFn(pOperator);

67,301,984✔

5188

  QUERY_CHECK_CODE(code, lino, _end);

67,303,953✔

5190

  int64_t st = taosGetTimestampUs();

67,298,314✔

5192

  int32_t tableListSize = 0;

67,298,314✔

5193

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

67,295,169✔

5194

  QUERY_CHECK_CODE(code, lino, _end);

67,323,219✔

5196

  if (!pInfo->hasGroupId) {

67,323,219✔

5197

    pInfo->hasGroupId = true;

25,708,953✔

5199

    if (tableListSize == 0) {

25,694,613✔

5200

      setOperatorCompleted(pOperator);

3,177,816✔

5201

      (*ppRes) = NULL;

3,179,184✔

5202

      return code;

3,179,184✔

5204

    pInfo->tableStartIndex = 0;

22,516,797✔

5205

    STableKeyInfo* tmp = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

22,517,315✔

5206

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

22,515,602✔

5207

    pInfo->groupId = tmp->groupId;

22,515,602✔

5208

    startGroupTableMergeScan(pOperator);

22,518,035✔

5211

  SSDataBlock* pBlock = NULL;

64,135,343✔

5212

  while (pInfo->tableStartIndex < tableListSize) {

80,086,137✔

5213

    if (isTaskKilled(pTaskInfo)) {

80,090,954✔

5214

      goto _end;

15,950✔

5217

    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,

80,064,384✔

5219

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

80,095,621✔

5220

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,903,346✔

5221

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

1,903,346✔

5222

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

1,903,346✔

5224

    if (pBlock != NULL) {

80,091,277✔

5225

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

41,630,114✔

5226

      pOperator->resultInfo.totalRows += pBlock->info.rows;

41,632,698✔

5227

      pInfo->bGroupProcessed = true;

41,629,340✔

5228

      break;

41,630,681✔

5230

      if (pInfo->bNewFilesetEvent) {

38,461,163✔

5231

        stopDurationForGroupTableMergeScan(pOperator);

9,450,434✔

5232

        code = startDurationForGroupTableMergeScan(pOperator);

9,447,137✔

5233

        QUERY_CHECK_CODE(code, lino, _end);

9,446,528✔

5236

        stopGroupTableMergeScan(pOperator);

29,011,381✔

5237

        if (pInfo->tableEndIndex >= tableListSize - 1) {

29,008,619✔

5238

          setOperatorCompleted(pOperator);

22,504,239✔

5239

          break;

22,503,930✔

5242

        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

6,504,384✔

5243

        STableKeyInfo* tmp = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

6,504,384✔

5244

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

6,504,384✔

5245

        pInfo->groupId = tmp->groupId;

6,504,384✔

5246

        startGroupTableMergeScan(pOperator);

6,504,384✔

5247

        resetLimitInfoForNextGroup(&pInfo->limitInfo);

6,504,266✔

5252

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

64,132,205✔

5254

_end:

64,150,048✔

5255

  if (code != TSDB_CODE_SUCCESS) {

64,150,823✔

5260

    (*ppRes) = pBlock;

64,150,823✔

5263

  return code;

64,145,843✔

5272

void destroyTableMergeScanOperatorInfo(void* param) {

25,329,949✔

5273

  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;

25,329,949✔

5276

  if (pTableScanInfo->base.readerAPI.tsdReaderClose != NULL) {

25,329,949✔

5277

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

25,330,889✔

5278

    pTableScanInfo->base.dataReader = NULL;

25,321,504✔

5281

  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {

25,326,262✔

5288

  tsortDestroySortHandle(pTableScanInfo->pSortHandle);

25,328,525✔

5289

  pTableScanInfo->pSortHandle = NULL;

25,320,230✔

5290

  taosHashCleanup(pTableScanInfo->mSkipTables);

25,320,213✔

5291

  pTableScanInfo->mSkipTables = NULL;

25,319,403✔

5292

  blockDataDestroy(pTableScanInfo->pSortInputBlock);

25,323,083✔

5293

  pTableScanInfo->pSortInputBlock = NULL;

25,326,453✔

5296

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

25,326,453✔

5298

  blockDataDestroy(pTableScanInfo->pResBlock);

25,322,441✔

5299

  pTableScanInfo->pResBlock = NULL;

25,329,335✔

5302

  blockDataDestroy(pTableScanInfo->pReaderBlock);

25,329,255✔

5303

  pTableScanInfo->pReaderBlock = NULL;

25,328,509✔

5304

  taosArrayDestroy(pTableScanInfo->pSortInfo);

25,328,783✔

5306

  stopSubTablesTableMergeScan(pTableScanInfo);

25,328,459✔

5308

  taosMemoryFreeClear(param);

25,322,816✔

5309

25,316,128✔

5311

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {

1,257,472✔

5312

  if (pOptr == NULL) {

1,257,472✔

5317

  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));

1,257,472✔

5318

  if (!execInfo) {

1,255,154✔

5321

  STableMergeScanInfo* pInfo = pOptr->info;

1,255,154✔

5322

  execInfo->blockRecorder = pInfo->base.readRecorder;

1,256,690✔

5323

  execInfo->sortExecInfo = pInfo->sortExecInfo;

1,257,074✔

5325

  *pOptrExplain = execInfo;

1,257,465✔

5326

  *len = sizeof(STableMergeScanExecInfo);

1,257,081✔

5328

  return TSDB_CODE_SUCCESS;

1,257,856✔

5331

static int32_t resetTableMergeScanOperatorState(SOperatorInfo* pOper) {

2,432,406✔

5332

  int32_t         code = TSDB_CODE_SUCCESS;

2,432,406✔

5333

  STableMergeScanInfo*   pInfo = pOper->info;

2,432,406✔

5334

  pOper->status = OP_NOT_OPENED;

2,432,406✔

5336

  pInfo->tableEndIndex = 0;

2,432,406✔

5337

  pInfo->tableStartIndex = 0;

2,432,406✔

5338

  pInfo->hasGroupId = false;

2,432,406✔

5339

  if (pInfo->base.readerAPI.tsdReaderClose) {

2,432,406✔

5340

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

2,432,406✔

5342

  pInfo->base.dataReader = NULL;

2,432,406✔

5343

  pInfo->base.scanFlag = MAIN_SCAN;

2,432,406✔

5345

  pInfo->base.limitInfo = (SLimitInfo){0};

2,432,406✔

5346

  pInfo->base.limitInfo.limit.limit = -1;

2,432,406✔

5347

  pInfo->base.limitInfo.slimit.limit = -1;

2,432,406✔

5349

  tableListDestroy(pInfo->base.pTableListInfo);

2,432,406✔

5350

  pInfo->base.pTableListInfo = tableListCreate();

2,432,406✔

5351

  if (!pInfo->base.pTableListInfo) {

2,432,406✔

5355

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

2,432,406✔

5357

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

2,432,406✔

5358

  code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

4,864,812✔

5360

                                  pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

4,864,812✔

5361

  if (code) {

2,432,406✔

5366

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

2,432,406✔

5367

  cleanupQueryTableDataCond(&pInfo->base.cond);

2,432,406✔

5368

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

2,432,406✔

5369

  if (code) {

2,432,406✔

5374

  tsortDestroySortHandle(pInfo->pSortHandle);

2,432,406✔

5375

  pInfo->pSortHandle = NULL;

2,432,138✔

5377

  taosHashCleanup(pInfo->mSkipTables);

2,432,138✔

5378

  pInfo->mSkipTables = NULL;

2,432,406✔

5379

  pOper->resultInfo.totalRows = 0;

2,432,406✔

5381

  pInfo->bGroupProcessed = false;

2,432,138✔

5382

  pInfo->bNewFilesetEvent = false;

2,432,136✔

5383

  pInfo->bNextDurationBlockEvent = false;

2,432,136✔

5384

  pInfo->rtnNextDurationBlocks = false;

2,432,136✔

5385

  pInfo->nextDurationBlocksIdx = 0;

2,432,406✔

5386

  pInfo->bSortRowId = false;

2,432,138✔

5388

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

2,432,138✔

5394

  pInfo->numNextDurationBlocks = 0;

2,432,138✔

5396

  stopSubTablesTableMergeScan(pInfo);

2,432,406✔

5397

  return code;

2,432,136✔

5400

int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

25,296,792✔

5403

  QRY_PARAM_CHECK(pOptrInfo);

25,296,792✔

5405

  int32_t              code = TSDB_CODE_SUCCESS;

25,307,479✔

5406

  int32_t              lino = 0;

25,307,479✔

5407

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

25,307,479✔

5408

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

25,280,753✔

5409

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

25,251,706✔

5414

  pOperator->pPhyNode = pTableScanNode;

25,252,224✔

5415

  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

25,252,224✔

5417

  int32_t numOfCols = 0;

25,260,966✔

5418

  code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,

25,269,669✔

5420

  QUERY_CHECK_CODE(code, lino, _error);

25,257,104✔

5422

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

25,257,104✔

5423

  QUERY_CHECK_CODE(code, lino, _error);

25,211,506✔

5425

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

25,211,506✔

5426

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

10,920,723✔

5427

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->pExprInfo, &pSup->numOfExprs);

10,931,123✔

5428

    QUERY_CHECK_CODE(code, lino, _error);

10,938,649✔

5430

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

10,938,649✔

5432

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

10,928,124✔

5435

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

25,236,783✔

5437

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

25,226,581✔

5442

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

25,226,581✔

5443

  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

25,267,730✔

5444

  pInfo->base.scanFlag = MAIN_SCAN;

25,168,381✔

5445

  pInfo->base.readHandle = *readHandle;

25,239,649✔

5447

  pInfo->readIdx = -1;

25,172,154✔

5449

  pInfo->base.limitInfo.limit.limit = -1;

25,197,599✔

5450

  pInfo->base.limitInfo.slimit.limit = -1;

25,197,261✔

5451

  pInfo->base.pTableListInfo = pTableListInfo;

25,234,010✔

5453

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

25,217,747✔

5454

  pInfo->sample.seed = taosGetTimestampSec();

25,203,444✔

5456

  int32_t sortOrder = pInfo->base.cond.order;

25,276,131✔

5457

  if (pTableScanNode->scan.node.dynamicOp) {

25,280,257✔

5458

    memcpy(&pInfo->base.orgCond, &pInfo->base.cond, sizeof(SQueryTableDataCond));

65,632✔

5459

    sortOrder = pInfo->base.orgCond.order;

65,632✔

5460

    memset(&pInfo->base.cond, 0, sizeof(SQueryTableDataCond));

65,632✔

5463

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

25,228,681✔

5464

                            pTaskInfo->pStreamRuntimeInfo);

25,260,225✔

5465

  QUERY_CHECK_CODE(code, lino, _error);

25,243,528✔

5467

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

25,243,528✔

5469

  pInfo->mergeLimit = -1;

25,228,388✔

5470

  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;

25,235,665✔

5471

  if (hasLimit) {

25,241,628✔

5472

    pInfo->mergeLimit = pInfo->limitInfo.limit.offset != -1

3,856,757✔

5473

                            ? pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset

792,833✔

5474

                            : pInfo->limitInfo.limit.limit;

2,720,693✔

5475

    pInfo->mSkipTables = NULL;

1,928,523✔

5478

  initResultSizeInfo(&pOperator->resultInfo, 1024);

25,241,772✔

5479

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

25,247,177✔

5480

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

25,293,787✔

5481

  code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

25,205,055✔

5482

  QUERY_CHECK_CODE(code, lino, _error);

25,317,857✔

5483

  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {

25,317,857✔

5484

    pInfo->bSortRowId = true;

5,426,242✔

5486

    pInfo->bSortRowId = false;

19,891,540✔

5489

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

25,303,830✔

5490

  QUERY_CHECK_CODE(code, lino, _error);

25,284,198✔

5492

  code = generateSortByTsPkInfo(pInfo->base.matchInfo.pList, sortOrder, &pInfo->pSortInfo);

25,284,198✔

5493

  QUERY_CHECK_CODE(code, lino, _error);

25,264,204✔

5495

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pReaderBlock);

25,264,204✔

5496

  QUERY_CHECK_CODE(code, lino, _error);

25,296,273✔

5498

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

25,296,273✔

5500

  int32_t  rowSize = pInfo->pResBlock->info.rowSize;

25,304,169✔

5501

  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);

25,287,942✔

5503

  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);

25,234,117✔

5506

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pSortInputBlock);

25,306,389✔

5507

  QUERY_CHECK_CODE(code, lino, _error);

25,309,459✔

5509

  if (!tsExperimental) {

25,309,459✔

5512

    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

25,309,459✔

5515

  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,

25,292,151✔

5517

  pOperator->exprSupp.numOfExprs = numOfCols;

25,266,248✔

5519

  pOperator->fpSet = createOperatorFpSet(

25,257,890✔

5520

      optrDummyOpenFn, (pTableScanNode->paraTablesSort || pTableScanNode->scan.node.dynamicOp) ? doTableMergeScanParaSubTablesNext : doTableMergeScanNext, NULL,

25,280,703✔

5523

  setOperatorResetStateFn(pOperator, resetTableMergeScanOperatorState);

25,266,305✔

5524

  pOperator->cost.openCost = 0;

25,247,141✔

5526

  *pOptrInfo = pOperator;

25,253,346✔

5527

  return code;

25,234,573✔

5566

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {

28,816✔

5567

  if (scanCols != NULL) {

28,816✔

5568

    SNode* pNode = NULL;

15,413✔

5569

    FOREACH(pNode, scanCols) {

36,012✔

5570

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

20,599✔

5573

      STargetNode* targetNode = (STargetNode*)pNode;

20,599✔

5574

      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {

20,599✔

5577

      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);

20,599✔

5578

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

20,599✔

5579

        supp->dbNameSlotId = targetNode->slotId;

10,192✔

5580

      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

10,407✔

5581

        supp->stbNameSlotId = targetNode->slotId;

10,407✔

5585

  return TSDB_CODE_SUCCESS;

28,847✔

5588

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {

28,847✔

5589

  if (pseudoCols != NULL) {

28,847✔

5590

    SNode* pNode = NULL;

28,847✔

5591

    FOREACH(pNode, pseudoCols) {

57,694✔

5592

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

28,876✔

5595

      STargetNode* targetNode = (STargetNode*)pNode;

28,816✔

5596

      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {

28,816✔

5599

      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);

28,876✔

5600

      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {

28,845✔

5601

        supp->tbCountSlotId = targetNode->slotId;

28,876✔

5605

  return TSDB_CODE_SUCCESS;

28,847✔

5608

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {

28,876✔

5609

  if (groupTags != NULL) {

28,876✔

5610

    SNode* pNode = NULL;

15,413✔

5611

    FOREACH(pNode, groupTags) {

36,012✔

5612

      if (nodeType(pNode) != QUERY_NODE_COLUMN) {

20,599✔

5615

      SColumnNode* colNode = (SColumnNode*)pNode;

20,599✔

5616

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

20,599✔

5617

        supp->groupByDbName = true;

10,192✔

5619

      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

20,599✔

5620

        supp->groupByStbName = true;

10,407✔

5624

    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);

13,463✔

5625

    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);

13,463✔

5627

  return TSDB_CODE_SUCCESS;

28,816✔

5630

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,

28,847✔

5632

  int32_t code = 0;

28,847✔

5633

  code = tblCountScanGetInputs(groupTags, tableName, supp);

28,847✔

5634

  if (code != TSDB_CODE_SUCCESS) {

28,816✔

5639

  supp->dbNameSlotId = -1;

28,816✔

5640

  supp->stbNameSlotId = -1;

28,876✔

5641

  supp->tbCountSlotId = -1;

28,876✔

5643

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);

28,876✔

5644

  if (code != TSDB_CODE_SUCCESS) {

28,876✔

5649

  code = tblCountScanGetCountSlotId(pseudoCols, supp);

28,876✔

5650

  if (code != TSDB_CODE_SUCCESS) {

28,847✔

5654

  return code;

28,876✔

5657

int32_t createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,

28,876✔

5659

  QRY_PARAM_CHECK(pOptrInfo);

28,876✔

5661

  int32_t                      code = TSDB_CODE_SUCCESS;

28,876✔

5662

  int32_t                      lino = 0;

28,876✔

5663

  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;

28,876✔

5664

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

28,876✔

5665

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

28,876✔

5666

  if (!pInfo || !pOperator) {

28,876✔

5671

  pInfo->readHandle = *readHandle;

28,876✔

5673

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

28,876✔

5674

  initResultSizeInfo(&pOperator->resultInfo, 1);

28,876✔

5675

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

28,876✔

5676

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

28,876✔

5678

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

28,847✔

5679

  QUERY_CHECK_CODE(code, lino, _error);

28,876✔

5681

  code = getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,

28,876✔

5684

  QUERY_CHECK_CODE(code, lino, _error);

28,876✔

5686

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,

28,876✔

5688

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScanNext, NULL, destoryTableCountScanOperator,

28,847✔

5690

  *pOptrInfo = pOperator;

28,845✔

5691

  return code;

28,845✔

5705

int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,

39,379✔

5707

  int32_t code = TSDB_CODE_SUCCESS;

39,379✔

5708

  int32_t lino = 0;

39,379✔

5709

  if (pSupp->dbNameSlotId != -1) {

39,379✔

5710

    QUERY_CHECK_CONDITION((strlen(dbName) > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

15,368✔

5711

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);

15,368✔

5712

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

15,368✔

5714

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

15,368✔

5715

    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

15,368✔

5717

    varDataSetLen(varDbName, strlen(dbName));

15,368✔

5718

    code = colDataSetVal(colInfoData, 0, varDbName, false);

15,368✔

5719

    QUERY_CHECK_CODE(code, lino, _end);

15,368✔

5722

  if (pSupp->stbNameSlotId != -1) {

39,379✔

5723

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);

19,890✔

5724

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

19,890✔

5725

    if (strlen(stbName) != 0) {

19,890✔

5726

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

14,288✔

5727

      tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);

14,288✔

5728

      varDataSetLen(varStbName, strlen(stbName));

14,288✔

5729

      code = colDataSetVal(colInfoData, 0, varStbName, false);

14,288✔

5730

      QUERY_CHECK_CODE(code, lino, _end);

14,288✔

5736

  if (pSupp->tbCountSlotId != -1) {

39,439✔

5737

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);

39,379✔

5738

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

39,319✔

5739

    code = colDataSetVal(colInfoData, 0, (char*)&count, false);

39,319✔

5740

    QUERY_CHECK_CODE(code, lino, _end);

39,379✔

5742

  pRes->info.rows = 1;

39,379✔

5744

_end:

39,308✔

5745

  if (code != TSDB_CODE_SUCCESS) {

39,308✔

5748

  return code;

39,339✔

5751

static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {

11,346✔

5752

  STableCountScanSupp* pSupp = &pInfo->supp;

11,346✔

5753

  SSDataBlock*         pRes = pInfo->pRes;

11,346✔

5755

  size_t infodbTableNum;

11,346✔

5756

  getInfosDbMeta(NULL, &infodbTableNum);

11,346✔

5757

  infodbTableNum -= 1;

11,346✔

5758

  size_t perfdbTableNum;

11,346✔

5759

  getPerfDbMeta(NULL, &perfdbTableNum);

11,346✔

5761

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

11,346✔

5762

    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);

9,729✔

5763

    return (pRes->info.rows > 0) ? pRes : NULL;

9,729✔

5765

    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);

1,617✔

5766

    return (pRes->info.rows > 0) ? pRes : NULL;

1,617✔

5770

static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,

1,617✔

5772

  int32_t        code = TSDB_CODE_SUCCESS;

1,617✔

5773

  int32_t        lino = 0;

1,617✔

5774

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,617✔

5775

  if (strlen(pSupp->stbNameFilter) != 0) {

1,617✔

5777

    code = fillTableCountScanDataBlock(pSupp, "", "", 0, pRes);

549✔

5778

    QUERY_CHECK_CODE(code, lino, _end);

549✔

5779

  } else if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {

1,068✔

5780

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

183✔

5781

    QUERY_CHECK_CODE(code, lino, _end);

183✔

5782

  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

885✔

5785

  } else if (strlen(pSupp->dbNameFilter) == 0) {

885✔

5786

    code = fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);

702✔

5787

    QUERY_CHECK_CODE(code, lino, _end);

702✔

5790

    code = fillTableCountScanDataBlock(pSupp, pSupp->dbNameFilter, pSupp->stbNameFilter, 0, pRes);

183✔

5791

    QUERY_CHECK_CODE(code, lino, _end);

183✔

5794

_end:

183✔

5795

  if (code != TSDB_CODE_SUCCESS) {

1,617✔

5800

  setOperatorCompleted(pOperator);

1,617✔

5801

1,617✔

5803

static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,729✔

5806

  int32_t        code = TSDB_CODE_SUCCESS;

9,729✔

5807

  int32_t        lino = 0;

9,729✔

5808

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,729✔

5809

  if (pInfo->currGrpIdx == 0) {

9,729✔

5810

    uint64_t groupId = 0;

3,243✔

5811

    if (pSupp->groupByDbName) {

3,243✔

5812

      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));

2,222✔

5814

      groupId = calcGroupId("", 0);

1,021✔

5817

    pRes->info.id.groupId = groupId;

3,243✔

5818

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

3,243✔

5819

    QUERY_CHECK_CODE(code, lino, _end);

3,243✔

5820

  } else if (pInfo->currGrpIdx == 1) {

6,486✔

5821

    uint64_t groupId = 0;

3,243✔

5822

    if (pSupp->groupByDbName) {

3,243✔

5823

      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));

2,222✔

5825

      groupId = calcGroupId("", 0);

1,021✔

5828

    pRes->info.id.groupId = groupId;

3,243✔

5829

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

3,243✔

5830

    QUERY_CHECK_CODE(code, lino, _end);

3,243✔

5832

    setOperatorCompleted(pOperator);

3,243✔

5834

  pInfo->currGrpIdx++;

9,729✔

5836

_end:

9,729✔

5837

  if (code != TSDB_CODE_SUCCESS) {

9,729✔

5842

9,729✔

5844

static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

67,512✔

5845

  int32_t                      code = TSDB_CODE_SUCCESS;

67,512✔

5846

  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;

67,512✔

5847

  STableCountScanOperatorInfo* pInfo = pOperator->info;

68,244✔

5848

  STableCountScanSupp*         pSupp = &pInfo->supp;

68,255✔

5849

  SSDataBlock*                 pRes = pInfo->pRes;

68,244✔

5851

  blockDataCleanup(pRes);

68,255✔

5852

  QRY_PARAM_CHECK(ppRes);

68,255✔

5854

  if (pOperator->status == OP_EXEC_DONE) {

68,255✔

5855

    return code;

17,449✔

5858

  if (pInfo->readHandle.mnd != NULL) {

50,806✔

5859

    (*ppRes) = buildSysDbTableCount(pOperator, pInfo);

11,346✔

5860

    return code;

11,346✔

5863

  code = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);

39,460✔

5864

  if (code != TSDB_CODE_SUCCESS) {

39,460✔

5869

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

39,460✔

5870

    *ppRes = pRes;

31,276✔

5873

  return code;

39,420✔

5876

static int32_t buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

39,460✔

5878

  int32_t        code = TSDB_CODE_SUCCESS;

39,460✔

5879

  int32_t        lino = 0;

39,460✔

5880

  const char*    db = NULL;

39,460✔

5881

  int32_t        vgId = 0;

39,460✔

5882

  char           dbName[TSDB_DB_NAME_LEN] = {0};

39,460✔

5883

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

39,460✔

5884

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

39,460✔

5887

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);

39,460✔

5888

  SName sn = {0};

39,460✔

5890

  code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

39,460✔

5891

  QUERY_CHECK_CODE(code, lino, _end);

39,460✔

5893

  code = tNameGetDbName(&sn, dbName);

39,460✔

5894

  QUERY_CHECK_CODE(code, lino, _end);

39,460✔

5896

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

39,460✔

5897

    code = buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);

27,614✔

5899

    code = buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);

11,846✔

5902

_end:

39,420✔

5903

  if (code != TSDB_CODE_SUCCESS) {

39,449✔

5908

  return code;

39,449✔

5911

static int32_t buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

27,614✔

5913

  int32_t        code = TSDB_CODE_SUCCESS;

27,614✔

5914

  int32_t        lino = 0;

27,614✔

5915

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

27,614✔

5916

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

27,614✔

5918

  if (pSupp->groupByStbName) {

27,614✔

5919

    if (pInfo->stbUidList == NULL) {

23,628✔

5920

      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));

8,184✔

5921

      QUERY_CHECK_NULL(pInfo->stbUidList, code, lino, _end, terrno);

8,184✔

5922

      code = pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList);

8,184✔

5923

      QUERY_CHECK_CODE(code, lino, _end);

8,184✔

5926

    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {

23,628✔

5927

      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);

14,288✔

5928

      code = buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);

14,288✔

5929

      QUERY_CHECK_CODE(code, lino, _end);

14,288✔

5931

      pInfo->currGrpIdx++;

14,288✔

5932

    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {

9,340✔

5933

      code = buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);

8,184✔

5934

      QUERY_CHECK_CODE(code, lino, _end);

8,184✔

5936

      pInfo->currGrpIdx++;

8,184✔

5938

      setOperatorCompleted(pOperator);

1,156✔

5941

    uint64_t groupId = calcGroupId(dbName, strlen(dbName));

3,986✔

5942

    pRes->info.id.groupId = groupId;

3,986✔

5944

    int64_t dbTableCount = 0;

3,986✔

5945

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);

3,986✔

5946

    code = fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);

3,986✔

5947

    QUERY_CHECK_CODE(code, lino, _end);

3,986✔

5948

    setOperatorCompleted(pOperator);

3,986✔

5951

_end:

27,614✔

5952

  if (code != TSDB_CODE_SUCCESS) {

27,614✔

5956

  return code;

27,614✔

5959

static int32_t buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

11,846✔

5961

  int32_t        code = TSDB_CODE_SUCCESS;

11,846✔

5962

  int32_t        lino = 0;

11,846✔

5963

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

11,846✔

5964

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

11,846✔

5966

  if (strlen(pSupp->dbNameFilter) != 0 && strcmp(dbName, pSupp->dbNameFilter) != 0) {

11,846✔

5970

  } else if (strlen(pSupp->stbNameFilter) != 0) {

11,846✔

5971

    uint64_t uid = 0;

5,816✔

5972

    code = pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid);

5,816✔

5973

    if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

5,776✔

5975

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, 0, pRes);

1,464✔

5976

      QUERY_CHECK_CODE(code, lino, _end);

1,464✔

5978

      QUERY_CHECK_CODE(code, lino, _end);

4,312✔

5980

      int64_t numOfChildTables = 0;

4,312✔

5981

      code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL, NULL);

4,312✔

5982

      QUERY_CHECK_CODE(code, lino, _end);

4,341✔

5984

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes);

4,341✔

5985

      QUERY_CHECK_CODE(code, lino, _end);

4,312✔

5988

    int64_t tbNumVnode = 0;

6,030✔

5989

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);

6,030✔

5990

    code = fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);

6,030✔

5991

    QUERY_CHECK_CODE(code, lino, _end);

6,030✔

5994

_end:

11,806✔

5995

  if (code != TSDB_CODE_SUCCESS) {

11,806✔

6000

  setOperatorCompleted(pOperator);

11,806✔

6001

  return code;

11,835✔

6004

static int32_t buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

8,184✔

6006

  int32_t code = TSDB_CODE_SUCCESS;

8,184✔

6007

  int32_t lino = 0;

8,184✔

6008

  char    fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

8,184✔

6009

  if (pSupp->groupByDbName) {

8,184✔

6010

    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");

3,984✔

6013

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

8,184✔

6014

  pRes->info.id.groupId = groupId;

8,184✔

6016

  int64_t numOfTables = 0;

8,184✔

6017

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);

8,184✔

6019

  if (numOfTables != 0) {

8,184✔

6020

    code = fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);

1,156✔

6023

_end:

7,028✔

6024

  if (code != TSDB_CODE_SUCCESS) {

8,184✔

6027

  return code;

8,184✔

6030

static int32_t buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

14,288✔

6032

  int32_t code = TSDB_CODE_SUCCESS;

14,288✔

6033

  int32_t lino = 0;

14,288✔

6034

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

14,288✔

6035

  code = pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);

14,288✔

6036

  QUERY_CHECK_CODE(code, lino, _end);

14,288✔

6038

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

14,288✔

6039

  if (pSupp->groupByDbName) {

14,288✔

6040

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));

6,620✔

6042

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));

7,668✔

6045

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

14,288✔

6046

  pRes->info.id.groupId = groupId;

14,288✔

6048

  int64_t ctbNum = 0;

14,288✔

6049

  code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL, NULL);

14,288✔

6050

  QUERY_CHECK_CODE(code, lino, _end);

14,288✔

6051

  code = fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);

14,288✔

6052

  QUERY_CHECK_CODE(code, lino, _end);

14,288✔

6054

_end:

14,288✔

6055

  if (code != TSDB_CODE_SUCCESS) {

14,288✔

6058

  return code;

14,288✔

6061

static void destoryTableCountScanOperator(void* param) {

28,876✔

6062

  STableCountScanOperatorInfo* pTableCountScanInfo = param;

28,876✔

6063

  blockDataDestroy(pTableCountScanInfo->pRes);

28,876✔

6065

  taosArrayDestroy(pTableCountScanInfo->stbUidList);

28,845✔

6066

  taosMemoryFreeClear(param);

28,845✔

6067

28,845✔

taosdata / TDengine / #4987

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

taosdata / TDengine / #4987

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

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