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

76

bool processBlockWithProbability(const SSampleExecInfo* pInfo) {

523,273,745✔

85

  return true;

523,273,745✔

89

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

66,128✔

90

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

66,128✔

93

66,128✔

95

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

567,631,314✔

96

  int32_t     code = TSDB_CODE_SUCCESS;

567,631,314✔

97

  STimeWindow w = {0};

567,631,314✔

100

  if (pInterval->interval == 0) {

567,730,372✔

101

    *overlap = false;

532,189,984✔

102

    return code;

532,210,716✔

105

  if (order == TSDB_ORDER_ASC) {

35,469,393✔

106

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

34,283,215✔

107

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

34,282,079✔

112

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

34,276,481✔

113

      *overlap = true;

29,958,926✔

114

      return code;

29,958,990✔

118

      getNextTimeWindow(pInterval, &w, order);

4,323,175✔

119

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

4,327,178✔

120

        break;

4,058,626✔

123

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

267,255✔

127

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

267,255✔

128

        *overlap = true;

267,255✔

129

        return code;

267,255✔

133

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

1,186,178✔

134

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

1,186,899✔

139

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

1,185,658✔

140

      *overlap = true;

1,020,850✔

141

      return code;

1,020,850✔

145

      getNextTimeWindow(pInterval, &w, order);

164,979✔

146

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

165,948✔

147

        break;

163,368✔

150

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

2,580✔

154

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

2,580✔

155

        *overlap = true;

2,580✔

156

        return code;

2,580✔

161

  *overlap = false;

4,221,994✔

162

  return code;

4,219,446✔

166

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

138,572,554✔

167

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

138,572,554✔

171

  int64_t buf[2] = {0};

138,584,091✔

172

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

138,580,182✔

174

  STableScanInfo* pTableScanInfo = pOperator->info;

138,559,962✔

176

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

138,581,843✔

179

  if (p1 == NULL) {

138,592,997✔

180

    return NULL;

47,822,147✔

183

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

90,770,850✔

184

  if (NULL == *pPage) {

90,762,653✔

188

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

90,762,677✔

191

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

11,352,280✔

192

  if (NULL == pTableScanInfo->pIgnoreTables) {

11,352,280✔

193

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

1,098,964✔

194

    pTableScanInfo->pIgnoreTables =

1,098,964✔

195

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

1,098,964✔

196

    if (NULL == pTableScanInfo->pIgnoreTables) {

1,098,964✔

201

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

11,352,689✔

203

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

11,351,488✔

207

  return TSDB_CODE_SUCCESS;

11,351,488✔

210

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

630,182,823✔

211

  STableScanInfo* pTableScanInfo = pOperator->info;

630,182,823✔

212

  int32_t         code = TSDB_CODE_SUCCESS;

630,337,201✔

214

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

630,337,201✔

215

    return TSDB_CODE_SUCCESS;

491,733,419✔

218

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

138,561,740✔

220

  SFilePage*  pPage = NULL;

138,567,033✔

221

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

138,567,387✔

223

  if (pRow == NULL) {

138,580,755✔

224

    return TSDB_CODE_SUCCESS;

47,811,554✔

227

  bool notLoadBlock = true;

90,769,201✔

228

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

2,147,483,647✔

229

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

2,147,483,647✔

231

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

2,147,483,647✔

233

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

2,147,483,647✔

234

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

2,147,483,647✔

235

      notLoadBlock = false;

79,411,640✔

236

      break;

79,411,640✔

241

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

89,729,525✔

243

  if (notLoadBlock) {

90,768,212✔

244

    *status = FUNC_DATA_REQUIRED_NOT_LOAD;

11,356,370✔

245

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

11,352,715✔

248

  return code;

90,728,450✔

251

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

8,827,373✔

253

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

8,827,373✔

258

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

8,827,907✔

261

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

124,297,149✔

263

  SStorageAPI* pAPI = &pTaskInfo->storageAPI;

124,297,149✔

264

  bool         allColumnsHaveAgg = true;

124,313,531✔

265

  bool         hasNullSMA = false;

124,323,160✔

266

  if (pLoad != NULL) {

124,305,391✔

267

    *pLoad = false;

124,334,834✔

270

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

124,306,446✔

272

  if (code != TSDB_CODE_SUCCESS) {

124,294,278✔

276

  if (!allColumnsHaveAgg || hasNullSMA) {

124,294,278✔

277

    *pLoad = false;

110,224,341✔

279

    *pLoad = true;

14,069,937✔

282

  return code;

124,277,289✔

285

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

662,957,396✔

287

  int32_t    code = 0;

662,957,396✔

288

  SExprSupp* pSup = &pTableScanInfo->pseudoSup;

662,957,396✔

289

  if (pSup->numOfExprs > 0) {

663,145,409✔

290

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

313,459,278✔

293

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

313,437,150✔

299

  return code;

663,024,833✔

302

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

1,106,717,185✔

303

  SLimit*     pLimit = &pLimitInfo->limit;

1,106,717,185✔

304

  const char* id = GET_TASKID(pTaskInfo);

1,106,800,223✔

306

  if (pLimitInfo->remainOffset > 0) {

1,106,861,209✔

307

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

4,944,598✔

308

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

2,707,193✔

309

      blockDataEmpty(pBlock);

2,707,193✔

310

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

2,707,193✔

311

      return false;

2,707,193✔

313

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

2,237,405✔

314

      if (code != TSDB_CODE_SUCCESS) {

2,237,405✔

319

      pLimitInfo->remainOffset = 0;

2,237,405✔

323

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

1,104,124,543✔

325

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

28,712,146✔

326

    blockDataKeepFirstNRows(pBlock, keep);

28,713,263✔

328

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

28,711,547✔

329

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

28,712,514✔

330

    return true;

28,712,460✔

333

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

1,075,273,713✔

334

  return false;

1,075,372,970✔

341

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

658,948,042✔

343

  int32_t        code = TSDB_CODE_SUCCESS;

658,948,042✔

344

  int32_t        lino = 0;

658,948,042✔

345

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

658,948,042✔

346

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

659,039,341✔

347

  bool           loadSMA = false;

659,009,022✔

349

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

659,009,022✔

351

  pCost->totalBlocks += 1;

658,943,858✔

352

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

658,992,713✔

353

  *status = pTableScanInfo->dataBlockLoadFlag;

658,961,789✔

355

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

658,991,608✔

356

    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

91,240,657✔

358

    bool overlap = false;

567,654,357✔

360

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

567,684,422✔

361

    if (ret != TSDB_CODE_SUCCESS) {

567,551,074✔

364

    if (overlap) {

567,551,074✔

365

      (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

31,245,319✔

369

  SDataBlockInfo* pBlockInfo = &pBlock->info;

658,927,107✔

370

  taosMemoryFreeClear(pBlock->pBlockAgg);

658,964,478✔

372

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {

658,946,316✔

379

  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

658,771,739✔

380

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

20,045,418✔

383

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

20,053,813✔

384

    pCost->skipBlocks += 1;

20,054,476✔

385

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

20,054,392✔

386

    return code;

20,053,942✔

387

  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {

638,716,657✔

388

    pCost->loadBlockStatis += 1;

33,069,452✔

389

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

33,074,269✔

390

    bool success = true;

33,074,269✔

391

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

33,071,952✔

392

    if (code) {

33,061,583✔

398

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

33,059,825✔

399

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

5,242,564✔

401

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

5,242,564✔

402

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

5,242,564✔

403

      return code;

5,242,564✔

405

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

27,817,261✔

406

      *status = FUNC_DATA_REQUIRED_DATA_LOAD;

27,836,918✔

410

  if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {

633,491,008✔

417

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

633,489,975✔

418

    bool success = true;

91,240,736✔

419

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

91,245,352✔

420

    if (code) {

91,209,486✔

423

      QUERY_CHECK_CODE(code, lino, _end);

18,605✔

426

    if (success) {

91,228,091✔

427

      size_t size = taosArrayGetSize(pBlock->pDataBlock);

8,827,373✔

428

      bool   keep = false;

8,827,907✔

429

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

8,827,373✔

430

      if (code) {

8,827,373✔

436

      if (!keep) {

8,827,373✔

437

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

3,285,691✔

439

        pCost->filterOutBlocks += 1;

3,285,691✔

440

        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

3,285,691✔

441

        taosMemoryFreeClear(pBlock->pBlockAgg);

3,285,691✔

443

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

3,285,691✔

444

        return TSDB_CODE_SUCCESS;

3,285,691✔

450

  taosMemoryFreeClear(pBlock->pBlockAgg);

630,311,548✔

453

  code = doDynamicPruneDataBlock(pOperator, pBlockInfo, status);

630,211,409✔

454

  if (code) {

630,218,245✔

459

  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

630,218,245✔

460

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

11,356,396✔

462

    pCost->skipBlocks += 1;

11,358,032✔

463

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

11,356,805✔

465

    STableScanInfo* p1 = pOperator->info;

11,356,805✔

466

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

11,356,805✔

467

      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;

2,721,146✔

469

      *status = FUNC_DATA_REQUIRED_FILTEROUT;

8,635,659✔

471

    return TSDB_CODE_SUCCESS;

11,356,491✔

474

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

618,843,775✔

475

  pCost->loadBlocks += 1;

618,946,921✔

477

  SSDataBlock* p = NULL;

618,932,671✔

478

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

618,915,538✔

479

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

619,002,809✔

483

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

619,002,809✔

484

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

485,145,147✔

488

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

592,024,575✔

489

    if (code) {

591,967,752✔

495

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

618,929,387✔

497

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

618,915,042✔

498

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

87,923,757✔

499

    QUERY_CHECK_CODE(code, lino, _end);

87,904,309✔

501

    int64_t st = taosGetTimestampUs();

87,792,517✔

502

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

87,800,403✔

503

    pTableScanInfo->readRecorder.filterTime += el;

87,800,403✔

505

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

87,786,598✔

506

      pCost->filterOutBlocks += 1;

15,908,547✔

507

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

15,906,558✔

510

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

71,861,735✔

514

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

618,815,456✔

515

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

618,831,745✔

516

    setOperatorCompleted(pOperator);

2,822,364✔

519

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

618,831,394✔

521

_end:

619,058,930✔

522

  if (code != TSDB_CODE_SUCCESS) {

618,994,546✔

523

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

128,960✔

525

  return code;

618,911,637✔

528

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

66,128✔

529

  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

66,128✔

531

  switchCtxOrder(pCtx, numOfOutput);

66,128✔

532

  pTableScanInfo->cond.order = TSDB_ORDER_DESC;

66,128✔

535

66,128✔

603

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

375,551,228✔

605

  int32_t          code = TSDB_CODE_SUCCESS;

375,551,228✔

606

  int32_t          lino = 0;

375,551,228✔

607

  bool             freeReader = false;

375,551,228✔

608

  LRUHandle*       h = NULL;

375,551,228✔

609

  STableCachedVal  val = {0};

375,551,228✔

610

  SMetaReader      mr = {0};

375,551,228✔

611

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

375,533,551✔

612

  int32_t          insertRet = TAOS_LRU_STATUS_OK;

375,582,260✔

613

  STableCachedVal* pVal = NULL;

375,582,260✔

616

  if (numOfExpr <= 0) {

375,557,054✔

617

    return TSDB_CODE_SUCCESS;

540,625✔

623

  int32_t backupRows = pBlock->info.rows;

375,016,429✔

624

  pBlock->info.rows = rows;

375,074,628✔

630

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

375,106,039✔

631

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

375,106,501✔

632

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

375,138,284✔

633

    if (code != TSDB_CODE_SUCCESS) {

374,923,351✔

648

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

374,923,351✔

650

    val.pName = mr.me.name;

375,084,877✔

651

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

375,084,877✔

653

    freeReader = true;

375,084,877✔

695

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

1,457,109,711✔

696

    const SExprInfo* pExpr1 = &pExpr[j];

1,081,873,801✔

697

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

1,081,967,019✔

699

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

1,081,857,847✔

700

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

1,082,129,354✔

702

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

1,082,110,951✔

705

    if (fmIsScanPseudoColumnFunc(functionId)) {

1,082,133,594✔

706

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

125,408,411✔

707

      if (fType == FUNCTION_TYPE_TBNAME) {

125,415,419✔

708

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

125,415,419✔

709

        QUERY_CHECK_CODE(code, lino, _end);

125,395,144✔

718

      STagVal tagVal = {0};

956,514,429✔

719

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

956,491,539✔

720

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

956,742,021✔

722

      char* data = NULL;

956,753,774✔

723

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

956,753,774✔

724

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

950,560,718✔

726

        data = (char*)p;

6,229,755✔

729

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

956,612,668✔

730

      if (isNullVal) {

956,582,797✔

731

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

5,748,894✔

732

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

950,833,903✔

733

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

950,441,610✔

734

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

950,502,268✔

735

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

363,728,030✔

736

          if (tmp != NULL) {

363,755,831✔

737

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

363,755,831✔

738

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

363,741,785✔

740

            taosMemoryFree(tmp);

363,792,683✔

742

          taosMemoryFree(data);

363,789,670✔

744

        QUERY_CHECK_CODE(code, lino, _end);

950,480,350✔

746

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

2,981,525✔

747

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

2,500,664✔

748

          QUERY_CHECK_CODE(code, lino, _end);

2,500,664✔

755

  pBlock->info.rows = backupRows;

375,235,910✔

757

_end:

375,094,960✔

759

  if (NULL != pVal) {

375,119,299✔

768

  if (freeReader) {

375,119,299✔

769

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

375,062,423✔

771

  if (code != TSDB_CODE_SUCCESS) {

375,101,515✔

774

  return code;

375,037,726✔

777

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

125,396,465✔

779

  int32_t                     code = TSDB_CODE_SUCCESS;

125,396,465✔

780

  int32_t                     lino = 0;

125,396,465✔

781

  struct SScalarFuncExecFuncs fpSet = {0};

125,396,465✔

782

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

125,409,522✔

783

  QUERY_CHECK_CODE(code, lino, _end);

125,412,789✔

785

  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;

125,412,789✔

786

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

125,412,789✔

787

  STR_TO_VARSTR(buf, name)

125,416,764✔

789

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

125,402,509✔

791

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

125,396,354✔

792

  QUERY_CHECK_CODE(code, lino, _end);

125,405,406✔

794

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

125,405,406✔

795

  QUERY_CHECK_CODE(code, lino, _end);

125,413,891✔

797

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

125,413,891✔

798

  SScalarParam param = {.columnData = pColInfoData};

125,412,566✔

800

  if (fpSet.process != NULL) {

125,390,206✔

801

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

125,390,206✔

802

    QUERY_CHECK_CODE(code, lino, _end);

125,419,903✔

807

  colDataDestroy(&infoData);

125,419,903✔

809

_end:

125,391,922✔

810

  if (code != TSDB_CODE_SUCCESS) {

125,407,171✔

813

  return code;

125,407,171✔

882

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

178,985,990✔

883

  int32_t code = TSDB_CODE_SUCCESS;

178,985,990✔

884

  int32_t lino = 0;

178,985,990✔

885

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

178,985,990✔

886

  QUERY_CHECK_CODE(code, lino, _end);

178,978,854✔

888

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

178,978,854✔

889

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

179,031,360✔

890

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

178,826,323✔

891

  taosMemoryFreeClear(pInfo->pResBlock->pBlockAgg);

178,998,635✔

893

  if (!pInfo->needCountEmptyTable) {

178,871,268✔

894

    pInfo->countState = TABLE_COUNT_STATE_END;

160,863,437✔

896

    pInfo->countState = TABLE_COUNT_STATE_SCAN;

18,107,155✔

899

_end:

178,925,034✔

900

  if (code != TSDB_CODE_SUCCESS) {

178,925,034✔

903

  return code;

178,966,439✔

906

void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {

541,904,989✔

907

  if (pInfo->countState == TABLE_COUNT_STATE_END) {

541,904,989✔

908

    return;

437,705,526✔

910

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

104,259,378✔

911

    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;

221,682✔

913

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

104,038,108✔

914

    if (code != TSDB_CODE_SUCCESS) {

104,043,621✔

915

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

41,223,425✔

920

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

45,782,158✔

922

  blockDataEmpty(pBlock);

45,782,158✔

923

  pBlock->info.rows = 1;

45,782,657✔

924

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

45,782,657✔

925

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

45,782,657✔

926

  pBlock->info.blankFill = true;

45,781,659✔

929

  int32_t col_num = blockDataGetNumOfCols(pBlock);

45,779,757✔

930

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

189,313,757✔

931

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

143,529,603✔

936

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

45,784,154✔

938

  return pBlock;

45,782,657✔

941

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

43,980,124✔

942

  STableScanInfo* pTableScanInfo = pOperator->info;

43,980,124✔

944

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

43,979,625✔

946

  pOperator->resultInfo.totalRows++;

43,978,128✔

947

  return pBlock;

43,979,625✔

954

static int32_t prepareTableScan(SOperatorInfo* pOperator) {

701,848,854✔

955

  int32_t code = TSDB_CODE_SUCCESS;

701,848,854✔

956

  int32_t lino = 0;

701,848,854✔

958

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

701,848,854✔

959

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

701,921,123✔

960

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

702,060,615✔

961

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

31,124,304✔

962

         NOTIFY_TYPE_SCAN_PARAM &&

2,943,188✔

963

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

2,943,188✔

964

    STableScanOperatorParam* pNotify =

2,943,188✔

967

    STableScanInfo* pTableScanInfo = pOperator->info;

2,943,188✔

968

    SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

2,943,188✔

969

    SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

2,943,188✔

970

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

2,943,188✔

972

    QUERY_CHECK_CODE(code, lino, _end);

2,943,188✔

974

    if (!pGetParam->reUse) {

2,943,188✔

975

      freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

90,616✔

981

      pGetParam->reUse = false;

2,852,572✔

986

    pOperator->pOperatorGetParam = NULL;

2,943,188✔

989

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

702,052,261✔

990

_end:

82,256,466✔

991

  if (code != TSDB_CODE_SUCCESS) {

702,055,472✔

995

  return code;

702,042,511✔

998

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

663,608,419✔

999

  int32_t         code = TSDB_CODE_SUCCESS;

663,608,419✔

1000

  int32_t         lino = 0;

663,608,419✔

1001

  STableScanInfo* pTableScanInfo = pOperator->info;

663,608,419✔

1002

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

663,740,589✔

1003

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

663,704,908✔

1004

  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;

663,632,147✔

1005

  bool            hasNext = false;

663,677,591✔

1006

  int64_t         st = taosGetTimestampUs();

663,641,020✔

1008

  QRY_PARAM_CHECK(ppRes);

663,641,020✔

1009

  pBlock->info.dataLoad = false;

663,688,611✔

1011

  while (true) {

22,500,773✔

1012

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

686,196,176✔

1013

    if (code != TSDB_CODE_SUCCESS) {

686,093,949✔

1015

      QUERY_CHECK_CODE(code, lino, _end);

42✔

1018

    if (!hasNext) {

686,054,053✔

1019

      break;

162,715,960✔

1022

    if (isTaskKilled(pTaskInfo)) {

523,338,093✔

1023

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

10,675✔

1024

      code = pTaskInfo->code;

10,675✔

1025

      QUERY_CHECK_CODE(code, lino, _end);

10,675✔

1028

    if (pOperator->status == OP_EXEC_DONE) {

523,528,170✔

1029

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

118,048✔

1030

      break;

118,048✔

1034

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

523,301,012✔

1035

    if (!processThisBlock) {

523,327,077✔

1039

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

523,327,077✔

1040

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

523,386,650✔

1043

    uint32_t status = 0;

523,433,420✔

1044

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

523,425,415✔

1045

    QUERY_CHECK_CODE(code, lino, _end);

523,310,672✔

1047

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

523,198,352✔

1048

      break;

2,720,832✔

1052

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

520,477,520✔

1053

      continue;

22,500,773✔

1056

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

498,021,721✔

1057

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

498,068,962✔

1059

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

498,088,086✔

1060

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

498,079,967✔

1062

    (*ppRes) = pBlock;

497,997,751✔

1063

    return code;

497,942,431✔

1066

_end:

165,685,365✔

1067

  if (code != TSDB_CODE_SUCCESS) {

165,692,968✔

1068

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

131,248✔

1069

    pTaskInfo->code = code;

131,248✔

1070

    T_LONG_JMP(pTaskInfo->env, code);

122,995✔

1072

  return code;

165,561,720✔

1075

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

710,568,246✔

1076

  int32_t         code = TSDB_CODE_SUCCESS;

710,568,246✔

1077

  int32_t         lino = 0;

710,568,246✔

1078

  STableScanInfo* pTableScanInfo = pOperator->info;

710,568,246✔

1079

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

710,626,260✔

1080

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

710,540,945✔

1082

  QRY_PARAM_CHECK(pBlock);

710,538,729✔

1085

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

710,573,615✔

1086

    return code;

3,038,199✔

1090

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

859,008,298✔

1091

    SSDataBlock* p = NULL;

578,879,895✔

1092

    code = doTableScanImplNext(pOperator, &p);

578,839,472✔

1093

    QUERY_CHECK_CODE(code, lino, _end);

578,679,763✔

1095

    if (p != NULL) {

578,679,763✔

1096

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

427,136,116✔

1097

      *pBlock = p;

427,082,900✔

1098

      return code;

427,068,567✔

1101

    pTableScanInfo->scanTimes += 1;

151,543,647✔

1102

    taosHashClear(pTableScanInfo->pIgnoreTables);

151,559,453✔

1104

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

151,527,619✔

1105

      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

83,076✔

1106

      pTableScanInfo->base.scanFlag = MAIN_SCAN;

83,076✔

1107

      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

83,076✔

1108

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

83,076✔

1111

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

83,076✔

1112

      QUERY_CHECK_CODE(code, lino, _end);

75,192✔

1116

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

280,211,263✔

1117

  if (pTableScanInfo->scanTimes < total) {

280,208,328✔

1118

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

84,829,676✔

1119

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

66,128✔

1120

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

66,128✔

1121

      QUERY_CHECK_CODE(code, lino, _end);

66,128✔

1122

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

66,128✔

1125

    while (pTableScanInfo->scanTimes < total) {

98,846,586✔

1126

      SSDataBlock* p = NULL;

84,825,471✔

1127

      code = doTableScanImplNext(pOperator, &p);

84,811,782✔

1128

      QUERY_CHECK_CODE(code, lino, _end);

84,789,519✔

1130

      if (p != NULL) {

84,789,519✔

1131

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

70,774,195✔

1132

        *pBlock = p;

70,792,255✔

1133

        return code;

70,776,844✔

1136

      pTableScanInfo->scanTimes += 1;

14,015,324✔

1137

      taosHashClear(pTableScanInfo->pIgnoreTables);

14,012,520✔

1139

      if (pTableScanInfo->scanTimes < total) {

14,013,883✔

1145

        QUERY_CHECK_CODE(code, lino, _end);

191✔

1150

  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {

209,374,413✔

1151

    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;

61,388,641✔

1152

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

61,389,735✔

1153

      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {

120,387✔

1154

        pTableScanInfo->countState = TABLE_COUNT_STATE_PROCESSED;

4,876✔

1156

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

4,876✔

1158

        if (NULL == pStart) {

4,876✔

1162

        *pBlock = getBlockForEmptyTable(pOperator, pStart);

4,876✔

1163

        return code;

4,876✔

1166

      int32_t numOfTables = 0;

61,266,544✔

1167

      code = tableListGetSize(pTableListInfo, &numOfTables);

61,268,041✔

1168

      QUERY_CHECK_CODE(code, lino, _end);

61,269,256✔

1170

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

61,269,256✔

1172

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

60,083,099✔

1173

        if (pIte != NULL) {

60,083,112✔

1174

          size_t        keySize = 0;

43,977,338✔

1175

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

43,977,338✔

1176

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

43,977,743✔

1177

          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);

43,973,845✔

1178

          markGroupProcessed(pTableScanInfo, *pGroupId);

43,974,749✔

1179

          *pBlock = getBlockForEmptyTable(pOperator, &info);

43,975,747✔

1181

          return code;

43,975,248✔

1185

    pTableScanInfo->countState = TABLE_COUNT_STATE_END;

17,406,809✔

1188

_end:

148,017,858✔

1189

  if (code != TSDB_CODE_SUCCESS) {

165,410,416✔

1194

  return code;

165,410,416✔

1197

static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {

938,298✔

1198

  STableScanInfo*          pInfo = pOperator->info;

938,298✔

1199

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

938,672✔

1200

  int32_t                  code = 0;

938,672✔

1201

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

938,672✔

1202

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

938,672✔

1203

  int32_t                  num = taosArrayGetSize(pParam->pUidList);

938,672✔

1204

  if (num <= 0) {

938,672✔

1209

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

938,672✔

1213

  if (pParam->tableSeq) {

938,672✔

1214

    pListInfo->oneTableForEachGroup = true;

697,959✔

1215

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

697,959✔

1221

    pListInfo->oneTableForEachGroup = false;

240,713✔

1222

    pListInfo->numOfOuputGroups = 1;

240,713✔

1224

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

938,672✔

1227

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

693,567✔

1228

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

693,567✔

1231

  STableKeyInfo info = {.groupId = 0};

938,672✔

1232

  int32_t       tableIdx = 0;

938,672✔

1233

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

8,890,605✔

1234

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

7,951,933✔

1235

    if (!pUid) {

7,951,933✔

1240

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

7,951,933✔

1241

      if (TSDB_CODE_DUP_KEY == terrno) {

6,670,576✔

1242

        continue;

6,670,576✔

1247

    info.uid = *pUid;

1,281,357✔

1248

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

1,281,357✔

1249

    if (p == NULL) {

1,281,357✔

1253

    tableIdx++;

1,281,357✔

1254

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

1,281,357✔

1257

  return code;

938,672✔

1260

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

165,203,626✔

1262

  const char* idStr = GET_TASKID(pTaskInfo);

165,203,626✔

1263

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

165,238,093✔

1264

  if (code) {

165,173,188✔

1269

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

165,173,188✔

1274

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

330,345,762✔

1275

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

165,100,805✔

1276

  if (code) {

165,121,072✔

1277

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

2,190✔

1280

  return code;

165,069,014✔

1283

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

294,372,342✔

1284

  SColIdPair* node1 = (SColIdPair*)elem1;

294,372,342✔

1285

  SColIdPair* node2 = (SColIdPair*)elem2;

294,372,342✔

1287

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

294,372,342✔

1288

    return 0;

107,816✔

1291

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

294,283,163✔

1294

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

396,086,974✔

1295

  SColIdSlotIdPair* node1 = (SColIdSlotIdPair*)elem1;

396,086,974✔

1296

  SColIdSlotIdPair* node2 = (SColIdSlotIdPair*)elem2;

396,086,974✔

1298

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

396,086,974✔

1299

    return 0;

164,696✔

1302

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

395,948,789✔

1305

static int compareColMatchItemByColId(const void* elem1, const void* elem2) {

705,439,537✔

1306

  const SColMatchItem* node1 = (const SColMatchItem*)elem1;

705,439,537✔

1307

  const SColMatchItem* node2 = (const SColMatchItem*)elem2;

705,439,537✔

1309

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

705,439,537✔

1313

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

705,439,537✔

1316

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

92,992,518✔

1317

  if (pList == NULL) {

92,992,518✔

1321

  int32_t left = 0;

92,992,518✔

1322

  int32_t right = taosArrayGetSize(pList) - 1;

92,992,518✔

1323

  while (left <= right) {

288,360,716✔

1324

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

288,360,716✔

1325

    SColMatchItem* pItem = taosArrayGet(pList, mid);

288,360,716✔

1326

    if (pItem->colId == colId) {

288,360,716✔

1328

      while (mid > 0) {

92,992,518✔

1329

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

69,741,869✔

1330

        if (pPrev->colId != colId) {

69,741,869✔

1331

          break;

69,741,869✔

1335

      return taosArrayGet(pList, mid);

92,992,518✔

1338

    if (pItem->colId < colId) {

195,368,198✔

1339

      left = mid + 1;

91,773,144✔

1341

      right = mid - 1;

103,595,054✔

1348

static bool isNewScanParam(STableScanOperatorParam* pParam) {

39,768,855✔

1349

  return pParam->isNewParam;

39,768,855✔

1352

static int32_t createVTableScanInfoFromBatchParam(SOperatorInfo* pOperator) {

17,137,615✔

1353

  int32_t                  code = 0;

17,137,615✔

1354

  int32_t                  lino = 0;

17,137,615✔

1355

  STableScanInfo*          pInfo = pOperator->info;

17,137,615✔

1356

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

17,145,254✔

1357

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

17,124,571✔

1358

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

17,135,607✔

1359

  SMetaReader              orgTable = {0};

17,148,004✔

1360

  SMetaReader              superTable = {0};

17,164,413✔

1361

  SSchemaWrapper*          schema = NULL;

17,146,913✔

1362

  SArray*                  pColArray = NULL;

17,146,913✔

1363

  SArray*                  pBlockColArray = NULL;

17,146,913✔

1364

  int32_t                  num = 0;

17,146,913✔

1365

  STableKeyInfo*           pList = NULL;

17,144,752✔

1369

  cleanupQueryTableDataCond(&pInfo->base.cond);

17,145,502✔

1371

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

17,149,069✔

1372

  if (pGetParam &&

17,157,526✔

1373

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

3,983,737✔

1374

    STableScanOperatorParam* pParam =

3,983,717✔

1376

    pInfo->pBatchColMap = pParam->pBatchTbInfo;

3,995,508✔

1377

    pParam->pBatchTbInfo = NULL;

3,981,201✔

1378

    pInfo->cachedTimeWindow = pParam->window;

3,976,282✔

1379

    pInfo->currentBatchIdx = 0;

3,991,234✔

1380

    pInfo->lastBatchIdx = -1;

3,970,234✔

1381

    pInfo->cachedTagList = pParam->pTagList;

3,980,250✔

1382

    pParam->pTagList = NULL;

3,987,825✔

1383

    pInfo->cachedGroupId = pParam->groupid;

3,984,548✔

1385

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

3,994,375✔

1387

    taosHashClear(pInfo->pIgnoreTables);

3,994,375✔

1389

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

3,990,775✔

1390

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

3,799,507✔

1391

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

47,728,937✔

1392

        SqlFunctionCtx ctx = pSup->pCtx[i];

43,937,361✔

1393

        ctx.skipDynDataCheck = false;

43,931,764✔

1398

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

17,159,477✔

1399

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

17,165,403✔

1401

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

17,165,403✔

1402

    isNewTable = false;

10,440,871✔

1404

    isNewTable = true;

6,709,596✔

1405

    pInfo->lastBatchIdx = pInfo->currentBatchIdx;

6,709,596✔

1408

  if (isNewTable) {

17,158,974✔

1409

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

6,718,731✔

1410

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

6,724,102✔

1411

    pAPI->metaReaderFn.readerReleaseLock(&orgTable);

6,698,025✔

1412

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

6,702,725✔

1413

    QUERY_CHECK_CODE(code, lino, _return);

6,707,361✔

1414

    switch (orgTable.me.type) {

6,707,361✔

1415

      case TSDB_CHILD_TABLE:

279,454✔

1416

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

279,454✔

1417

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

279,454✔

1418

        pAPI->metaReaderFn.readerReleaseLock(&superTable);

278,882✔

1419

        QUERY_CHECK_CODE(code, lino, _return);

279,454✔

1420

        schema = &superTable.me.stbEntry.schemaRow;

279,454✔

1421

        break;

279,454✔

1422

      case TSDB_NORMAL_TABLE:

6,391,629✔

1423

        schema = &orgTable.me.ntbEntry.schemaRow;

6,391,629✔

1424

        break;

6,391,629✔

1425

      default:

36,278✔

1426

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

36,278✔

1431

    pListInfo->oneTableForEachGroup = true;

6,671,083✔

1432

    taosHashClear(pListInfo->map);

6,707,173✔

1433

    taosArrayClear(pListInfo->pTableList);

6,728,702✔

1435

    uint64_t      pUid = orgTable.me.uid;

6,727,313✔

1436

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

6,727,313✔

1437

    int32_t       tableIdx = 0;

6,727,313✔

1438

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

6,726,808✔

1439

    QUERY_CHECK_CODE(code, lino, _return);

6,728,702✔

1440

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

13,453,186✔

1441

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

6,724,484✔

1443

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

6,724,484✔

1444

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

6,724,675✔

1445

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

6,724,675✔

1446

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

6,717,237✔

1449

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

6,717,237✔

1450

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

6,727,255✔

1452

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

25,688,830✔

1453

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

18,956,864✔

1454

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

151,012,725✔

1455

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

150,701,522✔

1456

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

18,965,424✔

1457

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

18,969,519✔

1458

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

18,969,519✔

1460

          break;

18,962,586✔

1465

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

32,413,207✔

1466

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

25,680,791✔

1467

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

153,756,636✔

1468

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

153,576,231✔

1469

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

153,614,235✔

1470

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

25,681,035✔

1471

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

25,696,392✔

1472

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

25,696,392✔

1474

          break;

25,676,946✔

1479

    taosArraySort(pColArray, compareColIdPair);

6,728,074✔

1480

    taosArraySort(pBlockColArray, compareColIdSlotIdPair);

6,720,450✔

1481

    if (pInfo->pBlockColMap) {

6,719,943✔

1482

      taosArrayDestroy(pInfo->pBlockColMap);

4,795,564✔

1483

      pInfo->pBlockColMap = NULL;

4,785,299✔

1485

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

6,712,572✔

1486

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

6,721,730✔

1489

    taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

6,714,036✔

1490

    taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

6,710,246✔

1492

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

6,723,344✔

1493

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

6,003,595✔

1494

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

74,519,717✔

1495

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

68,530,088✔

1496

        bool           needScan = false;

68,527,780✔

1497

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

137,986,842✔

1498

          SFunctParam pParam = ctx->param[j];

69,426,366✔

1499

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

333,376,038✔

1500

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

280,198,930✔

1501

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

280,780,927✔

1502

              needScan |= true;

17,290,163✔

1503

              break;

17,290,163✔

1507

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

68,526,634✔

1508

        if (!needScan) {

68,559,383✔

1509

          ctx->skipDynDataCheck = true;

51,281,164✔

1511

          ctx->skipDynDataCheck = false;

17,278,219✔

1518

    pColArray = pInfo->lastColArray;

10,440,243✔

1519

    pBlockColArray = pInfo->lastBlockColArray;

10,440,243✔

1522

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

17,161,699✔

1523

  QUERY_CHECK_CODE(code, lino, _return);

17,166,735✔

1525

  if (pInfo->pResBlock) {

17,166,735✔

1526

    blockDataDestroy(pInfo->pResBlock);

13,164,982✔

1527

    pInfo->pResBlock = NULL;

13,163,535✔

1530

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

17,166,131✔

1538

  if (isNewTable) {

17,154,144✔

1539

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

6,714,406✔

1540

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

6,006,119✔

1541

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

5,973,413✔

1543

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

720,212✔

1544

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

709,752✔

1547

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

10,439,738✔

1548

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

1,378,004✔

1550

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

9,061,734✔

1554

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

17,147,414✔

1555

  pInfo->currentGroupId = 0;

17,140,880✔

1556

  pInfo->ignoreTag = true;

17,140,794✔

1558

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

17,081,573✔

1559

  QUERY_CHECK_CODE(code, lino, _return);

17,155,667✔

1561

  if (isNewTable) {

17,155,667✔

1562

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

6,715,929✔

1563

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

4,795,127✔

1565

    pInfo->base.dataReader = NULL;

6,686,606✔

1567

    taosRLockLatch(&pTaskInfo->lock);

6,694,113✔

1568

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

6,728,702✔

1569

    taosRUnLockLatch(&pTaskInfo->lock);

6,679,333✔

1570

    QUERY_CHECK_CODE(code, lino, _return);

6,711,407✔

1572

    pInfo->newReader = true;

6,711,407✔

1573

    if (pInfo->lastColArray) {

6,713,291✔

1574

      taosArrayDestroy(pInfo->lastColArray);

2,724,367✔

1575

      pInfo->lastColArray = NULL;

2,724,367✔

1577

    if (pInfo->lastBlockColArray) {

6,703,267✔

1578

      taosArrayDestroy(pInfo->lastBlockColArray);

2,724,367✔

1579

      pInfo->lastBlockColArray = NULL;

2,724,367✔

1581

    pInfo->lastColArray = pColArray;

6,681,429✔

1582

    pInfo->lastBlockColArray = pBlockColArray;

6,677,870✔

1584

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

10,439,738✔

1585

    QUERY_CHECK_CODE(code, lino, _return);

10,440,871✔

1587

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

10,440,871✔

1588

    pInfo->newReader = false;

10,440,871✔

1589

    pInfo->scanTimes = 0;

10,440,871✔

1592

  pOperator->status = OP_OPENED;

17,140,501✔

1594

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

17,137,512✔

1598

  pInfo->currentGroupId = -1;

17,142,816✔

1599

_return:

17,067,001✔

1600

  if (code) {

17,095,451✔

1603

  pAPI->metaReaderFn.clearReader(&superTable);

17,095,451✔

1604

  pAPI->metaReaderFn.clearReader(&orgTable);

17,120,767✔

1605

  return code;

17,133,128✔

1608

static int32_t createVTableScanInfoFromParam(SOperatorInfo* pOperator) {

23,250,649✔

1609

  int32_t                  code = 0;

23,250,649✔

1610

  int32_t                  lino = 0;

23,250,649✔

1611

  STableScanInfo*          pInfo = pOperator->info;

23,250,649✔

1612

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

23,250,649✔

1613

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

23,250,649✔

1614

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

23,250,649✔

1615

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

23,250,649✔

1616

  SMetaReader              orgTable = {0};

23,250,649✔

1617

  SMetaReader              superTable = {0};

23,250,649✔

1618

  SSchemaWrapper*          schema = NULL;

23,250,649✔

1619

  SArray*                  pColArray = NULL;

23,250,649✔

1620

  SArray*                  pBlockColArray = NULL;

23,250,649✔

1621

  int32_t                  num = 0;

23,250,649✔

1622

  STableKeyInfo*           pList = NULL;

23,250,649✔

1624

  SArray*                  pMatchList = NULL;

23,250,649✔

1626

  cleanupQueryTableDataCond(&pInfo->base.cond);

23,250,649✔

1628

  pOrgTbInfo = pParam->pOrgTbInfo;

23,250,649✔

1630

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

23,250,649✔

1632

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

23,250,649✔

1633

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

23,250,649✔

1634

  pAPI->metaReaderFn.readerReleaseLock(&orgTable);

23,250,649✔

1635

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

23,250,649✔

1636

  QUERY_CHECK_CODE(code, lino, _return);

23,250,649✔

1637

  switch (orgTable.me.type) {

23,250,649✔

1638

    case TSDB_CHILD_TABLE:

6,351,635✔

1639

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

6,351,635✔

1640

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

6,351,635✔

1641

      pAPI->metaReaderFn.readerReleaseLock(&superTable);

6,351,635✔

1642

      QUERY_CHECK_CODE(code, lino, _return);

6,351,635✔

1643

      schema = &superTable.me.stbEntry.schemaRow;

6,351,635✔

1644

      break;

6,351,635✔

1645

    case TSDB_NORMAL_TABLE:

16,899,014✔

1646

      schema = &orgTable.me.ntbEntry.schemaRow;

16,899,014✔

1647

      break;

16,899,014✔

1654

  pListInfo->oneTableForEachGroup = true;

23,250,649✔

1655

  taosHashClear(pListInfo->map);

23,250,649✔

1656

  taosArrayClear(pListInfo->pTableList);

23,250,649✔

1658

  uint64_t      pUid = orgTable.me.uid;

23,250,649✔

1659

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

23,250,649✔

1660

  int32_t       tableIdx = 0;

23,250,649✔

1661

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

23,250,649✔

1662

  QUERY_CHECK_CODE(code, lino, _return);

23,250,649✔

1663

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

46,501,298✔

1664

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

23,250,649✔

1666

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

23,250,649✔

1667

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

23,250,649✔

1668

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

23,250,649✔

1669

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

23,250,649✔

1670

  if (pInfo->pBlockColMap) {

23,250,649✔

1671

    taosArrayDestroy(pInfo->pBlockColMap);

20,916,319✔

1672

    pInfo->pBlockColMap = NULL;

20,916,319✔

1674

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

23,250,649✔

1675

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

23,250,649✔

1678

  SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID,

23,250,649✔

1682

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

23,250,649✔

1684

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

92,992,518✔

1685

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

69,741,869✔

1686

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

588,433,024✔

1687

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

588,433,024✔

1688

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

69,741,869✔

1689

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

69,741,869✔

1690

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

69,741,869✔

1691

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

69,741,869✔

1692

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

69,741,869✔

1693

        break;

69,741,869✔

1698

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

23,250,649✔

1699

    pMatchList = taosArrayDup(pInfo->base.matchInfo.pList, NULL);

23,250,649✔

1700

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

23,250,649✔

1701

    taosArraySort(pMatchList, compareColMatchItemByColId);

23,250,649✔

1705

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

116,242,541✔

1706

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

92,992,518✔

1707

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

92,992,518✔

1708

    if (pItem) {

92,992,518✔

1709

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

92,992,518✔

1710

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

626✔

1712

        code = TSDB_CODE_VTABLE_COLUMN_TYPE_MISMATCH;

626✔

1713

        goto _return;

626✔

1715

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

92,991,892✔

1716

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

92,991,892✔

1717

      if (i > 0) {

92,991,892✔

1718

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

139,482,486✔

1723

  taosArraySort(pColArray, compareColIdPair);

23,250,023✔

1724

  taosArraySort(pBlockColArray, compareColIdSlotIdPair);

23,250,023✔

1725

  taosArraySort(pInfo->pBlockColMap, compareColIdSlotIdPair);

23,250,023✔

1727

  taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

23,250,023✔

1728

  taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

23,250,023✔

1730

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

23,250,023✔

1731

  QUERY_CHECK_CODE(code, lino, _return);

23,250,023✔

1733

  if (pInfo->pResBlock) {

23,250,023✔

1734

    blockDataDestroy(pInfo->pResBlock);

16,521,084✔

1735

    pInfo->pResBlock = NULL;

16,521,084✔

1738

  if (isNewScanParam(pParam)) {

23,250,023✔

1739

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

6,736,943✔

1740

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

6,736,943✔

1741

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

6,736,943✔

1747

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

16,513,080✔

1749

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

23,250,023✔

1750

  pInfo->currentGroupId = 0;

23,250,023✔

1751

  pInfo->ignoreTag = true;

23,250,023✔

1753

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

23,250,023✔

1754

  QUERY_CHECK_CODE(code, lino, _return);

23,250,023✔

1756

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

23,250,023✔

1757

  if (reader) {

23,250,023✔

1758

    if (isNewScanParam(pParam)) {

16,518,832✔

1759

      pAPI->tsdReader.tsdReaderClose(*reader);

5,752✔

1760

      pInfo->base.dataReader = NULL;

5,752✔

1762

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

5,752✔

1763

      QUERY_CHECK_CODE(code, lino, _return);

5,752✔

1765

      taosRLockLatch(&pTaskInfo->lock);

5,752✔

1766

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

5,752✔

1767

      taosRUnLockLatch(&pTaskInfo->lock);

5,752✔

1768

      QUERY_CHECK_CODE(code, lino, _return);

5,752✔

1770

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

5,752✔

1771

      QUERY_CHECK_CODE(code, lino, _return);

5,752✔

1772

      pInfo->newReader = true;

5,752✔

1774

      pInfo->base.dataReader = *reader;

16,513,080✔

1775

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

16,513,080✔

1776

      QUERY_CHECK_CODE(code, lino, _return);

16,513,080✔

1778

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

16,513,080✔

1779

      pInfo->newReader = false;

16,513,080✔

1780

      pInfo->scanTimes = 0;

16,513,080✔

1783

    pInfo->base.dataReader = NULL;

6,731,191✔

1785

    taosRLockLatch(&pTaskInfo->lock);

6,731,191✔

1786

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

6,731,191✔

1787

    taosRUnLockLatch(&pTaskInfo->lock);

6,731,191✔

1788

    QUERY_CHECK_CODE(code, lino, _return);

6,731,191✔

1790

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

6,731,191✔

1791

    QUERY_CHECK_CODE(code, lino, _return);

6,731,191✔

1792

    pInfo->newReader = true;

6,731,191✔

1794

  pOperator->status = OP_OPENED;

23,250,023✔

1796

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

23,250,023✔

1800

  pInfo->currentGroupId = -1;

23,250,023✔

1801

_return:

23,250,649✔

1802

  if (code) {

23,250,649✔

1803

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

626✔

1805

  taosArrayDestroy(pMatchList);

23,250,649✔

1806

  taosArrayDestroy(pColArray);

23,250,649✔

1807

  taosArrayDestroy(pBlockColArray);

23,250,649✔

1808

  pAPI->metaReaderFn.clearReader(&superTable);

23,250,649✔

1809

  pAPI->metaReaderFn.clearReader(&orgTable);

23,250,649✔

1810

  return code;

23,250,649✔

1813

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

162,856,236✔

1814

  int32_t         code = TSDB_CODE_SUCCESS;

162,856,236✔

1815

  int32_t         lino = 0;

162,856,236✔

1816

  STableScanInfo* pInfo = pOperator->info;

162,856,236✔

1817

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

162,867,552✔

1818

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

162,836,206✔

1819

  int32_t         numOfTables = 0;

162,831,732✔

1821

  QRY_PARAM_CHECK(pResult);

162,816,425✔

1823

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

162,851,785✔

1824

  QUERY_CHECK_CODE(code, lino, _end);

162,862,188✔

1826

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

162,862,188✔

1827

    setOperatorCompleted(pOperator);

149,060,347✔

1828

    if (pOperator->dynamicTask) {

149,066,899✔

1829

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

936,466✔

1830

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

936,466✔

1832

    return code;

149,067,968✔

1836

  pOperator->status = OP_OPENED;

13,720,651✔

1837

  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);

13,784,572✔

1839

  int32_t        num = 0;

13,776,371✔

1840

  STableKeyInfo* pList = NULL;

13,776,371✔

1841

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

13,780,275✔

1842

  QUERY_CHECK_CODE(code, lino, _end);

13,794,513✔

1844

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

13,794,513✔

1845

  QUERY_CHECK_CODE(code, lino, _end);

13,773,649✔

1847

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

13,773,649✔

1848

  QUERY_CHECK_CODE(code, lino, _end);

13,767,552✔

1849

  pInfo->scanTimes = 0;

13,767,552✔

1851

  code = doGroupedTableScan(pOperator, pResult);

13,771,703✔

1852

  QUERY_CHECK_CODE(code, lino, _end);

13,815,277✔

1854

  if (*pResult != NULL) {

13,815,277✔

1855

    if (pOperator->dynamicTask) {

11,183,970✔

1856

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

10,933,308✔

1858

    return code;

11,183,465✔

1861

_end:

2,630,791✔

1862

  if (code != TSDB_CODE_SUCCESS) {

2,633,821✔

1866

  return code;

2,632,372✔

1869

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

658,384,416✔

1870

  int32_t         code = TSDB_CODE_SUCCESS;

658,384,416✔

1871

  int32_t         lino = 0;

658,384,416✔

1872

  STableScanInfo* pInfo = pOperator->info;

658,384,416✔

1873

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

658,391,036✔

1874

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

658,386,812✔

1875

  int32_t         num = 0;

658,399,951✔

1876

  STableKeyInfo*  pList = NULL;

658,401,151✔

1877

  SSDataBlock*    pResult = NULL;

658,311,602✔

1878

  const char*     idStr = GET_TASKID(pTaskInfo);

658,294,670✔

1880

  QRY_PARAM_CHECK(pResBlock);

658,327,171✔

1882

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

658,365,839✔

1883

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

151,719,486✔

1888

    taosRLockLatch(&pTaskInfo->lock);

151,740,413✔

1889

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

151,807,782✔

1890

    taosRUnLockLatch(&pTaskInfo->lock);

151,710,173✔

1891

    QUERY_CHECK_CODE(code, lino, _end);

151,768,082✔

1893

    if (pInfo->filesetDelimited) {

151,765,892✔

1894

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

241,800✔

1897

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

151,719,604✔

1902

  code = doGroupedTableScan(pOperator, &pResult);

658,366,561✔

1903

  QUERY_CHECK_CODE(code, lino, _end);

658,064,221✔

1905

  if (pResult != NULL) {

658,064,221✔

1906

    if (pOperator->dynamicTask) {

508,929,488✔

1907

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

1,999,949✔

1910

    *pResBlock = pResult;

508,919,058✔

1911

    return code;

508,708,138✔

1915

    code = startNextGroupScan(pOperator, &pResult);

149,212,440✔

1916

    QUERY_CHECK_CODE(code, lino, _end);

149,411,308✔

1918

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

149,411,308✔

1919

      *pResBlock = pResult;

149,336,582✔

1920

      return code;

149,317,024✔

1924

_end:

2,190✔

1925

  if (code != TSDB_CODE_SUCCESS) {

2,190✔

1926

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

2,190✔

1927

    pTaskInfo->code = code;

2,190✔

1930

  return code;

2,190✔

1933

static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {

23,250,023✔

1934

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

23,250,023✔

1937

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

938,672✔

1938

  int32_t                  code = TSDB_CODE_SUCCESS;

938,672✔

1939

  int32_t                  lino = 0;

938,672✔

1940

  STableScanInfo*          pInfo = pOperator->info;

938,672✔

1941

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

938,672✔

1942

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

938,672✔

1943

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

938,672✔

1945

  code = createTableListInfoFromParam(pOperator);

938,672✔

1946

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

938,672✔

1947

  pOperator->pOperatorGetParam = NULL;

938,672✔

1948

  QUERY_CHECK_CODE(code, lino, _end);

938,672✔

1950

  if (pOperator->status == OP_EXEC_DONE) {

938,672✔

1951

    pInfo->currentGroupId = -1;

22,797✔

1952

    pOperator->status = OP_OPENED;

22,797✔

1953

    SSDataBlock* result = NULL;

22,797✔

1956

      code = startNextGroupScan(pOperator, &result);

22,797✔

1957

      QUERY_CHECK_CODE(code, lino, _end);

22,797✔

1959

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

22,797✔

1960

        (*ppRes) = result;

22,797✔

1961

        return code;

22,797✔

1965

_end:

915,875✔

1966

  if (code) {

915,875✔

1969

  return code;

915,875✔

1972

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

16,522,088✔

1973

  int32_t      code = TSDB_CODE_SUCCESS;

16,522,088✔

1974

  int32_t      lino = 0;

16,522,088✔

1975

  bool         found = false;

16,522,088✔

1977

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

16,522,088✔

1978

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

16,522,088✔

1980

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

82,233,460✔

1981

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

65,711,372✔

1982

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

65,711,372✔

1983

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

188,242,126✔

1984

      SColIdSlotIdPair* pColPair = taosArrayGet(pColMap, j);

171,722,998✔

1985

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

171,722,998✔

1986

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

171,722,998✔

1987

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

49,192,244✔

1988

        found = true;

49,192,244✔

1989

        break;

49,192,244✔

1992

    if (!found) {

65,711,372✔

1993

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

16,519,128✔

1994

        pSrcCol->info.slotId = -1;

16,519,128✔

2003

  return code;

16,522,088✔

2009

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

23,250,649✔

2010

  int32_t                  code = TSDB_CODE_SUCCESS;

23,250,649✔

2011

  int32_t                  lino = 0;

23,250,649✔

2012

  STableScanInfo*          pInfo = pOperator->info;

23,250,649✔

2013

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

23,250,649✔

2014

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

23,250,649✔

2015

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

23,250,649✔

2017

  QRY_PARAM_CHECK(ppRes);

23,250,649✔

2018

  qTrace("%s call", __FUNCTION__);

23,250,649✔

2020

  code = createVTableScanInfoFromParam(pOperator);

23,250,649✔

2021

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

23,250,649✔

2022

  pOperator->pOperatorGetParam = NULL;

23,250,649✔

2024

  QUERY_CHECK_CODE(code, lino, _end);

23,250,649✔

2026

  SSDataBlock* result = NULL;

23,250,023✔

2027

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

23,250,023✔

2028

    (*ppRes) = result;

18,116✔

2029

    return code;

18,116✔

2032

  if (pInfo->newReader) {

23,231,907✔

2033

    code = startNextGroupScan(pOperator, &result);

6,736,943✔

2034

    QUERY_CHECK_CODE(code, lino, _end);

6,736,943✔

2036

    code = doGroupedTableScan(pOperator, &result);

16,494,964✔

2037

    QUERY_CHECK_CODE(code, lino, _end);

16,494,964✔

2040

  if (result) {

23,231,907✔

2041

    SSDataBlock* res = NULL;

16,522,088✔

2042

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

16,522,088✔

2043

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

16,522,088✔

2044

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

16,522,088✔

2046

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

16,522,088✔

2047

    QUERY_CHECK_CODE(code, lino, _end);

16,522,088✔

2053

    (*ppRes) = result;

16,522,088✔

2055

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

6,709,819✔

2056

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

6,709,819✔

2058

    blockDataDestroy(pInfo->pResBlock);

6,709,819✔

2059

    pInfo->pResBlock = NULL;

6,709,819✔

2061

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

6,709,819✔

2062

    if (reader) {

6,709,819✔

2063

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

6,709,819✔

2064

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

6,709,819✔

2065

        pInfo->base.dataReader = NULL;

6,709,819✔

2067

      if (pAPI->tsdReader.tsdReaderClose) {

6,709,819✔

2068

        pAPI->tsdReader.tsdReaderClose(*reader);

6,709,819✔

2072

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

6,709,819✔

2073

    QUERY_CHECK_CODE(code, lino, _end);

6,709,819✔

2075

_end:

23,232,533✔

2076

  if (code) {

23,232,533✔

2077

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

626✔

2079

  return code;

23,232,533✔

2082

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

2,982,668✔

2083

  int32_t                  code = TSDB_CODE_SUCCESS;

2,982,668✔

2084

  int32_t                  lino = 0;

2,982,668✔

2085

  STableScanInfo*          pInfo = pOperator->info;

2,982,668✔

2086

  int32_t                  index = 0;

2,982,668✔

2087

  char*                    tagVal = NULL;

2,982,668✔

2089

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

20,892,124✔

2090

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

17,909,456✔

2091

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

17,909,456✔

2093

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

17,909,456✔

2094

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

17,909,456✔

2096

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

2,147,483,647✔

2097

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

2,147,483,647✔

2098

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

2,147,483,647✔

2099

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

2,147,483,647✔

2101

        varDataSetLen(tagVal, pTagVal->nData);

2,147,483,647✔

2102

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

2,147,483,647✔

2103

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

2,147,483,647✔

2104

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2105

        taosMemoryFreeClear(tagVal);

2,147,483,647✔

2107

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

2,147,483,647✔

2108

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2111

    index++;

17,909,456✔

2113

  return code;

2,982,668✔

2120

static void clearVstbBatchDynamicTableScanInfo(STableScanInfo* pInfo) {

4,001,195✔

2121

  if (pInfo->pBatchColMap) {

4,001,195✔

2122

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

10,725,310✔

2123

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

6,724,115✔

2124

      taosArrayDestroy(pOrgTbInfo->colMap);

6,723,173✔

2126

    taosArrayDestroy(pInfo->pBatchColMap);

4,000,062✔

2127

    pInfo->pBatchColMap = NULL;

3,999,625✔

2129

  if (pInfo->cachedTagList) {

4,000,253✔

2130

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

12,524,860✔

2131

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

10,671,728✔

2132

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

10,672,356✔

2133

        taosMemoryFreeClear(pTagVal->pData);

4,681,908✔

2136

    taosArrayDestroy(pInfo->cachedTagList);

1,852,818✔

2137

    pInfo->cachedTagList = NULL;

1,851,562✔

2139

  if (pInfo->lastColArray) {

3,998,997✔

2140

    taosArrayDestroy(pInfo->lastColArray);

4,000,567✔

2141

    pInfo->lastColArray = NULL;

3,997,113✔

2143

  if (pInfo->lastBlockColArray) {

3,997,113✔

2144

    taosArrayDestroy(pInfo->lastBlockColArray);

3,998,683✔

2145

    pInfo->lastBlockColArray = NULL;

3,999,625✔

2148

  blockDataDestroy(pInfo->pResBlock);

3,998,055✔

2149

  pInfo->pResBlock = NULL;

3,999,625✔

2150

3,999,939✔

2152

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

14,424,961✔

2153

  int32_t                  code = TSDB_CODE_SUCCESS;

14,424,961✔

2154

  int32_t                  lino = 0;

14,424,961✔

2155

  STableScanInfo*          pInfo = pOperator->info;

14,424,961✔

2156

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

14,425,857✔

2157

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

14,420,111✔

2158

  SSDataBlock*             result = NULL;

14,419,933✔

2160

  QRY_PARAM_CHECK(ppRes);

14,422,953✔

2161

  qTrace("%s call", __FUNCTION__);

14,425,217✔

2163

  while (!result) {

27,596,122✔

2164

    code = createVTableScanInfoFromBatchParam(pOperator);

17,154,814✔

2165

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

17,128,847✔

2166

    pOperator->pOperatorGetParam = NULL;

17,124,416✔

2167

    QUERY_CHECK_CODE(code, lino, _end);

17,134,376✔

2169

    if (pInfo->newReader) {

17,134,376✔

2170

      code = startNextGroupScan(pOperator, &result);

6,682,808✔

2171

      QUERY_CHECK_CODE(code, lino, _end);

6,723,156✔

2173

      code = doGroupedTableScan(pOperator, &result);

10,458,556✔

2174

      QUERY_CHECK_CODE(code, lino, _end);

10,439,738✔

2176

    if (!result) {

17,162,894✔

2177

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

6,723,782✔

2179

      pInfo->currentBatchIdx++;

6,723,782✔

2180

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

6,719,810✔

2182

        break;

3,999,625✔

2187

  if (result) {

14,440,933✔

2188

    SSDataBlock* res = NULL;

10,440,052✔

2189

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

10,440,366✔

2190

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

10,436,967✔

2191

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

10,439,929✔

2193

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

10,440,557✔

2194

    QUERY_CHECK_CODE(code, lino, _end);

10,440,871✔

2196

    if (pInfo->cachedTagList) {

10,440,871✔

2197

      code = setTagValFromTagList(pOperator, res);

2,982,668✔

2198

      QUERY_CHECK_CODE(code, lino, _end);

2,982,668✔

2201

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

10,440,871✔

2202

    pInfo->pResBlock = res;

10,440,871✔

2203

    blockDataDestroy(result);

10,440,871✔

2204

    (*ppRes) = res;

10,440,871✔

2206

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

10,440,871✔

2208

    clearVstbBatchDynamicTableScanInfo(pInfo);

4,000,881✔

2209

    setOperatorCompleted(pOperator);

3,999,939✔

2212

_end:

14,442,380✔

2213

  return code;

14,442,380✔

2216

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

38,624,826✔

2217

  int32_t                  code = TSDB_CODE_SUCCESS;

38,624,826✔

2218

  int32_t                  lino = 0;

38,624,826✔

2219

  STableScanInfo*          pInfo = pOperator->info;

38,624,826✔

2220

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

38,628,616✔

2221

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

38,623,246✔

2223

  QRY_PARAM_CHECK(ppRes);

38,626,969✔

2224

  qTrace("%s call", __FUNCTION__);

38,623,065✔

2226

  pOperator->dynamicTask = true;

38,623,065✔

2228

  switch (type) {

38,622,369✔

2229

    case DYN_TYPE_VSTB_WIN_SCAN:

938,672✔

2231

      code = doStbJoinDynamicTableScanNext(pOperator, ppRes);

938,672✔

2232

      QUERY_CHECK_CODE(code, lino, _end);

938,672✔

2233

      break;

938,672✔

2235

    case DYN_TYPE_VSTB_SINGLE_SCAN: {

23,250,649✔

2236

      code = doVstbSingleDynamicTableScanNext(pOperator, ppRes);

23,250,649✔

2237

      QUERY_CHECK_CODE(code, lino, _end);

23,250,649✔

2238

      break;

23,250,023✔

2240

    case DYN_TYPE_VSTB_BATCH_SCAN:

14,433,048✔

2241

      code = doVstbBatchDynamicTableScanNext(pOperator, ppRes);

14,433,048✔

2242

      QUERY_CHECK_CODE(code, lino, _end);

14,441,438✔

2243

      break;

14,441,438✔

2250

_end:

38,630,759✔

2251

  return code;

38,630,759✔

2254

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

701,903,053✔

2255

  int32_t         code = TSDB_CODE_SUCCESS;

701,903,053✔

2256

  int32_t         lino = 0;

701,903,053✔

2257

  STableScanInfo* pInfo = pOperator->info;

701,903,053✔

2258

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

701,996,543✔

2259

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

701,969,305✔

2260

  QRY_PARAM_CHECK(ppRes);

702,017,668✔

2261

  qTrace("%s call", __FUNCTION__);

702,017,630✔

2263

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

702,019,104✔

2264

  QUERY_CHECK_CODE(code, lino, _end);

702,044,887✔

2266

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

702,044,887✔

2267

  if (pGetParam || pInfo->pBatchColMap) {

702,058,066✔

2268

    ETableScanDynType type = pGetParam ?

38,620,604✔

2269

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

38,626,082✔

2272

    code = doDynamicTableScanNext(pOperator, type, ppRes);

38,620,604✔

2273

    QUERY_CHECK_CODE(code, lino, _end);

38,631,387✔

2275

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

38,630,761✔

2276

      return code;

37,717,084✔

2281

  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {

664,320,571✔

2282

    int32_t       numOfTables = 0;

5,962,002✔

2283

    STableKeyInfo tInfo = {0};

5,962,002✔

2284

    pInfo->countState = TABLE_COUNT_STATE_END;

5,962,002✔

2286

    while (1) {

5,567,543✔

2287

      SSDataBlock* result = NULL;

11,529,545✔

2288

      code = doGroupedTableScan(pOperator, &result);

11,527,994✔

2289

      QUERY_CHECK_CODE(code, lino, _end);

11,522,437✔

2291

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

11,522,437✔

2292

        (*ppRes) = result;

5,911,002✔

2293

        return code;

5,919,364✔

2297

      pInfo->currentTable++;

5,609,260✔

2299

      taosRLockLatch(&pTaskInfo->lock);

5,604,613✔

2300

      numOfTables = 0;

5,613,295✔

2301

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

5,613,295✔

2302

      if (code != TSDB_CODE_SUCCESS) {

5,608,646✔

2307

      if (pInfo->currentTable >= numOfTables) {

5,608,646✔

2308

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

42,016✔

2309

        taosRUnLockLatch(&pTaskInfo->lock);

42,016✔

2310

        (*ppRes) = NULL;

42,016✔

2311

        return code;

42,016✔

2314

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

5,555,773✔

2315

      if (!tmp) {

5,561,668✔

2321

      tInfo = *tmp;

5,561,668✔

2322

      taosRUnLockLatch(&pTaskInfo->lock);

5,560,426✔

2324

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

5,570,045✔

2325

      QUERY_CHECK_CODE(code, lino, _end);

5,565,374✔

2326

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

5,565,374✔

2329

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

5,571,884✔

2330

      QUERY_CHECK_CODE(code, lino, _end);

5,562,283✔

2331

      pInfo->scanTimes = 0;

5,562,283✔

2334

    code = groupSeqTableScan(pOperator, ppRes);

658,309,853✔

2335

    QUERY_CHECK_CODE(code, lino, _end);

658,085,931✔

2338

_end:

658,083,741✔

2339

  if (code != TSDB_CODE_SUCCESS) {

658,091,404✔

2340

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

2,816✔

2341

    pTaskInfo->code = code;

2,816✔

2342

    T_LONG_JMP(pTaskInfo->env, code);

2,816✔

2345

  return code;

658,088,588✔

2348

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

1,825,118✔

2349

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

1,825,118✔

2350

  if (!pRecorder) {

1,825,118✔

2353

  STableScanInfo* pTableScanInfo = pOptr->info;

1,825,118✔

2354

  *pRecorder = pTableScanInfo->base.readRecorder;

1,825,118✔

2355

  *pOptrExplain = pRecorder;

1,825,118✔

2356

  *len = sizeof(SFileBlockLoadRecorder);

1,825,118✔

2357

  return 0;

1,825,118✔

2360

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

186,684,340✔

2361

  cleanupQueryTableDataCond(&pBase->cond);

186,684,340✔

2362

  cleanupQueryTableDataCond(&pBase->orgCond);

186,665,214✔

2364

  if (pAPI->tsdReaderClose) {

186,683,860✔

2365

    pAPI->tsdReaderClose(pBase->dataReader);

186,688,840✔

2367

  pBase->dataReader = NULL;

186,620,113✔

2369

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

186,629,973✔

2370

    taosArrayDestroy(pBase->matchInfo.pList);

186,648,285✔

2373

  tableListDestroy(pBase->pTableListInfo);

186,573,223✔

2374

  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);

186,664,041✔

2375

  cleanupExprSupp(&pBase->pseudoSup);

186,654,731✔

2376

186,680,331✔

2378

static void cleanReaderForVTable(STableScanInfo* pInfo){

5,094,118✔

2379

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

5,094,118✔

2380

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

5,094,174✔

2381

    while (pIter != NULL) {

5,114,630✔

2382

      void **reader = pIter;

21,372✔

2383

      if (*reader) {

21,372✔

2384

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

21,372✔

2385

          pInfo->base.dataReader = NULL;

19,120✔

2387

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

21,372✔

2389

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

21,372✔

2392

5,093,516✔

2394

static void destroyTableScanOperatorInfo(void* param) {

161,838,462✔

2395

  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;

161,838,462✔

2396

  blockDataDestroy(pTableScanInfo->pResBlock);

161,838,462✔

2397

  blockDataDestroy(pTableScanInfo->pOrgBlock);

161,825,334✔

2398

  taosHashCleanup(pTableScanInfo->pIgnoreTables);

161,830,029✔

2399

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

161,820,606✔

2400

    cleanReaderForVTable(pTableScanInfo);

5,092,480✔

2401

    taosHashCleanup(pTableScanInfo->readerCache);

5,093,572✔

2403

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

161,821,618✔

2404

  taosArrayDestroy(pTableScanInfo->pBlockColMap);

161,825,916✔

2405

  taosMemoryFreeClear(param);

161,830,847✔

2406

161,784,601✔

2408

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

161,716,099✔

2409

  if (pBlock && dataRequireFlag == FUNC_DATA_REQUIRED_NOT_LOAD) {

161,716,099✔

2410

    int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

7,481,100✔

2411

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

19,048,889✔

2412

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

11,576,467✔

2413

      if (pCol) {

11,571,296✔

2414

        pCol->info.noData = true;

11,571,298✔

2418

161,707,421✔

2420

static int32_t resetTableScanOperatorState(SOperatorInfo* pOper) {

263,258✔

2421

  int32_t         code = TSDB_CODE_SUCCESS;

263,258✔

2422

  STableScanInfo*   pInfo = pOper->info;

263,258✔

2423

  pOper->status = OP_NOT_OPENED;

263,258✔

2425

  pInfo->scanTimes = 0;

263,258✔

2426

  pInfo->currentGroupId = -1;

263,258✔

2427

  pInfo->tableEndIndex = -1;

263,258✔

2428

  pInfo->tableStartIndex = 0;

263,258✔

2429

  pInfo->currentTable = 0;

263,258✔

2430

  pInfo->scanMode = 0;

263,258✔

2431

  pInfo->countState = 0;

263,258✔

2432

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

263,258✔

2434

  if (!pInfo->virtualStableScan) {

263,258✔

2440

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

124,646✔

2441

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

124,646✔

2443

    pInfo->base.dataReader = NULL;

124,482✔

2445

  tableListDestroy(pInfo->base.pTableListInfo);

263,094✔

2446

  pInfo->base.pTableListInfo = tableListCreate();

263,258✔

2447

  if (!pInfo->base.pTableListInfo) {

263,258✔

2451

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

263,258✔

2453

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

263,258✔

2454

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

263,258✔

2455

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

246,495✔

2457

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

246,495✔

2458

    if (code) {

123,654✔

2464

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

263,258✔

2465

  cleanupQueryTableDataCond(&pInfo->base.cond);

263,258✔

2466

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

263,258✔

2467

  if (code) {

263,258✔

2471

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

263,258✔

2472

    cleanupQueryTableDataCond(&pInfo->base.orgCond);

138,612✔

2473

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

138,612✔

2474

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

138,612✔

2477

  pOper->resultInfo.totalRows = 0;

263,258✔

2478

  blockDataEmpty(pInfo->pResBlock);

263,258✔

2479

  blockDataEmpty(pInfo->pOrgBlock);

263,258✔

2480

  taosHashClear(pInfo->pIgnoreTables);

263,258✔

2481

  taosArrayDestroy(pInfo->pBlockColMap);

263,258✔

2482

  pInfo->pBlockColMap = NULL;

263,258✔

2483

  return code;

263,258✔

2486

int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

161,835,623✔

2489

  QRY_PARAM_CHECK(pOptrInfo);

161,835,623✔

2491

  int32_t         code = TSDB_CODE_SUCCESS;

161,846,627✔

2492

  int32_t         lino = 0;

161,846,627✔

2493

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

161,846,627✔

2494

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

161,808,008✔

2495

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

161,796,415✔

2500

  pOperator->pPhyNode = pTableScanNode;

161,801,269✔

2501

  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;

161,800,787✔

2502

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

161,801,601✔

2504

  int32_t numOfCols = 0;

161,822,550✔

2506

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

161,826,826✔

2507

  QUERY_CHECK_CODE(code, lino, _error);

161,768,552✔

2509

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

161,768,552✔

2510

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

161,741,190✔

2511

  QUERY_CHECK_CODE(code, lino, _error);

161,666,322✔

2513

  if (pScanNode->pScanPseudoCols != NULL) {

161,666,322✔

2514

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

51,013,397✔

2515

    pSup->pExprInfo = NULL;

51,057,014✔

2516

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

51,066,582✔

2517

    QUERY_CHECK_CODE(code, lino, _error);

51,015,627✔

2519

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

51,015,627✔

2521

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

51,066,291✔

2524

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

161,685,136✔

2525

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

161,542,698✔

2527

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

161,699,653✔

2528

  pInfo->base.readHandle = *readHandle;

161,536,289✔

2529

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

161,707,464✔

2531

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

161,643,028✔

2532

  pInfo->sample.seed = taosGetTimestampSec();

161,749,405✔

2534

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

161,719,502✔

2535

  initResultSizeInfo(&pOperator->resultInfo, 4096);

161,730,569✔

2536

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

161,539,404✔

2537

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

161,717,723✔

2538

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

161,530,882✔

2540

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

161,541,435✔

2541

  QUERY_CHECK_CODE(code, lino, _error);

161,556,093✔

2543

  pInfo->virtualStableScan = pScanNode->virtualStableScan;

161,556,093✔

2544

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

161,585,823✔

2545

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

5,091,964✔

2546

    pInfo->pResBlock = NULL;

5,088,795✔

2547

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

5,093,321✔

2548

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

5,091,824✔

2551

  if (pInfo->virtualStableScan) {

161,600,981✔

2552

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

5,091,366✔

2553

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

5,096,193✔

2556

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

161,734,608✔

2557

                            pTaskInfo->pStreamRuntimeInfo);

161,716,764✔

2558

  QUERY_CHECK_CODE(code, lino, _error);

161,622,642✔

2560

  pInfo->currentGroupId = -1;

161,599,884✔

2562

  pInfo->tableEndIndex = -1;

161,685,838✔

2563

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

161,688,505✔

2564

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

161,571,915✔

2566

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

161,706,050✔

2568

  pOperator->exprSupp.numOfExprs = numOfCols;

161,581,865✔

2570

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

161,636,659✔

2571

  pInfo->ignoreTag = false;

161,718,389✔

2573

  pInfo->lastBatchIdx = -1;

161,627,290✔

2574

  pInfo->currentBatchIdx = 0;

161,549,851✔

2575

  pInfo->base.pTableListInfo = pTableListInfo;

161,731,404✔

2576

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

161,674,216✔

2586

  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

161,674,216✔

2588

  pOperator->fpSet = createOperatorFpSet(prepareTableScan, doTableScanNext,

161,650,002✔

2594

  setOperatorResetStateFn(pOperator, resetTableScanOperatorState);

161,543,580✔

2596

  pOperator->cost.openCost = 0;

161,565,019✔

2597

  *pOptrInfo = pOperator;

161,571,120✔

2598

  return TSDB_CODE_SUCCESS;

161,497,997✔

2600

_error:

22,758✔

2601

  if (pInfo != NULL) {

22,758✔

2602

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

22,758✔

2603

    destroyTableScanOperatorInfo(pInfo);

22,758✔

2606

  if (pOperator != NULL) {

22,758✔

2607

    pOperator->info = NULL;

22,758✔

2608

    destroyOperator(pOperator);

22,758✔

2610

  pTaskInfo->code = code;

22,758✔

2611

  return code;

22,758✔

2648

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

13,396✔

2649

  int32_t code = TSDB_CODE_SUCCESS;

13,396✔

2650

  int32_t lino = 0;

13,396✔

2651

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

13,396✔

2652

    return code;

1,624✔

2654

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

11,772✔

2655

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

11,772✔

2656

  bool hasUnqualified = false;

11,772✔

2658

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

11,772✔

2659

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

11,772✔

2661

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

11,772✔

2663

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

11,772✔

2666

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

11,772✔

2667

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

11,772✔

2668

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

42,279,242✔

2669

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

42,267,470✔

2670

    void*    data = colDataGetData(pColPk, i);

42,267,470✔

2671

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

42,269,742✔

2672

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

2,272✔

2675

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

2,272✔

2676

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

2,272✔

2677

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

2,272✔

2678

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

2,272✔

2679

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

2,272✔

2680

      taosMemoryFree(tmq);

2,272✔

2682

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

42,265,198✔

2685

    if (!p[i]) {

42,267,470✔

2686

      hasUnqualified = true;

15,069,752✔

2690

  if (hasUnqualified) {

11,772✔

2691

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

3,963✔

2692

    QUERY_CHECK_CODE(code, lino, _end);

3,963✔

2695

_end:

11,772✔

2696

  taosMemoryFree(p);

11,772✔

2697

  if (code != TSDB_CODE_SUCCESS) {

11,772✔

2700

  return code;

11,772✔

2703

int32_t colIdComparFn(const void* param1, const void* param2) {

1,389,898,704✔

2704

  int32_t p1 = *(int32_t*)param1;

1,389,898,704✔

2705

  int32_t p2 = *(int32_t*)param2;

1,389,943,494✔

2707

  if (p1 == p2) {

1,389,924,111✔

2710

    return (p1 < p2) ? -1 : 1;

1,389,924,111✔

2714

static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock) {

96,774,595✔

2715

  int32_t         code = TSDB_CODE_SUCCESS;

96,774,595✔

2716

  int32_t         lino = 0;

96,774,595✔

2717

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

96,774,595✔

2718

  SOperatorInfo*  pOperator = pInfo->pStreamScanOp;

96,780,051✔

2719

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

96,780,392✔

2720

  const char*     id = GET_TASKID(pTaskInfo);

96,778,377✔

2721

  SArray* pColList = NULL;

96,781,415✔

2723

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

96,781,415✔

2724

  QUERY_CHECK_CODE(code, lino, _end);

96,780,051✔

2726

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

96,780,051✔

2727

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

96,776,641✔

2728

  pBlockInfo->type = STREAM_NORMAL;

96,779,369✔

2729

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

96,780,392✔

2731

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

96,780,392✔

2732

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

96,781,756✔

2734

  pColList = taosArrayInit(4, sizeof(int32_t));

96,781,074✔

2735

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

96,778,693✔

2738

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

521,993,730✔

2739

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

425,195,722✔

2740

    if (!pColMatchInfo->needOutput) {

425,176,450✔

2741

      continue;

636✔

2744

    bool colExists = false;

425,192,110✔

2745

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

1,343,703,764✔

2746

      SColumnInfoData* pResCol = NULL;

1,343,113,986✔

2747

      code = bdGetColumnInfoData(pBlock, j, &pResCol);

1,343,046,574✔

2748

      QUERY_CHECK_CODE(code, lino, _end);

1,343,049,027✔

2750

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

1,343,049,027✔

2751

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

425,220,590✔

2752

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

425,143,493✔

2753

        QUERY_CHECK_CODE(code, lino, _end);

425,109,451✔

2755

        colExists = true;

425,109,451✔

2756

        void* tmp = taosArrayPush(pColList, &pColMatchInfo->dstSlotId);

425,109,451✔

2757

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

425,223,230✔

2758

        break;

425,223,230✔

2763

    if (!colExists) {

425,214,401✔

2772

  if (pInfo->numOfPseudoExpr > 0) {

96,780,051✔

2773

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

122,977,519✔

2774

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

61,489,015✔

2776

    if (code) {

61,487,651✔

2781

    terrno = 0;

61,487,651✔

2783

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

256,700,820✔

2784

      void* tmp = taosArrayPush(pColList, &pInfo->pPseudoExpr[i].base.resSchema.slotId);

195,193,050✔

2785

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

195,214,874✔

2789

  taosArraySort(pColList, colIdComparFn);

96,782,446✔

2791

  int32_t i = 0, j = 0;

96,777,762✔

2792

  while (i < taosArrayGetSize(pColList)) {

717,070,417✔

2793

    int32_t slot1 = *(int32_t*)taosArrayGet(pColList, i);

620,277,818✔

2794

    if (slot1 > j) {

620,292,655✔

2799

      i += 1;

620,292,655✔

2800

      j += 1;

620,292,655✔

2804

  while (j < taosArrayGetSize(pInfo->pRes->pDataBlock)) {

96,774,972✔

2805

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

636✔

2806

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

636✔

2807

    j += 1;

636✔

2810

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

96,776,331✔

2811

  QUERY_CHECK_CODE(code, lino, _end);

96,775,773✔

2813

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

96,775,773✔

2814

  QUERY_CHECK_CODE(code, lino, _end);

96,774,918✔

2816

_end:

96,774,918✔

2817

  taosArrayDestroy(pColList);

96,266,696✔

2818

  if (code != TSDB_CODE_SUCCESS) {

96,776,015✔

2821

  return code;

96,770,088✔

2824

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

6,635,661✔

2825

  int32_t code = TSDB_CODE_SUCCESS;

6,635,661✔

2826

  int32_t lino = 0;

6,635,661✔

2827

  SValue  val = {0};

6,635,661✔

2828

  if (hasPrimaryKey) {

6,636,247✔

2829

    code = doBlockDataPrimaryKeyFilter(pBlock, offset);

13,396✔

2830

    if (code != TSDB_CODE_SUCCESS) {

13,396✔

2834

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

13,396✔

2836

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

13,396✔

2837

      return code;

3,679✔

2839

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

9,717✔

2840

    val.type = pColPk->info.type;

9,717✔

2841

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

9,717✔

2842

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

619✔

2845

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

619✔

2846

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

619✔

2847

      val.nData = varDataLen(tmp);

619✔

2848

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

619✔

2850

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

9,098✔

2853

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

6,632,568✔

2855

_end:

6,630,477✔

2856

  if (code != TSDB_CODE_SUCCESS) {

6,631,375✔

2859

  return code;

6,631,046✔

2862

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

123,382,863✔

2863

  int32_t        code = TSDB_CODE_SUCCESS;

123,382,863✔

2864

  int32_t        lino = 0;

123,382,863✔

2865

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

123,382,863✔

2866

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

123,387,362✔

2868

  SStreamScanInfo* pInfo = pOperator->info;

123,388,361✔

2869

  const char*      id = GET_TASKID(pTaskInfo);

123,386,000✔

2871

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

123,386,117✔

2873

  if (isTaskKilled(pTaskInfo)) {

123,387,990✔

2878

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

123,390,030✔

2879

    while (1) {

3,679✔

2880

      SSDataBlock* pResult = NULL;

5,962,002✔

2881

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

5,962,002✔

2882

      QUERY_CHECK_CODE(code, lino, _end);

5,961,068✔

2884

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

5,961,068✔

2885

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

5,919,364✔

2886

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

5,919,986✔

2887

        QUERY_CHECK_CODE(code, lino, _end);

5,916,874✔

2888

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

5,916,874✔

2889

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

5,919,676✔

2890

          (*ppRes) = pResult;

5,916,307✔

2891

          return code;

5,916,307✔

2898

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

42,016✔

2899

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

42,016✔

2901

    pTSInfo->base.dataReader = NULL;

42,016✔

2902

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

42,016✔

2903

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

42,016✔

2904

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

42,016✔

2905

      (*ppRes) = NULL;

11,707✔

2906

      return code;

11,707✔

2909

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

30,309✔

2912

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

117,461,675✔

2913

    while (1) {

750,260✔

2915

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

118,211,935✔

2917

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

118,189,597✔

2918

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

118,200,452✔

2921

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

118,204,897✔

2924

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

118,201,796✔

2926

      if (hasResult) {

118,203,844✔

2927

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

96,773,324✔

2929

        blockDataCleanup(pInfo->pRes);

96,782,125✔

2930

        code = setBlockIntoRes(pInfo, pRes);

96,776,982✔

2931

        QUERY_CHECK_CODE(code, lino, _end);

96,769,455✔

2932

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

96,769,455✔

2934

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

96,752,615✔

2935

          (*ppRes) = pInfo->pRes;

96,029,450✔

2936

          return code;

96,028,086✔

2939

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

21,430,520✔

2940

        (*ppRes) = NULL;

21,430,520✔

2941

        return code;

21,430,520✔

2966

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

324,159✔

2967

  int32_t code = TSDB_CODE_SUCCESS;

324,159✔

2968

  int32_t lino = 0;

324,159✔

2969

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

324,159✔

2970

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

325,552✔

2973

  int32_t size = 0;

325,552✔

2974

  code = tableListGetSize(pTableListInfo, &size);

325,898✔

2975

  QUERY_CHECK_CODE(code, lino, _end);

325,898✔

2976

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

9,508,238✔

2977

    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);

9,182,340✔

2978

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

9,180,480✔

2979

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

9,180,480✔

2980

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

9,182,340✔

2983

  (*ppArrayRes) = tableIdList;

325,898✔

2985

_end:

325,576✔

2986

  if (code != TSDB_CODE_SUCCESS) {

325,576✔

2989

  return code;

325,576✔

2992

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

1,486,889✔

2993

  int32_t        code = TSDB_CODE_SUCCESS;

1,486,889✔

2994

  int32_t        lino = 0;

1,486,889✔

2995

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,486,889✔

2996

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,486,889✔

2998

  SStreamRawScanInfo* pInfo = pOperator->info;

1,486,889✔

2999

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

1,486,889✔

3000

  pTaskInfo->streamInfo.btMetaRsp.batchMetaLen = NULL;

1,486,889✔

3002

  qDebug("tmqsnap doRawScan called");

1,486,889✔

3003

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

1,486,596✔

3004

    bool hasNext = false;

1,481,801✔

3005

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

1,481,801✔

3006

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

1,481,251✔

3007

      if (code != TSDB_CODE_SUCCESS) {

1,479,493✔

3013

    if (pInfo->dataReader && hasNext) {

1,480,336✔

3014

      if (isTaskKilled(pTaskInfo)) {

717,834✔

3019

      SSDataBlock* pBlock = NULL;

717,541✔

3020

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

716,955✔

3021

      QUERY_CHECK_CODE(code, lino, _end);

716,955✔

3023

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

716,955✔

3024

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

717,541✔

3025

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

718,127✔

3026

        QUERY_CHECK_CODE(code, lino, _end);

717,248✔

3027

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

717,248✔

3028

        (*ppRes) = pBlock;

717,834✔

3029

        return code;

718,127✔

3033

    SMetaTableInfo mtInfo = {0};

762,795✔

3034

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

763,967✔

3035

    QUERY_CHECK_CODE(code, lino, _end);

763,381✔

3036

    if (code != 0) {

763,381✔

3040

    STqOffsetVal offset = {0};

763,381✔

3041

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

762,782✔

3042

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

11,127✔

3043

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

11,127✔

3045

      SValue val = {0};

752,534✔

3046

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

752,534✔

3047

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

752,534✔

3050

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

763,967✔

3051

    QUERY_CHECK_CODE(code, lino, _end);

763,661✔

3052

    (*ppRes) = NULL;

763,661✔

3053

    return code;

763,661✔

3054

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

4,795✔

3055

    SSnapContext* sContext = pInfo->sContext;

4,795✔

3056

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

39,828✔

3057

      void*   data = NULL;

40,094✔

3058

      int32_t dataLen = 0;

40,094✔

3059

      int16_t type = 0;

40,094✔

3060

      int64_t uid = 0;

40,094✔

3061

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

40,094✔

3067

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

40,094✔

3068

        STqOffsetVal offset = {0};

4,795✔

3069

        SValue       val = {0};

4,795✔

3071

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

4,795✔

3072

        QUERY_CHECK_CODE(code, lino, _end);

4,795✔

3073

        break;

4,795✔

3075

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

35,299✔

3076

        SMqMetaRsp tmpMetaRsp = {0};

35,299✔

3077

        tmpMetaRsp.resMsgType = type;

35,299✔

3078

        tmpMetaRsp.metaRspLen = dataLen;

35,299✔

3079

        tmpMetaRsp.metaRsp = data;

35,299✔

3080

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

35,299✔

3081

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

4,263✔

3082

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

4,263✔

3084

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

4,263✔

3085

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

4,263✔

3087

        int32_t  tempRes = TSDB_CODE_SUCCESS;

35,299✔

3088

        uint32_t len = 0;

35,299✔

3089

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

35,299✔

3090

        if (TSDB_CODE_SUCCESS != tempRes) {

35,299✔

3096

        int32_t tLen = sizeof(SMqRspHead) + len;

35,299✔

3097

        void*   tBuf = taosMemoryCalloc(1, tLen);

35,299✔

3098

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

35,033✔

3100

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

35,033✔

3101

        SEncoder encoder = {0};

35,033✔

3102

        tEncoderInit(&encoder, metaBuff, len);

35,033✔

3103

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

35,033✔

3104

        if (tempLen < 0) {

35,033✔

3111

        taosMemoryFreeClear(data);

35,033✔

3112

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

34,767✔

3113

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

35,299✔

3115

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

35,299✔

3116

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

35,299✔

3120

    (*ppRes) = NULL;

4,529✔

3121

    return code;

4,795✔

3135

static void destroyTmqRawScanOperatorInfo(void* param) {

115,610✔

3136

  SStreamRawScanInfo* pRawScan = (SStreamRawScanInfo*)param;

115,610✔

3137

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

115,610✔

3138

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

115,610✔

3139

  tableListDestroy(pRawScan->pTableListInfo);

115,610✔

3140

  taosMemoryFree(pRawScan);

115,344✔

3141

115,344✔

3146

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

115,327✔

3152

  QRY_PARAM_CHECK(pOptrInfo);

115,327✔

3153

  int32_t code = TSDB_CODE_SUCCESS;

115,610✔

3154

  int32_t lino = 0;

115,610✔

3156

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

115,610✔

3157

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

115,304✔

3158

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

115,304✔

3164

  pInfo->pTableListInfo = tableListCreate();

115,304✔

3165

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

115,610✔

3166

  pInfo->vnode = pHandle->vnode;

115,610✔

3167

  pInfo->pAPI = &pTaskInfo->storageAPI;

115,610✔

3169

  pInfo->sContext = pHandle->sContext;

115,610✔

3170

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

115,610✔

3171

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

115,610✔

3174

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

115,610✔

3176

  *pOptrInfo = pOperator;

115,610✔

3177

  return code;

115,610✔

3189

void destroyTmqScanOperatorInfo(void* param) {

325,576✔

3190

  if (param == NULL) {

325,576✔

3194

  SStreamScanInfo* pStreamScan = (SStreamScanInfo*)param;

325,576✔

3195

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

325,576✔

3196

    destroyOperator(pStreamScan->pTableScanOp);

325,898✔

3199

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

325,576✔

3200

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

325,576✔

3202

  if (pStreamScan->pVtableMergeHandles) {

325,898✔

3206

  if (pStreamScan->pVtableMergeBuf) {

325,898✔

3210

  if (pStreamScan->pVtableReadyHandles) {

325,898✔

3214

  if (pStreamScan->pTableListInfo) {

325,898✔

3218

  if (pStreamScan->matchInfo.pList) {

325,898✔

3219

    taosArrayDestroy(pStreamScan->matchInfo.pList);

325,898✔

3221

  if (pStreamScan->pPseudoExpr) {

325,898✔

3222

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

189,314✔

3223

    taosMemoryFree(pStreamScan->pPseudoExpr);

189,314✔

3226

  cleanupExprSupp(&pStreamScan->tbnameCalSup);

325,898✔

3227

  cleanupExprSupp(&pStreamScan->tagCalSup);

325,898✔

3229

  blockDataDestroy(pStreamScan->pRes);

325,898✔

3230

  blockDataDestroy(pStreamScan->pUpdateRes);

325,898✔

3231

  blockDataDestroy(pStreamScan->pDeleteDataRes);

325,898✔

3232

  blockDataDestroy(pStreamScan->pUpdateDataRes);

325,646✔

3233

  blockDataDestroy(pStreamScan->pCreateTbRes);

325,576✔

3234

  taosArrayDestroy(pStreamScan->pBlockLists);

325,282✔

3235

  blockDataDestroy(pStreamScan->pCheckpointRes);

325,576✔

3237

  taosMemoryFree(pStreamScan);

325,898✔

3257

 int32_t createTmqScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,

325,898✔

3260

  QRY_PARAM_CHECK(pOptrInfo);

325,898✔

3262

  int32_t          code = TSDB_CODE_SUCCESS;

325,898✔

3263

  int32_t          lino = 0;

325,898✔

3264

  SArray*          pColIds = NULL;

325,898✔

3265

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

325,898✔

3266

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

325,898✔

3267

  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;

325,898✔

3268

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

325,898✔

3270

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

325,898✔

3275

  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;

325,898✔

3276

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;

325,898✔

3278

  pInfo->pTagCond = pTagCond;

325,898✔

3279

  pInfo->pGroupTags = pTableScanNode->pGroupTags;

325,898✔

3281

  int32_t numOfCols = 0;

325,898✔

3282

  code = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);

325,898✔

3283

  if (code != TSDB_CODE_SUCCESS) {

325,898✔

3287

  SDataType pkType = {0};

325,898✔

3288

  pInfo->primaryKeyIndex = -1;

325,898✔

3289

  pInfo->basic.primaryPkIndex = -1;

325,898✔

3290

  int32_t numOfOutput = taosArrayGetSize(pInfo->matchInfo.pList);

325,898✔

3291

  pColIds = taosArrayInit(numOfOutput, sizeof(int16_t));

325,898✔

3292

  QUERY_CHECK_NULL(pColIds, code, lino, _error, terrno);

325,898✔

3294

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

2,045,865✔

3295

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

1,719,701✔

3296

    QUERY_CHECK_NULL(id, code, lino, _error, terrno);

1,719,967✔

3298

    int16_t colId = id->colId;

1,719,967✔

3299

    void*   tmp = taosArrayPush(pColIds, &colId);

1,719,967✔

3300

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

1,719,967✔

3302

    if (id->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

1,719,967✔

3303

      pInfo->primaryTsIndex = id->dstSlotId;

325,632✔

3305

    if (id->isPk) {

1,719,701✔

3306

      pInfo->primaryKeyIndex = id->dstSlotId;

1,469✔

3307

      pInfo->basic.primaryPkIndex = id->dstSlotId;

1,469✔

3308

      pkType = id->dataType;

1,469✔

3312

  pInfo->pPartTbnameSup = NULL;

326,164✔

3313

  if (pTableScanNode->pSubtable != NULL) {

325,898✔

3329

  if (pTableScanNode->pTags != NULL) {

325,898✔

3341

  pInfo->pBlockLists = taosArrayInit(4, sizeof(SPackedData));

325,898✔

3342

  TSDB_CHECK_NULL(pInfo->pBlockLists, code, lino, _error, terrno);

325,898✔

3344

  if (pHandle->vnode) {

325,898✔

3345

    SOperatorInfo* pTableScanOp = NULL;

325,898✔

3346

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

325,898✔

3347

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

325,286✔

3352

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

325,632✔

3353

    if (pHandle->version > 0) {

325,632✔

3354

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

325,632✔

3357

    STableKeyInfo* pList = NULL;

326,164✔

3358

    int32_t        num = 0;

325,898✔

3359

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

325,898✔

3360

    QUERY_CHECK_CODE(code, lino, _error);

325,898✔

3362

    if (pHandle->initTableReader) {

325,898✔

3363

      pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;

325,632✔

3364

      pTSInfo->base.dataReader = NULL;

325,632✔

3367

    if (pHandle->initTqReader) {

325,632✔

3368

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

325,286✔

3369

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

325,898✔

3371

      pInfo->tqReader = pHandle->tqReader;

346✔

3375

    pInfo->pUpdateInfo = NULL;

325,898✔

3376

    pInfo->pTableScanOp = pTableScanOp;

325,898✔

3377

    if (pInfo->pTableScanOp->pTaskInfo->streamInfo.pState) {

325,898✔

3382

    pInfo->readHandle = *pHandle;

325,898✔

3383

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

325,898✔

3384

    pInfo->pCreateTbRes = buildCreateTableBlock(&pInfo->tbnameCalSup, &pInfo->tagCalSup);

325,898✔

3385

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

325,576✔

3386

    pInfo->hasPart = false;

325,576✔

3388

    code = blockDataEnsureCapacity(pInfo->pCreateTbRes, 8);

325,898✔

3389

    QUERY_CHECK_CODE(code, lino, _error);

325,898✔

3392

    code = pAPI->tqReaderFn.tqReaderSetColIdList(pInfo->tqReader, pColIds, idstr);

325,898✔

3393

    QUERY_CHECK_CODE(code, lino, _error);

325,576✔

3395

    SArray* tableIdList = NULL;

325,576✔

3396

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

325,576✔

3397

    QUERY_CHECK_CODE(code, lino, _error);

325,898✔

3398

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

325,898✔

3399

    QUERY_CHECK_CODE(code, lino, _error);

325,898✔

3400

    taosArrayDestroy(tableIdList);

325,898✔

3401

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

325,898✔

3408

  pColIds = NULL;

325,898✔

3411

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

325,898✔

3412

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pInfo->pPseudoExpr, &pInfo->numOfPseudoExpr);

189,314✔

3413

    QUERY_CHECK_CODE(code, lino, _error);

189,314✔

3416

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

325,898✔

3417

  QUERY_CHECK_CODE(code, lino, _error);

325,898✔

3419

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

325,898✔

3420

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

325,898✔

3421

  code = createSpecialDataBlock(STREAM_CLEAR, &pInfo->pUpdateRes);

325,898✔

3422

  QUERY_CHECK_CODE(code, lino, _error);

325,898✔

3424

  pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;

325,898✔

3425

  pInfo->windowSup = (SWindowSupporter){.pStreamAggSup = NULL, .gap = -1, .parentType = QUERY_NODE_PHYSICAL_PLAN};

325,898✔

3426

  pInfo->groupId = 0;

325,576✔

3427

  pInfo->igCheckGroupId = false;

325,576✔

3428

  pInfo->pStreamScanOp = pOperator;

325,898✔

3429

  pInfo->deleteDataIndex = 0;

325,576✔

3430

  code = createSpecialDataBlock(STREAM_DELETE_DATA, &pInfo->pDeleteDataRes);

325,898✔

3431

  QUERY_CHECK_CODE(code, lino, _error);

325,898✔

3433

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

325,898✔

3434

  code = createSpecialDataBlock(STREAM_CLEAR, &pInfo->pUpdateDataRes);

325,898✔

3435

  QUERY_CHECK_CODE(code, lino, _error);

325,898✔

3437

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

325,898✔

3438

  pInfo->partitionSup.needCalc = false;

325,898✔

3439

  pInfo->igCheckUpdate = pTableScanNode->igCheckUpdate;

325,576✔

3440

  pInfo->igExpired = pTableScanNode->igExpired;

325,898✔

3441

  pInfo->twAggSup.maxTs = INT64_MIN;

325,898✔

3442

  pInfo->pState = pTaskInfo->streamInfo.pState;

325,898✔

3443

  pInfo->stateStore = pTaskInfo->storageAPI.stateStore;

325,576✔

3444

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

325,898✔

3445

  pInfo->pFillSup = NULL;

325,576✔

3446

  pInfo->useGetResultRange = false;

325,898✔

3447

  pInfo->pRangeScanRes = NULL;

325,898✔

3449

  code = createSpecialDataBlock(STREAM_CHECKPOINT, &pInfo->pCheckpointRes);

325,898✔

3450

  QUERY_CHECK_CODE(code, lino, _error);

325,898✔

3452

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

325,898✔

3454

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

325,898✔

3456

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

325,898✔

3459

  *pOptrInfo = pOperator;

325,898✔

3460

  return code;

325,898✔

3483

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

9,144,053✔

3484

  int32_t        code = TSDB_CODE_SUCCESS;

9,144,053✔

3485

  int32_t        lino = 0;

9,144,053✔

3486

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,144,053✔

3487

  STagScanInfo*  pInfo = pOperator->info;

9,148,160✔

3488

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

9,148,093✔

3489

  int32_t        count = pRes->info.rows;

9,147,655✔

3491

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

9,145,759✔

3492

  if (!item) {

9,143,616✔

3499

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

9,143,616✔

3500

  tDecoderClear(&(*mr).coder);

9,149,170✔

3501

  if (code != TSDB_CODE_SUCCESS) {

9,149,675✔

3507

  char str[512];

9,149,571✔

3508

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

41,331,839✔

3509

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

32,183,872✔

3512

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

32,186,828✔

3513

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

3,725,532✔

3514

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

3,723,705✔

3515

      QUERY_CHECK_CODE(code, lino, _end);

3,724,717✔

3516

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

28,464,763✔

3517

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

5,224,839✔

3518

      QUERY_CHECK_CODE(code, lino, _end);

5,224,334✔

3519

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

23,238,341✔

3520

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

15,055✔

3521

      QUERY_CHECK_CODE(code, lino, _end);

15,055✔

3523

      STagVal val = {0};

23,224,801✔

3524

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

23,224,105✔

3525

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

23,227,954✔

3527

      char* data = NULL;

23,224,487✔

3528

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

23,224,487✔

3529

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

23,224,637✔

3531

        data = (char*)p;

1,365✔

3534

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

23,223,791✔

3535

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

23,223,791✔

3536

      QUERY_CHECK_CODE(code, lino, _end);

23,226,125✔

3538

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

23,226,125✔

3540

        taosMemoryFree(data);

8,215,097✔

3545

_end:

9,152,064✔

3546

  if (code != TSDB_CODE_SUCCESS) {

9,149,170✔

3550

    pRes->info.rows++;

9,149,170✔

3553

  return code;

9,148,979✔

3556

static void tagScanFreeUidTag(void* p) {

7,052,241✔

3557

  STUidTagInfo* pInfo = p;

7,052,241✔

3558

  if (pInfo->pTagVal != NULL) {

7,052,241✔

3559

    taosMemoryFree(pInfo->pTagVal);

7,051,799✔

3561

7,046,351✔

3563

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

1,617,487✔

3564

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

1,617,487✔

3565

  if (pColumnData == NULL) {

1,617,360✔

3569

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

1,617,360✔

3570

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

1,616,868✔

3571

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

1,617,845✔

3572

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

1,614,590✔

3574

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

1,617,271✔

3575

  if (code != TSDB_CODE_SUCCESS) {

1,614,245✔

3581

  pParam->columnData = pColumnData;

1,614,245✔

3582

  pParam->colAlloced = true;

1,616,505✔

3583

  return TSDB_CODE_SUCCESS;

1,616,510✔

3586

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

9,896,279✔

3587

  int32_t                code = TSDB_CODE_SUCCESS;

9,896,279✔

3588

  int32_t                lino = 0;

9,896,279✔

3589

  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;

9,896,279✔

3590

  SColumnNode*           pSColumnNode = NULL;

9,896,279✔

3591

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

9,895,801✔

3592

    pSColumnNode = *(SColumnNode**)pNode;

2,772,467✔

3593

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

7,131,107✔

3594

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

496✔

3595

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

496✔

3608

      return DEAL_RES_CONTINUE;

496✔

3611

    return DEAL_RES_CONTINUE;

7,126,702✔

3614

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

2,772,939✔

3615

  if (!data) {

2,769,682✔

3616

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

1,618,518✔

3617

    if (code == TSDB_CODE_DUP_KEY) {

1,621,971✔

3620

    QUERY_CHECK_CODE(code, lino, _end);

1,621,971✔

3621

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

1,621,971✔

3622

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

1,619,048✔

3623

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

1,620,665✔

3624

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

1,619,070✔

3625

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

1,621,628✔

3626

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

1,622,461✔

3628

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

1,151,164✔

3629

    pSColumnNode->slotId = col->slotId;

1,151,164✔

3632

_end:

2,771,591✔

3633

  if (code != TSDB_CODE_SUCCESS) {

2,771,099✔

3634

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

1,783✔

3637

  return DEAL_RES_CONTINUE;

2,769,316✔

3640

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

1,619,024✔

3642

  int32_t code = TSDB_CODE_SUCCESS;

1,619,024✔

3643

  int32_t lino = 0;

1,619,024✔

3644

  int32_t numOfTables = taosArrayGetSize(aUidTags);

1,619,024✔

3645

  SArray* pBlockList = NULL;

1,619,024✔

3647

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

1,619,024✔

3648

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

1,616,999✔

3650

  pBlockList = taosArrayInit(1, POINTER_BYTES);

1,616,999✔

3651

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

1,617,166✔

3653

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

1,619,024✔

3654

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

1,619,024✔

3656

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

1,619,024✔

3658

  SScalarParam output = {0};

1,618,322✔

3659

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

1,618,322✔

3660

  QUERY_CHECK_CODE(code, lino, _end);

1,616,518✔

3662

  gTaskScalarExtra.pStreamInfo = (void*)pStreamRuntimeInfo;

1,616,518✔

3663

  gTaskScalarExtra.pStreamRange = NULL;

1,616,518✔

3664

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

1,616,518✔

3665

  QUERY_CHECK_CODE(code, lino, _end);

1,614,898✔

3667

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

1,614,898✔

3668

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

4,038,394✔

3669

    if (result[i]) {

2,423,908✔

3670

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

2,283,845✔

3671

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

2,283,845✔

3675

  colDataDestroy(output.columnData);

1,616,514✔

3676

  taosMemoryFreeClear(output.columnData);

1,613,248✔

3678

_end:

1,614,874✔

3679

  blockDataDestroy(pResBlock);

1,614,523✔

3680

  taosArrayDestroy(pBlockList);

1,611,755✔

3681

  if (code != TSDB_CODE_SUCCESS) {

1,611,398✔

3684

  return code;

1,611,398✔

3687

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

28,860,708✔

3690

  int32_t code = TSDB_CODE_SUCCESS;

28,860,708✔

3691

  int32_t lino = 0;

28,860,708✔

3692

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

28,860,708✔

3693

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

13,723,122✔

3699

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

13,716,799✔

3700

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

6,857,660✔

3701

      QUERY_CHECK_CODE(code, lino, _end);

6,859,591✔

3702

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

6,861,330✔

3703

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

6,858,286✔

3704

      QUERY_CHECK_CODE(code, lino, _end);

6,859,678✔

3707

    STagVal tagVal = {0};

15,137,055✔

3708

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

15,138,755✔

3709

    if (pUidTagInfo->pTagVal == NULL) {

15,136,881✔

3712

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

15,134,695✔

3714

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

15,141,416✔

3715

        colDataSetNULL(pColInfo, rowIndex);

7,523✔

3716

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

15,134,362✔

3719

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

18,966,959✔

3720

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

3,838,850✔

3723

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

3,836,489✔

3724

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

3,836,615✔

3726

        varDataSetLen(tmp, tagVal.nData);

3,836,615✔

3727

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

3,836,305✔

3728

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

3,839,266✔

3729

        taosMemoryFree(tmp);

3,836,801✔

3730

        QUERY_CHECK_CODE(code, lino, _end);

3,836,167✔

3732

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

11,292,166✔

3733

        QUERY_CHECK_CODE(code, lino, _end);

11,291,998✔

3738

_end:

28,859,549✔

3739

  if (code != TSDB_CODE_SUCCESS) {

28,859,549✔

3742

  return code;

28,852,861✔

3745

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

4,508,387✔

3747

  int32_t       code = TSDB_CODE_SUCCESS;

4,508,387✔

3748

  int32_t       lino = 0;

4,508,387✔

3749

  STagScanInfo* pInfo = pOperator->info;

4,508,387✔

3750

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

4,509,102✔

3751

  if (!ignoreFilterIdx) {

4,507,194✔

3752

    size_t szTables = taosArrayGetSize(aFilterIdxs);

1,613,683✔

3753

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

3,898,978✔

3754

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

2,280,178✔

3755

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);

2,281,979✔

3756

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

2,282,691✔

3757

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

12,985,785✔

3758

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

10,696,637✔

3759

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

10,704,033✔

3760

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

10,704,033✔

3761

        QUERY_CHECK_CODE(code, lino, _end);

10,703,094✔

3765

    size_t szTables = taosArrayGetSize(aUidTags);

2,893,511✔

3766

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

7,522,301✔

3767

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);

4,629,126✔

3768

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

4,629,126✔

3769

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

22,773,773✔

3770

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

18,148,771✔

3771

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

18,149,085✔

3772

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

18,149,085✔

3773

        QUERY_CHECK_CODE(code, lino, _end);

18,144,275✔

3778

_end:

2,893,175✔

3779

  if (code != TSDB_CODE_SUCCESS) {

4,511,975✔

3782

  return code;

4,510,793✔

3785

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

8,986,459✔

3786

  int32_t        code = TSDB_CODE_SUCCESS;

8,986,459✔

3787

  int32_t        lino = 0;

8,986,459✔

3788

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

8,986,459✔

3789

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

8,985,903✔

3791

  STagScanInfo* pInfo = pOperator->info;

8,982,604✔

3792

  SSDataBlock*  pRes = pInfo->pRes;

8,988,238✔

3794

  QRY_PARAM_CHECK(ppRes);

8,984,275✔

3796

  if (pOperator->status == OP_EXEC_DONE) {

8,984,275✔

3797

    return TSDB_CODE_SUCCESS;

4,406,523✔

3799

  blockDataCleanup(pRes);

4,578,955✔

3801

  if (pInfo->pCtbCursor == NULL) {

4,581,788✔

3802

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

4,582,214✔

3803

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

4,583,841✔

3809

  SArray* aUidTags = pInfo->aUidTags;

4,582,901✔

3810

  SArray* aFilterIdxs = pInfo->aFilterIdxs;

4,584,213✔

3811

  int32_t count = 0;

4,582,901✔

3812

  bool    ctbCursorFinished = false;

4,582,901✔

3813

  while (1) {

92,015✔

3814

    taosArrayClearEx(aUidTags, tagScanFreeUidTag);

4,674,916✔

3815

    taosArrayClear(aFilterIdxs);

4,677,728✔

3817

    int32_t numTables = 0;

4,678,100✔

3818

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

11,725,194✔

3819

      SMCtbCursor* pCur = pInfo->pCtbCursor;

11,731,578✔

3820

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

11,731,185✔

3821

      if (uid == 0) {

11,732,222✔

3822

        ctbCursorFinished = true;

4,677,634✔

3823

        break;

4,677,634✔

3825

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

7,054,588✔

3826

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

7,053,562✔

3827

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

7,048,008✔

3829

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

7,048,008✔

3830

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

7,054,849✔

3831

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

7,054,849✔

3832

      ++numTables;

7,054,849✔

3835

    if (numTables == 0) {

4,678,100✔

3836

      break;

165,193✔

3838

    bool ignoreFilterIdx = true;

4,512,907✔

3839

    if (pInfo->pTagCond != NULL) {

4,512,907✔

3840

      ignoreFilterIdx = false;

1,619,024✔

3841

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

1,619,024✔

3842

      QUERY_CHECK_CODE(code, lino, _end);

1,613,263✔

3844

      ignoreFilterIdx = true;

2,893,511✔

3847

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

4,506,774✔

3848

    QUERY_CHECK_CODE(code, lino, _end);

4,512,076✔

3850

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

4,512,076✔

3852

    if (count != 0) {

4,510,323✔

3853

      break;

4,418,308✔

3857

  if (count > 0) {

4,583,501✔

3858

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

4,418,659✔

3860

  if (count == 0 || ctbCursorFinished) {

4,583,751✔

3861

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

4,584,217✔

3862

    pInfo->pCtbCursor = NULL;

4,577,476✔

3863

    setOperatorCompleted(pOperator);

4,578,455✔

3866

  pRes->info.rows = count;

4,576,159✔

3867

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

4,575,692✔

3868

  if (bLimitReached) {

4,579,926✔

3872

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

4,579,926✔

3873

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

4,578,936✔

3875

_end:

4,580,768✔

3876

  if (code != TSDB_CODE_SUCCESS) {

4,580,768✔

3882

  return code;

4,580,768✔

3885

static int32_t createTagScanTableListInfoFromParam(SOperatorInfo* pOperator) {

3,409,353✔

3886

  STagScanInfo*          pInfo = pOperator->info;

3,409,353✔

3887

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

3,409,353✔

3888

  int32_t                code = 0;

3,409,353✔

3889

  STableListInfo*        pListInfo = pInfo->pTableListInfo;

3,409,353✔

3890

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

3,409,353✔

3891

  tb_uid_t               pUid = pParam->vcUid;

3,409,353✔

3897

  STableKeyInfo info = {.groupId = 0};

3,409,353✔

3899

  int32_t tableIdx = 0;

3,409,353✔

3900

  taosHashClear(pListInfo->map);

3,409,353✔

3901

  taosArrayClear(pListInfo->pTableList);

3,409,353✔

3903

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

3,409,353✔

3910

  info.uid = pUid;

3,409,353✔

3911

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

3,409,353✔

3912

  if (p == NULL) {

3,409,353✔

3916

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

3,409,353✔

3918

  return code;

3,409,353✔

3921

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

6,602,639✔

3922

  int32_t       code = TSDB_CODE_SUCCESS;

6,602,639✔

3923

  int32_t       lino = 0;

6,602,639✔

3924

  STagScanInfo* pInfo = pOperator->info;

6,602,639✔

3925

  if (pOperator->pOperatorGetParam) {

6,604,535✔

3926

    pOperator->resultInfo.totalRows = 0;

3,409,353✔

3927

    pOperator->dynamicTask = true;

3,409,353✔

3928

    pInfo->curPos = 0;

3,409,353✔

3929

    code = createTagScanTableListInfoFromParam(pOperator);

3,409,353✔

3930

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

3,409,353✔

3931

    pOperator->pOperatorGetParam = NULL;

3,409,353✔

3932

    QUERY_CHECK_CODE(code, lino, _end);

3,409,353✔

3934

    if (pOperator->status == OP_EXEC_DONE) {

3,409,353✔

3935

      pOperator->status = OP_OPENED;

1,835,143✔

3939

  if (pOperator->status == OP_EXEC_DONE) {

6,601,629✔

3940

    (*ppRes) = NULL;

1,581,724✔

3941

    return code;

1,581,219✔

3944

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

5,019,467✔

3945

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

5,020,915✔

3947

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

5,019,086✔

3948

  SSDataBlock* pRes = pInfo->pRes;

5,019,905✔

3949

  blockDataCleanup(pRes);

5,017,257✔

3951

  int32_t size = 0;

5,020,096✔

3952

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

5,020,668✔

3953

  QUERY_CHECK_CODE(code, lino, _end);

5,018,076✔

3955

  if (size == 0) {

5,018,076✔

3956

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

18,701✔

3957

    (*ppRes) = NULL;

18,701✔

3958

    return code;

18,701✔

3961

  SMetaReader mr = {0};

4,999,375✔

3962

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

5,000,890✔

3963

  pRes->info.rows = 0;

5,001,204✔

3965

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

14,149,622✔

3966

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

9,145,445✔

3967

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

9,148,665✔

3970

      code = TSDB_CODE_SUCCESS;

9,149,484✔

3972

    QUERY_CHECK_CODE(code, lino, _end);

9,148,665✔

3974

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

9,148,665✔

3975

      setOperatorCompleted(pOperator);

5,002,786✔

3979

  pAPI->metaReaderFn.clearReader(&mr);

5,005,120✔

3980

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

5,002,281✔

3981

  if (bLimitReached) {

5,002,281✔

3986

  if (pOperator->status == OP_EXEC_DONE) {

5,002,281✔

3987

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

5,002,281✔

3990

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

5,003,291✔

3992

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

5,002,786✔

3993

_end:

5,001,462✔

3994

  if (code != TSDB_CODE_SUCCESS) {

5,002,786✔

3999

  return code;

5,002,786✔

4002

static void destroyTagScanOperatorInfo(void* param) {

8,094,449✔

4003

  STagScanInfo* pInfo = (STagScanInfo*)param;

8,094,449✔

4004

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

8,094,449✔

4007

  taosHashCleanup(pInfo->filterCtx.colHash);

8,098,823✔

4008

  taosArrayDestroy(pInfo->filterCtx.cInfoList);

8,087,307✔

4009

  taosArrayDestroy(pInfo->aFilterIdxs);

8,082,117✔

4010

  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);

8,087,274✔

4012

  blockDataDestroy(pInfo->pRes);

8,092,597✔

4013

  taosArrayDestroy(pInfo->matchInfo.pList);

8,094,648✔

4014

  tableListDestroy(pInfo->pTableListInfo);

8,092,485✔

4016

  pInfo->pRes = NULL;

8,094,694✔

4017

  pInfo->pTableListInfo = NULL;

8,095,045✔

4018

  taosMemoryFreeClear(param);

8,093,919✔

4019

8,092,120✔

4021

static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {

4,779✔

4022

  int32_t         code = TSDB_CODE_SUCCESS;

4,779✔

4023

  STagScanInfo*   pInfo = pOper->info;

4,779✔

4025

  pOper->status = OP_NOT_OPENED;

4,779✔

4026

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

4,779✔

4027

  if (pTagScanNode->onlyMetaCtbIdx) {

4,779✔

4028

      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

1,984✔

4029

      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,984✔

4030

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

1,984✔

4031

      pInfo->pCtbCursor = NULL;

1,984✔

4033

      pInfo->curPos = 0;

2,795✔

4036

  return code;

4,779✔

4039

int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,

8,093,403✔

4042

  QRY_PARAM_CHECK(pOptrInfo);

8,093,403✔

4044

  int32_t         code = TSDB_CODE_SUCCESS;

8,098,645✔

4045

  int32_t         lino = 0;

8,098,645✔

4046

  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;

8,098,645✔

4047

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

8,098,645✔

4048

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

8,084,972✔

4049

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

8,086,155✔

4053

  pOperator->pPhyNode = pTagScanNode;

8,086,155✔

4054

  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

8,086,155✔

4056

  int32_t    numOfExprs = 0;

8,090,668✔

4057

  SExprInfo* pExprInfo = NULL;

8,094,023✔

4059

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

8,090,348✔

4060

  QUERY_CHECK_CODE(code, lino, _error);

8,092,893✔

4062

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

8,092,893✔

4063

  QUERY_CHECK_CODE(code, lino, _error);

8,087,792✔

4065

  int32_t num = 0;

8,087,792✔

4066

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

8,090,472✔

4067

  QUERY_CHECK_CODE(code, lino, _error);

8,088,945✔

4069

  pInfo->pTagCond = pTagCond;

8,088,945✔

4070

  pInfo->pTagIndexCond = pTagIndexCond;

8,090,978✔

4071

  pInfo->suid = pPhyNode->suid;

8,087,872✔

4072

  pInfo->pStorageAPI = &pTaskInfo->storageAPI;

8,088,076✔

4074

  pInfo->pTableListInfo = pTableListInfo;

8,078,515✔

4075

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

8,091,867✔

4076

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

8,092,203✔

4078

  pInfo->readHandle = *pReadHandle;

8,072,374✔

4079

  pInfo->curPos = 0;

8,094,213✔

4081

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

8,078,047✔

4082

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

8,087,166✔

4084

  initResultSizeInfo(&pOperator->resultInfo, 4096);

8,087,785✔

4085

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

8,076,785✔

4086

  QUERY_CHECK_CODE(code, lino, _error);

8,098,410✔

4088

  if (pTagScanNode->onlyMetaCtbIdx) {

8,098,410✔

4089

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

4,588,513✔

4090

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

4,589,569✔

4092

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

4,588,117✔

4093

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

4,590,668✔

4095

    pInfo->filterCtx.colHash =

4,587,614✔

4096

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

4,590,198✔

4097

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

4,589,084✔

4099

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

4,587,273✔

4100

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

4,589,079✔

4102

    if (pInfo->pTagCond != NULL) {

4,587,607✔

4103

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

1,621,969✔

4106

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

8,098,041✔

4107

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

8,097,942✔

4109

  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);

8,095,287✔

4111

  *pOptrInfo = pOperator;

8,098,347✔

4112

  return code;

8,094,829✔

4129

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

232,976,976✔

4130

  int32_t                 left = *(int32_t*)pLeft;

232,976,976✔

4131

  int32_t                 right = *(int32_t*)pRight;

232,976,274✔

4132

  STmsSubTablesMergeInfo* pInfo = (STmsSubTablesMergeInfo*)param;

232,976,274✔

4134

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

232,976,274✔

4135

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

232,977,678✔

4137

  if (leftIdx == -1) {

232,975,221✔

4138

    return 1;

199,717,941✔

4139

  } else if (rightIdx == -1) {

33,257,280✔

4140

    return -1;

329,280✔

4143

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

32,928,000✔

4144

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

32,928,000✔

4145

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

32,928,000✔

4146

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

32,928,000✔

4149

  return ret;

32,928,000✔

4179

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

2,947,946✔

4180

  int32_t code = TSDB_CODE_SUCCESS;

2,947,946✔

4181

  int32_t lino = 0;

2,947,946✔

4182

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

2,947,946✔

4183

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

2,947,946✔

4184

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

2,945,545✔

4185

  for (int i = 0; i < src->numOfCols; i++) {

10,158,702✔

4186

    dst->colList[i] = src->colList[i];

7,229,645✔

4189

_end:

2,946,917✔

4190

  if (code != TSDB_CODE_SUCCESS) {

2,946,917✔

4193

  return code;

2,938,956✔

4196

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

4,439,699✔

4198

  int32_t code = 0;

4,439,699✔

4200

  STableMergeScanInfo*    pInfo = pOperator->info;

4,439,699✔

4201

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

4,440,361✔

4202

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

4,437,250✔

4203

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

4,434,466✔

4204

  const SStorageAPI*      pAPI = &pTaskInfo->storageAPI;

4,439,356✔

4206

  blockDataCleanup(pInput->pReaderBlock);

4,441,063✔

4207

  if (!pInput->bInMemReader) {

4,436,261✔

4208

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

4,441,497✔

4209

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

2,224,362✔

4210

    if (code != 0) {

2,221,610✔

4215

  pInfo->base.dataReader = pInput->pReader;

4,436,269✔

4217

  while (true) {

7,371✔

4218

    bool hasNext = false;

4,442,898✔

4219

    code = pAPI->tsdReader.tsdNextDataBlock(pInfo->base.dataReader, &hasNext);

4,442,547✔

4220

    if (code != 0) {

4,451,577✔

4226

    if (!hasNext || isTaskKilled(pTaskInfo)) {

4,451,577✔

4227

      if (isTaskKilled(pTaskInfo)) {

2,965,854✔

4233

      *pSubTableHasBlock = false;

2,966,540✔

4234

      break;

2,966,189✔

4237

    if (pInput->tblCond.order == TSDB_ORDER_ASC) {

1,485,723✔

4238

      pInput->tblCond.twindows.skey = pInput->pReaderBlock->info.window.ekey + 1;

1,461,855✔

4240

      pInput->tblCond.twindows.ekey = pInput->pReaderBlock->info.window.skey - 1;

23,868✔

4243

    uint32_t status = 0;

1,485,723✔

4244

    code = loadDataBlock(pOperator, &pInfo->base, pInput->pReaderBlock, &status);

1,485,723✔

4245

    if (code != 0) {

1,485,029✔

4250

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

1,485,029✔

4254

    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pInput->pReaderBlock->info.rows == 0) {

1,485,029✔

4255

      continue;

7,371✔

4258

    *pSubTableHasBlock = true;

1,478,009✔

4259

    break;

1,478,009✔

4262

  if (*pSubTableHasBlock) {

4,443,169✔

4263

    pInput->pReaderBlock->info.id.groupId =

2,956,361✔

4264

        tableListGetTableGroupId(pInfo->base.pTableListInfo, pInput->pReaderBlock->info.id.uid);

2,956,361✔

4265

    pOperator->resultInfo.totalRows += pInput->pReaderBlock->info.rows;

1,478,352✔

4267

  if (!pInput->bInMemReader || !*pSubTableHasBlock) {

4,444,206✔

4268

    pAPI->tsdReader.tsdReaderClose(pInput->pReader);

3,339,158✔

4269

    pInput->pReader = NULL;

3,334,691✔

4272

  pInfo->base.dataReader = NULL;

4,440,074✔

4273

  return TSDB_CODE_SUCCESS;

4,440,768✔

4276

static int32_t setGroupStartEndIndex(STableMergeScanInfo* pInfo) {

1,836,604✔

4277

  pInfo->bGroupProcessed = false;

1,836,604✔

4279

  int32_t numOfTables = 0;

1,838,000✔

4280

  int32_t code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

1,802,455✔

4281

  if (code != TSDB_CODE_SUCCESS) {

1,805,542✔

4286

  int32_t i = pInfo->tableStartIndex + 1;

1,805,542✔

4287

  for (; i < numOfTables; ++i) {

2,921,319✔

4288

    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

1,164,444✔

4289

    if (!tableKeyInfo) {

1,164,093✔

4293

    if (tableKeyInfo->groupId != pInfo->groupId) {

1,164,093✔

4294

      break;

54,849✔

4297

  pInfo->tableEndIndex = i - 1;

1,811,724✔

4298

  return TSDB_CODE_SUCCESS;

1,844,764✔

4301

static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {

1,854,496✔

4302

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,819,640✔

4303

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

2,966,540✔

4304

    if (pInput->rowIdx == -1) {

2,963,453✔

4305

      continue;

1,487,502✔

4308

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,478,009✔

4309

      pInput->rowIdx = 0;

1,477,307✔

4310

      pInput->pageIdx = -1;

1,478,001✔

4313

    pInput->pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,479,054✔

4314

    SColumnInfoData* col = taosArrayGet(pInput->pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

1,478,352✔

4315

    if (!col) {

1,477,299✔

4319

    pInput->aTs = (int64_t*)col->pData;

1,477,299✔

4322

  __merge_compare_fn_t mergeCompareFn = (!pSubTblsInfo->pPkOrderInfo) ? subTblRowCompareTsFn : subTblRowCompareTsPkFn;

1,854,831✔

4323

  return tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, mergeCompareFn);

1,854,831✔

4326

static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {

1,835,567✔

4327

  int32_t code = setGroupStartEndIndex(pInfo);

1,835,567✔

4328

  if (code != TSDB_CODE_SUCCESS) {

1,843,735✔

4332

  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));

1,843,735✔

4333

  if (pSubTblsInfo == NULL) {

1,826,848✔

4336

  pSubTblsInfo->pTsOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);

1,826,848✔

4337

  if (!pSubTblsInfo->pTsOrderInfo) {

1,836,173✔

4341

  if (taosArrayGetSize(pInfo->pSortInfo) == 2) {

1,837,968✔

4348

    pSubTblsInfo->pPkOrderInfo = NULL;

1,818,257✔

4350

  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

1,845,546✔

4351

  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));

1,826,553✔

4352

  if (pSubTblsInfo->aInputs == NULL) {

1,816,646✔

4356

  int32_t bufPageSize = pInfo->bufPageSize;

1,811,405✔

4357

  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;

1,840,042✔

4358

  code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);

1,843,488✔

4359

  if (code != TSDB_CODE_SUCCESS) {

1,826,609✔

4364

  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;

1,826,609✔

4365

  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;

1,844,501✔

4367

  pInfo->pSubTablesMergeInfo = pSubTblsInfo;

1,814,907✔

4368

  return TSDB_CODE_SUCCESS;

1,817,308✔

4371

static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {

1,835,918✔

4372

  int32_t        code = TSDB_CODE_SUCCESS;

1,835,918✔

4373

  int32_t        lino = 0;

1,835,918✔

4374

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,835,918✔

4375

  SReadHandle*   pHandle = &pInfo->base.readHandle;

1,839,348✔

4376

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,837,633✔

4378

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

1,847,947✔

4380

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,777,562✔

4381

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

2,953,474✔

4382

    pInput->type = SUB_TABLE_MEM_BLOCK;

2,930,014✔

4384

    code = dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);

2,951,384✔

4385

    QUERY_CHECK_CODE(code, lino, _end);

2,930,030✔

4387

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pReaderBlock);

2,930,030✔

4388

    QUERY_CHECK_CODE(code, lino, _end);

2,965,511✔

4390

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pPageBlock);

2,965,511✔

4391

    QUERY_CHECK_CODE(code, lino, _end);

2,965,854✔

4393

    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);

2,965,854✔

4394

    pInput->pKeyInfo = keyInfo;

2,962,767✔

4396

    if (isTaskKilled(pTaskInfo)) {

2,961,706✔

4400

    if (i + 1 < pSubTblsInfo->numInMemReaders) {

2,947,595✔

4401

      code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, keyInfo, 1, pInput->pReaderBlock,

2,220,243✔

4402

                                           (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

1,111,350✔

4403

      QUERY_CHECK_CODE(code, lino, _end);

1,108,893✔

4404

      pInput->bInMemReader = true;

1,108,893✔

4406

      pInput->pReader = NULL;

1,850,356✔

4407

      pInput->bInMemReader = false;

1,850,356✔

4409

    bool hasNext = true;

2,957,893✔

4410

    code = fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);

2,961,315✔

4411

    QUERY_CHECK_CODE(code, lino, _end);

2,964,139✔

4412

    if (!hasNext) {

2,964,139✔

4413

      pInput->rowIdx = -1;

1,485,787✔

4414

      ++pSubTblsInfo->numSubTablesCompleted;

1,487,845✔

4415

      continue;

1,486,130✔

4417

      pInput->rowIdx = 0;

1,478,352✔

4418

      pInput->pageIdx = -1;

1,478,352✔

4422

_end:

1,853,802✔

4423

  if (code != TSDB_CODE_SUCCESS) {

1,854,153✔

4426

  return code;

1,854,488✔

4429

static void adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

1,478,352✔

4430

  int32_t              code = TSDB_CODE_SUCCESS;

1,478,352✔

4431

  int32_t              lino = 0;

1,478,352✔

4432

  SExecTaskInfo*       pTaskInfo = pOperatorInfo->pTaskInfo;

1,478,352✔

4433

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

1,478,352✔

4434

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,478,352✔

4435

  bool                 hasNext = true;

1,478,352✔

4436

  code = fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);

1,478,352✔

4437

  QUERY_CHECK_CODE(code, lino, _end);

1,478,009✔

4439

  if (!hasNext) {

1,478,009✔

4440

    pInput->rowIdx = -1;

1,478,009✔

4441

    ++pSubTblsInfo->numSubTablesCompleted;

1,478,009✔

4446

_end:

1,478,009✔

4447

  if (code != TSDB_CODE_SUCCESS) {

1,478,009✔

4452

1,478,352✔

4454

static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

269,546,652✔

4455

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

269,546,652✔

4456

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

269,550,130✔

4458

  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

269,555,227✔

4459

  if (pInput->rowIdx < pInputBlock->info.rows - 1) {

269,587,438✔

4460

    ++pInput->rowIdx;

268,103,175✔

4461

  } else if (pInput->rowIdx == pInputBlock->info.rows - 1) {

1,478,352✔

4462

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,478,352✔

4463

      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);

1,478,352✔

4465

    if (pInput->rowIdx != -1) {

1,478,352✔

4476

  return tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));

269,580,147✔

4479

static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {

269,459,351✔

4480

  int32_t            code = TSDB_CODE_SUCCESS;

269,459,351✔

4481

  int32_t            lino = 0;

269,459,351✔

4482

  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

269,459,351✔

4483

  SSDataBlock*       pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

269,478,703✔

4485

  for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {

1,132,082,244✔

4486

    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);

861,648,951✔

4487

    QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);

861,982,886✔

4489

    SColumnInfoData* pSrcColInfo = taosArrayGet(pInputBlock->pDataBlock, i);

861,982,886✔

4490

    QUERY_CHECK_NULL(pSrcColInfo, code, lino, _end, terrno);

860,914,521✔

4491

    bool isNull = colDataIsNull(pSrcColInfo, pInputBlock->info.rows, pInput->rowIdx, NULL);

860,914,521✔

4493

    if (isNull) {

861,959,667✔

4494

      code = colDataSetVal(pColInfo, pBlock->info.rows, NULL, true);

101,418,072✔

4495

      QUERY_CHECK_CODE(code, lino, _end);

101,418,423✔

4497

      if (pSrcColInfo->pData != NULL) {

760,541,595✔

4498

        char* pData = colDataGetData(pSrcColInfo, pInput->rowIdx);

760,540,861✔

4499

        code = colDataSetVal(pColInfo, pBlock->info.rows, pData, false);

761,330,470✔

4500

        QUERY_CHECK_CODE(code, lino, _end);

761,251,132✔

4504

  pBlock->info.dataLoad = 1;

269,554,206✔

4505

  pBlock->info.scanFlag = pInputBlock->info.scanFlag;

269,558,075✔

4506

  pBlock->info.rows += 1;

269,559,782✔

4508

_end:

269,555,315✔

4509

  if (code != TSDB_CODE_SUCCESS) {

269,555,315✔

4512

  return code;

269,553,592✔

4515

static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,

3,004,254✔

4517

  int32_t                 code = TSDB_CODE_SUCCESS;

3,004,254✔

4518

  int32_t                 lino = 0;

3,004,254✔

4519

  STableMergeScanInfo*    pInfo = pOperator->info;

3,004,254✔

4520

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

3,004,254✔

4521

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

3,003,911✔

4522

  bool                    finished = false;

3,001,502✔

4524

  QRY_PARAM_CHECK(pResBlock);

3,001,502✔

4526

  blockDataCleanup(pBlock);

3,001,502✔

4530

      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {

272,494,868✔

4531

        finished = true;

3,003,568✔

4532

        break;

3,003,568✔

4535

      code = appendChosenRowToDataBlock(pSubTblsInfo, pBlock);

269,485,995✔

4536

      QUERY_CHECK_CODE(code, lino, _end);

269,546,724✔

4538

      code = adjustSubTableForNextRow(pOperator, pSubTblsInfo);

269,546,724✔

4539

      QUERY_CHECK_CODE(code, lino, _end);

269,492,010✔

4541

      if (pBlock->info.rows >= capacity) {

269,492,010✔

4546

    if (isTaskKilled(pTaskInfo)) {

3,003,568✔

4550

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pBlock, pTaskInfo);

3,004,254✔

4551

    if (finished || limitReached || pBlock->info.rows > 0) {

3,004,254✔

4556

  if (pBlock->info.rows > 0) {

3,004,254✔

4557

    *pResBlock = pBlock;

1,149,072✔

4560

_end:

1,855,182✔

4561

  if (code != TSDB_CODE_SUCCESS) {

3,004,254✔

4565

  return code;

3,004,254✔

4568

static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {

1,837,976✔

4569

  int32_t              code = TSDB_CODE_SUCCESS;

1,837,976✔

4570

  int32_t              lino = 0;

1,837,976✔

4571

  STableMergeScanInfo* pInfo = pOperator->info;

1,837,976✔

4573

  code = initSubTablesMergeInfo(pInfo);

1,844,517✔

4574

  QUERY_CHECK_CODE(code, lino, _end);

1,811,086✔

4576

  code = initSubTableInputs(pOperator, pInfo);

1,811,086✔

4577

  QUERY_CHECK_CODE(code, lino, _end);

1,854,153✔

4579

  code = openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);

1,854,153✔

4580

  QUERY_CHECK_CODE(code, lino, _end);

1,853,459✔

4582

_end:

1,853,459✔

4583

  if (code != TSDB_CODE_SUCCESS) {

1,853,459✔

4586

  return code;

1,853,810✔

4589

static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {

29,094,750✔

4590

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

29,094,750✔

4591

  if (pSubTblsInfo != NULL) {

29,098,940✔

4592

    tMergeTreeDestroy(&pSubTblsInfo->pTree);

1,853,810✔

4594

    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,820,350✔

4595

      STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

2,966,189✔

4596

      taosMemoryFree(pInput->tblCond.colList);

2,963,453✔

4597

      blockDataDestroy(pInput->pReaderBlock);

2,965,168✔

4598

      blockDataDestroy(pInput->pPageBlock);

2,966,540✔

4599

      taosArrayDestroy(pInput->aBlockPages);

2,966,197✔

4600

      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);

2,966,197✔

4601

      pInput->pReader = NULL;

2,965,854✔

4604

    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);

1,854,153✔

4605

    taosMemoryFree(pSubTblsInfo->aInputs);

1,851,401✔

4607

    taosMemoryFree(pSubTblsInfo);

1,852,430✔

4608

    pInfo->pSubTablesMergeInfo = NULL;

1,852,095✔

4612

29,097,568✔

4614

int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

2,972,205✔

4615

  QRY_PARAM_CHECK(ppRes);

2,972,205✔

4617

  int32_t lino = 0;

2,978,411✔

4618

  int32_t tableListSize = 0;

2,978,411✔

4619

  int64_t st = taosGetTimestampUs();

2,964,340✔

4621

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

2,964,340✔

4622

  STableMergeScanInfo* pInfo = pOperator->info;

2,973,250✔

4624

  if (pOperator->status == OP_EXEC_DONE) {

2,984,258✔

4628

  int32_t code = pOperator->fpSet._openFn(pOperator);

2,941,566✔

4629

  QUERY_CHECK_CODE(code, lino, _end);

2,966,709✔

4631

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

2,966,709✔

4632

  QUERY_CHECK_CODE(code, lino, _end);

2,960,822✔

4634

  if (!pInfo->hasGroupId) {

2,960,822✔

4635

    pInfo->hasGroupId = true;

1,821,761✔

4637

    if (tableListSize == 0) {

1,797,248✔

4638

      setOperatorCompleted(pOperator);

41,418✔

4639

      (*ppRes) = NULL;

41,067✔

4640

      return code;

41,418✔

4643

    pInfo->tableStartIndex = 0;

1,755,830✔

4644

    STableKeyInfo* pTmpGpId = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,756,564✔

4645

    QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

1,746,928✔

4646

    pInfo->groupId = pTmpGpId->groupId;

1,746,928✔

4647

    code = startSubTablesTableMergeScan(pOperator);

1,783,725✔

4648

    QUERY_CHECK_CODE(code, lino, _end);

1,798,610✔

4651

  SSDataBlock* pBlock = NULL;

2,941,460✔

4652

  while (pInfo->tableStartIndex < tableListSize) {

3,002,188✔

4653

    if (isTaskKilled(pTaskInfo)) {

3,003,568✔

4657

    code = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity, &pBlock);

3,001,502✔

4658

    QUERY_CHECK_CODE(code, lino, _end);

3,004,254✔

4660

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

3,004,254✔

4667

    if (pBlock != NULL) {

3,003,225✔

4668

      pBlock->info.id.groupId = pInfo->groupId;

1,149,072✔

4669

      pOperator->resultInfo.totalRows += pBlock->info.rows;

1,149,072✔

4670

      pInfo->bGroupProcessed = true;

1,149,072✔

4671

      break;

1,149,072✔

4674

      stopSubTablesTableMergeScan(pInfo);

1,854,153✔

4675

      if (pInfo->tableEndIndex >= tableListSize - 1) {

1,853,810✔

4676

        setOperatorCompleted(pOperator);

1,796,887✔

4677

        break;

1,799,288✔

4680

      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

55,200✔

4681

      STableKeyInfo* pTmpGpId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

55,200✔

4682

      QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

55,200✔

4684

      pInfo->groupId = pTmpGpId->groupId;

55,200✔

4685

      code = startSubTablesTableMergeScan(pOperator);

55,200✔

4686

      QUERY_CHECK_CODE(code, lino, _end);

55,200✔

4687

      resetLimitInfoForNextGroup(&pInfo->limitInfo);

55,200✔

4691

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

2,948,017✔

4693

_end:

2,944,252✔

4694

  if (code != TSDB_CODE_SUCCESS) {

2,946,996✔

4699

    (*ppRes) = pBlock;

2,946,996✔

4702

  return code;

2,948,360✔

4705

static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {

3,469,082✔

4706

  int32_t              code = TSDB_CODE_SUCCESS;

3,469,082✔

4707

  int32_t              lino = 0;

3,469,082✔

4708

  SOperatorInfo*       pOperator = (SOperatorInfo*)pTableMergeOpInfo;

3,469,082✔

4709

  STableMergeScanInfo* pInfo = pOperator->info;

3,469,082✔

4710

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,469,082✔

4712

  if (pInfo->mSkipTables == NULL) {

3,469,082✔

4713

    pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,

338,611✔

4715

    QUERY_CHECK_NULL(pInfo->mSkipTables, code, lino, _end, terrno);

338,611✔

4717

  int bSkip = 1;

3,469,082✔

4718

  if (pInfo->mSkipTables != NULL) {

3,469,082✔

4719

    code = taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));

3,469,082✔

4720

    if (code == TSDB_CODE_DUP_KEY) {

3,469,082✔

4723

    QUERY_CHECK_CODE(code, lino, _end);

3,469,082✔

4726

_end:

3,469,082✔

4727

  if (code != TSDB_CODE_SUCCESS) {

3,469,082✔

4732

3,469,082✔

4734

static int32_t doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {

162,439,076✔

4735

  STableMergeScanInfo* pInfo = pOperator->info;

162,439,076✔

4736

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

162,450,325✔

4737

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

162,443,992✔

4738

  SSDataBlock*         pBlock = pInfo->pReaderBlock;

162,445,903✔

4739

  int32_t              code = 0;

162,423,780✔

4740

  bool                 hasNext = false;

162,423,780✔

4741

  STsdbReader*         reader = pInfo->base.dataReader;

162,419,296✔

4743

  code = pAPI->tsdReader.tsdNextDataBlock(reader, &hasNext);

162,421,507✔

4744

  if (code != 0) {

162,409,707✔

4751

  if (!hasNext || isTaskKilled(pTaskInfo)) {

162,409,707✔

4752

    if (isTaskKilled(pTaskInfo)) {

28,322,344✔

4756

    *pFinished = true;

28,336,065✔

4757

    return code;

28,328,076✔

4760

  uint32_t status = 0;

134,097,032✔

4761

  code = loadDataBlock(pOperator, &pInfo->base, pBlock, &status);

134,090,284✔

4763

  if (code != TSDB_CODE_SUCCESS) {

134,090,626✔

4764

    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));

16,640✔

4765

    pTaskInfo->code = code;

16,640✔

4766

    return code;

16,640✔

4769

  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

134,073,986✔

4775

  if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {

134,073,986✔

4776

    *pSkipped = true;

6,216,162✔

4777

    return code;

6,221,112✔

4780

  return code;

127,841,249✔

4783

static int32_t getBlockForTableMergeScan(void* param, SSDataBlock** ppBlock) {

164,772,944✔

4784

  STableMergeScanSortSourceParam* source = param;

164,772,944✔

4786

  SOperatorInfo*       pOperator = source->pOperator;

164,772,944✔

4787

  STableMergeScanInfo* pInfo = pOperator->info;

164,787,197✔

4788

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

164,784,594✔

4789

  SSDataBlock*         pBlock = NULL;

164,768,265✔

4790

  int64_t              st = taosGetTimestampUs();

164,771,697✔

4791

  int32_t              code = TSDB_CODE_SUCCESS;

164,771,697✔

4794

    if (pInfo->rtnNextDurationBlocks) {

178,821,900✔

4795

      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d",

16,389,965✔

4798

      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {

16,399,881✔

4799

        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];

7,833,266✔

4800

        ++pInfo->nextDurationBlocksIdx;

7,833,266✔

4802

        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

16,388,560✔

4803

          blockDataDestroy(pInfo->nextDurationBlocks[i]);

7,832,226✔

4804

          pInfo->nextDurationBlocks[i] = NULL;

7,829,306✔

4807

        pInfo->rtnNextDurationBlocks = false;

8,557,064✔

4808

        pInfo->nextDurationBlocksIdx = 0;

8,559,254✔

4809

        pInfo->numNextDurationBlocks = 0;

8,555,969✔

4810

        continue;

8,557,794✔

4813

      bool bFinished = false;

162,431,961✔

4814

      bool bSkipped = false;

162,440,302✔

4816

      code = doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);

162,456,332✔

4817

      if (code != 0) {

162,402,474✔

4818

        return code;

16,640✔

4821

      pBlock = pInfo->pReaderBlock;

162,385,834✔

4822

      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d",

162,404,127✔

4824

      if (bFinished) {

162,452,944✔

4825

        pInfo->bNewFilesetEvent = false;

28,343,998✔

4826

        break;

28,340,554✔

4829

      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {

134,108,946✔

4830

        if (!bSkipped) {

8,558,971✔

4831

          code = createOneDataBlock(pBlock, true, &pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks]);

7,833,266✔

4832

          if (code) {

7,823,926✔

4837

          ++pInfo->numNextDurationBlocks;

7,823,926✔

4838

          if (pInfo->numNextDurationBlocks > 2) {

7,825,536✔

4846

        if (pInfo->bNewFilesetEvent) {

8,556,566✔

4847

          pInfo->rtnNextDurationBlocks = true;

8,554,389✔

4848

          *ppBlock = NULL;

8,556,794✔

4849

          return code;

8,541,464✔

4857

      if (bSkipped) continue;

125,549,982✔

4860

    pBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pBlock->info.id.uid);

127,884,458✔

4862

    pOperator->resultInfo.totalRows += pBlock->info.rows;

127,892,008✔

4863

    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

127,884,637✔

4864

    *ppBlock = pBlock;

127,886,267✔

4866

    return code;

127,883,827✔

4869

  *ppBlock = NULL;

28,342,522✔

4870

  return code;

28,343,998✔

4873

int32_t generateSortByTsPkInfo(SArray* colMatchInfo, int32_t order, SArray** ppSortArray) {

24,821,278✔

4874

  int32_t code = TSDB_CODE_SUCCESS;

24,821,278✔

4875

  int32_t lino = 0;

24,821,278✔

4876

  SArray* pSortInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));

24,821,278✔

4877

  QUERY_CHECK_NULL(pSortInfo, code, lino, _end, terrno);

24,783,484✔

4878

  SBlockOrderInfo biTs = {0};

24,783,484✔

4879

  SBlockOrderInfo biPk = {0};

24,788,678✔

4881

  int32_t tsTargetSlotId = 0;

24,795,768✔

4882

  int32_t pkTargetSlotId = -1;

24,795,768✔

4883

  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {

99,708,378✔

4884

    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);

74,989,446✔

4885

    QUERY_CHECK_NULL(colInfo, code, lino, _end, terrno);

74,985,128✔

4886

    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

74,985,128✔

4887

      tsTargetSlotId = colInfo->dstSlotId;

24,820,557✔

4888

      biTs.order = order;

24,824,856✔

4889

      biTs.slotId = tsTargetSlotId;

24,824,856✔

4890

      biTs.nullFirst = (order == TSDB_ORDER_ASC);

24,824,856✔

4891

      biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, order);

24,824,856✔

4894

    if (colInfo->isPk) {

74,940,936✔

4895

      pkTargetSlotId = colInfo->dstSlotId;

2,231,945✔

4896

      biPk.order = order;

2,231,945✔

4897

      biPk.slotId = pkTargetSlotId;

2,231,945✔

4898

      biPk.nullFirst = (order == TSDB_ORDER_ASC);

2,231,945✔

4899

      biPk.compFn = getKeyComparFunc(colInfo->dataType.type, order);

2,231,945✔

4903

  void* tmp = taosArrayPush(pSortInfo, &biTs);

24,834,403✔

4904

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

24,834,403✔

4905

  if (pkTargetSlotId != -1) {

24,834,403✔

4906

    tmp = taosArrayPush(pSortInfo, &biPk);

2,234,426✔

4907

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

2,234,426✔

4910

  (*ppSortArray) = pSortInfo;

24,835,644✔

4912

_end:

24,805,877✔

4913

  if (code != TSDB_CODE_SUCCESS) {

24,813,710✔

4916

  return code;

24,813,710✔

4919

void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {

30,354,098✔

4920

  STableMergeScanInfo* pTmsInfo = param;

30,354,098✔

4921

  if (type == TSD_READER_NOTIFY_DURATION_START) {

30,354,098✔

4922

    pTmsInfo->bNewFilesetEvent = true;

30,362,883✔

4926

  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);

30,342,512✔

4928

  return;

30,373,601✔

4931

int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

37,097,025✔

4932

  STableMergeScanInfo* pInfo = pOperator->info;

37,097,025✔

4933

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

37,119,049✔

4935

  int32_t code = TSDB_CODE_SUCCESS;

37,107,769✔

4936

  int32_t lino = 0;

37,107,769✔

4937

  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

37,107,769✔

4939

  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));

37,082,990✔

4940

  pInfo->bNewFilesetEvent = false;

37,107,817✔

4941

  pInfo->bNextDurationBlockEvent = false;

37,127,275✔

4943

  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;

37,126,070✔

4944

  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;

37,127,378✔

4946

  pInfo->pSortHandle = NULL;

37,118,352✔

4947

  code = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,

74,241,884✔

4948

                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0, &pInfo->pSortHandle);

37,108,924✔

4949

  if (code) {

37,079,123✔

4953

  if (pInfo->bSortRowId && numOfTable != 1) {

37,079,123✔

4954

    int32_t memSize = 512 * 1024 * 1024;

2,748,449✔

4955

    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);

2,748,449✔

4956

    QUERY_CHECK_CODE(code, lino, _end);

2,749,029✔

4959

  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);

37,088,700✔

4960

  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pOperator);

37,084,021✔

4961

  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);

37,093,974✔

4963

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);

37,095,836✔

4964

  QUERY_CHECK_CODE(code, lino, _end);

37,057,894✔

4966

  STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));

37,057,894✔

4967

  QUERY_CHECK_NULL(param, code, lino, _end, terrno);

37,072,977✔

4968

  param->pOperator = pOperator;

37,072,977✔

4970

  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));

37,084,523✔

4971

  if (ps == NULL) {

37,042,354✔

4976

  ps->param = param;

37,042,354✔

4977

  ps->onlyRef = false;

37,087,558✔

4978

  code = tsortAddSource(pInfo->pSortHandle, ps);

37,097,224✔

4979

  QUERY_CHECK_CODE(code, lino, _end);

37,111,234✔

4981

  if (numOfTable == 1) {

37,111,234✔

4982

    tsortSetSingleTableMerge(pInfo->pSortHandle);

25,395,790✔

4984

    code = tsortOpen(pInfo->pSortHandle);

11,715,444✔

4985

    QUERY_CHECK_CODE(code, lino, _end);

11,716,236✔

4988

_end:

11,707,916✔

4989

  if (code != TSDB_CODE_SUCCESS) {

37,100,386✔

4990

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,320✔

4992

  return code;

37,114,821✔

4995

void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

37,119,300✔

4996

  STableMergeScanInfo* pInfo = pOperator->info;

37,119,300✔

4997

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

37,120,661✔

4998

  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));

37,121,594✔

5000

  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

37,127,076✔

5001

  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;

37,116,789✔

5002

  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;

37,118,277✔

5003

  pInfo->sortExecInfo.loops += sortExecInfo.loops;

37,115,865✔

5004

  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;

37,115,689✔

5005

  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

37,106,824✔

5007

  tsortDestroySortHandle(pInfo->pSortHandle);

37,107,421✔

5008

  pInfo->pSortHandle = NULL;

37,116,902✔

5009

37,113,245✔

5011

void startGroupTableMergeScan(SOperatorInfo* pOperator) {

28,546,083✔

5012

  int32_t              code = TSDB_CODE_SUCCESS;

28,546,083✔

5013

  int32_t              lino = 0;

28,546,083✔

5014

  STableMergeScanInfo* pInfo = pOperator->info;

28,546,083✔

5015

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,546,061✔

5016

  SReadHandle*         pHandle = &pInfo->base.readHandle;

28,523,868✔

5017

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,514,945✔

5018

  qDebug("%s table merge scan start group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,544,903✔

5021

    int32_t numOfTables = 0;

28,551,243✔

5022

    code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

28,523,741✔

5023

    QUERY_CHECK_CODE(code, lino, _end);

28,553,531✔

5025

    int32_t i = pInfo->tableStartIndex + 1;

28,553,531✔

5026

    for (; i < numOfTables; ++i) {

132,449,578✔

5027

      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

110,047,985✔

5028

      QUERY_CHECK_NULL(tableKeyInfo, code, lino, _end, terrno);

110,046,784✔

5029

      if (tableKeyInfo->groupId != pInfo->groupId) {

110,046,784✔

5030

        break;

6,147,867✔

5033

    pInfo->tableEndIndex = i - 1;

28,549,460✔

5035

  pInfo->bGroupProcessed = false;

28,569,764✔

5036

  int32_t tableStartIdx = pInfo->tableStartIndex;

28,571,146✔

5037

  int32_t tableEndIdx = pInfo->tableEndIndex;

28,563,222✔

5039

  int32_t        numOfTable = tableEndIdx - tableStartIdx + 1;

28,572,101✔

5040

  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);

28,572,101✔

5041

  code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,

57,107,171✔

5042

                                       (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->mSkipTables);

28,555,794✔

5043

  QUERY_CHECK_CODE(code, lino, _end);

28,546,999✔

5044

  if (pInfo->filesetDelimited) {

28,546,999✔

5045

    pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);

28,541,301✔

5047

  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);

28,564,274✔

5049

  code = startDurationForGroupTableMergeScan(pOperator);

28,507,433✔

5050

  QUERY_CHECK_CODE(code, lino, _end);

28,526,867✔

5052

_end:

28,518,547✔

5053

  if (code != TSDB_CODE_SUCCESS) {

28,526,839✔

5054

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,320✔

5055

    pTaskInfo->code = code;

8,320✔

5056

    T_LONG_JMP(pTaskInfo->env, code);

8,320✔

5058

28,518,519✔

5060

void stopGroupTableMergeScan(SOperatorInfo* pOperator) {

28,563,049✔

5061

  STableMergeScanInfo* pInfo = pOperator->info;

28,563,049✔

5062

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,563,569✔

5063

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,562,585✔

5065

  stopDurationForGroupTableMergeScan(pOperator);

28,563,541✔

5067

  if (pInfo->base.dataReader != NULL) {

28,556,973✔

5068

    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);

28,557,703✔

5069

    pInfo->base.dataReader = NULL;

28,502,326✔

5071

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

28,508,184✔

5072

    if (pInfo->nextDurationBlocks[i]) {

1,040✔

5073

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

1,040✔

5074

      pInfo->nextDurationBlocks[i] = NULL;

1,040✔

5076

    pInfo->numNextDurationBlocks = 0;

1,040✔

5077

    pInfo->nextDurationBlocksIdx = 0;

1,040✔

5079

  resetLimitInfoForNextGroup(&pInfo->limitInfo);

28,539,438✔

5080

  taosHashCleanup(pInfo->mSkipTables);

28,514,104✔

5081

  pInfo->mSkipTables = NULL;

28,516,759✔

5082

  qDebug("%s table merge scan stop group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,516,227✔

5083

28,567,862✔

5087

SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,

76,633,984✔

5089

  STableMergeScanInfo* pInfo = pOperator->info;

76,633,984✔

5090

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

76,640,423✔

5091

  STupleHandle*        pTupleHandle = NULL;

76,642,876✔

5093

  blockDataCleanup(pResBlock);

76,659,432✔

5096

    while (1) {

2,147,483,647✔

5097

      pTupleHandle = NULL;

2,147,483,647✔

5098

      int32_t code = tsortNextTuple(pHandle, &pTupleHandle);

2,147,483,647✔

5099

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5100

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

82✔

5101

        T_LONG_JMP(pOperator->pTaskInfo->env, code);

82✔

5103

      if (pTupleHandle == NULL) {

2,147,483,647✔

5104

        break;

60,899,466✔

5107

      code = tsortAppendTupleToBlock(pInfo->pSortHandle, pResBlock, pTupleHandle);

2,147,483,647✔

5108

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5113

      if (pResBlock->info.rows >= capacity) {

2,147,483,647✔

5114

        break;

15,747,514✔

5118

    if (tsortIsClosed(pHandle)) {

76,646,980✔

5123

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);

76,654,428✔

5124

    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,

76,651,227✔

5126

    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {

76,663,964✔

5130

  return (pResBlock->info.rows > 0) ? pResBlock : NULL;

76,663,964✔

5133

int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

64,953,470✔

5134

  if (pOperator->status == OP_EXEC_DONE) {

64,953,470✔

5135

    (*ppRes) = NULL;

41,589✔

5136

    return TSDB_CODE_SUCCESS;

41,589✔

5139

  int32_t              code = TSDB_CODE_SUCCESS;

64,896,854✔

5140

  int32_t              lino = 0;

64,896,854✔

5141

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

64,896,854✔

5142

  STableMergeScanInfo* pInfo = pOperator->info;

64,908,210✔

5144

  code = pOperator->fpSet._openFn(pOperator);

64,899,992✔

5145

  QUERY_CHECK_CODE(code, lino, _end);

64,885,770✔

5147

  int64_t st = taosGetTimestampUs();

64,913,907✔

5149

  int32_t tableListSize = 0;

64,913,907✔

5150

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

64,901,030✔

5151

  QUERY_CHECK_CODE(code, lino, _end);

64,908,961✔

5153

  if (!pInfo->hasGroupId) {

64,908,961✔

5154

    pInfo->hasGroupId = true;

25,372,383✔

5156

    if (tableListSize == 0) {

25,341,790✔

5157

      setOperatorCompleted(pOperator);

2,973,198✔

5158

      (*ppRes) = NULL;

2,972,826✔

5159

      return code;

2,973,570✔

5161

    pInfo->tableStartIndex = 0;

22,368,592✔

5162

    STableKeyInfo* tmp = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

22,405,196✔

5163

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

22,373,963✔

5164

    pInfo->groupId = tmp->groupId;

22,373,963✔

5165

    startGroupTableMergeScan(pOperator);

22,400,923✔

5168

  SSDataBlock* pBlock = NULL;

61,934,963✔

5169

  while (pInfo->tableStartIndex < tableListSize) {

76,639,774✔

5170

    if (isTaskKilled(pTaskInfo)) {

76,650,026✔

5174

    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,

76,639,023✔

5176

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

76,664,140✔

5177

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,800,537✔

5178

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

1,800,537✔

5179

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

1,800,537✔

5181

    if (pBlock != NULL) {

76,661,592✔

5182

      pBlock->info.id.groupId = pInfo->groupId;

39,539,189✔

5183

      pOperator->resultInfo.totalRows += pBlock->info.rows;

39,538,690✔

5184

      pInfo->bGroupProcessed = true;

39,537,581✔

5185

      break;

39,538,817✔

5187

      if (pInfo->bNewFilesetEvent) {

37,122,403✔

5188

        stopDurationForGroupTableMergeScan(pOperator);

8,557,009✔

5189

        code = startDurationForGroupTableMergeScan(pOperator);

8,556,644✔

5190

        QUERY_CHECK_CODE(code, lino, _end);

8,556,944✔

5193

        stopGroupTableMergeScan(pOperator);

28,564,426✔

5194

        if (pInfo->tableEndIndex >= tableListSize - 1) {

28,562,601✔

5195

          setOperatorCompleted(pOperator);

22,411,240✔

5196

          break;

22,412,317✔

5199

        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

6,147,867✔

5200

        STableKeyInfo* tmp = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

6,147,867✔

5201

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

6,147,867✔

5202

        pInfo->groupId = tmp->groupId;

6,147,867✔

5203

        startGroupTableMergeScan(pOperator);

6,147,867✔

5204

        resetLimitInfoForNextGroup(&pInfo->limitInfo);

6,147,867✔

5209

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

61,943,706✔

5211

_end:

61,943,739✔

5212

  if (code != TSDB_CODE_SUCCESS) {

61,946,850✔

5217

    (*ppRes) = pBlock;

61,946,850✔

5220

  return code;

61,944,134✔

5229

void destroyTableMergeScanOperatorInfo(void* param) {

24,853,002✔

5230

  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;

24,853,002✔

5233

  if (pTableScanInfo->base.readerAPI.tsdReaderClose != NULL) {

24,853,002✔

5234

    pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);

24,851,878✔

5235

    pTableScanInfo->base.dataReader = NULL;

24,839,495✔

5238

  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {

24,843,458✔

5245

  tsortDestroySortHandle(pTableScanInfo->pSortHandle);

24,847,956✔

5246

  pTableScanInfo->pSortHandle = NULL;

24,843,377✔

5247

  taosHashCleanup(pTableScanInfo->mSkipTables);

24,851,057✔

5248

  pTableScanInfo->mSkipTables = NULL;

24,838,997✔

5249

  blockDataDestroy(pTableScanInfo->pSortInputBlock);

24,840,901✔

5250

  pTableScanInfo->pSortInputBlock = NULL;

24,847,255✔

5253

  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);

24,848,827✔

5255

  blockDataDestroy(pTableScanInfo->pResBlock);

24,849,420✔

5256

  pTableScanInfo->pResBlock = NULL;

24,849,230✔

5259

  blockDataDestroy(pTableScanInfo->pReaderBlock);

24,848,865✔

5260

  pTableScanInfo->pReaderBlock = NULL;

24,844,098✔

5261

  taosArrayDestroy(pTableScanInfo->pSortInfo);

24,846,310✔

5263

  stopSubTablesTableMergeScan(pTableScanInfo);

24,845,617✔

5265

  taosMemoryFreeClear(param);

24,836,000✔

5266

24,836,822✔

5268

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {

1,196,366✔

5269

  if (pOptr == NULL) {

1,196,366✔

5274

  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));

1,196,366✔

5275

  if (!execInfo) {

1,193,769✔

5278

  STableMergeScanInfo* pInfo = pOptr->info;

1,193,769✔

5279

  execInfo->blockRecorder = pInfo->base.readRecorder;

1,193,397✔

5280

  execInfo->sortExecInfo = pInfo->sortExecInfo;

1,194,878✔

5282

  *pOptrExplain = execInfo;

1,195,987✔

5283

  *len = sizeof(STableMergeScanExecInfo);

1,193,411✔

5285

  return TSDB_CODE_SUCCESS;

1,193,432✔

5288

static int32_t resetTableMergeScanOperatorState(SOperatorInfo* pOper) {

2,393,282✔

5289

  int32_t         code = TSDB_CODE_SUCCESS;

2,393,282✔

5290

  STableMergeScanInfo*   pInfo = pOper->info;

2,393,282✔

5291

  pOper->status = OP_NOT_OPENED;

2,393,282✔

5293

  pInfo->tableEndIndex = 0;

2,393,282✔

5294

  pInfo->tableStartIndex = 0;

2,393,282✔

5295

  pInfo->hasGroupId = false;

2,393,282✔

5296

  if (pInfo->base.readerAPI.tsdReaderClose) {

2,393,282✔

5297

    pInfo->base.readerAPI.tsdReaderClose(pInfo->base.dataReader);

2,393,282✔

5299

  pInfo->base.dataReader = NULL;

2,393,282✔

5300

  pInfo->base.scanFlag = MAIN_SCAN;

2,393,282✔

5302

  pInfo->base.limitInfo = (SLimitInfo){0};

2,393,282✔

5303

  pInfo->base.limitInfo.limit.limit = -1;

2,393,282✔

5304

  pInfo->base.limitInfo.slimit.limit = -1;

2,393,282✔

5306

  tableListDestroy(pInfo->base.pTableListInfo);

2,393,282✔

5307

  pInfo->base.pTableListInfo = tableListCreate();

2,393,282✔

5308

  if (!pInfo->base.pTableListInfo) {

2,393,044✔

5312

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

2,393,282✔

5314

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

2,393,282✔

5315

  code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

4,786,564✔

5317

                                  pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

4,786,564✔

5318

  if (code) {

2,393,282✔

5323

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

2,393,282✔

5324

  cleanupQueryTableDataCond(&pInfo->base.cond);

2,393,282✔

5325

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, &pInfo->base.readHandle, false);

2,393,282✔

5326

  if (code) {

2,393,282✔

5331

  tsortDestroySortHandle(pInfo->pSortHandle);

2,393,282✔

5332

  pInfo->pSortHandle = NULL;

2,393,282✔

5334

  taosHashCleanup(pInfo->mSkipTables);

2,393,282✔

5335

  pInfo->mSkipTables = NULL;

2,393,282✔

5336

  pOper->resultInfo.totalRows = 0;

2,393,282✔

5338

  pInfo->bGroupProcessed = false;

2,393,282✔

5339

  pInfo->bNewFilesetEvent = false;

2,393,282✔

5340

  pInfo->bNextDurationBlockEvent = false;

2,393,282✔

5341

  pInfo->rtnNextDurationBlocks = false;

2,393,282✔

5342

  pInfo->nextDurationBlocksIdx = 0;

2,393,282✔

5343

  pInfo->bSortRowId = false;

2,393,282✔

5345

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

2,393,282✔

5351

  pInfo->numNextDurationBlocks = 0;

2,393,282✔

5353

  stopSubTablesTableMergeScan(pInfo);

2,393,282✔

5354

  return code;

2,393,282✔

5357

int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

24,800,547✔

5360

  QRY_PARAM_CHECK(pOptrInfo);

24,800,547✔

5362

  int32_t              code = TSDB_CODE_SUCCESS;

24,817,173✔

5363

  int32_t              lino = 0;

24,817,173✔

5364

  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));

24,817,173✔

5365

  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

24,788,288✔

5366

  if (pInfo == NULL || pOperator == NULL) {

24,751,024✔

5371

  pOperator->pPhyNode = pTableScanNode;

24,752,261✔

5372

  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

24,774,698✔

5374

  int32_t numOfCols = 0;

24,778,079✔

5375

  code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,

24,781,812✔

5377

  QUERY_CHECK_CODE(code, lino, _error);

24,784,785✔

5379

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle, false);

24,784,785✔

5380

  QUERY_CHECK_CODE(code, lino, _error);

24,761,589✔

5382

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

24,761,589✔

5383

    SExprSupp* pSup = &pInfo->base.pseudoSup;

10,622,441✔

5384

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->pExprInfo, &pSup->numOfExprs);

10,612,639✔

5385

    QUERY_CHECK_CODE(code, lino, _error);

10,609,480✔

5387

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

10,609,480✔

5389

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

10,585,290✔

5392

  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};

24,723,982✔

5394

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

24,689,687✔

5399

  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;

24,689,687✔

5400

  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

24,787,143✔

5401

  pInfo->base.scanFlag = MAIN_SCAN;

24,682,294✔

5402

  pInfo->base.readHandle = *readHandle;

24,723,426✔

5404

  pInfo->readIdx = -1;

24,798,518✔

5406

  pInfo->base.limitInfo.limit.limit = -1;

24,672,831✔

5407

  pInfo->base.limitInfo.slimit.limit = -1;

24,689,121✔

5408

  pInfo->base.pTableListInfo = pTableListInfo;

24,733,855✔

5410

  pInfo->sample.sampleRatio = pTableScanNode->ratio;

24,797,049✔

5411

  pInfo->sample.seed = taosGetTimestampSec();

24,681,752✔

5413

  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,

24,796,202✔

5414

                            pTaskInfo->pStreamRuntimeInfo);

24,784,195✔

5415

  QUERY_CHECK_CODE(code, lino, _error);

24,757,898✔

5417

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

24,757,898✔

5419

  pInfo->mergeLimit = -1;

24,733,327✔

5420

  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;

24,792,761✔

5421

  if (hasLimit) {

24,701,326✔

5422

    pInfo->mergeLimit = pInfo->limitInfo.limit.offset != -1

4,080,203✔

5423

                            ? pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset

812,980✔

5424

                            : pInfo->limitInfo.limit.limit;

2,850,978✔

5425

    pInfo->mSkipTables = NULL;

2,040,101✔

5428

  initResultSizeInfo(&pOperator->resultInfo, 1024);

24,698,661✔

5429

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

24,765,459✔

5430

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

24,792,824✔

5431

  code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

24,730,099✔

5432

  QUERY_CHECK_CODE(code, lino, _error);

24,838,029✔

5433

  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {

24,838,029✔

5434

    pInfo->bSortRowId = true;

5,217,620✔

5436

    pInfo->bSortRowId = false;

19,611,628✔

5439

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

24,808,053✔

5440

  QUERY_CHECK_CODE(code, lino, _error);

24,797,439✔

5442

  code = generateSortByTsPkInfo(pInfo->base.matchInfo.pList, pInfo->base.cond.order, &pInfo->pSortInfo);

24,797,439✔

5443

  QUERY_CHECK_CODE(code, lino, _error);

24,792,876✔

5445

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pReaderBlock);

24,792,876✔

5446

  QUERY_CHECK_CODE(code, lino, _error);

24,802,410✔

5448

  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;

24,802,410✔

5450

  int32_t  rowSize = pInfo->pResBlock->info.rowSize;

24,830,393✔

5451

  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);

24,806,901✔

5453

  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);

24,737,640✔

5456

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pSortInputBlock);

24,811,966✔

5457

  QUERY_CHECK_CODE(code, lino, _error);

24,819,474✔

5459

  if (!tsExperimental) {

24,819,474✔

5462

    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

24,819,474✔

5465

  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,

24,810,381✔

5467

  pOperator->exprSupp.numOfExprs = numOfCols;

24,784,174✔

5469

  pOperator->fpSet = createOperatorFpSet(

24,772,577✔

5470

      optrDummyOpenFn, pTableScanNode->paraTablesSort ? doTableMergeScanParaSubTablesNext : doTableMergeScanNext, NULL,

24,817,174✔

5473

  setOperatorResetStateFn(pOperator, resetTableMergeScanOperatorState);

24,763,285✔

5474

  pOperator->cost.openCost = 0;

24,800,919✔

5476

  *pOptrInfo = pOperator;

24,810,925✔

5477

  return code;

24,733,861✔

5516

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {

25,102✔

5517

  if (scanCols != NULL) {

25,102✔

5518

    SNode* pNode = NULL;

14,430✔

5519

    FOREACH(pNode, scanCols) {

33,696✔

5520

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

19,266✔

5523

      STargetNode* targetNode = (STargetNode*)pNode;

19,266✔

5524

      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {

19,266✔

5527

      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);

19,266✔

5528

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

19,266✔

5529

        supp->dbNameSlotId = targetNode->slotId;

9,502✔

5530

      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

9,764✔

5531

        supp->stbNameSlotId = targetNode->slotId;

9,764✔

5535

  return TSDB_CODE_SUCCESS;

25,102✔

5538

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {

25,102✔

5539

  if (pseudoCols != NULL) {

25,102✔

5540

    SNode* pNode = NULL;

25,102✔

5541

    FOREACH(pNode, pseudoCols) {

50,204✔

5542

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

25,102✔

5545

      STargetNode* targetNode = (STargetNode*)pNode;

25,102✔

5546

      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {

25,102✔

5549

      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);

25,102✔

5550

      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {

25,102✔

5551

        supp->tbCountSlotId = targetNode->slotId;

25,102✔

5555

  return TSDB_CODE_SUCCESS;

25,102✔

5558

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {

25,102✔

5559

  if (groupTags != NULL) {

25,102✔

5560

    SNode* pNode = NULL;

14,430✔

5561

    FOREACH(pNode, groupTags) {

33,696✔

5562

      if (nodeType(pNode) != QUERY_NODE_COLUMN) {

19,266✔

5565

      SColumnNode* colNode = (SColumnNode*)pNode;

19,266✔

5566

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

19,266✔

5567

        supp->groupByDbName = true;

9,502✔

5569

      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

19,266✔

5570

        supp->groupByStbName = true;

9,764✔

5574

    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);

10,672✔

5575

    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);

10,672✔

5577

  return TSDB_CODE_SUCCESS;

25,102✔

5580

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,

25,102✔

5582

  int32_t code = 0;

25,102✔

5583

  code = tblCountScanGetInputs(groupTags, tableName, supp);

25,102✔

5584

  if (code != TSDB_CODE_SUCCESS) {

25,102✔

5589

  supp->dbNameSlotId = -1;

25,102✔

5590

  supp->stbNameSlotId = -1;

25,102✔

5591

  supp->tbCountSlotId = -1;

25,102✔

5593

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);

25,102✔

5594

  if (code != TSDB_CODE_SUCCESS) {

25,102✔

5599

  code = tblCountScanGetCountSlotId(pseudoCols, supp);

25,102✔

5600

  if (code != TSDB_CODE_SUCCESS) {

25,102✔

5604

  return code;

25,102✔

5607

int32_t createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,

25,102✔

5609

  QRY_PARAM_CHECK(pOptrInfo);

25,102✔

5611

  int32_t                      code = TSDB_CODE_SUCCESS;

25,102✔

5612

  int32_t                      lino = 0;

25,102✔

5613

  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;

25,102✔

5614

  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));

25,102✔

5615

  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

25,102✔

5616

  if (!pInfo || !pOperator) {

25,102✔

5621

  pInfo->readHandle = *readHandle;

25,102✔

5623

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

25,102✔

5624

  initResultSizeInfo(&pOperator->resultInfo, 1);

25,102✔

5625

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

25,102✔

5626

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

25,102✔

5628

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

25,102✔

5629

  QUERY_CHECK_CODE(code, lino, _error);

25,102✔

5631

  code = getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,

25,102✔

5634

  QUERY_CHECK_CODE(code, lino, _error);

25,102✔

5636

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,

25,102✔

5638

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScanNext, NULL, destoryTableCountScanOperator,

25,102✔

5640

  *pOptrInfo = pOperator;

25,102✔

5641

  return code;

25,102✔

5655

int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,

34,982✔

5657

  int32_t code = TSDB_CODE_SUCCESS;

34,982✔

5658

  int32_t lino = 0;

34,982✔

5659

  if (pSupp->dbNameSlotId != -1) {

34,982✔

5660

    QUERY_CHECK_CONDITION((strlen(dbName) > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

14,338✔

5661

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);

14,338✔

5662

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

14,338✔

5664

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

14,338✔

5665

    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

14,338✔

5667

    varDataSetLen(varDbName, strlen(dbName));

14,189✔

5668

    code = colDataSetVal(colInfoData, 0, varDbName, false);

14,338✔

5669

    QUERY_CHECK_CODE(code, lino, _end);

14,338✔

5672

  if (pSupp->stbNameSlotId != -1) {

34,982✔

5673

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);

18,694✔

5674

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

18,524✔

5675

    if (strlen(stbName) != 0) {

18,524✔

5676

      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

13,428✔

5677

      tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);

13,428✔

5678

      varDataSetLen(varStbName, strlen(stbName));

13,258✔

5679

      code = colDataSetVal(colInfoData, 0, varStbName, false);

13,428✔

5680

      QUERY_CHECK_CODE(code, lino, _end);

13,428✔

5686

  if (pSupp->tbCountSlotId != -1) {

34,982✔

5687

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);

34,982✔

5688

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

34,982✔

5689

    code = colDataSetVal(colInfoData, 0, (char*)&count, false);

34,982✔

5690

    QUERY_CHECK_CODE(code, lino, _end);

34,982✔

5692

  pRes->info.rows = 1;

34,982✔

5694

_end:

34,982✔

5695

  if (code != TSDB_CODE_SUCCESS) {

34,982✔

5698

  return code;

34,812✔

5701

static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {

10,610✔

5702

  STableCountScanSupp* pSupp = &pInfo->supp;

10,610✔

5703

  SSDataBlock*         pRes = pInfo->pRes;

10,610✔

5705

  size_t infodbTableNum;

10,610✔

5706

  getInfosDbMeta(NULL, &infodbTableNum);

10,610✔

5707

  infodbTableNum -= 1;

10,610✔

5708

  size_t perfdbTableNum;

10,610✔

5709

  getPerfDbMeta(NULL, &perfdbTableNum);

10,610✔

5711

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

10,610✔

5712

    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);

9,108✔

5713

    return (pRes->info.rows > 0) ? pRes : NULL;

9,108✔

5715

    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);

1,502✔

5716

    return (pRes->info.rows > 0) ? pRes : NULL;

1,502✔

5720

static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,

1,502✔

5722

  int32_t        code = TSDB_CODE_SUCCESS;

1,502✔

5723

  int32_t        lino = 0;

1,502✔

5724

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,502✔

5725

  if (strlen(pSupp->stbNameFilter) != 0) {

1,502✔

5727

    code = fillTableCountScanDataBlock(pSupp, "", "", 0, pRes);

510✔

5728

    QUERY_CHECK_CODE(code, lino, _end);

510✔

5729

  } else if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {

992✔

5730

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

170✔

5731

    QUERY_CHECK_CODE(code, lino, _end);

170✔

5732

  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

822✔

5735

  } else if (strlen(pSupp->dbNameFilter) == 0) {

822✔

5736

    code = fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);

652✔

5737

    QUERY_CHECK_CODE(code, lino, _end);

652✔

5740

    code = fillTableCountScanDataBlock(pSupp, pSupp->dbNameFilter, pSupp->stbNameFilter, 0, pRes);

170✔

5741

    QUERY_CHECK_CODE(code, lino, _end);

170✔

5744

_end:

170✔

5745

  if (code != TSDB_CODE_SUCCESS) {

1,502✔

5750

  setOperatorCompleted(pOperator);

1,502✔

5751

1,502✔

5753

static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,108✔

5756

  int32_t        code = TSDB_CODE_SUCCESS;

9,108✔

5757

  int32_t        lino = 0;

9,108✔

5758

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,108✔

5759

  if (pInfo->currGrpIdx == 0) {

9,108✔

5760

    uint64_t groupId = 0;

3,036✔

5761

    if (pSupp->groupByDbName) {

3,036✔

5762

      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));

2,070✔

5764

      groupId = calcGroupId("", 0);

966✔

5767

    pRes->info.id.groupId = groupId;

3,036✔

5768

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

3,036✔

5769

    QUERY_CHECK_CODE(code, lino, _end);

3,036✔

5770

  } else if (pInfo->currGrpIdx == 1) {

6,072✔

5771

    uint64_t groupId = 0;

3,036✔

5772

    if (pSupp->groupByDbName) {

3,036✔

5773

      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));

2,070✔

5775

      groupId = calcGroupId("", 0);

966✔

5778

    pRes->info.id.groupId = groupId;

3,036✔

5779

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

3,036✔

5780

    QUERY_CHECK_CODE(code, lino, _end);

3,036✔

5782

    setOperatorCompleted(pOperator);

3,036✔

5784

  pInfo->currGrpIdx++;

9,108✔

5786

_end:

9,108✔

5787

  if (code != TSDB_CODE_SUCCESS) {

9,108✔

5792

9,108✔

5794

static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

60,084✔

5795

  int32_t                      code = TSDB_CODE_SUCCESS;

60,084✔

5796

  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;

60,084✔

5797

  STableCountScanOperatorInfo* pInfo = pOperator->info;

60,084✔

5798

  STableCountScanSupp*         pSupp = &pInfo->supp;

60,084✔

5799

  SSDataBlock*                 pRes = pInfo->pRes;

60,084✔

5801

  blockDataCleanup(pRes);

60,084✔

5802

  QRY_PARAM_CHECK(ppRes);

60,084✔

5804

  if (pOperator->status == OP_EXEC_DONE) {

60,084✔

5805

    return code;

14,388✔

5808

  if (pInfo->readHandle.mnd != NULL) {

45,696✔

5809

    (*ppRes) = buildSysDbTableCount(pOperator, pInfo);

10,610✔

5810

    return code;

10,610✔

5813

  code = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);

35,086✔

5814

  if (code != TSDB_CODE_SUCCESS) {

34,916✔

5819

  if (pRes->info.rows > 0) {

34,916✔

5820

    *ppRes = pRes;

27,238✔

5823

  return code;

35,086✔

5826

static int32_t buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

35,086✔

5828

  int32_t        code = TSDB_CODE_SUCCESS;

35,086✔

5829

  int32_t        lino = 0;

35,086✔

5830

  const char*    db = NULL;

35,086✔

5831

  int32_t        vgId = 0;

35,086✔

5832

  char           dbName[TSDB_DB_NAME_LEN] = {0};

35,086✔

5833

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

35,086✔

5834

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

35,086✔

5837

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);

35,086✔

5838

  SName sn = {0};

35,086✔

5840

  code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

35,086✔

5841

  QUERY_CHECK_CODE(code, lino, _end);

35,086✔

5843

  code = tNameGetDbName(&sn, dbName);

35,086✔

5844

  QUERY_CHECK_CODE(code, lino, _end);

35,086✔

5846

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

35,086✔

5847

    code = buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);

25,916✔

5849

    code = buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);

9,170✔

5852

_end:

35,086✔

5853

  if (code != TSDB_CODE_SUCCESS) {

35,086✔

5858

  return code;

35,086✔

5861

static int32_t buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

25,916✔

5863

  int32_t        code = TSDB_CODE_SUCCESS;

25,916✔

5864

  int32_t        lino = 0;

25,916✔

5865

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

25,916✔

5866

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

25,916✔

5868

  if (pSupp->groupByStbName) {

25,916✔

5869

    if (pInfo->stbUidList == NULL) {

22,200✔

5870

      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));

7,529✔

5871

      QUERY_CHECK_NULL(pInfo->stbUidList, code, lino, _end, terrno);

7,678✔

5872

      code = pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList);

7,678✔

5873

      QUERY_CHECK_CODE(code, lino, _end);

7,678✔

5876

    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {

22,349✔

5877

      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);

13,428✔

5878

      code = buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);

13,428✔

5879

      QUERY_CHECK_CODE(code, lino, _end);

13,428✔

5881

      pInfo->currGrpIdx++;

13,428✔

5882

    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {

8,772✔

5883

      code = buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);

7,678✔

5884

      QUERY_CHECK_CODE(code, lino, _end);

7,678✔

5886

      pInfo->currGrpIdx++;

7,678✔

5888

      setOperatorCompleted(pOperator);

1,094✔

5891

    uint64_t groupId = calcGroupId(dbName, strlen(dbName));

3,716✔

5892

    pRes->info.id.groupId = groupId;

3,716✔

5894

    int64_t dbTableCount = 0;

3,716✔

5895

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);

3,716✔

5896

    code = fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);

3,716✔

5897

    QUERY_CHECK_CODE(code, lino, _end);

3,716✔

5898

    setOperatorCompleted(pOperator);

3,716✔

5901

_end:

25,916✔

5902

  if (code != TSDB_CODE_SUCCESS) {

25,916✔

5906

  return code;

25,916✔

5909

static int32_t buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,170✔

5911

  int32_t        code = TSDB_CODE_SUCCESS;

9,170✔

5912

  int32_t        lino = 0;

9,170✔

5913

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,170✔

5914

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

9,170✔

5916

  if (strlen(pSupp->dbNameFilter) != 0 && strcmp(dbName, pSupp->dbNameFilter) != 0) {

9,170✔

5920

  } else if (strlen(pSupp->stbNameFilter) != 0) {

9,170✔

5921

    uint64_t uid = 0;

3,004✔

5922

    code = pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid);

3,004✔

5923

    if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

3,004✔

5925

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, 0, pRes);

1,360✔

5926

      QUERY_CHECK_CODE(code, lino, _end);

1,360✔

5928

      QUERY_CHECK_CODE(code, lino, _end);

1,644✔

5930

      int64_t numOfChildTables = 0;

1,644✔

5931

      code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL, NULL);

1,644✔

5932

      QUERY_CHECK_CODE(code, lino, _end);

1,644✔

5934

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes);

1,644✔

5935

      QUERY_CHECK_CODE(code, lino, _end);

1,644✔

5938

    int64_t tbNumVnode = 0;

6,166✔

5939

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);

6,166✔

5940

    code = fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);

6,166✔

5941

    QUERY_CHECK_CODE(code, lino, _end);

6,166✔

5944

_end:

9,170✔

5945

  if (code != TSDB_CODE_SUCCESS) {

9,170✔

5950

  setOperatorCompleted(pOperator);

9,170✔

5951

  return code;

9,170✔

5954

static int32_t buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

7,678✔

5956

  int32_t code = TSDB_CODE_SUCCESS;

7,678✔

5957

  int32_t lino = 0;

7,678✔

5958

  char    fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

7,678✔

5959

  if (pSupp->groupByDbName) {

7,678✔

5960

    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");

3,716✔

5963

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

7,678✔

5964

  pRes->info.id.groupId = groupId;

7,678✔

5966

  int64_t numOfTables = 0;

7,678✔

5967

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);

7,678✔

5969

  if (numOfTables != 0) {

7,678✔

5970

    code = fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);

1,094✔

5973

_end:

6,584✔

5974

  if (code != TSDB_CODE_SUCCESS) {

7,678✔

5977

  return code;

7,678✔

5980

static int32_t buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

13,428✔

5982

  int32_t code = TSDB_CODE_SUCCESS;

13,428✔

5983

  int32_t lino = 0;

13,428✔

5984

  char    stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

13,428✔

5985

  code = pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);

13,428✔

5986

  QUERY_CHECK_CODE(code, lino, _end);

13,428✔

5988

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

13,428✔

5989

  if (pSupp->groupByDbName) {

13,428✔

5990

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));

6,184✔

5992

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));

7,244✔

5995

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

13,428✔

5996

  pRes->info.id.groupId = groupId;

13,279✔

5998

  int64_t ctbNum = 0;

13,428✔

5999

  code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL, NULL);

13,428✔

6000

  QUERY_CHECK_CODE(code, lino, _end);

13,428✔

6001

  code = fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);

13,428✔

6002

  QUERY_CHECK_CODE(code, lino, _end);

13,258✔

6004

_end:

13,258✔

6005

  if (code != TSDB_CODE_SUCCESS) {

13,258✔

6008

  return code;

13,258✔

6011

static void destoryTableCountScanOperator(void* param) {

25,102✔

6012

  STableCountScanOperatorInfo* pTableCountScanInfo = param;

25,102✔

6013

  blockDataDestroy(pTableCountScanInfo->pRes);

25,102✔

6015

  taosArrayDestroy(pTableCountScanInfo->stbUidList);

25,102✔

6016

  taosMemoryFreeClear(param);

25,102✔

6017

25,102✔

taosdata / TDengine / #4947

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4947

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