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

76

bool processBlockWithProbability(const SSampleExecInfo* pInfo) {

538,685,187✔

85

  return true;

538,685,187✔

89

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

67,867✔

90

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

67,867✔

93

67,867✔

95

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

583,941,364✔

96

  int32_t     code = TSDB_CODE_SUCCESS;

583,941,364✔

97

  STimeWindow w = {0};

583,941,364✔

100

  if (pInterval->interval == 0) {

584,040,490✔

101

    *overlap = false;

547,781,165✔

102

    return code;

547,806,966✔

105

  if (order == TSDB_ORDER_ASC) {

36,193,080✔

106

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

34,980,224✔

107

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

34,986,822✔

112

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

34,974,634✔

113

      *overlap = true;

30,452,433✔

114

      return code;

30,451,064✔

118

      getNextTimeWindow(pInterval, &w, order);

4,531,649✔

119

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

4,536,493✔

120

        break;

4,266,486✔

123

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

270,007✔

127

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

270,007✔

128

        *overlap = true;

270,007✔

129

        return code;

270,007✔

133

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

1,212,856✔

134

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

1,213,845✔

139

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

1,213,212✔

140

      *overlap = true;

1,044,225✔

141

      return code;

1,044,225✔

145

      getNextTimeWindow(pInterval, &w, order);

169,620✔

146

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

169,620✔

147

        break;

166,968✔

150

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

2,652✔

154

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

2,652✔

155

        *overlap = true;

2,652✔

156

        return code;

2,652✔

161

  *overlap = false;

4,433,454✔

162

  return code;

4,432,908✔

166

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

142,563,004✔

167

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

142,563,004✔

171

  int64_t buf[2] = {0};

142,580,014✔

172

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

142,561,141✔

174

  STableScanInfo* pTableScanInfo = pOperator->info;

142,563,373✔

176

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

142,553,697✔

179

  if (p1 == NULL) {

142,589,626✔

180

    return NULL;

48,871,079✔

183

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

93,718,547✔

184

  if (NULL == *pPage) {

93,711,482✔

188

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

93,713,356✔

191

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

11,679,131✔

192

  if (NULL == pTableScanInfo->pIgnoreTables) {

11,679,131✔

193

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

1,136,731✔

194

    pTableScanInfo->pIgnoreTables =

1,136,731✔

195

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

1,136,731✔

196

    if (NULL == pTableScanInfo->pIgnoreTables) {

1,136,731✔

201

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

11,679,131✔

203

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

11,679,575✔

207

  return TSDB_CODE_SUCCESS;

11,679,575✔

210

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

648,038,225✔

211

  STableScanInfo* pTableScanInfo = pOperator->info;

648,038,225✔

212

  int32_t         code = TSDB_CODE_SUCCESS;

648,217,462✔

214

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

648,217,462✔

215

    return TSDB_CODE_SUCCESS;

505,648,233✔

218

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

142,567,493✔

220

  SFilePage*  pPage = NULL;

142,591,764✔

221

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

142,584,647✔

223

  if (pRow == NULL) {

142,564,664✔

224

    return TSDB_CODE_SUCCESS;

48,866,087✔

227

  bool notLoadBlock = true;

93,698,577✔

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;

82,037,405✔

236

      break;

82,037,405✔

241

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

93,545,788✔

243

  if (notLoadBlock) {

93,714,453✔

244

    *status = FUNC_DATA_REQUIRED_NOT_LOAD;

11,680,387✔

245

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

11,680,387✔

248

  return code;

93,705,280✔

251

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

9,220,751✔

253

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

9,220,751✔

258

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

9,220,891✔

261

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

127,638,914✔

263

  SStorageAPI* pAPI = &pTaskInfo->storageAPI;

127,638,914✔

264

  bool         allColumnsHaveAgg = true;

127,661,561✔

265

  bool         hasNullSMA = false;

127,668,097✔

266

  if (pLoad != NULL) {

127,660,852✔

267

    *pLoad = false;

127,633,181✔

270

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

127,653,849✔

272

  if (code != TSDB_CODE_SUCCESS) {

127,629,853✔

276

  if (!allColumnsHaveAgg || hasNullSMA) {

127,629,853✔

277

    *pLoad = false;

112,874,645✔

279

    *pLoad = true;

14,755,208✔

282

  return code;

127,637,707✔

285

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

681,506,932✔

287

  int32_t    code = 0;

681,506,932✔

288

  SExprSupp* pSup = &pTableScanInfo->pseudoSup;

681,506,932✔

289

  if (pSup->numOfExprs > 0) {

681,681,802✔

290

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

321,674,091✔

293

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

321,580,965✔

299

  return code;

681,533,078✔

302

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

1,132,102,685✔

303

  SLimit*     pLimit = &pLimitInfo->limit;

1,132,102,685✔

304

  const char* id = GET_TASKID(pTaskInfo);

1,132,181,267✔

306

  if (pLimitInfo->remainOffset > 0) {

1,132,250,026✔

307

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

5,541,691✔

308

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

2,692,093✔

309

      blockDataEmpty(pBlock);

2,692,093✔

310

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

2,692,093✔

311

      return false;

2,692,093✔

313

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

2,849,598✔

314

      if (code != TSDB_CODE_SUCCESS) {

2,849,598✔

319

      pLimitInfo->remainOffset = 0;

2,849,598✔

323

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

1,129,548,070✔

325

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

26,583,276✔

326

    blockDataKeepFirstNRows(pBlock, keep);

26,586,821✔

328

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

26,584,129✔

329

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

26,585,753✔

330

    return true;

26,586,298✔

333

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

1,102,826,205✔

334

  return false;

1,102,939,175✔

341

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

677,415,953✔

343

  int32_t        code = TSDB_CODE_SUCCESS;

677,415,953✔

344

  int32_t        lino = 0;

677,415,953✔

345

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

677,415,953✔

346

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

677,568,295✔

347

  bool           loadSMA = false;

677,536,467✔

349

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

677,536,467✔

351

  pCost->totalBlocks += 1;

677,626,496✔

352

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

677,490,861✔

353

  *status = pTableScanInfo->dataBlockLoadFlag;

677,572,860✔

355

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

677,518,781✔

356

    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

93,430,844✔

358

    bool overlap = false;

584,038,336✔

360

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

584,115,096✔

361

    if (ret != TSDB_CODE_SUCCESS) {

583,859,319✔

364

    if (overlap) {

583,859,319✔

365

      (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

31,765,310✔

369

  SDataBlockInfo* pBlockInfo = &pBlock->info;

677,297,207✔

370

  taosMemoryFreeClear(pBlock->pBlockAgg);

677,483,651✔

372

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {

677,564,746✔

379

  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

677,350,855✔

380

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

20,374,545✔

383

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

20,377,628✔

384

    pCost->skipBlocks += 1;

20,376,451✔

385

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

20,374,421✔

386

    return code;

20,372,198✔

387

  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {

656,962,160✔

388

    pCost->loadBlockStatis += 1;

34,215,749✔

389

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

34,207,047✔

390

    bool success = true;

34,207,047✔

391

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

34,211,382✔

392

    if (code) {

34,202,079✔

398

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

34,194,508✔

399

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

5,535,140✔

401

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

5,535,140✔

402

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

5,535,000✔

403

      return code;

5,535,000✔

405

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

28,659,368✔

406

      *status = FUNC_DATA_REQUIRED_DATA_LOAD;

28,670,382✔

410

  if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {

651,425,678✔

417

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

651,430,963✔

418

    bool success = true;

93,447,540✔

419

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

93,453,461✔

420

    if (code) {

93,416,604✔

423

      QUERY_CHECK_CODE(code, lino, _end);

8,752✔

426

    if (success) {

93,425,356✔

427

      size_t size = taosArrayGetSize(pBlock->pDataBlock);

9,220,891✔

428

      bool   keep = false;

9,220,628✔

429

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

9,220,768✔

430

      if (code) {

9,220,891✔

436

      if (!keep) {

9,220,331✔

437

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

3,377,440✔

439

        pCost->filterOutBlocks += 1;

3,377,440✔

440

        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

3,377,440✔

441

        taosMemoryFreeClear(pBlock->pBlockAgg);

3,377,440✔

443

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

3,377,440✔

444

        return TSDB_CODE_SUCCESS;

3,377,440✔

450

  taosMemoryFreeClear(pBlock->pBlockAgg);

648,180,086✔

453

  code = doDynamicPruneDataBlock(pOperator, pBlockInfo, status);

648,068,656✔

454

  if (code) {

648,091,082✔

459

  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

648,091,082✔

460

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

11,677,875✔

462

    pCost->skipBlocks += 1;

11,679,981✔

463

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

11,680,793✔

465

    STableScanInfo* p1 = pOperator->info;

11,680,793✔

466

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

11,680,793✔

467

      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;

2,790,104✔

469

      *status = FUNC_DATA_REQUIRED_FILTEROUT;

8,890,689✔

471

    return TSDB_CODE_SUCCESS;

11,680,793✔

474

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

636,350,135✔

475

  pCost->loadBlocks += 1;

636,565,277✔

477

  SSDataBlock* p = NULL;

636,420,719✔

478

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

636,430,034✔

479

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

636,458,475✔

483

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

636,458,475✔

484

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

499,603,178✔

488

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

608,669,254✔

489

    if (code) {

608,887,280✔

495

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

636,437,268✔

497

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

636,480,714✔

498

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

90,046,022✔

499

    QUERY_CHECK_CODE(code, lino, _end);

90,019,543✔

501

    int64_t st = taosGetTimestampUs();

89,904,811✔

502

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

89,904,541✔

503

    pTableScanInfo->readRecorder.filterTime += el;

89,904,541✔

505

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

89,891,754✔

506

      pCost->filterOutBlocks += 1;

15,999,679✔

507

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

16,002,672✔

510

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

73,882,855✔

514

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

636,334,294✔

515

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

636,324,888✔

516

    setOperatorCompleted(pOperator);

3,453,646✔

519

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

636,324,888✔

521

_end:

636,515,642✔

522

  if (code != TSDB_CODE_SUCCESS) {

636,537,451✔

523

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

132,432✔

525

  return code;

636,383,852✔

528

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

67,867✔

529

  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

67,867✔

531

  switchCtxOrder(pCtx, numOfOutput);

67,867✔

532

  pTableScanInfo->cond.order = TSDB_ORDER_DESC;

67,867✔

535

67,867✔

603

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

388,374,193✔

605

  int32_t          code = TSDB_CODE_SUCCESS;

388,374,193✔

606

  int32_t          lino = 0;

388,374,193✔

607

  bool             freeReader = false;

388,374,193✔

608

  LRUHandle*       h = NULL;

388,374,193✔

609

  STableCachedVal  val = {0};

388,374,193✔

610

  SMetaReader      mr = {0};

388,374,193✔

611

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

388,362,130✔

612

  int32_t          insertRet = TAOS_LRU_STATUS_OK;

388,387,650✔

613

  STableCachedVal* pVal = NULL;

388,387,650✔

616

  if (numOfExpr <= 0) {

388,402,601✔

617

    return TSDB_CODE_SUCCESS;

533,090✔

623

  int32_t backupRows = pBlock->info.rows;

387,869,511✔

624

  pBlock->info.rows = rows;

387,920,304✔

630

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

387,889,940✔

631

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

387,886,273✔

632

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

387,940,268✔

633

    if (code != TSDB_CODE_SUCCESS) {

387,732,250✔

648

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

387,732,250✔

650

    val.pName = mr.me.name;

387,889,036✔

651

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

387,889,036✔

653

    freeReader = true;

387,889,036✔

695

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

1,527,657,325✔

696

    const SExprInfo* pExpr1 = &pExpr[j];

1,139,769,281✔

697

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

1,139,729,925✔

699

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

1,139,906,504✔

700

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

1,139,876,100✔

702

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

1,139,852,715✔

705

    if (fmIsScanPseudoColumnFunc(functionId)) {

1,139,943,817✔

706

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

128,461,615✔

707

      if (fType == FUNCTION_TYPE_TBNAME) {

128,475,220✔

708

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

128,475,220✔

709

        QUERY_CHECK_CODE(code, lino, _end);

128,456,217✔

718

      STagVal tagVal = {0};

1,011,297,142✔

719

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

1,011,278,856✔

720

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

1,011,453,757✔

722

      char* data = NULL;

1,011,439,032✔

723

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

1,011,439,032✔

724

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

1,005,099,787✔

726

        data = (char*)p;

6,369,763✔

729

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

1,011,331,436✔

730

      if (isNullVal) {

1,011,376,777✔

731

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

5,874,429✔

732

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

1,005,502,348✔

733

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

1,004,912,196✔

734

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

1,005,071,208✔

735

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

377,607,706✔

736

          if (tmp != NULL) {

377,651,097✔

737

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

377,651,097✔

738

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

377,647,712✔

740

            taosMemoryFree(tmp);

377,701,175✔

742

          taosMemoryFree(data);

377,695,857✔

744

        QUERY_CHECK_CODE(code, lino, _end);

1,004,981,686✔

746

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

3,045,014✔

747

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

2,549,680✔

748

          QUERY_CHECK_CODE(code, lino, _end);

2,549,680✔

755

  pBlock->info.rows = backupRows;

387,888,044✔

757

_end:

387,925,562✔

759

  if (NULL != pVal) {

387,869,911✔

768

  if (freeReader) {

387,869,911✔

769

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

387,871,829✔

771

  if (code != TSDB_CODE_SUCCESS) {

387,907,574✔

774

  return code;

387,860,213✔

777

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

128,443,510✔

779

  int32_t                     code = TSDB_CODE_SUCCESS;

128,443,510✔

780

  int32_t                     lino = 0;

128,443,510✔

781

  struct SScalarFuncExecFuncs fpSet = {0};

128,443,510✔

782

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

128,453,896✔

783

  QUERY_CHECK_CODE(code, lino, _end);

128,449,615✔

785

  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;

128,449,615✔

786

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

128,449,615✔

787

  STR_TO_VARSTR(buf, name)

128,439,495✔

789

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

128,443,815✔

791

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

128,436,202✔

792

  QUERY_CHECK_CODE(code, lino, _end);

128,440,382✔

794

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

128,440,382✔

795

  QUERY_CHECK_CODE(code, lino, _end);

128,455,275✔

797

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

128,455,275✔

798

  SScalarParam param = {.columnData = pColInfoData};

128,452,500✔

800

  if (fpSet.process != NULL) {

128,427,564✔

801

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

128,427,564✔

802

    QUERY_CHECK_CODE(code, lino, _end);

128,460,392✔

807

  colDataDestroy(&infoData);

128,460,392✔

809

_end:

128,440,862✔

810

  if (code != TSDB_CODE_SUCCESS) {

128,466,168✔

813

  return code;

128,466,168✔

882

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

180,207,097✔

883

  int32_t code = TSDB_CODE_SUCCESS;

180,207,097✔

884

  int32_t lino = 0;

180,207,097✔

885

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

180,207,097✔

886

  QUERY_CHECK_CODE(code, lino, _end);

180,215,267✔

888

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

180,215,267✔

889

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

180,156,072✔

890

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

180,159,943✔

891

  taosMemoryFreeClear(pInfo->pResBlock->pBlockAgg);

180,122,454✔

893

  if (!pInfo->needCountEmptyTable) {

180,244,447✔

894

    pInfo->countState = TABLE_COUNT_STATE_END;

163,764,719✔

896

    pInfo->countState = TABLE_COUNT_STATE_SCAN;

16,460,442✔

899

_end:

180,156,247✔

900

  if (code != TSDB_CODE_SUCCESS) {

180,156,247✔

903

  return code;

180,185,677✔

906

void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {

557,850,689✔

907

  if (pInfo->countState == TABLE_COUNT_STATE_END) {

557,850,689✔

908

    return;

450,985,553✔

910

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

106,891,399✔

911

    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;

221,762✔

913

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

106,671,087✔

914

    if (code != TSDB_CODE_SUCCESS) {

106,666,498✔

915

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

42,399,800✔

920

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

46,777,969✔

922

  blockDataEmpty(pBlock);

46,777,969✔

923

  pBlock->info.rows = 1;

46,784,030✔

924

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

46,784,030✔

925

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

46,784,030✔

926

  pBlock->info.blankFill = true;

46,781,029✔

929

  int32_t col_num = blockDataGetNumOfCols(pBlock);

46,780,519✔

930

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

193,414,321✔

931

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

146,633,802✔

936

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

46,780,519✔

938

  return pBlock;

46,776,890✔

941

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

44,946,493✔

942

  STableScanInfo* pTableScanInfo = pOperator->info;

44,946,493✔

944

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

44,951,024✔

946

  pOperator->resultInfo.totalRows++;

44,940,373✔

947

  return pBlock;

44,952,044✔

954

static int32_t prepareTableScan(SOperatorInfo* pOperator) {

717,943,688✔

955

  int32_t code = TSDB_CODE_SUCCESS;

717,943,688✔

956

  int32_t lino = 0;

717,943,688✔

958

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

717,943,688✔

959

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

718,003,624✔

960

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

718,164,187✔

961

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

31,820,014✔

962

         NOTIFY_TYPE_SCAN_PARAM &&

3,030,745✔

963

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

3,030,745✔

964

    STableScanOperatorParam* pNotify =

3,030,745✔

967

    STableScanInfo* pTableScanInfo = pOperator->info;

3,030,745✔

968

    SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

3,030,745✔

969

    SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

3,030,745✔

970

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

3,030,745✔

972

    QUERY_CHECK_CODE(code, lino, _end);

3,030,745✔

974

    if (!pGetParam->reUse) {

3,030,745✔

975

      freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

96,286✔

981

      pGetParam->reUse = false;

2,934,459✔

986

    pOperator->pOperatorGetParam = NULL;

3,030,745✔

989

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

718,182,589✔

990

_end:

84,231,183✔

991

  if (code != TSDB_CODE_SUCCESS) {

718,168,718✔

995

  return code;

718,178,063✔

998

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

679,014,849✔

999

  int32_t         code = TSDB_CODE_SUCCESS;

679,014,849✔

1000

  int32_t         lino = 0;

679,014,849✔

1001

  STableScanInfo* pTableScanInfo = pOperator->info;

679,014,849✔

1002

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

679,057,945✔

1003

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

679,020,803✔

1004

  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;

679,079,091✔

1005

  bool            hasNext = false;

678,941,871✔

1006

  int64_t         st = taosGetTimestampUs();

678,971,206✔

1008

  QRY_PARAM_CHECK(ppRes);

678,971,206✔

1009

  pBlock->info.dataLoad = false;

678,990,654✔

1011

  while (true) {

22,870,546✔

1012

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

701,632,680✔

1013

    if (code != TSDB_CODE_SUCCESS) {

701,772,270✔

1015

      QUERY_CHECK_CODE(code, lino, _end);

5,154✔

1018

    if (!hasNext) {

701,769,204✔

1019

      break;

163,008,743✔

1022

    if (isTaskKilled(pTaskInfo)) {

538,760,461✔

1023

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

14,954✔

1024

      code = pTaskInfo->code;

14,954✔

1025

      QUERY_CHECK_CODE(code, lino, _end);

14,954✔

1028

    if (pOperator->status == OP_EXEC_DONE) {

538,954,139✔

1029

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

121,408✔

1030

      break;

121,408✔

1034

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

538,747,740✔

1035

    if (!processThisBlock) {

538,770,269✔

1039

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

538,770,269✔

1040

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

538,830,944✔

1043

    uint32_t status = 0;

538,834,600✔

1044

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

538,856,868✔

1045

    QUERY_CHECK_CODE(code, lino, _end);

538,723,654✔

1047

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

538,608,310✔

1048

      break;

2,790,104✔

1052

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

535,818,206✔

1053

      continue;

22,870,546✔

1056

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

512,950,774✔

1057

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

512,996,904✔

1059

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

512,974,504✔

1060

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

513,006,136✔

1062

    (*ppRes) = pBlock;

513,019,824✔

1063

    return code;

512,920,634✔

1066

_end:

166,062,473✔

1067

  if (code != TSDB_CODE_SUCCESS) {

166,070,983✔

1068

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

141,522✔

1069

    pTaskInfo->code = code;

141,522✔

1070

    T_LONG_JMP(pTaskInfo->env, code);

130,298✔

1072

  return code;

165,929,461✔

1075

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

726,843,963✔

1076

  int32_t         code = TSDB_CODE_SUCCESS;

726,843,963✔

1077

  int32_t         lino = 0;

726,843,963✔

1078

  STableScanInfo* pTableScanInfo = pOperator->info;

726,843,963✔

1079

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

726,864,467✔

1080

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

726,808,205✔

1082

  QRY_PARAM_CHECK(pBlock);

726,815,938✔

1085

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

726,785,086✔

1086

    return code;

2,990,466✔

1090

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

877,267,038✔

1091

    SSDataBlock* p = NULL;

593,574,396✔

1092

    code = doTableScanImplNext(pOperator, &p);

593,485,228✔

1093

    QUERY_CHECK_CODE(code, lino, _end);

593,421,641✔

1095

    if (p != NULL) {

593,421,641✔

1096

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

439,988,790✔

1097

      *pBlock = p;

440,061,541✔

1098

      return code;

439,956,299✔

1101

    pTableScanInfo->scanTimes += 1;

153,432,851✔

1102

    taosHashClear(pTableScanInfo->pIgnoreTables);

153,437,736✔

1104

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

153,440,453✔

1105

      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

86,834✔

1106

      pTableScanInfo->base.scanFlag = MAIN_SCAN;

86,834✔

1107

      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

86,834✔

1108

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

86,834✔

1111

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

86,834✔

1112

      QUERY_CHECK_CODE(code, lino, _end);

83,493✔

1116

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

283,639,227✔

1117

  if (pTableScanInfo->scanTimes < total) {

283,652,259✔

1118

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

85,398,695✔

1119

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

67,867✔

1120

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

67,867✔

1121

      QUERY_CHECK_CODE(code, lino, _end);

67,867✔

1122

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

67,867✔

1125

    while (pTableScanInfo->scanTimes < total) {

97,902,329✔

1126

      SSDataBlock* p = NULL;

85,414,126✔

1127

      code = doTableScanImplNext(pOperator, &p);

85,393,637✔

1128

      QUERY_CHECK_CODE(code, lino, _end);

85,364,067✔

1130

      if (p != NULL) {

85,364,067✔

1131

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

72,874,420✔

1132

        *pBlock = p;

72,883,595✔

1133

        return code;

72,890,122✔

1136

      pTableScanInfo->scanTimes += 1;

12,489,647✔

1137

      taosHashClear(pTableScanInfo->pIgnoreTables);

12,489,346✔

1139

      if (pTableScanInfo->scanTimes < total) {

12,490,832✔

1150

  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {

210,727,289✔

1151

    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;

60,742,339✔

1152

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

60,740,647✔

1153

      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {

110,598✔

1154

        pTableScanInfo->countState = TABLE_COUNT_STATE_PROCESSED;

970✔

1156

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

970✔

1158

        if (NULL == pStart) {

970✔

1162

        *pBlock = getBlockForEmptyTable(pOperator, pStart);

970✔

1163

        return code;

970✔

1166

      int32_t numOfTables = 0;

60,626,801✔

1167

      code = tableListGetSize(pTableListInfo, &numOfTables);

60,629,187✔

1168

      QUERY_CHECK_CODE(code, lino, _end);

60,631,231✔

1170

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

60,631,231✔

1172

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

59,940,256✔

1173

        if (pIte != NULL) {

59,938,811✔

1174

          size_t        keySize = 0;

44,951,163✔

1175

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

44,949,633✔

1176

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

44,951,584✔

1177

          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);

44,952,094✔

1178

          markGroupProcessed(pTableScanInfo, *pGroupId);

44,949,544✔

1179

          *pBlock = getBlockForEmptyTable(pOperator, &info);

44,946,033✔

1181

          return code;

44,950,054✔

1185

    pTableScanInfo->countState = TABLE_COUNT_STATE_END;

15,788,251✔

1188

_end:

150,000,639✔

1189

  if (code != TSDB_CODE_SUCCESS) {

165,773,691✔

1194

  return code;

165,773,691✔

1197

static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {

966,595✔

1198

  STableScanInfo*          pInfo = pOperator->info;

966,595✔

1199

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

966,595✔

1200

  int32_t                  code = 0;

966,595✔

1201

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

966,595✔

1202

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

966,595✔

1203

  int32_t                  num = taosArrayGetSize(pParam->pUidList);

966,595✔

1204

  if (num <= 0) {

966,595✔

1209

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

966,595✔

1213

  if (pParam->tableSeq) {

966,595✔

1214

    pListInfo->oneTableForEachGroup = true;

706,005✔

1215

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

706,005✔

1221

    pListInfo->oneTableForEachGroup = false;

260,590✔

1222

    pListInfo->numOfOuputGroups = 1;

260,590✔

1224

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

966,595✔

1227

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

701,445✔

1228

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

701,445✔

1231

  STableKeyInfo info = {.groupId = 0};

966,595✔

1232

  int32_t       tableIdx = 0;

966,595✔

1233

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

9,046,874✔

1234

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

8,080,279✔

1235

    if (!pUid) {

8,078,231✔

1240

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

8,078,231✔

1241

      if (TSDB_CODE_DUP_KEY == terrno) {

6,764,300✔

1242

        continue;

6,763,788✔

1247

    info.uid = *pUid;

1,318,539✔

1248

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

1,318,539✔

1249

    if (p == NULL) {

1,318,155✔

1253

    tableIdx++;

1,318,155✔

1254

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

1,318,155✔

1257

  return code;

966,595✔

1260

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

166,150,050✔

1262

  const char* idStr = GET_TASKID(pTaskInfo);

166,150,050✔

1263

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

166,182,881✔

1264

  if (code) {

165,937,980✔

1269

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

165,937,980✔

1274

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

332,184,224✔

1275

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

166,056,188✔

1276

  if (code) {

166,050,248✔

1277

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

2,265✔

1280

  return code;

166,078,907✔

1283

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

300,949,975✔

1284

  SColIdPair* node1 = (SColIdPair*)elem1;

300,949,975✔

1285

  SColIdPair* node2 = (SColIdPair*)elem2;

300,949,975✔

1287

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

300,949,975✔

1288

    return 0;

109,490✔

1291

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

300,935,985✔

1294

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

405,012,848✔

1295

  SColIdSlotIdPair* node1 = (SColIdSlotIdPair*)elem1;

405,012,848✔

1296

  SColIdSlotIdPair* node2 = (SColIdSlotIdPair*)elem2;

405,012,848✔

1298

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

405,012,848✔

1299

    return 0;

167,258✔

1302

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

404,936,380✔

1316

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

94,980,885✔

1317

  if (pList == NULL) {

94,980,885✔

1321

  int32_t left = 0;

94,980,885✔

1322

  int32_t right = taosArrayGetSize(pList) - 1;

94,980,885✔

1323

  while (left <= right) {

294,735,003✔

1324

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

294,735,003✔

1325

    SColMatchItem* pItem = taosArrayGet(pList, mid);

294,735,003✔

1326

    if (pItem->colId == colId) {

294,735,003✔

1328

      while (mid > 0) {

94,980,885✔

1329

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

71,264,781✔

1330

        if (pPrev->colId != colId) {

71,264,781✔

1331

          break;

71,264,781✔

1335

      return taosArrayGet(pList, mid);

94,980,885✔

1338

    if (pItem->colId < colId) {

199,754,118✔

1339

      left = mid + 1;

93,826,391✔

1341

      right = mid - 1;

105,927,727✔

1348

static bool isNewScanParam(STableScanOperatorParam* pParam) {

40,568,497✔

1349

  return pParam->isNewParam;

40,568,497✔

1352

static int32_t createVTableScanInfoFromBatchParam(SOperatorInfo* pOperator) {

17,550,352✔

1353

  int32_t                  code = 0;

17,550,352✔

1354

  int32_t                  lino = 0;

17,550,352✔

1355

  STableScanInfo*          pInfo = pOperator->info;

17,550,352✔

1356

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

17,550,580✔

1357

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

17,546,546✔

1358

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

17,549,532✔

1359

  SMetaReader              orgTable = {0};

17,553,998✔

1360

  SMetaReader              superTable = {0};

17,573,322✔

1361

  SSchemaWrapper*          schema = NULL;

17,560,115✔

1362

  SArray*                  pColArray = NULL;

17,560,115✔

1363

  SArray*                  pBlockColArray = NULL;

17,560,115✔

1364

  int32_t                  num = 0;

17,560,115✔

1365

  STableKeyInfo*           pList = NULL;

17,558,895✔

1369

  cleanupQueryTableDataCond(&pInfo->base.cond);

17,560,396✔

1371

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

17,565,846✔

1372

  if (pGetParam &&

17,567,722✔

1373

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

4,091,825✔

1374

    STableScanOperatorParam* pParam =

4,094,615✔

1376

    pInfo->pBatchColMap = pParam->pBatchTbInfo;

4,096,033✔

1377

    pParam->pBatchTbInfo = NULL;

4,104,565✔

1378

    pInfo->cachedTimeWindow = pParam->window;

4,074,289✔

1379

    pInfo->currentBatchIdx = 0;

4,086,091✔

1380

    pInfo->lastBatchIdx = -1;

4,098,668✔

1381

    pInfo->cachedTagList = pParam->pTagList;

4,099,601✔

1382

    pParam->pTagList = NULL;

4,085,300✔

1383

    pInfo->cachedGroupId = pParam->groupid;

4,103,911✔

1385

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

4,093,516✔

1387

    taosHashClear(pInfo->pIgnoreTables);

4,093,516✔

1389

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

4,098,995✔

1390

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

3,899,122✔

1391

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

49,019,286✔

1392

        SqlFunctionCtx ctx = pSup->pCtx[i];

45,143,497✔

1393

        ctx.skipDynDataCheck = false;

45,122,595✔

1398

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

17,555,793✔

1399

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

17,575,120✔

1401

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

17,575,120✔

1402

    isNewTable = false;

10,692,707✔

1404

    isNewTable = true;

6,878,639✔

1405

    pInfo->lastBatchIdx = pInfo->currentBatchIdx;

6,878,639✔

1408

  if (isNewTable) {

17,560,884✔

1409

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

6,868,504✔

1410

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

6,880,727✔

1411

    pAPI->metaReaderFn.readerReleaseLock(&orgTable);

6,857,279✔

1412

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

6,872,147✔

1413

    QUERY_CHECK_CODE(code, lino, _return);

6,848,586✔

1414

    switch (orgTable.me.type) {

6,848,586✔

1415

      case TSDB_CHILD_TABLE:

283,733✔

1416

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

283,733✔

1417

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

283,733✔

1418

        pAPI->metaReaderFn.readerReleaseLock(&superTable);

283,733✔

1419

        QUERY_CHECK_CODE(code, lino, _return);

283,733✔

1420

        schema = &superTable.me.stbEntry.schemaRow;

283,733✔

1421

        break;

283,733✔

1422

      case TSDB_NORMAL_TABLE:

6,582,935✔

1423

        schema = &orgTable.me.ntbEntry.schemaRow;

6,582,935✔

1424

        break;

6,582,935✔

1431

    pListInfo->oneTableForEachGroup = true;

6,866,668✔

1432

    taosHashClear(pListInfo->map);

6,866,014✔

1433

    taosArrayClear(pListInfo->pTableList);

6,885,040✔

1435

    uint64_t      pUid = orgTable.me.uid;

6,882,011✔

1436

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

6,882,011✔

1437

    int32_t       tableIdx = 0;

6,880,845✔

1438

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

6,882,153✔

1439

    QUERY_CHECK_CODE(code, lino, _return);

6,885,107✔

1440

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

13,767,185✔

1441

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

6,882,078✔

1443

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

6,882,078✔

1444

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

6,881,893✔

1445

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

6,881,893✔

1446

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

6,874,830✔

1449

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

6,874,830✔

1450

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

6,880,258✔

1452

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

26,283,420✔

1453

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

19,405,462✔

1454

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

154,297,544✔

1455

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

154,137,711✔

1456

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

19,415,715✔

1457

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

19,409,614✔

1458

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

19,409,614✔

1460

          break;

19,395,938✔

1465

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

33,177,472✔

1466

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

26,288,540✔

1467

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

157,559,760✔

1468

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

157,308,680✔

1469

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

157,345,923✔

1470

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

26,296,614✔

1471

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

26,302,358✔

1472

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

26,302,358✔

1474

          break;

26,288,095✔

1479

    taosArraySort(pColArray, compareColIdPair);

6,882,078✔

1480

    taosArraySort(pBlockColArray, compareColIdSlotIdPair);

6,874,983✔

1481

    if (pInfo->pBlockColMap) {

6,878,722✔

1482

      taosArrayDestroy(pInfo->pBlockColMap);

4,898,383✔

1483

      pInfo->pBlockColMap = NULL;

4,886,977✔

1485

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

6,870,616✔

1486

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

6,882,448✔

1489

    taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

6,854,801✔

1490

    taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

6,873,873✔

1492

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

6,878,470✔

1493

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

6,158,162✔

1494

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

76,273,225✔

1495

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

70,138,486✔

1496

        bool           needScan = false;

70,133,095✔

1497

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

141,221,544✔

1498

          SFunctParam pParam = ctx->param[j];

71,095,398✔

1499

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

340,844,275✔

1500

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

286,223,488✔

1501

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

287,108,403✔

1502

              needScan |= true;

17,706,594✔

1503

              break;

17,706,594✔

1507

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

70,148,980✔

1508

        if (!needScan) {

70,184,287✔

1509

          ctx->skipDynDataCheck = true;

52,485,558✔

1511

          ctx->skipDynDataCheck = false;

17,698,729✔

1518

    pColArray = pInfo->lastColArray;

10,692,380✔

1519

    pBlockColArray = pInfo->lastBlockColArray;

10,692,380✔

1522

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

17,572,496✔

1523

  QUERY_CHECK_CODE(code, lino, _return);

17,569,695✔

1525

  if (pInfo->pResBlock) {

17,569,695✔

1526

    blockDataDestroy(pInfo->pResBlock);

13,467,888✔

1527

    pInfo->pResBlock = NULL;

13,467,234✔

1530

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

17,566,709✔

1538

  if (isNewTable) {

17,563,168✔

1539

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

6,871,115✔

1540

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

6,132,814✔

1541

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

6,136,051✔

1543

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

743,292✔

1544

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

719,697✔

1547

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

10,692,053✔

1548

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

1,411,057✔

1550

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

9,280,996✔

1554

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

17,562,002✔

1555

  pInfo->currentGroupId = 0;

17,554,698✔

1556

  pInfo->ignoreTag = true;

17,546,431✔

1558

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

17,535,756✔

1559

  QUERY_CHECK_CODE(code, lino, _return);

17,561,546✔

1561

  if (isNewTable) {

17,561,546✔

1562

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

6,869,166✔

1563

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

4,896,705✔

1565

    pInfo->base.dataReader = NULL;

6,861,883✔

1567

    taosRLockLatch(&pTaskInfo->lock);

6,861,846✔

1568

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

6,875,318✔

1569

    taosRUnLockLatch(&pTaskInfo->lock);

6,837,287✔

1570

    QUERY_CHECK_CODE(code, lino, _return);

6,859,550✔

1572

    pInfo->newReader = true;

6,859,550✔

1573

    if (pInfo->lastColArray) {

6,849,526✔

1574

      taosArrayDestroy(pInfo->lastColArray);

2,775,181✔

1575

      pInfo->lastColArray = NULL;

2,774,602✔

1577

    if (pInfo->lastBlockColArray) {

6,850,526✔

1578

      taosArrayDestroy(pInfo->lastBlockColArray);

2,774,090✔

1579

      pInfo->lastBlockColArray = NULL;

2,774,669✔

1581

    pInfo->lastColArray = pColArray;

6,859,321✔

1582

    pInfo->lastBlockColArray = pBlockColArray;

6,858,788✔

1584

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

10,692,380✔

1585

    QUERY_CHECK_CODE(code, lino, _return);

10,692,380✔

1587

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

10,692,380✔

1588

    pInfo->newReader = false;

10,692,707✔

1589

    pInfo->scanTimes = 0;

10,692,707✔

1592

  pOperator->status = OP_OPENED;

17,532,400✔

1594

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

17,553,414✔

1598

  pInfo->currentGroupId = -1;

17,529,607✔

1599

_return:

17,459,725✔

1600

  if (code) {

17,469,962✔

1603

  pAPI->metaReaderFn.clearReader(&superTable);

17,469,962✔

1604

  pAPI->metaReaderFn.clearReader(&orgTable);

17,537,717✔

1605

  return code;

17,548,857✔

1608

static int32_t createVTableScanInfoFromParam(SOperatorInfo* pOperator) {

23,716,104✔

1609

  int32_t                  code = 0;

23,716,104✔

1610

  int32_t                  lino = 0;

23,716,104✔

1611

  STableScanInfo*          pInfo = pOperator->info;

23,716,104✔

1612

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

23,716,104✔

1613

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

23,716,104✔

1614

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

23,716,104✔

1615

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

23,716,104✔

1616

  SMetaReader              orgTable = {0};

23,716,104✔

1617

  SMetaReader              superTable = {0};

23,716,104✔

1618

  SSchemaWrapper*          schema = NULL;

23,716,104✔

1619

  SArray*                  pColArray = NULL;

23,716,104✔

1620

  SArray*                  pBlockColArray = NULL;

23,716,104✔

1621

  int32_t                  num = 0;

23,716,104✔

1622

  STableKeyInfo*           pList = NULL;

23,716,104✔

1624

  SArray*                  pMatchList = NULL;

23,716,104✔

1626

  cleanupQueryTableDataCond(&pInfo->base.cond);

23,716,104✔

1628

  pOrgTbInfo = pParam->pOrgTbInfo;

23,716,104✔

1630

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

23,716,104✔

1632

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

23,716,104✔

1633

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

23,716,104✔

1634

  pAPI->metaReaderFn.readerReleaseLock(&orgTable);

23,716,104✔

1635

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

23,716,104✔

1636

  QUERY_CHECK_CODE(code, lino, _return);

23,716,104✔

1637

  switch (orgTable.me.type) {

23,716,104✔

1638

    case TSDB_CHILD_TABLE:

6,457,318✔

1639

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

6,457,318✔

1640

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

6,457,318✔

1641

      pAPI->metaReaderFn.readerReleaseLock(&superTable);

6,457,318✔

1642

      QUERY_CHECK_CODE(code, lino, _return);

6,457,318✔

1643

      schema = &superTable.me.stbEntry.schemaRow;

6,457,318✔

1644

      break;

6,457,318✔

1645

    case TSDB_NORMAL_TABLE:

17,258,786✔

1646

      schema = &orgTable.me.ntbEntry.schemaRow;

17,258,786✔

1647

      break;

17,258,786✔

1654

  pListInfo->oneTableForEachGroup = true;

23,716,104✔

1655

  taosHashClear(pListInfo->map);

23,716,104✔

1656

  taosArrayClear(pListInfo->pTableList);

23,716,104✔

1658

  uint64_t      pUid = orgTable.me.uid;

23,716,104✔

1659

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

23,716,104✔

1660

  int32_t       tableIdx = 0;

23,716,104✔

1661

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

23,716,104✔

1662

  QUERY_CHECK_CODE(code, lino, _return);

23,716,104✔

1663

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

47,432,208✔

1664

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

23,716,104✔

1666

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

23,716,104✔

1667

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

23,716,104✔

1668

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

23,716,104✔

1669

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

23,716,104✔

1670

  if (pInfo->pBlockColMap) {

23,716,104✔

1671

    taosArrayDestroy(pInfo->pBlockColMap);

21,349,872✔

1672

    pInfo->pBlockColMap = NULL;

21,349,872✔

1674

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

23,716,104✔

1675

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

23,716,104✔

1678

  SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID,

23,716,104✔

1682

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

23,716,104✔

1684

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

94,980,885✔

1685

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

71,264,781✔

1686

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

601,779,653✔

1687

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

601,779,653✔

1688

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

71,264,781✔

1689

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

71,264,781✔

1690

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

71,264,781✔

1691

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

71,264,781✔

1692

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

71,264,781✔

1693

        break;

71,264,781✔

1698

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

23,716,104✔

1699

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

23,716,104✔

1704

      pMatchList = pInfo->base.matchInfo.pList;

23,716,104✔

1709

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

118,696,356✔

1710

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

94,980,885✔

1711

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

94,980,885✔

1712

    if (pItem) {

94,980,885✔

1713

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

94,980,885✔

1714

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

633✔

1716

        code = TSDB_CODE_VTABLE_COLUMN_TYPE_MISMATCH;

633✔

1717

        goto _return;

633✔

1719

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

94,980,252✔

1720

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

94,980,252✔

1721

      if (i > 0) {

94,980,252✔

1722

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

142,528,296✔

1727

  taosArraySort(pColArray, compareColIdPair);

23,715,471✔

1728

  taosArraySort(pBlockColArray, compareColIdSlotIdPair);

23,715,471✔

1729

  taosArraySort(pInfo->pBlockColMap, compareColIdSlotIdPair);

23,715,471✔

1731

  taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

23,715,471✔

1732

  taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

23,715,471✔

1734

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

23,715,471✔

1735

  QUERY_CHECK_CODE(code, lino, _return);

23,715,471✔

1737

  if (pInfo->pResBlock) {

23,715,471✔

1738

    blockDataDestroy(pInfo->pResBlock);

16,855,344✔

1739

    pInfo->pResBlock = NULL;

16,855,344✔

1742

  if (isNewScanParam(pParam)) {

23,715,471✔

1743

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

6,867,135✔

1744

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

6,867,135✔

1745

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

6,867,135✔

1751

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

16,848,336✔

1753

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

23,715,471✔

1754

  pInfo->currentGroupId = 0;

23,715,471✔

1755

  pInfo->ignoreTag = true;

23,715,471✔

1757

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

23,715,471✔

1758

  QUERY_CHECK_CODE(code, lino, _return);

23,715,471✔

1760

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

23,715,471✔

1761

  if (reader) {

23,715,471✔

1762

    if (isNewScanParam(pParam)) {

16,853,026✔

1763

      pAPI->tsdReader.tsdReaderClose(*reader);

4,690✔

1764

      pInfo->base.dataReader = NULL;

4,690✔

1766

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

4,690✔

1767

      QUERY_CHECK_CODE(code, lino, _return);

4,690✔

1769

      taosRLockLatch(&pTaskInfo->lock);

4,690✔

1770

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

4,690✔

1771

      taosRUnLockLatch(&pTaskInfo->lock);

4,690✔

1772

      QUERY_CHECK_CODE(code, lino, _return);

4,690✔

1774

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

4,690✔

1775

      QUERY_CHECK_CODE(code, lino, _return);

4,690✔

1776

      pInfo->newReader = true;

4,690✔

1778

      pInfo->base.dataReader = *reader;

16,848,336✔

1779

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

16,848,336✔

1780

      QUERY_CHECK_CODE(code, lino, _return);

16,848,336✔

1782

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

16,848,336✔

1783

      pInfo->newReader = false;

16,848,336✔

1784

      pInfo->scanTimes = 0;

16,848,336✔

1787

    pInfo->base.dataReader = NULL;

6,862,445✔

1789

    taosRLockLatch(&pTaskInfo->lock);

6,862,445✔

1790

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

6,862,445✔

1791

    taosRUnLockLatch(&pTaskInfo->lock);

6,862,445✔

1792

    QUERY_CHECK_CODE(code, lino, _return);

6,862,445✔

1794

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

6,862,445✔

1795

    QUERY_CHECK_CODE(code, lino, _return);

6,862,445✔

1796

    pInfo->newReader = true;

6,862,445✔

1798

  pOperator->status = OP_OPENED;

23,715,471✔

1800

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

23,715,471✔

1804

  pInfo->currentGroupId = -1;

23,715,471✔

1805

_return:

23,716,104✔

1806

  if (code) {

23,716,104✔

1807

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

633✔

1809

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

23,716,104✔

1812

  taosArrayDestroy(pColArray);

23,716,104✔

1813

  taosArrayDestroy(pBlockColArray);

23,716,104✔

1814

  pAPI->metaReaderFn.clearReader(&superTable);

23,716,104✔

1815

  pAPI->metaReaderFn.clearReader(&orgTable);

23,716,104✔

1816

  return code;

23,716,104✔

1819

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

163,052,802✔

1820

  int32_t         code = TSDB_CODE_SUCCESS;

163,052,802✔

1821

  int32_t         lino = 0;

163,052,802✔

1822

  STableScanInfo* pInfo = pOperator->info;

163,052,802✔

1823

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

163,041,928✔

1824

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

163,057,167✔

1825

  int32_t         numOfTables = 0;

163,056,269✔

1827

  QRY_PARAM_CHECK(pResult);

162,982,929✔

1829

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

162,981,143✔

1830

  QUERY_CHECK_CODE(code, lino, _end);

163,018,035✔

1832

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

163,018,035✔

1833

    setOperatorCompleted(pOperator);

148,960,061✔

1834

    if (pOperator->dynamicTask) {

148,959,419✔

1835

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

965,451✔

1836

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

965,074✔

1838

    return code;

148,965,718✔

1842

  pOperator->status = OP_OPENED;

14,002,320✔

1843

  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);

14,079,531✔

1845

  int32_t        num = 0;

14,047,980✔

1846

  STableKeyInfo* pList = NULL;

14,049,331✔

1847

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

14,037,683✔

1848

  QUERY_CHECK_CODE(code, lino, _end);

14,084,854✔

1850

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

14,084,854✔

1851

  QUERY_CHECK_CODE(code, lino, _end);

14,082,576✔

1853

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

14,082,576✔

1854

  QUERY_CHECK_CODE(code, lino, _end);

14,040,076✔

1855

  pInfo->scanTimes = 0;

14,040,076✔

1857

  code = doGroupedTableScan(pOperator, pResult);

14,064,145✔

1858

  QUERY_CHECK_CODE(code, lino, _end);

14,108,512✔

1860

  if (*pResult != NULL) {

14,108,512✔

1861

    if (pOperator->dynamicTask) {

11,428,343✔

1862

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

11,175,915✔

1864

    return code;

11,428,016✔

1867

_end:

2,681,662✔

1868

  if (code != TSDB_CODE_SUCCESS) {

2,681,335✔

1872

  return code;

2,681,335✔

1875

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

673,542,425✔

1876

  int32_t         code = TSDB_CODE_SUCCESS;

673,542,425✔

1877

  int32_t         lino = 0;

673,542,425✔

1878

  STableScanInfo* pInfo = pOperator->info;

673,542,425✔

1879

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

673,531,880✔

1880

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

673,464,610✔

1881

  int32_t         num = 0;

673,498,130✔

1882

  STableKeyInfo*  pList = NULL;

673,489,925✔

1883

  SSDataBlock*    pResult = NULL;

673,533,285✔

1884

  const char*     idStr = GET_TASKID(pTaskInfo);

673,549,919✔

1886

  QRY_PARAM_CHECK(pResBlock);

673,470,533✔

1888

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

673,424,554✔

1889

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

152,423,541✔

1894

    taosRLockLatch(&pTaskInfo->lock);

152,387,586✔

1895

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

152,424,355✔

1896

    taosRUnLockLatch(&pTaskInfo->lock);

152,349,388✔

1897

    QUERY_CHECK_CODE(code, lino, _end);

152,415,209✔

1899

    if (pInfo->filesetDelimited) {

152,412,944✔

1900

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

261,725✔

1903

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

152,412,876✔

1908

  code = doGroupedTableScan(pOperator, &pResult);

673,436,188✔

1909

  QUERY_CHECK_CODE(code, lino, _end);

673,223,447✔

1911

  if (pResult != NULL) {

673,223,447✔

1912

    if (pOperator->dynamicTask) {

524,164,112✔

1913

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

2,035,038✔

1916

    *pResBlock = pResult;

524,162,541✔

1917

    return code;

523,970,714✔

1921

    code = startNextGroupScan(pOperator, &pResult);

149,138,936✔

1922

    QUERY_CHECK_CODE(code, lino, _end);

149,310,789✔

1924

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

149,310,789✔

1925

      *pResBlock = pResult;

149,208,485✔

1926

      return code;

149,225,262✔

1930

_end:

2,265✔

1931

  if (code != TSDB_CODE_SUCCESS) {

2,265✔

1932

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

2,265✔

1933

    pTaskInfo->code = code;

2,265✔

1936

  return code;

2,265✔

1939

static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {

23,715,471✔

1940

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

23,715,471✔

1943

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

966,595✔

1944

  int32_t                  code = TSDB_CODE_SUCCESS;

966,595✔

1945

  int32_t                  lino = 0;

966,595✔

1946

  STableScanInfo*          pInfo = pOperator->info;

966,595✔

1947

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

966,595✔

1948

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

966,595✔

1949

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

966,595✔

1951

  code = createTableListInfoFromParam(pOperator);

966,595✔

1952

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

966,595✔

1953

  pOperator->pOperatorGetParam = NULL;

966,595✔

1954

  QUERY_CHECK_CODE(code, lino, _end);

966,595✔

1956

  if (pOperator->status == OP_EXEC_DONE) {

966,595✔

1957

    pInfo->currentGroupId = -1;

23,115✔

1958

    pOperator->status = OP_OPENED;

23,115✔

1959

    SSDataBlock* result = NULL;

23,115✔

1962

      code = startNextGroupScan(pOperator, &result);

23,115✔

1963

      QUERY_CHECK_CODE(code, lino, _end);

23,115✔

1965

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

23,115✔

1966

        (*ppRes) = result;

23,115✔

1967

        return code;

23,115✔

1971

_end:

943,480✔

1972

  if (code) {

943,480✔

1975

  return code;

943,480✔

1978

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

16,857,608✔

1979

  int32_t      code = TSDB_CODE_SUCCESS;

16,857,608✔

1980

  int32_t      lino = 0;

16,857,608✔

1981

  bool         found = false;

16,857,608✔

1983

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

16,857,608✔

1984

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

16,857,608✔

1986

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

83,987,219✔

1987

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

67,129,611✔

1988

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

67,129,611✔

1989

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

192,429,627✔

1990

      SColIdSlotIdPair* pColPair = taosArrayGet(pColMap, j);

175,575,024✔

1991

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

175,575,024✔

1992

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

175,575,024✔

1993

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

50,275,008✔

1994

        found = true;

50,275,008✔

1995

        break;

50,275,008✔

1998

    if (!found) {

67,129,611✔

1999

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

16,854,603✔

2000

        pSrcCol->info.slotId = -1;

16,854,603✔

2009

  return code;

16,857,608✔

2015

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

23,716,104✔

2016

  int32_t                  code = TSDB_CODE_SUCCESS;

23,716,104✔

2017

  int32_t                  lino = 0;

23,716,104✔

2018

  STableScanInfo*          pInfo = pOperator->info;

23,716,104✔

2019

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

23,716,104✔

2020

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

23,716,104✔

2021

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

23,716,104✔

2023

  QRY_PARAM_CHECK(ppRes);

23,716,104✔

2024

  qTrace("%s call", __FUNCTION__);

23,716,104✔

2026

  code = createVTableScanInfoFromParam(pOperator);

23,716,104✔

2027

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

23,716,104✔

2028

  pOperator->pOperatorGetParam = NULL;

23,716,104✔

2030

  QUERY_CHECK_CODE(code, lino, _end);

23,716,104✔

2032

  SSDataBlock* result = NULL;

23,715,471✔

2033

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

23,715,471✔

2034

    (*ppRes) = result;

16,972✔

2035

    return code;

16,972✔

2038

  if (pInfo->newReader) {

23,698,499✔

2039

    code = startNextGroupScan(pOperator, &result);

6,867,135✔

2040

    QUERY_CHECK_CODE(code, lino, _end);

6,867,135✔

2042

    code = doGroupedTableScan(pOperator, &result);

16,831,364✔

2043

    QUERY_CHECK_CODE(code, lino, _end);

16,831,364✔

2046

  if (result) {

23,698,499✔

2047

    SSDataBlock* res = NULL;

16,857,608✔

2048

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

16,857,608✔

2049

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

16,857,608✔

2050

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

16,857,608✔

2052

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

16,857,608✔

2053

    QUERY_CHECK_CODE(code, lino, _end);

16,857,608✔

2059

    (*ppRes) = result;

16,857,608✔

2061

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

6,840,891✔

2062

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

6,840,891✔

2064

    blockDataDestroy(pInfo->pResBlock);

6,840,891✔

2065

    pInfo->pResBlock = NULL;

6,840,891✔

2067

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

6,840,891✔

2068

    if (reader) {

6,840,891✔

2069

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

6,840,891✔

2070

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

6,840,891✔

2071

        pInfo->base.dataReader = NULL;

6,840,891✔

2073

      if (pAPI->tsdReader.tsdReaderClose) {

6,840,891✔

2074

        pAPI->tsdReader.tsdReaderClose(*reader);

6,840,891✔

2078

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

6,840,891✔

2079

    QUERY_CHECK_CODE(code, lino, _end);

6,840,891✔

2081

_end:

23,699,132✔

2082

  if (code) {

23,699,132✔

2083

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

633✔

2085

  return code;

23,699,132✔

2088

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

3,056,650✔

2089

  int32_t                  code = TSDB_CODE_SUCCESS;

3,056,650✔

2090

  int32_t                  lino = 0;

3,056,650✔

2091

  STableScanInfo*          pInfo = pOperator->info;

3,056,650✔

2092

  int32_t                  index = 0;

3,056,650✔

2093

  char*                    tagVal = NULL;

3,056,650✔

2095

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

21,406,904✔

2096

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

18,350,254✔

2097

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

18,350,254✔

2099

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

18,350,254✔

2100

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

18,350,254✔

2102

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

2,147,483,647✔

2103

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

2,147,483,647✔

2104

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

2,147,483,647✔

2105

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

2,147,483,647✔

2107

        varDataSetLen(tagVal, pTagVal->nData);

2,147,483,647✔

2108

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

2,147,483,647✔

2109

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

2,147,483,647✔

2110

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2111

        taosMemoryFreeClear(tagVal);

2,147,483,647✔

2113

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

2,147,483,647✔

2114

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2117

    index++;

18,350,254✔

2119

  return code;

3,056,650✔

2126

static void clearVstbBatchDynamicTableScanInfo(STableScanInfo* pInfo) {

4,108,476✔

2127

  if (pInfo->pBatchColMap) {

4,108,476✔

2128

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

10,991,152✔

2129

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

6,883,003✔

2130

      taosArrayDestroy(pOrgTbInfo->colMap);

6,883,984✔

2132

    taosArrayDestroy(pInfo->pBatchColMap);

4,109,784✔

2133

    pInfo->pBatchColMap = NULL;

4,107,168✔

2135

  if (pInfo->cachedTagList) {

4,109,457✔

2136

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

12,860,020✔

2137

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

10,960,034✔

2138

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

10,959,053✔

2139

        taosMemoryFreeClear(pTagVal->pData);

4,807,299✔

2142

    taosArrayDestroy(pInfo->cachedTagList);

1,902,602✔

2143

    pInfo->cachedTagList = NULL;

1,902,275✔

2145

  if (pInfo->lastColArray) {

4,107,495✔

2146

    taosArrayDestroy(pInfo->lastColArray);

4,107,822✔

2147

    pInfo->lastColArray = NULL;

4,108,149✔

2149

  if (pInfo->lastBlockColArray) {

4,108,476✔

2150

    taosArrayDestroy(pInfo->lastBlockColArray);

4,108,149✔

2151

    pInfo->lastBlockColArray = NULL;

4,107,495✔

2154

  blockDataDestroy(pInfo->pResBlock);

4,107,822✔

2155

  pInfo->pResBlock = NULL;

4,108,149✔

2156

4,108,476✔

2158

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

14,783,596✔

2159

  int32_t                  code = TSDB_CODE_SUCCESS;

14,783,596✔

2160

  int32_t                  lino = 0;

14,783,596✔

2161

  STableScanInfo*          pInfo = pOperator->info;

14,783,596✔

2162

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

14,782,430✔

2163

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

14,792,029✔

2164

  SSDataBlock*             result = NULL;

14,790,906✔

2166

  QRY_PARAM_CHECK(ppRes);

14,786,242✔

2167

  qTrace("%s call", __FUNCTION__);

14,785,349✔

2169

  while (!result) {

28,260,504✔

2170

    code = createVTableScanInfoFromBatchParam(pOperator);

17,568,963✔

2171

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

17,547,431✔

2172

    pOperator->pOperatorGetParam = NULL;

17,521,529✔

2173

    QUERY_CHECK_CODE(code, lino, _end);

17,525,984✔

2175

    if (pInfo->newReader) {

17,525,984✔

2176

      code = startNextGroupScan(pOperator, &result);

6,850,561✔

2177

      QUERY_CHECK_CODE(code, lino, _end);

6,882,708✔

2179

      code = doGroupedTableScan(pOperator, &result);

10,682,430✔

2180

      QUERY_CHECK_CODE(code, lino, _end);

10,691,868✔

2182

    if (!result) {

17,574,576✔

2183

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

6,881,239✔

2185

      pInfo->currentBatchIdx++;

6,881,239✔

2186

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

6,876,701✔

2188

        break;

4,108,476✔

2193

  if (result) {

14,800,017✔

2194

    SSDataBlock* res = NULL;

10,691,868✔

2195

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

10,691,868✔

2196

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

10,691,868✔

2197

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

10,691,541✔

2199

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

10,691,029✔

2200

    QUERY_CHECK_CODE(code, lino, _end);

10,692,380✔

2202

    if (pInfo->cachedTagList) {

10,692,380✔

2203

      code = setTagValFromTagList(pOperator, res);

3,056,650✔

2204

      QUERY_CHECK_CODE(code, lino, _end);

3,056,650✔

2207

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

10,692,380✔

2208

    pInfo->pResBlock = res;

10,692,380✔

2209

    blockDataDestroy(result);

10,692,380✔

2210

    (*ppRes) = res;

10,692,707✔

2212

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

10,692,707✔

2214

    clearVstbBatchDynamicTableScanInfo(pInfo);

4,108,149✔

2215

    setOperatorCompleted(pOperator);

4,108,149✔

2218

_end:

14,802,164✔

2219

  return code;

14,800,202✔

2222

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

39,480,810✔

2223

  int32_t                  code = TSDB_CODE_SUCCESS;

39,480,810✔

2224

  int32_t                  lino = 0;

39,480,810✔

2225

  STableScanInfo*          pInfo = pOperator->info;

39,480,810✔

2226

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

39,479,815✔

2227

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

39,470,519✔

2229

  QRY_PARAM_CHECK(ppRes);

39,469,126✔

2230

  qTrace("%s call", __FUNCTION__);

39,477,583✔

2232

  pOperator->dynamicTask = true;

39,477,967✔

2234

  switch (type) {

39,464,703✔

2235

    case DYN_TYPE_VSTB_WIN_SCAN:

966,595✔

2237

      code = doStbJoinDynamicTableScanNext(pOperator, ppRes);

966,595✔

2238

      QUERY_CHECK_CODE(code, lino, _end);

966,595✔

2239

      break;

966,595✔

2241

    case DYN_TYPE_VSTB_SINGLE_SCAN: {

23,716,104✔

2242

      code = doVstbSingleDynamicTableScanNext(pOperator, ppRes);

23,716,104✔

2243

      QUERY_CHECK_CODE(code, lino, _end);

23,716,104✔

2244

      break;

23,715,471✔

2246

    case DYN_TYPE_VSTB_BATCH_SCAN:

14,782,004✔

2247

      code = doVstbBatchDynamicTableScanNext(pOperator, ppRes);

14,782,004✔

2248

      QUERY_CHECK_CODE(code, lino, _end);

14,800,529✔

2249

      break;

14,800,529✔

2256

_end:

39,483,228✔

2257

  return code;

39,483,228✔

2260

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

717,968,889✔

2261

  int32_t         code = TSDB_CODE_SUCCESS;

717,968,889✔

2262

  int32_t         lino = 0;

717,968,889✔

2263

  STableScanInfo* pInfo = pOperator->info;

717,968,889✔

2264

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

718,112,143✔

2265

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

718,126,763✔

2266

  QRY_PARAM_CHECK(ppRes);

718,052,690✔

2267

  qTrace("%s call", __FUNCTION__);

718,032,240✔

2269

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

718,034,153✔

2270

  QUERY_CHECK_CODE(code, lino, _end);

718,176,637✔

2272

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

718,176,637✔

2273

  if (pGetParam || pInfo->pBatchColMap) {

718,171,053✔

2274

    ETableScanDynType type = pGetParam ?

39,472,506✔

2275

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

39,466,424✔

2278

    code = doDynamicTableScanNext(pOperator, type, ppRes);

39,472,506✔

2279

    QUERY_CHECK_CODE(code, lino, _end);

39,482,901✔

2281

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

39,482,268✔

2282

      return code;

38,540,423✔

2287

  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {

679,663,611✔

2288

    int32_t       numOfTables = 0;

6,090,922✔

2289

    STableKeyInfo tInfo = {0};

6,090,922✔

2290

    pInfo->countState = TABLE_COUNT_STATE_END;

6,090,922✔

2292

    while (1) {

5,685,790✔

2293

      SSDataBlock* result = NULL;

11,776,712✔

2294

      code = doGroupedTableScan(pOperator, &result);

11,774,468✔

2295

      QUERY_CHECK_CODE(code, lino, _end);

11,773,459✔

2297

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

11,773,459✔

2298

        (*ppRes) = result;

6,042,388✔

2299

        return code;

6,045,245✔

2303

      pInfo->currentTable++;

5,730,790✔

2305

      taosRLockLatch(&pTaskInfo->lock);

5,728,848✔

2306

      numOfTables = 0;

5,730,123✔

2307

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

5,730,123✔

2308

      if (code != TSDB_CODE_SUCCESS) {

5,729,171✔

2313

      if (pInfo->currentTable >= numOfTables) {

5,729,171✔

2314

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

44,394✔

2315

        taosRUnLockLatch(&pTaskInfo->lock);

44,394✔

2316

        (*ppRes) = NULL;

44,394✔

2317

        return code;

44,394✔

2320

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

5,681,273✔

2321

      if (!tmp) {

5,682,559✔

2327

      tInfo = *tmp;

5,682,559✔

2328

      taosRUnLockLatch(&pTaskInfo->lock);

5,683,498✔

2330

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

5,688,296✔

2331

      QUERY_CHECK_CODE(code, lino, _end);

5,684,153✔

2332

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

5,684,153✔

2335

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

5,690,854✔

2336

      QUERY_CHECK_CODE(code, lino, _end);

5,683,175✔

2337

      pInfo->scanTimes = 0;

5,683,175✔

2340

    code = groupSeqTableScan(pOperator, ppRes);

673,423,330✔

2341

    QUERY_CHECK_CODE(code, lino, _end);

673,217,829✔

2344

_end:

673,215,564✔

2345

  if (code != TSDB_CODE_SUCCESS) {

673,189,149✔

2346

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

2,898✔

2347

    pTaskInfo->code = code;

2,898✔

2348

    T_LONG_JMP(pTaskInfo->env, code);

2,898✔

2351

  return code;

673,186,251✔

2354

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

1,873,481✔

2355

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

1,873,481✔

2356

  if (!pRecorder) {

1,873,481✔

2359

  STableScanInfo* pTableScanInfo = pOptr->info;

1,873,481✔

2360

  *pRecorder = pTableScanInfo->base.readRecorder;

1,873,481✔

2361

  *pOptrExplain = pRecorder;

1,873,481✔

2362

  *len = sizeof(SFileBlockLoadRecorder);

1,873,481✔

2363

  return 0;

1,873,481✔

2366

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

187,706,954✔

2367

  cleanupQueryTableDataCond(&pBase->cond);

187,706,954✔

2368

  cleanupQueryTableDataCond(&pBase->orgCond);

187,698,687✔

2370

  if (pAPI->tsdReaderClose) {

187,715,450✔

2371

    pAPI->tsdReaderClose(pBase->dataReader);

187,720,875✔

2373

  pBase->dataReader = NULL;

187,604,262✔

2375

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

187,637,315✔

2376

    taosArrayDestroy(pBase->matchInfo.pList);

187,662,510✔

2379

  tableListDestroy(pBase->pTableListInfo);

187,621,878✔

2380

  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);

187,701,706✔

2381

  cleanupExprSupp(&pBase->pseudoSup);

187,652,930✔

2382

187,718,464✔

2384

static void cleanReaderForVTable(STableScanInfo* pInfo){

5,225,638✔

2385

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

5,225,638✔

2386

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

5,226,021✔

2387

    while (pIter != NULL) {

5,247,192✔

2388

      void **reader = pIter;

21,554✔

2389

      if (*reader) {

21,554✔

2390

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

21,554✔

2391

          pInfo->base.dataReader = NULL;

19,236✔

2393

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

21,554✔

2395

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

21,554✔

2398

5,225,767✔

2400

static void destroyTableScanOperatorInfo(void* param) {

162,709,738✔

2401

  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;

162,709,738✔

2402

  blockDataDestroy(pTableScanInfo->pResBlock);

162,709,738✔

2403

  blockDataDestroy(pTableScanInfo->pOrgBlock);

162,691,424✔

2404

  taosHashCleanup(pTableScanInfo->pIgnoreTables);

162,697,059✔

2405

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

162,668,906✔

2406

    cleanReaderForVTable(pTableScanInfo);

5,223,574✔

2407

    taosHashCleanup(pTableScanInfo->readerCache);

5,225,638✔

2409

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

162,676,828✔

2410

  taosArrayDestroy(pTableScanInfo->pBlockColMap);

162,703,535✔

2411

  taosMemoryFreeClear(param);

162,705,725✔

2412

162,675,999✔

2414

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

162,587,360✔

2415

  if (pBlock && dataRequireFlag == FUNC_DATA_REQUIRED_NOT_LOAD) {

162,587,360✔

2416

    int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

7,598,724✔

2417

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

19,314,380✔

2418

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

11,729,646✔

2419

      if (pCol) {

11,727,337✔

2420

        pCol->info.noData = true;

11,727,341✔

2424

162,573,370✔

2426

static int32_t resetTableScanOperatorState(SOperatorInfo* pOper) {

276,820✔

2427

  int32_t         code = TSDB_CODE_SUCCESS;

276,820✔

2428

  STableScanInfo*   pInfo = pOper->info;

276,820✔

2429

  pOper->status = OP_NOT_OPENED;

276,820✔

2431

  pInfo->scanTimes = 0;

276,820✔

2432

  pInfo->currentGroupId = -1;

276,820✔

2433

  pInfo->tableEndIndex = -1;

276,820✔

2434

  pInfo->tableStartIndex = 0;

276,820✔

2435

  pInfo->currentTable = 0;

276,820✔

2436

  pInfo->scanMode = 0;

276,820✔

2437

  pInfo->countState = 0;

276,820✔

2438

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

276,820✔

2440

  if (!pInfo->virtualStableScan) {

276,820✔

2446

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

150,968✔

2447

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

150,968✔

2449

    pInfo->base.dataReader = NULL;

150,968✔

2451

  tableListDestroy(pInfo->base.pTableListInfo);

276,820✔

2452

  pInfo->base.pTableListInfo = tableListCreate();

276,820✔

2453

  if (!pInfo->base.pTableListInfo) {

276,820✔

2457

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

276,820✔

2459

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

276,820✔

2460

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

276,820✔

2461

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

299,050✔

2463

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

299,050✔

2464

    if (code) {

149,936✔

2470

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

276,820✔

2471

  cleanupQueryTableDataCond(&pInfo->base.cond);

276,820✔

2472

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

276,820✔

2473

  if (code) {

276,820✔

2477

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

276,820✔

2478

    cleanupQueryTableDataCond(&pInfo->base.orgCond);

125,852✔

2479

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

125,852✔

2480

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

125,852✔

2483

  pOper->resultInfo.totalRows = 0;

276,820✔

2484

  blockDataEmpty(pInfo->pResBlock);

276,820✔

2485

  blockDataEmpty(pInfo->pOrgBlock);

276,820✔

2486

  taosHashClear(pInfo->pIgnoreTables);

276,820✔

2487

  taosArrayDestroy(pInfo->pBlockColMap);

276,820✔

2488

  pInfo->pBlockColMap = NULL;

276,820✔

2489

  return code;

276,820✔

2492

int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

162,702,584✔

2495

  QRY_PARAM_CHECK(pOptrInfo);

162,702,584✔

2497

  int32_t         code = TSDB_CODE_SUCCESS;

162,712,948✔

2498

  int32_t         lino = 0;

162,712,948✔

2499

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

162,712,948✔

2500

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

162,666,984✔

2501

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

162,664,080✔

2506

  pOperator->pPhyNode = pTableScanNode;

162,666,274✔

2507

  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;

162,668,985✔

2508

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

162,671,078✔

2510

  int32_t numOfCols = 0;

162,672,830✔

2512

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

162,687,868✔

2513

  QUERY_CHECK_CODE(code, lino, _error);

162,598,537✔

2515

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

162,598,537✔

2516

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

162,561,879✔

2517

  QUERY_CHECK_CODE(code, lino, _error);

162,519,030✔

2519

  if (pScanNode->pScanPseudoCols != NULL) {

162,519,030✔

2520

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

50,129,490✔

2521

    pSup->pExprInfo = NULL;

50,131,852✔

2522

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

50,117,595✔

2523

    QUERY_CHECK_CODE(code, lino, _error);

50,101,281✔

2525

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

50,101,281✔

2527

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

50,107,594✔

2530

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

162,574,389✔

2531

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

162,425,181✔

2533

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

162,587,319✔

2534

  pInfo->base.readHandle = *readHandle;

162,318,204✔

2535

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

162,376,011✔

2537

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

162,605,578✔

2538

  pInfo->sample.seed = taosGetTimestampSec();

162,383,337✔

2540

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

162,536,991✔

2541

  initResultSizeInfo(&pOperator->resultInfo, 4096);

162,518,593✔

2542

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

162,546,863✔

2543

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

162,597,003✔

2544

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

162,580,568✔

2546

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

162,428,947✔

2547

  QUERY_CHECK_CODE(code, lino, _error);

162,649,874✔

2549

  pInfo->virtualStableScan = pScanNode->virtualStableScan;

162,649,874✔

2550

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

162,580,979✔

2551

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

5,222,013✔

2552

    pInfo->pResBlock = NULL;

5,224,141✔

2553

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

5,219,579✔

2554

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

5,223,399✔

2557

  if (pInfo->virtualStableScan) {

162,542,593✔

2558

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

5,224,646✔

2559

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

5,227,372✔

2562

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

162,516,864✔

2563

                            pTaskInfo->pStreamRuntimeInfo);

162,584,125✔

2564

  QUERY_CHECK_CODE(code, lino, _error);

162,414,844✔

2566

  pInfo->currentGroupId = -1;

162,391,836✔

2568

  pInfo->tableEndIndex = -1;

162,418,544✔

2569

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

162,410,535✔

2570

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

162,584,131✔

2572

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

162,410,818✔

2574

  pOperator->exprSupp.numOfExprs = numOfCols;

162,357,544✔

2576

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

162,342,743✔

2577

  pInfo->ignoreTag = false;

162,438,228✔

2579

  pInfo->lastBatchIdx = -1;

162,593,283✔

2580

  pInfo->currentBatchIdx = 0;

162,482,239✔

2581

  pInfo->base.pTableListInfo = pTableListInfo;

162,307,613✔

2582

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

162,334,587✔

2592

  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

162,334,587✔

2594

  pOperator->fpSet = createOperatorFpSet(prepareTableScan, doTableScanNext,

162,489,991✔

2600

  setOperatorResetStateFn(pOperator, resetTableScanOperatorState);

162,419,524✔

2602

  pOperator->cost.openCost = 0;

162,389,395✔

2603

  *pOptrInfo = pOperator;

162,487,583✔

2604

  return TSDB_CODE_SUCCESS;

162,266,087✔

2606

_error:

23,008✔

2607

  if (pInfo != NULL) {

23,008✔

2608

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

23,008✔

2609

    destroyTableScanOperatorInfo(pInfo);

23,008✔

2612

  if (pOperator != NULL) {

23,008✔

2613

    pOperator->info = NULL;

23,008✔

2614

    destroyOperator(pOperator);

23,008✔

2616

  pTaskInfo->code = code;

23,008✔

2617

  return code;

23,008✔

2654

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

13,795✔

2655

  int32_t code = TSDB_CODE_SUCCESS;

13,795✔

2656

  int32_t lino = 0;

13,795✔

2657

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

13,795✔

2658

    return code;

1,696✔

2660

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

12,099✔

2661

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

12,099✔

2662

  bool hasUnqualified = false;

12,099✔

2664

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

12,099✔

2665

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

12,099✔

2667

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

12,099✔

2669

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

12,099✔

2672

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

12,099✔

2673

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

12,099✔

2674

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

43,373,137✔

2675

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

43,361,038✔

2676

    void*    data = colDataGetData(pColPk, i);

43,361,038✔

2677

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

43,363,374✔

2678

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

2,336✔

2681

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

2,336✔

2682

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

2,336✔

2683

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

2,336✔

2684

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

2,336✔

2685

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

2,336✔

2686

      taosMemoryFree(tmq);

2,336✔

2688

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

43,358,702✔

2691

    if (!p[i]) {

43,361,038✔

2692

      hasUnqualified = true;

15,442,504✔

2696

  if (hasUnqualified) {

12,099✔

2697

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

4,062✔

2698

    QUERY_CHECK_CODE(code, lino, _end);

4,062✔

2701

_end:

12,099✔

2702

  taosMemoryFree(p);

12,099✔

2703

  if (code != TSDB_CODE_SUCCESS) {

12,099✔

2706

  return code;

12,099✔

2709

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

1,459,391,247✔

2710

  int32_t p1 = *(int32_t*)param1;

1,459,391,247✔

2711

  int32_t p2 = *(int32_t*)param2;

1,459,465,146✔

2713

  if (p1 == p2) {

1,459,472,630✔

2716

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

1,459,472,630✔

2720

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

102,711,457✔

2721

  int32_t         code = TSDB_CODE_SUCCESS;

102,711,457✔

2722

  int32_t         lino = 0;

102,711,457✔

2723

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

102,711,457✔

2724

  SOperatorInfo*  pOperator = pInfo->pStreamScanOp;

102,724,201✔

2725

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

102,724,909✔

2726

  const char*     id = GET_TASKID(pTaskInfo);

102,725,263✔

2727

  SArray* pColList = NULL;

102,725,256✔

2729

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

102,725,256✔

2730

  QUERY_CHECK_CODE(code, lino, _end);

102,723,139✔

2732

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

102,723,139✔

2733

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

102,722,785✔

2734

  pBlockInfo->type = STREAM_NORMAL;

102,723,139✔

2735

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

102,720,307✔

2737

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

102,724,555✔

2738

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

102,725,610✔

2740

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

102,725,683✔

2741

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

102,722,070✔

2744

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

550,930,876✔

2745

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

448,197,795✔

2746

    if (!pColMatchInfo->needOutput) {

448,173,656✔

2747

      continue;

660✔

2750

    bool colExists = false;

448,178,370✔

2751

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

1,408,789,836✔

2752

      SColumnInfoData* pResCol = NULL;

1,408,360,764✔

2753

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

1,408,324,174✔

2754

      QUERY_CHECK_CODE(code, lino, _end);

1,408,301,182✔

2756

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

1,408,301,182✔

2757

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

448,216,188✔

2758

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

448,218,585✔

2759

        QUERY_CHECK_CODE(code, lino, _end);

448,062,622✔

2761

        colExists = true;

448,062,622✔

2762

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

448,062,622✔

2763

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

448,212,343✔

2764

        break;

448,212,343✔

2769

    if (!colExists) {

448,208,146✔

2778

  if (pInfo->numOfPseudoExpr > 0) {

102,723,466✔

2779

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

132,120,680✔

2780

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

66,060,418✔

2782

    if (code) {

66,059,710✔

2787

    terrno = 0;

66,059,710✔

2789

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

271,137,535✔

2790

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

205,069,329✔

2791

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

205,078,902✔

2795

  taosArraySort(pColList, colIdComparFn);

102,708,598✔

2797

  int32_t i = 0, j = 0;

102,720,081✔

2798

  while (i < taosArrayGetSize(pColList)) {

755,820,777✔

2799

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

653,087,476✔

2800

    if (slot1 > j) {

653,100,696✔

2805

      i += 1;

653,100,696✔

2806

      j += 1;

653,100,696✔

2810

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

102,721,346✔

2811

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

660✔

2812

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

660✔

2813

    j += 1;

660✔

2816

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

102,722,800✔

2817

  QUERY_CHECK_CODE(code, lino, _end);

102,722,173✔

2819

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

102,722,173✔

2820

  QUERY_CHECK_CODE(code, lino, _end);

102,721,178✔

2822

_end:

102,721,178✔

2823

  taosArrayDestroy(pColList);

101,524,685✔

2824

  if (code != TSDB_CODE_SUCCESS) {

102,721,129✔

2827

  return code;

102,718,627✔

2830

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

6,804,056✔

2831

  int32_t code = TSDB_CODE_SUCCESS;

6,804,056✔

2832

  int32_t lino = 0;

6,804,056✔

2833

  SValue  val = {0};

6,804,056✔

2834

  if (hasPrimaryKey) {

6,806,222✔

2835

    code = doBlockDataPrimaryKeyFilter(pBlock, offset);

13,795✔

2836

    if (code != TSDB_CODE_SUCCESS) {

13,795✔

2840

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

13,795✔

2842

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

13,795✔

2843

      return code;

3,770✔

2845

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

10,025✔

2846

    val.type = pColPk->info.type;

10,025✔

2847

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

10,025✔

2848

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

643✔

2851

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

643✔

2852

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

643✔

2853

      val.nData = varDataLen(tmp);

643✔

2854

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

643✔

2856

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

9,382✔

2859

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

6,802,452✔

2861

_end:

6,800,864✔

2862

  if (code != TSDB_CODE_SUCCESS) {

6,801,185✔

2865

  return code;

6,799,034✔

2868

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

130,024,782✔

2869

  int32_t        code = TSDB_CODE_SUCCESS;

130,024,782✔

2870

  int32_t        lino = 0;

130,024,782✔

2871

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

130,024,782✔

2872

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

130,026,552✔

2874

  SStreamScanInfo* pInfo = pOperator->info;

130,026,201✔

2875

  const char*      id = GET_TASKID(pTaskInfo);

130,026,913✔

2877

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

130,027,000✔

2879

  if (isTaskKilled(pTaskInfo)) {

130,027,888✔

2884

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

130,029,644✔

2885

    while (1) {

3,770✔

2886

      SSDataBlock* pResult = NULL;

6,090,922✔

2887

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

6,090,922✔

2888

      QUERY_CHECK_CODE(code, lino, _end);

6,090,281✔

2890

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

6,090,281✔

2891

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

6,045,887✔

2892

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

6,046,207✔

2893

        QUERY_CHECK_CODE(code, lino, _end);

6,044,925✔

2894

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

6,044,925✔

2895

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

6,046,507✔

2896

          (*ppRes) = pResult;

6,042,758✔

2897

          return code;

6,042,758✔

2904

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

44,394✔

2905

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

44,394✔

2907

    pTSInfo->base.dataReader = NULL;

44,394✔

2908

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

44,394✔

2909

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

44,394✔

2910

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

44,394✔

2911

      (*ppRes) = NULL;

13,101✔

2912

      return code;

13,101✔

2915

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

31,293✔

2918

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

123,973,785✔

2919

    while (1) {

779,600✔

2921

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

124,753,031✔

2923

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

124,728,880✔

2924

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

124,739,017✔

2927

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

124,746,013✔

2930

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

124,742,117✔

2932

      if (hasResult) {

124,737,230✔

2933

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

102,709,455✔

2935

        blockDataCleanup(pInfo->pRes);

102,718,270✔

2936

        code = setBlockIntoRes(pInfo, pRes);

102,720,661✔

2937

        QUERY_CHECK_CODE(code, lino, _end);

102,702,732✔

2938

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

102,702,732✔

2940

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

102,705,119✔

2941

          (*ppRes) = pInfo->pRes;

101,944,948✔

2942

          return code;

101,942,824✔

2945

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

22,027,775✔

2946

        (*ppRes) = NULL;

22,027,775✔

2947

        return code;

22,027,775✔

2972

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

336,581✔

2973

  int32_t code = TSDB_CODE_SUCCESS;

336,581✔

2974

  int32_t lino = 0;

336,581✔

2975

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

336,581✔

2976

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

337,479✔

2979

  int32_t size = 0;

337,479✔

2980

  code = tableListGetSize(pTableListInfo, &size);

337,479✔

2981

  QUERY_CHECK_CODE(code, lino, _end);

337,479✔

2982

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

9,614,949✔

2983

    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);

9,277,795✔

2984

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

9,268,157✔

2985

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

9,268,157✔

2986

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

9,277,470✔

2989

  (*ppArrayRes) = tableIdList;

337,154✔

2991

_end:

337,479✔

2992

  if (code != TSDB_CODE_SUCCESS) {

337,479✔

2995

  return code;

337,479✔

2998

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

1,533,817✔

2999

  int32_t        code = TSDB_CODE_SUCCESS;

1,533,817✔

3000

  int32_t        lino = 0;

1,533,817✔

3001

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,533,817✔

3002

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,533,817✔

3004

  SStreamRawScanInfo* pInfo = pOperator->info;

1,533,817✔

3005

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

1,533,817✔

3006

  pTaskInfo->streamInfo.btMetaRsp.batchMetaLen = NULL;

1,533,817✔

3008

  qDebug("tmqsnap doRawScan called");

1,533,817✔

3009

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

1,533,512✔

3010

    bool hasNext = false;

1,528,865✔

3011

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

1,528,865✔

3012

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

1,527,997✔

3013

      if (code != TSDB_CODE_SUCCESS) {

1,525,851✔

3019

    if (pInfo->dataReader && hasNext) {

1,526,719✔

3020

      if (isTaskKilled(pTaskInfo)) {

759,999✔

3025

      SSDataBlock* pBlock = NULL;

759,999✔

3026

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

759,999✔

3027

      QUERY_CHECK_CODE(code, lino, _end);

758,474✔

3029

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

758,474✔

3030

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

759,084✔

3031

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

759,389✔

3032

        QUERY_CHECK_CODE(code, lino, _end);

759,084✔

3033

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

759,084✔

3034

        (*ppRes) = pBlock;

759,999✔

3035

        return code;

759,999✔

3039

    SMetaTableInfo mtInfo = {0};

767,330✔

3040

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

768,866✔

3041

    QUERY_CHECK_CODE(code, lino, _end);

767,894✔

3042

    if (code != 0) {

767,894✔

3046

    STqOffsetVal offset = {0};

767,894✔

3047

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

768,256✔

3048

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

10,001✔

3049

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

10,001✔

3051

      SValue val = {0};

758,560✔

3052

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

758,560✔

3053

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

759,216✔

3056

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

768,866✔

3057

    QUERY_CHECK_CODE(code, lino, _end);

768,866✔

3058

    (*ppRes) = NULL;

768,866✔

3059

    return code;

768,866✔

3060

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

4,952✔

3061

    SSnapContext* sContext = pInfo->sContext;

4,952✔

3062

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

41,328✔

3063

      void*   data = NULL;

41,328✔

3064

      int32_t dataLen = 0;

41,328✔

3065

      int16_t type = 0;

41,328✔

3066

      int64_t uid = 0;

41,328✔

3067

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

41,328✔

3073

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

41,328✔

3074

        STqOffsetVal offset = {0};

4,952✔

3075

        SValue       val = {0};

4,952✔

3077

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

4,952✔

3078

        QUERY_CHECK_CODE(code, lino, _end);

4,952✔

3079

        break;

4,952✔

3081

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

36,376✔

3082

        SMqMetaRsp tmpMetaRsp = {0};

36,376✔

3083

        tmpMetaRsp.resMsgType = type;

36,376✔

3084

        tmpMetaRsp.metaRspLen = dataLen;

36,376✔

3085

        tmpMetaRsp.metaRsp = data;

36,376✔

3086

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

36,376✔

3087

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

4,404✔

3088

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

4,404✔

3090

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

4,404✔

3091

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

4,404✔

3093

        int32_t  tempRes = TSDB_CODE_SUCCESS;

36,376✔

3094

        uint32_t len = 0;

36,376✔

3095

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

36,376✔

3096

        if (TSDB_CODE_SUCCESS != tempRes) {

36,376✔

3102

        int32_t tLen = sizeof(SMqRspHead) + len;

36,376✔

3103

        void*   tBuf = taosMemoryCalloc(1, tLen);

36,376✔

3104

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

36,376✔

3106

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

36,376✔

3107

        SEncoder encoder = {0};

36,376✔

3108

        tEncoderInit(&encoder, metaBuff, len);

36,376✔

3109

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

36,376✔

3110

        if (tempLen < 0) {

36,376✔

3117

        taosMemoryFreeClear(data);

36,376✔

3118

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

36,102✔

3119

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

36,376✔

3121

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

36,376✔

3122

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

36,376✔

3126

    (*ppRes) = NULL;

4,952✔

3127

    return code;

4,952✔

3141

static void destroyTmqRawScanOperatorInfo(void* param) {

122,575✔

3142

  SStreamRawScanInfo* pRawScan = (SStreamRawScanInfo*)param;

122,575✔

3143

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

122,575✔

3144

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

122,575✔

3145

  tableListDestroy(pRawScan->pTableListInfo);

122,224✔

3146

  taosMemoryFree(pRawScan);

121,873✔

3147

122,224✔

3152

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

122,575✔

3158

  QRY_PARAM_CHECK(pOptrInfo);

122,575✔

3159

  int32_t code = TSDB_CODE_SUCCESS;

122,575✔

3160

  int32_t lino = 0;

122,575✔

3162

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

122,575✔

3163

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

122,575✔

3164

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

122,575✔

3170

  pInfo->pTableListInfo = tableListCreate();

122,575✔

3171

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

122,575✔

3172

  pInfo->vnode = pHandle->vnode;

122,575✔

3173

  pInfo->pAPI = &pTaskInfo->storageAPI;

122,575✔

3175

  pInfo->sContext = pHandle->sContext;

122,575✔

3176

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

122,575✔

3177

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

122,575✔

3180

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

122,575✔

3182

  *pOptrInfo = pOperator;

122,575✔

3183

  return code;

122,575✔

3195

void destroyTmqScanOperatorInfo(void* param) {

337,479✔

3196

  if (param == NULL) {

337,479✔

3200

  SStreamScanInfo* pStreamScan = (SStreamScanInfo*)param;

337,479✔

3201

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

337,479✔

3202

    destroyOperator(pStreamScan->pTableScanOp);

337,479✔

3205

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

337,479✔

3206

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

337,479✔

3208

  if (pStreamScan->pVtableMergeHandles) {

337,479✔

3212

  if (pStreamScan->pVtableMergeBuf) {

337,479✔

3216

  if (pStreamScan->pVtableReadyHandles) {

337,479✔

3220

  if (pStreamScan->pTableListInfo) {

337,479✔

3224

  if (pStreamScan->matchInfo.pList) {

337,479✔

3225

    taosArrayDestroy(pStreamScan->matchInfo.pList);

337,479✔

3227

  if (pStreamScan->pPseudoExpr) {

337,479✔

3228

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

195,381✔

3229

    taosMemoryFree(pStreamScan->pPseudoExpr);

195,381✔

3232

  cleanupExprSupp(&pStreamScan->tbnameCalSup);

337,479✔

3233

  cleanupExprSupp(&pStreamScan->tagCalSup);

337,479✔

3235

  blockDataDestroy(pStreamScan->pRes);

337,479✔

3236

  blockDataDestroy(pStreamScan->pUpdateRes);

337,479✔

3237

  blockDataDestroy(pStreamScan->pDeleteDataRes);

337,479✔

3238

  blockDataDestroy(pStreamScan->pUpdateDataRes);

337,479✔

3239

  blockDataDestroy(pStreamScan->pCreateTbRes);

337,479✔

3240

  taosArrayDestroy(pStreamScan->pBlockLists);

337,479✔

3241

  blockDataDestroy(pStreamScan->pCheckpointRes);

337,479✔

3243

  taosMemoryFree(pStreamScan);

337,479✔

3263

 int32_t createTmqScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,

336,917✔

3266

  QRY_PARAM_CHECK(pOptrInfo);

336,917✔

3268

  int32_t          code = TSDB_CODE_SUCCESS;

337,479✔

3269

  int32_t          lino = 0;

337,479✔

3270

  SArray*          pColIds = NULL;

337,479✔

3271

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

337,479✔

3272

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

337,479✔

3273

  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;

337,479✔

3274

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

337,479✔

3276

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

337,479✔

3281

  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;

337,479✔

3282

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;

337,479✔

3284

  pInfo->pTagCond = pTagCond;

337,479✔

3285

  pInfo->pGroupTags = pTableScanNode->pGroupTags;

337,479✔

3287

  int32_t numOfCols = 0;

337,479✔

3288

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

337,479✔

3289

  if (code != TSDB_CODE_SUCCESS) {

337,479✔

3293

  SDataType pkType = {0};

337,479✔

3294

  pInfo->primaryKeyIndex = -1;

337,479✔

3295

  pInfo->basic.primaryPkIndex = -1;

337,479✔

3296

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

337,479✔

3297

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

337,479✔

3298

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

337,479✔

3300

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

2,116,120✔

3301

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

1,778,641✔

3302

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

1,778,641✔

3304

    int16_t colId = id->colId;

1,778,641✔

3305

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

1,778,641✔

3306

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

1,778,641✔

3308

    if (id->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

1,778,641✔

3309

      pInfo->primaryTsIndex = id->dstSlotId;

337,479✔

3311

    if (id->isPk) {

1,778,641✔

3312

      pInfo->primaryKeyIndex = id->dstSlotId;

1,515✔

3313

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

1,515✔

3314

      pkType = id->dataType;

1,515✔

3318

  pInfo->pPartTbnameSup = NULL;

337,479✔

3319

  if (pTableScanNode->pSubtable != NULL) {

337,479✔

3335

  if (pTableScanNode->pTags != NULL) {

337,479✔

3347

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

337,479✔

3348

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

337,479✔

3350

  if (pHandle->vnode) {

337,479✔

3351

    SOperatorInfo* pTableScanOp = NULL;

337,479✔

3352

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

337,479✔

3353

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

337,158✔

3354

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

883✔

3358

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

336,275✔

3359

    if (pHandle->version > 0) {

336,596✔

3360

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

336,556✔

3363

    STableKeyInfo* pList = NULL;

336,222✔

3364

    int32_t        num = 0;

336,556✔

3365

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

336,877✔

3366

    QUERY_CHECK_CODE(code, lino, _error);

337,479✔

3368

    if (pHandle->initTableReader) {

337,479✔

3369

      pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;

336,824✔

3370

      pTSInfo->base.dataReader = NULL;

336,824✔

3373

    if (pHandle->initTqReader) {

337,479✔

3374

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

337,158✔

3375

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

337,479✔

3381

    pInfo->pUpdateInfo = NULL;

337,479✔

3382

    pInfo->pTableScanOp = pTableScanOp;

337,198✔

3383

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

337,479✔

3388

    pInfo->readHandle = *pHandle;

337,479✔

3389

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

337,479✔

3390

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

337,198✔

3391

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

337,479✔

3392

    pInfo->hasPart = false;

337,479✔

3394

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

337,479✔

3395

    QUERY_CHECK_CODE(code, lino, _error);

337,479✔

3398

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

337,479✔

3399

    QUERY_CHECK_CODE(code, lino, _error);

337,479✔

3401

    SArray* tableIdList = NULL;

337,479✔

3402

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

337,479✔

3403

    QUERY_CHECK_CODE(code, lino, _error);

337,479✔

3404

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

337,479✔

3405

    QUERY_CHECK_CODE(code, lino, _error);

337,479✔

3406

    taosArrayDestroy(tableIdList);

337,479✔

3407

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

337,479✔

3414

  pColIds = NULL;

337,479✔

3417

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

337,479✔

3418

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

195,381✔

3419

    QUERY_CHECK_CODE(code, lino, _error);

195,381✔

3422

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

337,479✔

3423

  QUERY_CHECK_CODE(code, lino, _error);

337,479✔

3425

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

337,479✔

3426

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

337,479✔

3427

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

337,479✔

3428

  QUERY_CHECK_CODE(code, lino, _error);

337,479✔

3430

  pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;

337,479✔

3431

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

337,479✔

3432

  pInfo->groupId = 0;

337,479✔

3433

  pInfo->igCheckGroupId = false;

337,479✔

3434

  pInfo->pStreamScanOp = pOperator;

337,479✔

3435

  pInfo->deleteDataIndex = 0;

337,479✔

3436

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

336,247✔

3437

  QUERY_CHECK_CODE(code, lino, _error);

337,479✔

3439

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

337,479✔

3440

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

337,479✔

3441

  QUERY_CHECK_CODE(code, lino, _error);

337,125✔

3443

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

337,125✔

3444

  pInfo->partitionSup.needCalc = false;

337,125✔

3445

  pInfo->igCheckUpdate = pTableScanNode->igCheckUpdate;

337,125✔

3446

  pInfo->igExpired = pTableScanNode->igExpired;

337,125✔

3447

  pInfo->twAggSup.maxTs = INT64_MIN;

336,844✔

3448

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

337,479✔

3449

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

337,479✔

3450

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

337,163✔

3451

  pInfo->pFillSup = NULL;

337,479✔

3452

  pInfo->useGetResultRange = false;

337,163✔

3453

  pInfo->pRangeScanRes = NULL;

337,125✔

3455

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

337,163✔

3456

  QUERY_CHECK_CODE(code, lino, _error);

337,479✔

3458

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

337,479✔

3460

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

337,479✔

3462

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

337,479✔

3465

  *pOptrInfo = pOperator;

337,479✔

3466

  return code;

337,479✔

3489

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

9,312,222✔

3490

  int32_t        code = TSDB_CODE_SUCCESS;

9,312,222✔

3491

  int32_t        lino = 0;

9,312,222✔

3492

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,312,222✔

3493

  STagScanInfo*  pInfo = pOperator->info;

9,313,246✔

3494

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

9,312,037✔

3495

  int32_t        count = pRes->info.rows;

9,309,662✔

3497

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

9,312,222✔

3498

  if (!item) {

9,310,359✔

3505

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

9,310,359✔

3506

  tDecoderClear(&(*mr).coder);

9,313,242✔

3507

  if (code != TSDB_CODE_SUCCESS) {

9,311,705✔

3513

  char str[512];

9,311,601✔

3514

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

42,225,786✔

3515

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

32,912,398✔

3518

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

32,912,203✔

3519

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

3,762,846✔

3520

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

3,763,363✔

3521

      QUERY_CHECK_CODE(code, lino, _end);

3,762,851✔

3522

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

29,147,684✔

3523

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

5,352,135✔

3524

      QUERY_CHECK_CODE(code, lino, _end);

5,352,974✔

3525

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

23,796,203✔

3526

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

15,612✔

3527

      QUERY_CHECK_CODE(code, lino, _end);

15,612✔

3529

      STagVal val = {0};

23,781,942✔

3530

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

23,782,824✔

3531

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

23,783,521✔

3533

      char* data = NULL;

23,784,545✔

3534

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

23,784,545✔

3535

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

23,782,950✔

3537

        data = (char*)p;

1,410✔

3540

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

23,786,266✔

3541

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

23,786,266✔

3542

      QUERY_CHECK_CODE(code, lino, _end);

23,785,711✔

3544

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

23,785,711✔

3546

        taosMemoryFree(data);

8,416,114✔

3551

_end:

9,317,342✔

3552

  if (code != TSDB_CODE_SUCCESS) {

9,313,242✔

3556

    pRes->info.rows++;

9,313,242✔

3559

  return code;

9,313,061✔

3562

static void tagScanFreeUidTag(void* p) {

7,199,204✔

3563

  STUidTagInfo* pInfo = p;

7,199,204✔

3564

  if (pInfo->pTagVal != NULL) {

7,199,204✔

3565

    taosMemoryFree(pInfo->pTagVal);

7,197,750✔

3567

7,197,719✔

3569

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

1,604,692✔

3570

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

1,604,692✔

3571

  if (pColumnData == NULL) {

1,604,328✔

3575

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

1,604,328✔

3576

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

1,604,338✔

3577

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

1,603,219✔

3578

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

1,603,365✔

3580

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

1,604,338✔

3581

  if (code != TSDB_CODE_SUCCESS) {

1,602,994✔

3587

  pParam->columnData = pColumnData;

1,602,994✔

3588

  pParam->colAlloced = true;

1,604,197✔

3589

  return TSDB_CODE_SUCCESS;

1,604,197✔

3592

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

9,706,454✔

3593

  int32_t                code = TSDB_CODE_SUCCESS;

9,706,454✔

3594

  int32_t                lino = 0;

9,706,454✔

3595

  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;

9,706,454✔

3596

  SColumnNode*           pSColumnNode = NULL;

9,706,454✔

3597

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

9,706,811✔

3598

    pSColumnNode = *(SColumnNode**)pNode;

2,710,850✔

3599

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

6,997,940✔

3600

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

516✔

3601

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

516✔

3614

      return DEAL_RES_CONTINUE;

516✔

3617

    return DEAL_RES_CONTINUE;

6,996,575✔

3620

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

2,711,342✔

3621

  if (!data) {

2,709,011✔

3622

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

1,605,413✔

3623

    if (code == TSDB_CODE_DUP_KEY) {

1,607,390✔

3626

    QUERY_CHECK_CODE(code, lino, _end);

1,607,390✔

3627

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

1,607,390✔

3628

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

1,606,398✔

3629

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

1,604,923✔

3630

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

1,606,410✔

3631

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

1,604,559✔

3632

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

1,607,247✔

3634

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

1,103,598✔

3635

    pSColumnNode->slotId = col->slotId;

1,103,598✔

3638

_end:

2,709,986✔

3639

  if (code != TSDB_CODE_SUCCESS) {

2,710,842✔

3640

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

500✔

3643

  return DEAL_RES_CONTINUE;

2,710,342✔

3646

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

1,604,692✔

3648

  int32_t code = TSDB_CODE_SUCCESS;

1,604,692✔

3649

  int32_t lino = 0;

1,604,692✔

3650

  int32_t numOfTables = taosArrayGetSize(aUidTags);

1,604,692✔

3651

  SArray* pBlockList = NULL;

1,604,692✔

3653

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

1,604,692✔

3654

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

1,603,464✔

3656

  pBlockList = taosArrayInit(1, POINTER_BYTES);

1,603,464✔

3657

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

1,603,464✔

3659

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

1,604,692✔

3660

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

1,604,692✔

3662

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

1,604,692✔

3664

  SScalarParam output = {0};

1,604,692✔

3665

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

1,604,214✔

3666

  QUERY_CHECK_CODE(code, lino, _end);

1,604,197✔

3668

  gTaskScalarExtra.pStreamInfo = (void*)pStreamRuntimeInfo;

1,604,197✔

3669

  gTaskScalarExtra.pStreamRange = NULL;

1,604,197✔

3670

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

1,603,702✔

3671

  QUERY_CHECK_CODE(code, lino, _end);

1,601,142✔

3673

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

1,601,142✔

3674

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

4,008,331✔

3675

    if (result[i]) {

2,404,843✔

3676

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

2,245,263✔

3677

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

2,245,263✔

3681

  colDataDestroy(output.columnData);

1,603,202✔

3682

  taosMemoryFreeClear(output.columnData);

1,603,464✔

3684

_end:

1,601,771✔

3685

  blockDataDestroy(pResBlock);

1,601,390✔

3686

  taosArrayDestroy(pBlockList);

1,601,157✔

3687

  if (code != TSDB_CODE_SUCCESS) {

1,597,300✔

3690

  return code;

1,597,300✔

3693

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

29,275,743✔

3696

  int32_t code = TSDB_CODE_SUCCESS;

29,275,743✔

3697

  int32_t lino = 0;

29,275,743✔

3698

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

29,275,743✔

3699

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

13,982,162✔

3705

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

13,975,633✔

3706

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

6,982,832✔

3707

      QUERY_CHECK_CODE(code, lino, _end);

6,988,416✔

3708

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

6,990,404✔

3709

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

6,987,050✔

3710

      QUERY_CHECK_CODE(code, lino, _end);

6,990,268✔

3713

    STagVal tagVal = {0};

15,285,872✔

3714

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

15,290,574✔

3715

    if (pUidTagInfo->pTagVal == NULL) {

15,291,880✔

3718

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

15,286,971✔

3720

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

15,290,635✔

3721

        colDataSetNULL(pColInfo, rowIndex);

6,542✔

3722

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

15,286,291✔

3725

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

19,101,981✔

3726

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

3,823,285✔

3729

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

3,824,617✔

3730

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

3,823,278✔

3732

        varDataSetLen(tmp, tagVal.nData);

3,823,278✔

3733

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

3,823,968✔

3734

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

3,822,190✔

3735

        taosMemoryFree(tmp);

3,824,052✔

3736

        QUERY_CHECK_CODE(code, lino, _end);

3,824,248✔

3738

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

11,447,746✔

3739

        QUERY_CHECK_CODE(code, lino, _end);

11,459,481✔

3744

_end:

29,265,914✔

3745

  if (code != TSDB_CODE_SUCCESS) {

29,265,914✔

3748

  return code;

29,264,027✔

3751

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

4,605,000✔

3753

  int32_t       code = TSDB_CODE_SUCCESS;

4,605,000✔

3754

  int32_t       lino = 0;

4,605,000✔

3755

  STagScanInfo* pInfo = pOperator->info;

4,605,000✔

3756

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

4,605,839✔

3757

  if (!ignoreFilterIdx) {

4,600,933✔

3758

    size_t szTables = taosArrayGetSize(aFilterIdxs);

1,598,193✔

3759

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

3,840,619✔

3760

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

2,240,716✔

3761

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);

2,240,093✔

3762

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

2,236,558✔

3763

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

12,741,450✔

3764

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

10,498,674✔

3765

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

10,505,491✔

3766

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

10,505,491✔

3767

        QUERY_CHECK_CODE(code, lino, _end);

10,504,892✔

3771

    size_t szTables = taosArrayGetSize(aUidTags);

3,002,740✔

3772

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

7,803,823✔

3773

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);

4,800,817✔

3774

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

4,800,817✔

3775

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

23,560,529✔

3776

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

18,759,067✔

3777

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

18,759,945✔

3778

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

18,759,945✔

3779

        QUERY_CHECK_CODE(code, lino, _end);

18,759,712✔

3784

_end:

3,003,006✔

3785

  if (code != TSDB_CODE_SUCCESS) {

4,602,909✔

3788

  return code;

4,605,359✔

3791

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

9,171,862✔

3792

  int32_t        code = TSDB_CODE_SUCCESS;

9,171,862✔

3793

  int32_t        lino = 0;

9,171,862✔

3794

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,171,862✔

3795

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

9,172,404✔

3797

  STagScanInfo* pInfo = pOperator->info;

9,173,193✔

3798

  SSDataBlock*  pRes = pInfo->pRes;

9,175,259✔

3800

  QRY_PARAM_CHECK(ppRes);

9,172,344✔

3802

  if (pOperator->status == OP_EXEC_DONE) {

9,171,346✔

3803

    return TSDB_CODE_SUCCESS;

4,489,629✔

3805

  blockDataCleanup(pRes);

4,679,528✔

3807

  if (pInfo->pCtbCursor == NULL) {

4,681,080✔

3808

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

4,681,434✔

3809

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

4,680,571✔

3815

  SArray* aUidTags = pInfo->aUidTags;

4,679,576✔

3816

  SArray* aFilterIdxs = pInfo->aFilterIdxs;

4,681,922✔

3817

  int32_t count = 0;

4,680,602✔

3818

  bool    ctbCursorFinished = false;

4,680,602✔

3819

  while (1) {

101,693✔

3820

    taosArrayClearEx(aUidTags, tagScanFreeUidTag);

4,782,295✔

3821

    taosArrayClear(aFilterIdxs);

4,783,238✔

3823

    int32_t numTables = 0;

4,782,238✔

3824

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

11,993,304✔

3825

      SMCtbCursor* pCur = pInfo->pCtbCursor;

11,986,696✔

3826

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

11,983,994✔

3827

      if (uid == 0) {

11,985,299✔

3828

        ctbCursorFinished = true;

4,784,080✔

3829

        break;

4,784,080✔

3831

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

7,201,219✔

3832

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

7,200,868✔

3833

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

7,196,990✔

3835

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

7,196,990✔

3836

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

7,204,045✔

3837

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

7,204,045✔

3838

      ++numTables;

7,204,045✔

3841

    if (numTables == 0) {

4,784,464✔

3842

      break;

176,271✔

3844

    bool ignoreFilterIdx = true;

4,608,193✔

3845

    if (pInfo->pTagCond != NULL) {

4,608,193✔

3846

      ignoreFilterIdx = false;

1,604,692✔

3847

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

1,604,692✔

3848

      QUERY_CHECK_CODE(code, lino, _end);

1,601,848✔

3850

      ignoreFilterIdx = true;

3,003,501✔

3853

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

4,605,349✔

3854

    QUERY_CHECK_CODE(code, lino, _end);

4,605,359✔

3856

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

4,605,359✔

3858

    if (count != 0) {

4,605,238✔

3859

      break;

4,503,545✔

3863

  if (count > 0) {

4,679,816✔

3864

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

4,503,669✔

3866

  if (count == 0 || ctbCursorFinished) {

4,681,535✔

3867

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

4,681,535✔

3868

    pInfo->pCtbCursor = NULL;

4,676,458✔

3869

    setOperatorCompleted(pOperator);

4,676,290✔

3872

  pRes->info.rows = count;

4,676,036✔

3873

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

4,675,063✔

3874

  if (bLimitReached) {

4,678,763✔

3878

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

4,678,763✔

3879

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

4,675,667✔

3881

_end:

4,669,954✔

3882

  if (code != TSDB_CODE_SUCCESS) {

4,669,954✔

3888

  return code;

4,669,954✔

3891

static int32_t createTagScanTableListInfoFromParam(SOperatorInfo* pOperator) {

3,487,889✔

3892

  STagScanInfo*          pInfo = pOperator->info;

3,487,889✔

3893

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

3,487,889✔

3894

  int32_t                code = 0;

3,487,889✔

3895

  STableListInfo*        pListInfo = pInfo->pTableListInfo;

3,487,889✔

3896

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

3,487,889✔

3897

  tb_uid_t               pUid = pParam->vcUid;

3,487,889✔

3903

  STableKeyInfo info = {.groupId = 0};

3,487,889✔

3905

  int32_t tableIdx = 0;

3,487,889✔

3906

  taosHashClear(pListInfo->map);

3,487,889✔

3907

  taosArrayClear(pListInfo->pTableList);

3,487,889✔

3909

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

3,487,889✔

3916

  info.uid = pUid;

3,487,889✔

3917

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

3,487,889✔

3918

  if (p == NULL) {

3,487,889✔

3922

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

3,487,889✔

3924

  return code;

3,487,889✔

3927

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

6,668,003✔

3928

  int32_t       code = TSDB_CODE_SUCCESS;

6,668,003✔

3929

  int32_t       lino = 0;

6,668,003✔

3930

  STagScanInfo* pInfo = pOperator->info;

6,668,003✔

3931

  if (pOperator->pOperatorGetParam) {

6,668,515✔

3932

    pOperator->resultInfo.totalRows = 0;

3,487,889✔

3933

    pOperator->dynamicTask = true;

3,487,889✔

3934

    pInfo->curPos = 0;

3,487,889✔

3935

    code = createTagScanTableListInfoFromParam(pOperator);

3,487,889✔

3936

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

3,487,889✔

3937

    pOperator->pOperatorGetParam = NULL;

3,487,889✔

3938

    QUERY_CHECK_CODE(code, lino, _end);

3,487,889✔

3940

    if (pOperator->status == OP_EXEC_DONE) {

3,487,889✔

3941

      pOperator->status = OP_OPENED;

1,878,015✔

3945

  if (pOperator->status == OP_EXEC_DONE) {

6,663,950✔

3946

    (*ppRes) = NULL;

1,575,966✔

3947

    return code;

1,575,966✔

3950

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

5,090,686✔

3951

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

5,090,501✔

3953

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

5,088,496✔

3954

  SSDataBlock* pRes = pInfo->pRes;

5,092,549✔

3955

  blockDataCleanup(pRes);

5,092,549✔

3957

  int32_t size = 0;

5,092,549✔

3958

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

5,092,549✔

3959

  QUERY_CHECK_CODE(code, lino, _end);

5,091,198✔

3961

  if (size == 0) {

5,091,198✔

3962

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

19,060✔

3963

    (*ppRes) = NULL;

19,060✔

3964

    return code;

19,060✔

3967

  SMetaReader mr = {0};

5,072,138✔

3968

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

5,072,465✔

3969

  pRes->info.rows = 0;

5,074,001✔

3971

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

14,386,730✔

3972

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

9,309,520✔

3973

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

9,314,085✔

3976

      code = TSDB_CODE_SUCCESS;

9,314,597✔

3978

    QUERY_CHECK_CODE(code, lino, _end);

9,314,085✔

3980

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

9,314,085✔

3981

      setOperatorCompleted(pOperator);

5,074,001✔

3985

  pAPI->metaReaderFn.clearReader(&mr);

5,074,845✔

3986

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

5,074,001✔

3987

  if (bLimitReached) {

5,074,001✔

3992

  if (pOperator->status == OP_EXEC_DONE) {

5,074,001✔

3993

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

5,074,001✔

3996

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

5,074,001✔

3998

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

5,074,001✔

3999

_end:

5,074,001✔

4000

  if (code != TSDB_CODE_SUCCESS) {

5,074,001✔

4005

  return code;

5,074,001✔

4008

static void destroyTagScanOperatorInfo(void* param) {

8,228,283✔

4009

  STagScanInfo* pInfo = (STagScanInfo*)param;

8,228,283✔

4010

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

8,228,283✔

4013

  taosHashCleanup(pInfo->filterCtx.colHash);

8,231,152✔

4014

  taosArrayDestroy(pInfo->filterCtx.cInfoList);

8,216,194✔

4015

  taosArrayDestroy(pInfo->aFilterIdxs);

8,222,575✔

4016

  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);

8,217,863✔

4018

  blockDataDestroy(pInfo->pRes);

8,226,260✔

4019

  taosArrayDestroy(pInfo->matchInfo.pList);

8,224,913✔

4020

  tableListDestroy(pInfo->pTableListInfo);

8,224,106✔

4022

  pInfo->pRes = NULL;

8,230,078✔

4023

  pInfo->pTableListInfo = NULL;

8,230,382✔

4024

  taosMemoryFreeClear(param);

8,233,144✔

4025

8,227,493✔

4027

static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {

4,740✔

4028

  int32_t         code = TSDB_CODE_SUCCESS;

4,740✔

4029

  STagScanInfo*   pInfo = pOper->info;

4,740✔

4031

  pOper->status = OP_NOT_OPENED;

4,740✔

4032

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

4,740✔

4033

  if (pTagScanNode->onlyMetaCtbIdx) {

4,740✔

4034

      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

2,064✔

4035

      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,064✔

4036

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

2,064✔

4037

      pInfo->pCtbCursor = NULL;

2,064✔

4039

      pInfo->curPos = 0;

2,676✔

4042

  return code;

4,740✔

4045

int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,

8,228,500✔

4048

  QRY_PARAM_CHECK(pOptrInfo);

8,228,500✔

4050

  int32_t         code = TSDB_CODE_SUCCESS;

8,222,642✔

4051

  int32_t         lino = 0;

8,222,642✔

4052

  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;

8,222,642✔

4053

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

8,222,642✔

4054

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

8,216,933✔

4055

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

8,212,904✔

4059

  pOperator->pPhyNode = pTagScanNode;

8,213,268✔

4060

  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

8,217,923✔

4062

  int32_t    numOfExprs = 0;

8,219,672✔

4063

  SExprInfo* pExprInfo = NULL;

8,218,150✔

4065

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

8,217,866✔

4066

  QUERY_CHECK_CODE(code, lino, _error);

8,224,794✔

4068

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

8,224,794✔

4069

  QUERY_CHECK_CODE(code, lino, _error);

8,217,448✔

4071

  int32_t num = 0;

8,217,448✔

4072

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

8,219,971✔

4073

  QUERY_CHECK_CODE(code, lino, _error);

8,217,680✔

4075

  pInfo->pTagCond = pTagCond;

8,217,680✔

4076

  pInfo->pTagIndexCond = pTagIndexCond;

8,225,383✔

4077

  pInfo->suid = pPhyNode->suid;

8,213,300✔

4078

  pInfo->pStorageAPI = &pTaskInfo->storageAPI;

8,212,540✔

4080

  pInfo->pTableListInfo = pTableListInfo;

8,219,239✔

4081

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

8,212,279✔

4082

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

8,225,678✔

4084

  pInfo->readHandle = *pReadHandle;

8,216,883✔

4085

  pInfo->curPos = 0;

8,224,587✔

4087

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

8,205,689✔

4088

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

8,208,907✔

4090

  initResultSizeInfo(&pOperator->resultInfo, 4096);

8,211,060✔

4091

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

8,224,775✔

4092

  QUERY_CHECK_CODE(code, lino, _error);

8,233,954✔

4094

  if (pTagScanNode->onlyMetaCtbIdx) {

8,233,954✔

4095

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

4,687,142✔

4096

    QUERY_CHECK_NULL(pInfo->aUidTags, code, lino, _error, terrno);

4,688,254✔

4098

    pInfo->aFilterIdxs = taosArrayInit(pOperator->resultInfo.capacity, sizeof(int32_t));

4,687,405✔

4099

    QUERY_CHECK_NULL(pInfo->aFilterIdxs, code, lino, _error, terrno);

4,687,902✔

4101

    pInfo->filterCtx.colHash =

4,686,081✔

4102

        taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);

4,685,719✔

4103

    QUERY_CHECK_NULL(pInfo->filterCtx.colHash, code, lino, _error, terrno);

4,686,517✔

4105

    pInfo->filterCtx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));

4,685,193✔

4106

    QUERY_CHECK_NULL(pInfo->filterCtx.cInfoList, code, lino, _error, terrno);

4,686,916✔

4108

    if (pInfo->pTagCond != NULL) {

4,684,334✔

4109

      nodesRewriteExprPostOrder(&pTagCond, tagScanRewriteTagColumn, (void*)&pInfo->filterCtx);

1,607,885✔

4112

  __optr_fn_t tagScanNextFn = (pTagScanNode->onlyMetaCtbIdx) ? doTagScanFromCtbIdxNext : doTagScanFromMetaEntryNext;

8,234,179✔

4113

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, tagScanNextFn, NULL, destroyTagScanOperatorInfo,

8,231,985✔

4115

  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);

8,226,764✔

4117

  *pOptrInfo = pOperator;

8,229,984✔

4118

  return code;

8,224,043✔

4135

static int32_t subTblRowCompareTsFn(const void* pLeft, const void* pRight, void* param) {

240,278,016✔

4136

  int32_t                 left = *(int32_t*)pLeft;

240,278,016✔

4137

  int32_t                 right = *(int32_t*)pRight;

240,279,108✔

4138

  STmsSubTablesMergeInfo* pInfo = (STmsSubTablesMergeInfo*)param;

240,278,380✔

4140

  int32_t leftIdx = pInfo->aInputs[left].rowIdx;

240,278,380✔

4141

  int32_t rightIdx = pInfo->aInputs[right].rowIdx;

240,278,744✔

4143

  if (leftIdx == -1) {

240,279,108✔

4144

    return 1;

205,955,268✔

4145

  } else if (rightIdx == -1) {

34,323,840✔

4146

    return -1;

339,840✔

4149

  int64_t leftTs = pInfo->aInputs[left].aTs[leftIdx];

33,984,000✔

4150

  int64_t rightTs = pInfo->aInputs[right].aTs[rightIdx];

33,984,000✔

4151

  int32_t ret = leftTs > rightTs ? 1 : ((leftTs < rightTs) ? -1 : 0);

33,984,000✔

4152

  if (pInfo->pTsOrderInfo->order == TSDB_ORDER_DESC) {

33,984,000✔

4155

  return ret;

33,984,000✔

4185

int32_t dumpQueryTableCond(const SQueryTableDataCond* src, SQueryTableDataCond* dst) {

3,049,164✔

4186

  int32_t code = TSDB_CODE_SUCCESS;

3,049,164✔

4187

  int32_t lino = 0;

3,049,164✔

4188

  memcpy((void*)dst, (void*)src, sizeof(SQueryTableDataCond));

3,049,164✔

4189

  dst->colList = taosMemoryCalloc(src->numOfCols, sizeof(SColumnInfo));

3,049,164✔

4190

  QUERY_CHECK_NULL(dst->colList, code, lino, _end, terrno);

3,031,080✔

4191

  for (int i = 0; i < src->numOfCols; i++) {

9,992,064✔

4192

    dst->colList[i] = src->colList[i];

6,965,796✔

4195

_end:

3,050,236✔

4196

  if (code != TSDB_CODE_SUCCESS) {

3,050,236✔

4199

  return code;

3,049,518✔

4202

static int32_t fetchNextSubTableBlockFromReader(SOperatorInfo* pOperator, STmsSubTableInput* pInput,

4,582,972✔

4204

  int32_t code = 0;

4,582,972✔

4206

  STableMergeScanInfo*    pInfo = pOperator->info;

4,582,972✔

4207

  SReadHandle*            pHandle = &pInfo->base.readHandle;

4,582,972✔

4208

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

4,583,336✔

4209

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

4,584,034✔

4210

  const SStorageAPI*      pAPI = &pTaskInfo->storageAPI;

4,581,182✔

4212

  blockDataCleanup(pInput->pReaderBlock);

4,580,484✔

4213

  if (!pInput->bInMemReader) {

4,583,700✔

4214

    code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, pInput->pKeyInfo, 1, pInput->pReaderBlock,

4,594,838✔

4215

                                         (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

2,297,232✔

4216

    if (code != 0) {

2,295,118✔

4221

  pInfo->base.dataReader = pInput->pReader;

4,581,576✔

4223

  while (true) {

8,736✔

4224

    bool hasNext = false;

4,590,980✔

4225

    code = pAPI->tsdReader.tsdNextDataBlock(pInfo->base.dataReader, &hasNext);

4,590,262✔

4226

    if (code != 0) {

4,593,154✔

4232

    if (!hasNext || isTaskKilled(pTaskInfo)) {

4,593,154✔

4233

      if (isTaskKilled(pTaskInfo)) {

3,060,542✔

4239

      *pSubTableHasBlock = false;

3,062,686✔

4240

      break;

3,061,604✔

4243

    if (pInput->tblCond.order == TSDB_ORDER_ASC) {

1,532,612✔

4244

      pInput->tblCond.twindows.skey = pInput->pReaderBlock->info.window.ekey + 1;

1,510,044✔

4246

      pInput->tblCond.twindows.ekey = pInput->pReaderBlock->info.window.skey - 1;

22,568✔

4249

    uint32_t status = 0;

1,532,612✔

4250

    code = loadDataBlock(pOperator, &pInfo->base, pInput->pReaderBlock, &status);

1,532,612✔

4251

    if (code != 0) {

1,532,612✔

4256

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

1,532,612✔

4260

    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pInput->pReaderBlock->info.rows == 0) {

1,532,612✔

4261

      continue;

8,736✔

4264

    *pSubTableHasBlock = true;

1,523,512✔

4265

    break;

1,523,512✔

4268

  if (*pSubTableHasBlock) {

4,586,562✔

4269

    pInput->pReaderBlock->info.id.groupId =

3,047,752✔

4270

        tableListGetTableGroupId(pInfo->base.pTableListInfo, pInput->pReaderBlock->info.id.uid);

3,047,388✔

4271

    pOperator->resultInfo.totalRows += pInput->pReaderBlock->info.rows;

1,523,876✔

4273

  if (!pInput->bInMemReader || !*pSubTableHasBlock) {

4,586,926✔

4274

    pAPI->tsdReader.tsdReaderClose(pInput->pReader);

3,445,704✔

4275

    pInput->pReader = NULL;

3,443,580✔

4278

  pInfo->base.dataReader = NULL;

4,583,002✔

4279

  return TSDB_CODE_SUCCESS;

4,584,418✔

4282

static int32_t setGroupStartEndIndex(STableMergeScanInfo* pInfo) {

1,903,200✔

4283

  pInfo->bGroupProcessed = false;

1,903,200✔

4285

  int32_t numOfTables = 0;

1,903,100✔

4286

  int32_t code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

1,882,194✔

4287

  if (code != TSDB_CODE_SUCCESS) {

1,872,556✔

4292

  int32_t i = pInfo->tableStartIndex + 1;

1,872,556✔

4293

  for (; i < numOfTables; ++i) {

3,024,174✔

4294

    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

1,201,912✔

4295

    if (!tableKeyInfo) {

1,198,636✔

4299

    if (tableKeyInfo->groupId != pInfo->groupId) {

1,198,636✔

4300

      break;

57,040✔

4303

  pInfo->tableEndIndex = i - 1;

1,879,302✔

4304

  return TSDB_CODE_SUCCESS;

1,911,352✔

4307

static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {

1,914,568✔

4308

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,972,612✔

4309

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,062,322✔

4310

    if (pInput->rowIdx == -1) {

3,061,604✔

4311

      continue;

1,537,030✔

4314

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,522,430✔

4315

      pInput->rowIdx = 0;

1,523,876✔

4316

      pInput->pageIdx = -1;

1,523,148✔

4319

    pInput->pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,522,056✔

4320

    SColumnInfoData* col = taosArrayGet(pInput->pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

1,523,148✔

4321

    if (!col) {

1,523,512✔

4325

    pInput->aTs = (int64_t*)col->pData;

1,523,512✔

4328

  __merge_compare_fn_t mergeCompareFn = (!pSubTblsInfo->pPkOrderInfo) ? subTblRowCompareTsFn : subTblRowCompareTsPkFn;

1,912,808✔

4329

  return tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, mergeCompareFn);

1,915,276✔

4332

static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {

1,902,836✔

4333

  int32_t code = setGroupStartEndIndex(pInfo);

1,902,836✔

4334

  if (code != TSDB_CODE_SUCCESS) {

1,903,494✔

4338

  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));

1,903,494✔

4339

  if (pSubTblsInfo == NULL) {

1,888,790✔

4342

  pSubTblsInfo->pTsOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);

1,888,790✔

4343

  if (!pSubTblsInfo->pTsOrderInfo) {

1,899,580✔

4347

  if (taosArrayGetSize(pInfo->pSortInfo) == 2) {

1,886,706✔

4354

    pSubTblsInfo->pPkOrderInfo = NULL;

1,883,560✔

4356

  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

1,899,206✔

4357

  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));

1,901,036✔

4358

  if (pSubTblsInfo->aInputs == NULL) {

1,880,050✔

4362

  int32_t bufPageSize = pInfo->bufPageSize;

1,897,506✔

4363

  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;

1,900,732✔

4364

  code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);

1,883,216✔

4365

  if (code != TSDB_CODE_SUCCESS) {

1,889,274✔

4370

  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;

1,889,274✔

4371

  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;

1,902,068✔

4373

  pInfo->pSubTablesMergeInfo = pSubTblsInfo;

1,903,868✔

4374

  return TSDB_CODE_SUCCESS;

1,896,080✔

4377

static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {

1,901,410✔

4378

  int32_t        code = TSDB_CODE_SUCCESS;

1,901,410✔

4379

  int32_t        lino = 0;

1,901,410✔

4380

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,901,410✔

4381

  SReadHandle*   pHandle = &pInfo->base.readHandle;

1,902,452✔

4382

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,894,594✔

4384

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

1,880,344✔

4386

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,962,326✔

4387

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,051,662✔

4388

    pInput->type = SUB_TABLE_MEM_BLOCK;

3,039,506✔

4390

    code = dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);

3,036,340✔

4391

    QUERY_CHECK_CODE(code, lino, _end);

3,051,308✔

4393

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pReaderBlock);

3,051,308✔

4394

    QUERY_CHECK_CODE(code, lino, _end);

3,053,766✔

4396

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pPageBlock);

3,053,766✔

4397

    QUERY_CHECK_CODE(code, lino, _end);

3,062,332✔

4399

    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);

3,062,332✔

4400

    pInput->pKeyInfo = keyInfo;

3,059,814✔

4402

    if (isTaskKilled(pTaskInfo)) {

3,060,178✔

4406

    if (i + 1 < pSubTblsInfo->numInMemReaders) {

3,051,622✔

4407

      code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, keyInfo, 1, pInput->pReaderBlock,

2,290,472✔

4408

                                           (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

1,144,144✔

4409

      QUERY_CHECK_CODE(code, lino, _end);

1,144,144✔

4410

      pInput->bInMemReader = true;

1,144,144✔

4412

      pInput->pReader = NULL;

1,914,214✔

4413

      pInput->bInMemReader = false;

1,913,496✔

4415

    bool hasNext = true;

3,059,076✔

4416

    code = fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);

3,058,732✔

4417

    QUERY_CHECK_CODE(code, lino, _end);

3,061,604✔

4418

    if (!hasNext) {

3,061,604✔

4419

      pInput->rowIdx = -1;

1,537,728✔

4420

      ++pSubTblsInfo->numSubTablesCompleted;

1,538,102✔

4421

      continue;

1,537,394✔

4423

      pInput->rowIdx = 0;

1,523,876✔

4424

      pInput->pageIdx = -1;

1,523,876✔

4428

_end:

1,914,942✔

4429

  if (code != TSDB_CODE_SUCCESS) {

1,915,296✔

4432

  return code;

1,914,922✔

4435

static void adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

1,523,876✔

4436

  int32_t              code = TSDB_CODE_SUCCESS;

1,523,876✔

4437

  int32_t              lino = 0;

1,523,876✔

4438

  SExecTaskInfo*       pTaskInfo = pOperatorInfo->pTaskInfo;

1,523,876✔

4439

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

1,523,876✔

4440

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,523,876✔

4441

  bool                 hasNext = true;

1,523,876✔

4442

  code = fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);

1,523,876✔

4443

  QUERY_CHECK_CODE(code, lino, _end);

1,523,876✔

4445

  if (!hasNext) {

1,523,876✔

4446

    pInput->rowIdx = -1;

1,523,876✔

4447

    ++pSubTblsInfo->numSubTablesCompleted;

1,523,876✔

4452

_end:

1,523,876✔

4453

  if (code != TSDB_CODE_SUCCESS) {

1,523,876✔

4458

1,523,876✔

4460

static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

277,422,260✔

4461

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

277,422,260✔

4462

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

277,425,466✔

4464

  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

277,428,662✔

4465

  if (pInput->rowIdx < pInputBlock->info.rows - 1) {

277,438,270✔

4466

    ++pInput->rowIdx;

275,912,564✔

4467

  } else if (pInput->rowIdx == pInputBlock->info.rows - 1) {

1,523,876✔

4468

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,523,876✔

4469

      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);

1,523,876✔

4471

    if (pInput->rowIdx != -1) {

1,523,876✔

4482

  return tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));

277,438,978✔

4485

static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {

277,277,844✔

4486

  int32_t            code = TSDB_CODE_SUCCESS;

277,277,844✔

4487

  int32_t            lino = 0;

277,277,844✔

4488

  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

277,277,844✔

4489

  SSDataBlock*       pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

277,308,014✔

4491

  for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {

1,119,781,186✔

4492

    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);

841,714,456✔

4493

    QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);

841,552,320✔

4495

    SColumnInfoData* pSrcColInfo = taosArrayGet(pInputBlock->pDataBlock, i);

841,552,320✔

4496

    QUERY_CHECK_NULL(pSrcColInfo, code, lino, _end, terrno);

841,194,246✔

4497

    bool isNull = colDataIsNull(pSrcColInfo, pInputBlock->info.rows, pInput->rowIdx, NULL);

841,194,246✔

4499

    if (isNull) {

841,810,914✔

4500

      code = colDataSetVal(pColInfo, pBlock->info.rows, NULL, true);

104,237,408✔

4501

      QUERY_CHECK_CODE(code, lino, _end);

104,233,040✔

4503

      if (pSrcColInfo->pData != NULL) {

737,573,506✔

4504

        char* pData = colDataGetData(pSrcColInfo, pInput->rowIdx);

737,545,794✔

4505

        code = colDataSetVal(pColInfo, pBlock->info.rows, pData, false);

738,481,476✔

4506

        QUERY_CHECK_CODE(code, lino, _end);

738,291,956✔

4510

  pBlock->info.dataLoad = 1;

277,431,130✔

4511

  pBlock->info.scanFlag = pInputBlock->info.scanFlag;

277,428,318✔

4512

  pBlock->info.rows += 1;

277,426,234✔

4514

_end:

277,427,994✔

4515

  if (code != TSDB_CODE_SUCCESS) {

277,427,994✔

4518

  return code;

277,425,132✔

4521

static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,

3,098,948✔

4523

  int32_t                 code = TSDB_CODE_SUCCESS;

3,098,948✔

4524

  int32_t                 lino = 0;

3,098,948✔

4525

  STableMergeScanInfo*    pInfo = pOperator->info;

3,098,948✔

4526

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

3,100,030✔

4527

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

3,097,532✔

4528

  bool                    finished = false;

3,096,116✔

4530

  QRY_PARAM_CHECK(pResBlock);

3,096,116✔

4532

  blockDataCleanup(pBlock);

3,097,168✔

4536

      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {

280,382,536✔

4537

        finished = true;

3,100,394✔

4538

        break;

3,100,394✔

4541

      code = appendChosenRowToDataBlock(pSubTblsInfo, pBlock);

277,279,958✔

4542

      QUERY_CHECK_CODE(code, lino, _end);

277,416,262✔

4544

      code = adjustSubTableForNextRow(pOperator, pSubTblsInfo);

277,416,262✔

4545

      QUERY_CHECK_CODE(code, lino, _end);

277,281,414✔

4547

      if (pBlock->info.rows >= capacity) {

277,281,414✔

4552

    if (isTaskKilled(pTaskInfo)) {

3,100,394✔

4556

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pBlock, pTaskInfo);

3,100,394✔

4557

    if (finished || limitReached || pBlock->info.rows > 0) {

3,100,040✔

4562

  if (pBlock->info.rows > 0) {

3,100,040✔

4563

    *pResBlock = pBlock;

1,184,036✔

4566

_end:

1,916,712✔

4567

  if (code != TSDB_CODE_SUCCESS) {

3,100,748✔

4571

  return code;

3,100,040✔

4574

static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {

1,902,816✔

4575

  int32_t              code = TSDB_CODE_SUCCESS;

1,902,816✔

4576

  int32_t              lino = 0;

1,902,816✔

4577

  STableMergeScanInfo* pInfo = pOperator->info;

1,902,816✔

4579

  code = initSubTablesMergeInfo(pInfo);

1,908,126✔

4580

  QUERY_CHECK_CODE(code, lino, _end);

1,883,944✔

4582

  code = initSubTableInputs(pOperator, pInfo);

1,883,944✔

4583

  QUERY_CHECK_CODE(code, lino, _end);

1,913,860✔

4585

  code = openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);

1,913,860✔

4586

  QUERY_CHECK_CODE(code, lino, _end);

1,912,758✔

4588

_end:

1,912,758✔

4589

  if (code != TSDB_CODE_SUCCESS) {

1,912,758✔

4592

  return code;

1,914,548✔

4595

static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {

29,227,515✔

4596

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

29,227,515✔

4597

  if (pSubTblsInfo != NULL) {

29,231,703✔

4598

    tMergeTreeDestroy(&pSubTblsInfo->pTree);

1,914,234✔

4600

    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,977,982✔

4601

      STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,062,332✔

4602

      taosMemoryFree(pInput->tblCond.colList);

3,060,208✔

4603

      blockDataDestroy(pInput->pReaderBlock);

3,060,208✔

4604

      blockDataDestroy(pInput->pPageBlock);

3,062,332✔

4605

      taosArrayDestroy(pInput->aBlockPages);

3,062,332✔

4606

      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);

3,062,332✔

4607

      pInput->pReader = NULL;

3,062,332✔

4610

    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);

1,916,358✔

4611

    taosMemoryFree(pSubTblsInfo->aInputs);

1,913,880✔

4613

    taosMemoryFree(pSubTblsInfo);

1,914,588✔

4614

    pInfo->pSubTablesMergeInfo = NULL;

1,913,880✔

4618

29,232,411✔

4620

int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

3,075,312✔

4621

  QRY_PARAM_CHECK(ppRes);

3,075,312✔

4623

  int32_t lino = 0;

3,075,970✔

4624

  int32_t tableListSize = 0;

3,075,970✔

4625

  int64_t st = taosGetTimestampUs();

3,078,528✔

4627

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,078,528✔

4628

  STableMergeScanInfo* pInfo = pOperator->info;

3,068,526✔

4630

  if (pOperator->status == OP_EXEC_DONE) {

3,059,970✔

4634

  int32_t code = pOperator->fpSet._openFn(pOperator);

3,074,948✔

4635

  QUERY_CHECK_CODE(code, lino, _end);

3,053,284✔

4637

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

3,053,284✔

4638

  QUERY_CHECK_CODE(code, lino, _end);

3,063,904✔

4640

  if (!pInfo->hasGroupId) {

3,063,904✔

4641

    pInfo->hasGroupId = true;

1,880,920✔

4643

    if (tableListSize == 0) {

1,869,168✔

4644

      setOperatorCompleted(pOperator);

44,044✔

4645

      (*ppRes) = NULL;

44,044✔

4646

      return code;

44,044✔

4649

    pInfo->tableStartIndex = 0;

1,825,124✔

4650

    STableKeyInfo* pTmpGpId = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,844,744✔

4651

    QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

1,822,980✔

4652

    pInfo->groupId = pTmpGpId->groupId;

1,822,980✔

4653

    code = startSubTablesTableMergeScan(pOperator);

1,843,966✔

4654

    QUERY_CHECK_CODE(code, lino, _end);

1,856,800✔

4657

  SSDataBlock* pBlock = NULL;

3,041,898✔

4658

  while (pInfo->tableStartIndex < tableListSize) {

3,098,220✔

4659

    if (isTaskKilled(pTaskInfo)) {

3,099,312✔

4663

    code = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity, &pBlock);

3,098,604✔

4664

    QUERY_CHECK_CODE(code, lino, _end);

3,100,394✔

4666

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

3,100,394✔

4673

    if (pBlock != NULL) {

3,099,332✔

4674

      pBlock->info.id.groupId = pInfo->groupId;

1,184,036✔

4675

      pOperator->resultInfo.totalRows += pBlock->info.rows;

1,184,036✔

4676

      pInfo->bGroupProcessed = true;

1,184,036✔

4677

      break;

1,184,036✔

4680

      stopSubTablesTableMergeScan(pInfo);

1,915,296✔

4681

      if (pInfo->tableEndIndex >= tableListSize - 1) {

1,914,234✔

4682

        setOperatorCompleted(pOperator);

1,857,902✔

4683

        break;

1,857,184✔

4686

      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

57,040✔

4687

      STableKeyInfo* pTmpGpId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

57,040✔

4688

      QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

57,040✔

4690

      pInfo->groupId = pTmpGpId->groupId;

57,040✔

4691

      code = startSubTablesTableMergeScan(pOperator);

57,040✔

4692

      QUERY_CHECK_CODE(code, lino, _end);

57,040✔

4693

      resetLimitInfoForNextGroup(&pInfo->limitInfo);

57,040✔

4697

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

3,041,584✔

4699

_end:

3,043,708✔

4700

  if (code != TSDB_CODE_SUCCESS) {

3,042,646✔

4705

    (*ppRes) = pBlock;

3,042,646✔

4708

  return code;

3,042,646✔

4711

static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {

3,560,802✔

4712

  int32_t              code = TSDB_CODE_SUCCESS;

3,560,802✔

4713

  int32_t              lino = 0;

3,560,802✔

4714

  SOperatorInfo*       pOperator = (SOperatorInfo*)pTableMergeOpInfo;

3,560,802✔

4715

  STableMergeScanInfo* pInfo = pOperator->info;

3,560,802✔

4716

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,561,344✔

4718

  if (pInfo->mSkipTables == NULL) {

3,561,344✔

4719

    pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,

335,544✔

4721

    QUERY_CHECK_NULL(pInfo->mSkipTables, code, lino, _end, terrno);

335,544✔

4723

  int bSkip = 1;

3,561,344✔

4724

  if (pInfo->mSkipTables != NULL) {

3,558,634✔

4725

    code = taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));

3,561,344✔

4726

    if (code == TSDB_CODE_DUP_KEY) {

3,561,344✔

4729

    QUERY_CHECK_CODE(code, lino, _end);

3,561,344✔

4732

_end:

3,561,344✔

4733

  if (code != TSDB_CODE_SUCCESS) {

3,561,344✔

4738

3,561,344✔

4740

static int32_t doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {

165,757,279✔

4741

  STableMergeScanInfo* pInfo = pOperator->info;

165,757,279✔

4742

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

165,766,331✔

4743

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

165,759,813✔

4744

  SSDataBlock*         pBlock = pInfo->pReaderBlock;

165,728,263✔

4745

  int32_t              code = 0;

165,754,189✔

4746

  bool                 hasNext = false;

165,754,189✔

4747

  STsdbReader*         reader = pInfo->base.dataReader;

165,754,332✔

4749

  code = pAPI->tsdReader.tsdNextDataBlock(reader, &hasNext);

165,751,562✔

4750

  if (code != 0) {

165,730,728✔

4757

  if (!hasNext || isTaskKilled(pTaskInfo)) {

165,730,728✔

4758

    if (isTaskKilled(pTaskInfo)) {

28,560,521✔

4762

    *pFinished = true;

28,559,649✔

4763

    return code;

28,546,337✔

4766

  uint32_t status = 0;

137,180,868✔

4767

  code = loadDataBlock(pOperator, &pInfo->base, pBlock, &status);

137,190,511✔

4769

  if (code != TSDB_CODE_SUCCESS) {

137,184,311✔

4770

    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));

17,088✔

4771

    pTaskInfo->code = code;

17,088✔

4772

    return code;

17,088✔

4775

  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

137,167,223✔

4781

  if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {

137,167,223✔

4782

    *pSkipped = true;

6,337,608✔

4783

    return code;

6,319,302✔

4786

  return code;

130,819,833✔

4789

static int32_t getBlockForTableMergeScan(void* param, SSDataBlock** ppBlock) {

166,027,361✔

4790

  STableMergeScanSortSourceParam* source = param;

166,027,361✔

4792

  SOperatorInfo*       pOperator = source->pOperator;

166,027,361✔

4793

  STableMergeScanInfo* pInfo = pOperator->info;

166,025,816✔

4794

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

166,012,709✔

4795

  SSDataBlock*         pBlock = NULL;

166,041,520✔

4796

  int64_t              st = taosGetTimestampUs();

166,031,433✔

4797

  int32_t              code = TSDB_CODE_SUCCESS;

166,031,433✔

4800

    if (pInfo->rtnNextDurationBlocks) {

178,443,579✔

4801

      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d",

12,683,693✔

4804

      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {

12,689,327✔

4805

        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];

6,096,926✔

4806

        ++pInfo->nextDurationBlocksIdx;

6,096,926✔

4808

        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

12,682,229✔

4809

          blockDataDestroy(pInfo->nextDurationBlocks[i]);

6,095,860✔

4810

          pInfo->nextDurationBlocks[i] = NULL;

6,093,221✔

4813

        pInfo->rtnNextDurationBlocks = false;

6,586,746✔

4814

        pInfo->nextDurationBlocksIdx = 0;

6,587,500✔

4815

        pInfo->numNextDurationBlocks = 0;

6,587,877✔

4816

        continue;

6,587,500✔

4819

      bool bFinished = false;

165,741,091✔

4820

      bool bSkipped = false;

165,759,802✔

4822

      code = doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);

165,779,568✔

4823

      if (code != 0) {

165,693,752✔

4824

        return code;

17,088✔

4827

      pBlock = pInfo->pReaderBlock;

165,676,664✔

4828

      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d",

165,733,130✔

4830

      if (bFinished) {

165,775,263✔

4831

        pInfo->bNewFilesetEvent = false;

28,564,386✔

4832

        break;

28,563,864✔

4835

      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {

137,210,877✔

4836

        if (!bSkipped) {

6,589,667✔

4837

          code = createOneDataBlock(pBlock, true, &pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks]);

6,096,926✔

4838

          if (code) {

6,090,140✔

4843

          ++pInfo->numNextDurationBlocks;

6,090,140✔

4844

          if (pInfo->numNextDurationBlocks > 2) {

6,091,271✔

4852

        if (pInfo->bNewFilesetEvent) {

6,580,996✔

4853

          pInfo->rtnNextDurationBlocks = true;

6,584,197✔

4854

          *ppBlock = NULL;

6,581,780✔

4855

          return code;

6,575,371✔

4858

        if (pInfo->bNextDurationBlockEvent) {

2,107✔

4863

      if (bSkipped) continue;

130,622,876✔

4866

    pBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pBlock->info.id.uid);

130,894,253✔

4868

    pOperator->resultInfo.totalRows += pBlock->info.rows;

130,894,400✔

4869

    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

130,888,519✔

4870

    *ppBlock = pBlock;

130,890,898✔

4872

    return code;

130,886,820✔

4875

  *ppBlock = NULL;

28,563,386✔

4876

  return code;

28,564,386✔

4879

int32_t generateSortByTsPkInfo(SArray* colMatchInfo, int32_t order, SArray** ppSortArray) {

24,979,453✔

4880

  int32_t code = TSDB_CODE_SUCCESS;

24,979,453✔

4881

  int32_t lino = 0;

24,979,453✔

4882

  SArray* pSortInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));

24,979,453✔

4883

  QUERY_CHECK_NULL(pSortInfo, code, lino, _end, terrno);

24,938,212✔

4884

  SBlockOrderInfo biTs = {0};

24,938,212✔

4885

  SBlockOrderInfo biPk = {0};

24,934,811✔

4887

  int32_t tsTargetSlotId = 0;

24,960,861✔

4888

  int32_t pkTargetSlotId = -1;

24,960,861✔

4889

  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {

95,115,989✔

4890

    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);

70,196,023✔

4891

    QUERY_CHECK_NULL(colInfo, code, lino, _end, terrno);

70,195,028✔

4892

    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

70,195,028✔

4893

      tsTargetSlotId = colInfo->dstSlotId;

24,992,016✔

4894

      biTs.order = order;

24,982,524✔

4895

      biTs.slotId = tsTargetSlotId;

24,982,524✔

4896

      biTs.nullFirst = (order == TSDB_ORDER_ASC);

24,982,524✔

4897

      biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, order);

24,982,524✔

4900

    if (colInfo->isPk) {

70,164,890✔

4901

      pkTargetSlotId = colInfo->dstSlotId;

1,716,157✔

4902

      biPk.order = order;

1,716,157✔

4903

      biPk.slotId = pkTargetSlotId;

1,716,157✔

4904

      biPk.nullFirst = (order == TSDB_ORDER_ASC);

1,716,157✔

4905

      biPk.compFn = getKeyComparFunc(colInfo->dataType.type, order);

1,716,157✔

4909

  void* tmp = taosArrayPush(pSortInfo, &biTs);

25,002,850✔

4910

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

25,002,850✔

4911

  if (pkTargetSlotId != -1) {

25,002,850✔

4912

    tmp = taosArrayPush(pSortInfo, &biPk);

1,715,558✔

4913

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

1,715,558✔

4916

  (*ppSortArray) = pSortInfo;

25,004,681✔

4918

_end:

24,964,695✔

4919

  if (code != TSDB_CODE_SUCCESS) {

24,965,105✔

4922

  return code;

24,965,105✔

4925

void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {

21,069,141✔

4926

  STableMergeScanInfo* pTmsInfo = param;

21,069,141✔

4927

  if (type == TSD_READER_NOTIFY_DURATION_START) {

21,069,141✔

4928

    pTmsInfo->bNewFilesetEvent = true;

21,077,026✔

4932

  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);

21,053,352✔

4934

  return;

21,087,230✔

4937

int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

35,358,377✔

4938

  STableMergeScanInfo* pInfo = pOperator->info;

35,358,377✔

4939

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

35,369,354✔

4941

  int32_t code = TSDB_CODE_SUCCESS;

35,357,522✔

4942

  int32_t lino = 0;

35,357,522✔

4943

  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

35,357,522✔

4945

  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));

35,344,688✔

4946

  pInfo->bNewFilesetEvent = false;

35,389,388✔

4947

  pInfo->bNextDurationBlockEvent = false;

35,392,866✔

4949

  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;

35,391,400✔

4950

  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;

35,374,173✔

4952

  pInfo->pSortHandle = NULL;

35,369,882✔

4953

  code = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,

70,765,852✔

4954

                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0, &pInfo->pSortHandle);

35,380,042✔

4955

  if (code) {

35,332,146✔

4959

  if (pInfo->bSortRowId && numOfTable != 1) {

35,332,146✔

4960

    int32_t memSize = 512 * 1024 * 1024;

2,949,517✔

4961

    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);

2,949,517✔

4962

    QUERY_CHECK_CODE(code, lino, _end);

2,953,469✔

4965

  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);

35,345,200✔

4966

  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pOperator);

35,345,102✔

4967

  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);

35,334,737✔

4969

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);

35,326,183✔

4970

  QUERY_CHECK_CODE(code, lino, _end);

35,315,138✔

4972

  STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));

35,315,138✔

4973

  QUERY_CHECK_NULL(param, code, lino, _end, terrno);

35,316,924✔

4974

  param->pOperator = pOperator;

35,316,924✔

4976

  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));

35,312,340✔

4977

  if (ps == NULL) {

35,329,617✔

4982

  ps->param = param;

35,329,617✔

4983

  ps->onlyRef = false;

35,356,523✔

4984

  code = tsortAddSource(pInfo->pSortHandle, ps);

35,345,169✔

4985

  QUERY_CHECK_CODE(code, lino, _end);

35,356,390✔

4987

  if (numOfTable == 1) {

35,356,390✔

4988

    tsortSetSingleTableMerge(pInfo->pSortHandle);

23,115,383✔

4990

    code = tsortOpen(pInfo->pSortHandle);

12,241,007✔

4991

    QUERY_CHECK_CODE(code, lino, _end);

12,248,868✔

4994

_end:

12,240,324✔

4995

  if (code != TSDB_CODE_SUCCESS) {

35,373,821✔

4996

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,544✔

4998

  return code;

35,369,797✔

5001

void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

35,379,071✔

5002

  STableMergeScanInfo* pInfo = pOperator->info;

35,379,071✔

5003

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

35,377,708✔

5004

  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));

35,380,839✔

5006

  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

35,382,354✔

5007

  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;

35,374,542✔

5008

  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;

35,376,926✔

5009

  pInfo->sortExecInfo.loops += sortExecInfo.loops;

35,374,766✔

5010

  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;

35,367,663✔

5011

  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

35,372,435✔

5013

  tsortDestroySortHandle(pInfo->pSortHandle);

35,363,496✔

5014

  pInfo->pSortHandle = NULL;

35,373,257✔

5015

35,375,117✔

5017

void startGroupTableMergeScan(SOperatorInfo* pOperator) {

28,783,466✔

5018

  int32_t              code = TSDB_CODE_SUCCESS;

28,783,466✔

5019

  int32_t              lino = 0;

28,783,466✔

5020

  STableMergeScanInfo* pInfo = pOperator->info;

28,783,466✔

5021

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,767,381✔

5022

  SReadHandle*         pHandle = &pInfo->base.readHandle;

28,778,907✔

5023

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,776,599✔

5024

  qDebug("%s table merge scan start group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,765,523✔

5027

    int32_t numOfTables = 0;

28,776,756✔

5028

    code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

28,790,479✔

5029

    QUERY_CHECK_CODE(code, lino, _end);

28,754,715✔

5031

    int32_t i = pInfo->tableStartIndex + 1;

28,754,715✔

5032

    for (; i < numOfTables; ++i) {

135,353,826✔

5033

      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

112,902,425✔

5034

      QUERY_CHECK_NULL(tableKeyInfo, code, lino, _end, terrno);

112,900,441✔

5035

      if (tableKeyInfo->groupId != pInfo->groupId) {

112,900,441✔

5036

        break;

6,294,050✔

5039

    pInfo->tableEndIndex = i - 1;

28,745,451✔

5041

  pInfo->bGroupProcessed = false;

28,797,563✔

5042

  int32_t tableStartIdx = pInfo->tableStartIndex;

28,798,497✔

5043

  int32_t tableEndIdx = pInfo->tableEndIndex;

28,791,501✔

5045

  int32_t        numOfTable = tableEndIdx - tableStartIdx + 1;

28,789,305✔

5046

  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);

28,789,305✔

5047

  code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,

57,556,672✔

5048

                                       (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->mSkipTables);

28,772,011✔

5049

  QUERY_CHECK_CODE(code, lino, _end);

28,762,183✔

5050

  if (pInfo->filesetDelimited) {

28,762,183✔

5051

    pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);

28,762,895✔

5053

  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);

28,779,077✔

5055

  code = startDurationForGroupTableMergeScan(pOperator);

28,736,569✔

5056

  QUERY_CHECK_CODE(code, lino, _end);

28,791,278✔

5058

_end:

28,782,734✔

5059

  if (code != TSDB_CODE_SUCCESS) {

28,785,622✔

5060

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,544✔

5061

    pTaskInfo->code = code;

8,544✔

5062

    T_LONG_JMP(pTaskInfo->env, code);

8,544✔

5064

28,777,078✔

5066

void stopGroupTableMergeScan(SOperatorInfo* pOperator) {

28,788,975✔

5067

  STableMergeScanInfo* pInfo = pOperator->info;

28,788,975✔

5068

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,791,019✔

5069

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,790,519✔

5071

  stopDurationForGroupTableMergeScan(pOperator);

28,790,519✔

5073

  if (pInfo->base.dataReader != NULL) {

28,784,046✔

5074

    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);

28,784,592✔

5075

    pInfo->base.dataReader = NULL;

28,743,349✔

5077

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

28,747,908✔

5078

    if (pInfo->nextDurationBlocks[i]) {

1,066✔

5079

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

1,066✔

5080

      pInfo->nextDurationBlocks[i] = NULL;

1,066✔

5082

    pInfo->numNextDurationBlocks = 0;

1,066✔

5083

    pInfo->nextDurationBlocksIdx = 0;

1,066✔

5085

  resetLimitInfoForNextGroup(&pInfo->limitInfo);

28,756,739✔

5086

  taosHashCleanup(pInfo->mSkipTables);

28,717,533✔

5087

  pInfo->mSkipTables = NULL;

28,740,602✔

5088

  qDebug("%s table merge scan stop group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,750,035✔

5089

28,774,620✔

5093

SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,

73,903,002✔

5095

  STableMergeScanInfo* pInfo = pOperator->info;

73,903,002✔

5096

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

73,901,308✔

5097

  STupleHandle*        pTupleHandle = NULL;

73,901,129✔

5099

  blockDataCleanup(pResBlock);

73,894,079✔

5102

    while (1) {

2,147,483,647✔

5103

      pTupleHandle = NULL;

2,147,483,647✔

5104

      int32_t code = tsortNextTuple(pHandle, &pTupleHandle);

2,147,483,647✔

5105

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5106

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

1,378,502✔

5107

        T_LONG_JMP(pOperator->pTaskInfo->env, code);

1,378,502✔

5109

      if (pTupleHandle == NULL) {

2,147,483,647✔

5110

        break;

57,749,494✔

5113

      code = tsortAppendTupleToBlock(pInfo->pSortHandle, pResBlock, pTupleHandle);

2,147,483,647✔

5114

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5119

      if (pResBlock->info.rows >= capacity) {

2,147,483,647✔

5120

        break;

16,162,040✔

5124

    if (tsortIsClosed(pHandle)) {

73,911,534✔

5129

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);

73,917,108✔

5130

    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,

73,919,322✔

5132

    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {

73,926,326✔

5136

  return (pResBlock->info.rows > 0) ? pResBlock : NULL;

73,923,074✔

5139

int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

63,965,950✔

5140

  if (pOperator->status == OP_EXEC_DONE) {

63,965,950✔

5141

    (*ppRes) = NULL;

42,897✔

5142

    return TSDB_CODE_SUCCESS;

42,897✔

5145

  int32_t              code = TSDB_CODE_SUCCESS;

63,890,208✔

5146

  int32_t              lino = 0;

63,890,208✔

5147

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

63,890,208✔

5148

  STableMergeScanInfo* pInfo = pOperator->info;

63,903,975✔

5150

  code = pOperator->fpSet._openFn(pOperator);

63,922,581✔

5151

  QUERY_CHECK_CODE(code, lino, _end);

63,913,798✔

5153

  int64_t st = taosGetTimestampUs();

63,914,674✔

5155

  int32_t tableListSize = 0;

63,914,674✔

5156

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

63,912,562✔

5157

  QUERY_CHECK_CODE(code, lino, _end);

63,924,140✔

5159

  if (!pInfo->hasGroupId) {

63,924,140✔

5160

    pInfo->hasGroupId = true;

25,382,201✔

5162

    if (tableListSize == 0) {

25,362,195✔

5163

      setOperatorCompleted(pOperator);

2,900,947✔

5164

      (*ppRes) = NULL;

2,901,816✔

5165

      return code;

2,901,816✔

5167

    pInfo->tableStartIndex = 0;

22,461,248✔

5168

    STableKeyInfo* tmp = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

22,470,130✔

5169

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

22,459,991✔

5170

    pInfo->groupId = tmp->groupId;

22,459,991✔

5171

    startGroupTableMergeScan(pOperator);

22,466,468✔

5174

  SSDataBlock* pBlock = NULL;

60,998,519✔

5175

  while (pInfo->tableStartIndex < tableListSize) {

73,882,003✔

5176

    if (isTaskKilled(pTaskInfo)) {

73,902,879✔

5180

    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,

73,896,774✔

5182

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

73,925,292✔

5183

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,830,456✔

5184

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

1,830,456✔

5185

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

1,830,456✔

5187

    if (pBlock != NULL) {

73,923,127✔

5188

      pBlock->info.id.groupId = pInfo->groupId;

38,541,657✔

5189

      pOperator->resultInfo.totalRows += pBlock->info.rows;

38,543,822✔

5190

      pInfo->bGroupProcessed = true;

38,542,436✔

5191

      break;

38,540,882✔

5193

      if (pInfo->bNewFilesetEvent) {

35,381,470✔

5194

        stopDurationForGroupTableMergeScan(pOperator);

6,590,074✔

5195

        code = startDurationForGroupTableMergeScan(pOperator);

6,587,435✔

5196

        QUERY_CHECK_CODE(code, lino, _end);

6,589,475✔

5199

        stopGroupTableMergeScan(pOperator);

28,790,420✔

5200

        if (pInfo->tableEndIndex >= tableListSize - 1) {

28,789,295✔

5201

          setOperatorCompleted(pOperator);

22,487,297✔

5202

          break;

22,491,161✔

5205

        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

6,294,050✔

5206

        STableKeyInfo* tmp = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

6,294,050✔

5207

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

6,294,050✔

5208

        pInfo->groupId = tmp->groupId;

6,294,050✔

5209

        startGroupTableMergeScan(pOperator);

6,294,050✔

5210

        resetLimitInfoForNextGroup(&pInfo->limitInfo);

6,293,499✔

5215

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

61,031,301✔

5217

_end:

61,029,512✔

5218

  if (code != TSDB_CODE_SUCCESS) {

61,030,659✔

5223

    (*ppRes) = pBlock;

61,030,659✔

5226

  return code;

61,031,760✔

5235

void destroyTableMergeScanOperatorInfo(void* param) {

25,009,296✔

5236

  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;

25,009,296✔

5239

  if (pTableScanInfo->base.readerAPI.tsdReaderClose != NULL) {

25,009,296✔

5240

    pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);

25,005,889✔

5241

    pTableScanInfo->base.dataReader = NULL;

25,004,569✔

5244

  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {

25,005,028✔

5251

  tsortDestroySortHandle(pTableScanInfo->pSortHandle);

25,007,925✔

5252

  pTableScanInfo->pSortHandle = NULL;

24,999,896✔

5253

  taosHashCleanup(pTableScanInfo->mSkipTables);

25,000,280✔

5254

  pTableScanInfo->mSkipTables = NULL;

25,006,205✔

5255

  blockDataDestroy(pTableScanInfo->pSortInputBlock);

25,007,714✔

5256

  pTableScanInfo->pSortInputBlock = NULL;

25,003,190✔

5259

  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);

25,004,335✔

5261

  blockDataDestroy(pTableScanInfo->pResBlock);

25,006,046✔

5262

  pTableScanInfo->pResBlock = NULL;

25,005,799✔

5265

  blockDataDestroy(pTableScanInfo->pReaderBlock);

25,005,436✔

5266

  pTableScanInfo->pReaderBlock = NULL;

25,003,807✔

5267

  taosArrayDestroy(pTableScanInfo->pSortInfo);

25,004,186✔

5269

  stopSubTablesTableMergeScan(pTableScanInfo);

25,005,273✔

5271

  taosMemoryFreeClear(param);

25,003,404✔

5272

24,995,996✔

5274

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {

1,262,723✔

5275

  if (pOptr == NULL) {

1,262,723✔

5280

  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));

1,262,723✔

5281

  if (!execInfo) {

1,259,288✔

5284

  STableMergeScanInfo* pInfo = pOptr->info;

1,259,288✔

5285

  execInfo->blockRecorder = pInfo->base.readRecorder;

1,260,440✔

5286

  execInfo->sortExecInfo = pInfo->sortExecInfo;

1,263,093✔

5288

  *pOptrExplain = execInfo;

1,263,477✔

5289

  *len = sizeof(STableMergeScanExecInfo);

1,263,854✔

5291

  return TSDB_CODE_SUCCESS;

1,262,716✔

5294

static int32_t resetTableMergeScanOperatorState(SOperatorInfo* pOper) {

2,306,677✔

5295

  int32_t         code = TSDB_CODE_SUCCESS;

2,306,677✔

5296

  STableMergeScanInfo*   pInfo = pOper->info;

2,306,677✔

5297

  pOper->status = OP_NOT_OPENED;

2,306,677✔

5299

  pInfo->tableEndIndex = 0;

2,306,677✔

5300

  pInfo->tableStartIndex = 0;

2,306,677✔

5301

  pInfo->hasGroupId = false;

2,306,677✔

5302

  if (pInfo->base.readerAPI.tsdReaderClose) {

2,306,677✔

5303

    pInfo->base.readerAPI.tsdReaderClose(pInfo->base.dataReader);

2,306,677✔

5305

  pInfo->base.dataReader = NULL;

2,306,677✔

5306

  pInfo->base.scanFlag = MAIN_SCAN;

2,306,677✔

5308

  pInfo->base.limitInfo = (SLimitInfo){0};

2,306,677✔

5309

  pInfo->base.limitInfo.limit.limit = -1;

2,306,677✔

5310

  pInfo->base.limitInfo.slimit.limit = -1;

2,306,677✔

5312

  tableListDestroy(pInfo->base.pTableListInfo);

2,306,677✔

5313

  pInfo->base.pTableListInfo = tableListCreate();

2,306,424✔

5314

  if (!pInfo->base.pTableListInfo) {

2,306,424✔

5318

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

2,306,677✔

5320

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

2,306,677✔

5321

  code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

4,613,354✔

5323

                                  pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

4,613,101✔

5324

  if (code) {

2,306,677✔

5329

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

2,306,677✔

5330

  cleanupQueryTableDataCond(&pInfo->base.cond);

2,306,677✔

5331

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, &pInfo->base.readHandle, false);

2,306,677✔

5332

  if (code) {

2,306,424✔

5337

  tsortDestroySortHandle(pInfo->pSortHandle);

2,306,424✔

5338

  pInfo->pSortHandle = NULL;

2,306,424✔

5340

  taosHashCleanup(pInfo->mSkipTables);

2,306,424✔

5341

  pInfo->mSkipTables = NULL;

2,306,424✔

5342

  pOper->resultInfo.totalRows = 0;

2,306,677✔

5344

  pInfo->bGroupProcessed = false;

2,306,424✔

5345

  pInfo->bNewFilesetEvent = false;

2,306,677✔

5346

  pInfo->bNextDurationBlockEvent = false;

2,306,424✔

5347

  pInfo->rtnNextDurationBlocks = false;

2,306,424✔

5348

  pInfo->nextDurationBlocksIdx = 0;

2,306,424✔

5349

  pInfo->bSortRowId = false;

2,306,424✔

5351

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

2,306,424✔

5357

  pInfo->numNextDurationBlocks = 0;

2,306,424✔

5359

  stopSubTablesTableMergeScan(pInfo);

2,306,424✔

5360

  return code;

2,306,177✔

5363

int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

24,968,195✔

5366

  QRY_PARAM_CHECK(pOptrInfo);

24,968,195✔

5368

  int32_t              code = TSDB_CODE_SUCCESS;

24,981,867✔

5369

  int32_t              lino = 0;

24,981,867✔

5370

  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));

24,981,867✔

5371

  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

24,943,153✔

5372

  if (pInfo == NULL || pOperator == NULL) {

24,908,401✔

5377

  pOperator->pPhyNode = pTableScanNode;

24,909,200✔

5378

  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

24,909,453✔

5380

  int32_t numOfCols = 0;

24,947,810✔

5381

  code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,

24,965,212✔

5383

  QUERY_CHECK_CODE(code, lino, _error);

24,924,380✔

5385

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle, false);

24,924,380✔

5386

  QUERY_CHECK_CODE(code, lino, _error);

24,892,290✔

5388

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

24,892,290✔

5389

    SExprSupp* pSup = &pInfo->base.pseudoSup;

10,653,920✔

5390

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->pExprInfo, &pSup->numOfExprs);

10,675,560✔

5391

    QUERY_CHECK_CODE(code, lino, _error);

10,653,922✔

5393

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

10,653,922✔

5395

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

10,667,999✔

5398

  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};

24,891,403✔

5400

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

24,871,201✔

5405

  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;

24,871,201✔

5406

  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

24,938,730✔

5407

  pInfo->base.scanFlag = MAIN_SCAN;

24,901,037✔

5408

  pInfo->base.readHandle = *readHandle;

24,845,134✔

5410

  pInfo->readIdx = -1;

24,893,879✔

5412

  pInfo->base.limitInfo.limit.limit = -1;

24,818,420✔

5413

  pInfo->base.limitInfo.slimit.limit = -1;

24,958,198✔

5414

  pInfo->base.pTableListInfo = pTableListInfo;

24,919,705✔

5416

  pInfo->sample.sampleRatio = pTableScanNode->ratio;

24,877,902✔

5417

  pInfo->sample.seed = taosGetTimestampSec();

24,954,010✔

5419

  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,

24,928,918✔

5420

                            pTaskInfo->pStreamRuntimeInfo);

24,949,154✔

5421

  QUERY_CHECK_CODE(code, lino, _error);

24,903,209✔

5423

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

24,903,209✔

5425

  pInfo->mergeLimit = -1;

24,881,765✔

5426

  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;

24,894,728✔

5427

  if (hasLimit) {

24,930,374✔

5428

    pInfo->mergeLimit = pInfo->limitInfo.limit.offset != -1

4,173,244✔

5429

                            ? pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset

822,739✔

5430

                            : pInfo->limitInfo.limit.limit;

2,907,388✔

5431

    pInfo->mSkipTables = NULL;

2,082,458✔

5434

  initResultSizeInfo(&pOperator->resultInfo, 1024);

24,929,000✔

5435

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

24,935,112✔

5436

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

24,958,937✔

5437

  code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

24,861,769✔

5438

  QUERY_CHECK_CODE(code, lino, _error);

24,995,432✔

5439

  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {

24,995,432✔

5440

    pInfo->bSortRowId = true;

5,185,981✔

5442

    pInfo->bSortRowId = false;

19,807,817✔

5445

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

24,961,044✔

5446

  QUERY_CHECK_CODE(code, lino, _error);

24,992,977✔

5448

  code = generateSortByTsPkInfo(pInfo->base.matchInfo.pList, pInfo->base.cond.order, &pInfo->pSortInfo);

24,992,977✔

5449

  QUERY_CHECK_CODE(code, lino, _error);

24,950,539✔

5451

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pReaderBlock);

24,950,539✔

5452

  QUERY_CHECK_CODE(code, lino, _error);

24,954,506✔

5454

  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;

24,954,506✔

5456

  int32_t  rowSize = pInfo->pResBlock->info.rowSize;

24,966,674✔

5457

  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);

24,932,498✔

5459

  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);

24,963,813✔

5462

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pSortInputBlock);

24,986,403✔

5463

  QUERY_CHECK_CODE(code, lino, _error);

24,978,217✔

5465

  if (!tsExperimental) {

24,978,217✔

5468

    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

24,978,217✔

5471

  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,

24,985,044✔

5473

  pOperator->exprSupp.numOfExprs = numOfCols;

24,913,855✔

5475

  pOperator->fpSet = createOperatorFpSet(

24,962,203✔

5476

      optrDummyOpenFn, pTableScanNode->paraTablesSort ? doTableMergeScanParaSubTablesNext : doTableMergeScanNext, NULL,

24,939,412✔

5479

  setOperatorResetStateFn(pOperator, resetTableMergeScanOperatorState);

24,910,918✔

5480

  pOperator->cost.openCost = 0;

24,927,030✔

5482

  *pOptrInfo = pOperator;

24,925,503✔

5483

  return code;

24,884,252✔

5522

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {

31,280✔

5523

  if (scanCols != NULL) {

31,280✔

5524

    SNode* pNode = NULL;

15,829✔

5525

    FOREACH(pNode, scanCols) {

36,999✔

5526

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

21,170✔

5529

      STargetNode* targetNode = (STargetNode*)pNode;

21,170✔

5530

      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {

21,170✔

5533

      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);

21,170✔

5534

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

21,170✔

5535

        supp->dbNameSlotId = targetNode->slotId;

10,507✔

5536

      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

10,663✔

5537

        supp->stbNameSlotId = targetNode->slotId;

10,663✔

5541

  return TSDB_CODE_SUCCESS;

31,280✔

5544

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {

31,280✔

5545

  if (pseudoCols != NULL) {

31,280✔

5546

    SNode* pNode = NULL;

31,280✔

5547

    FOREACH(pNode, pseudoCols) {

62,560✔

5548

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

31,280✔

5551

      STargetNode* targetNode = (STargetNode*)pNode;

31,317✔

5552

      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {

31,317✔

5555

      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);

31,317✔

5556

      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {

31,280✔

5557

        supp->tbCountSlotId = targetNode->slotId;

31,317✔

5561

  return TSDB_CODE_SUCCESS;

31,280✔

5564

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {

31,280✔

5565

  if (groupTags != NULL) {

31,280✔

5566

    SNode* pNode = NULL;

15,829✔

5567

    FOREACH(pNode, groupTags) {

36,999✔

5568

      if (nodeType(pNode) != QUERY_NODE_COLUMN) {

21,170✔

5571

      SColumnNode* colNode = (SColumnNode*)pNode;

21,170✔

5572

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

21,170✔

5573

        supp->groupByDbName = true;

10,507✔

5575

      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

21,170✔

5576

        supp->groupByStbName = true;

10,663✔

5580

    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);

15,451✔

5581

    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);

15,451✔

5583

  return TSDB_CODE_SUCCESS;

31,317✔

5586

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,

31,280✔

5588

  int32_t code = 0;

31,280✔

5589

  code = tblCountScanGetInputs(groupTags, tableName, supp);

31,280✔

5590

  if (code != TSDB_CODE_SUCCESS) {

31,317✔

5595

  supp->dbNameSlotId = -1;

31,317✔

5596

  supp->stbNameSlotId = -1;

31,280✔

5597

  supp->tbCountSlotId = -1;

31,280✔

5599

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);

31,280✔

5600

  if (code != TSDB_CODE_SUCCESS) {

31,280✔

5605

  code = tblCountScanGetCountSlotId(pseudoCols, supp);

31,280✔

5606

  if (code != TSDB_CODE_SUCCESS) {

31,280✔

5610

  return code;

31,280✔

5613

int32_t createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,

31,280✔

5615

  QRY_PARAM_CHECK(pOptrInfo);

31,280✔

5617

  int32_t                      code = TSDB_CODE_SUCCESS;

31,317✔

5618

  int32_t                      lino = 0;

31,317✔

5619

  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;

31,317✔

5620

  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));

31,317✔

5621

  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

31,317✔

5622

  if (!pInfo || !pOperator) {

31,317✔

5627

  pInfo->readHandle = *readHandle;

31,317✔

5629

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

31,317✔

5630

  initResultSizeInfo(&pOperator->resultInfo, 1);

31,317✔

5631

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

31,317✔

5632

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

31,317✔

5634

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

31,317✔

5635

  QUERY_CHECK_CODE(code, lino, _error);

31,317✔

5637

  code = getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,

31,317✔

5640

  QUERY_CHECK_CODE(code, lino, _error);

31,280✔

5642

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,

31,280✔

5644

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScanNext, NULL, destoryTableCountScanOperator,

31,280✔

5646

  *pOptrInfo = pOperator;

31,280✔

5647

  return code;

31,317✔

5661

int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,

42,109✔

5663

  int32_t code = TSDB_CODE_SUCCESS;

42,109✔

5664

  int32_t lino = 0;

42,109✔

5665

  if (pSupp->dbNameSlotId != -1) {

42,109✔

5666

    QUERY_CHECK_CONDITION((strlen(dbName) > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

15,866✔

5667

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);

15,866✔

5668

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

15,866✔

5670

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

15,866✔

5671

    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

15,866✔

5673

    varDataSetLen(varDbName, strlen(dbName));

15,866✔

5674

    code = colDataSetVal(colInfoData, 0, varDbName, false);

15,866✔

5675

    QUERY_CHECK_CODE(code, lino, _end);

15,866✔

5678

  if (pSupp->stbNameSlotId != -1) {

42,109✔

5679

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);

20,411✔

5680

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

20,411✔

5681

    if (strlen(stbName) != 0) {

20,411✔

5682

      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

14,714✔

5683

      tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);

14,714✔

5684

      varDataSetLen(varStbName, strlen(stbName));

14,714✔

5685

      code = colDataSetVal(colInfoData, 0, varStbName, false);

14,714✔

5686

      QUERY_CHECK_CODE(code, lino, _end);

14,714✔

5692

  if (pSupp->tbCountSlotId != -1) {

42,109✔

5693

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);

42,109✔

5694

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

41,940✔

5695

    code = colDataSetVal(colInfoData, 0, (char*)&count, false);

41,940✔

5696

    QUERY_CHECK_CODE(code, lino, _end);

42,109✔

5698

  pRes->info.rows = 1;

42,109✔

5700

_end:

42,109✔

5701

  if (code != TSDB_CODE_SUCCESS) {

42,109✔

5704

  return code;

42,109✔

5707

static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {

11,526✔

5708

  STableCountScanSupp* pSupp = &pInfo->supp;

11,526✔

5709

  SSDataBlock*         pRes = pInfo->pRes;

11,526✔

5711

  size_t infodbTableNum;

11,526✔

5712

  getInfosDbMeta(NULL, &infodbTableNum);

11,526✔

5713

  infodbTableNum -= 1;

11,526✔

5714

  size_t perfdbTableNum;

11,526✔

5715

  getPerfDbMeta(NULL, &perfdbTableNum);

11,526✔

5717

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

11,526✔

5718

    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);

9,900✔

5719

    return (pRes->info.rows > 0) ? pRes : NULL;

9,900✔

5721

    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);

1,626✔

5722

    return (pRes->info.rows > 0) ? pRes : NULL;

1,626✔

5726

static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,

1,626✔

5728

  int32_t        code = TSDB_CODE_SUCCESS;

1,626✔

5729

  int32_t        lino = 0;

1,626✔

5730

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,626✔

5731

  if (strlen(pSupp->stbNameFilter) != 0) {

1,626✔

5733

    code = fillTableCountScanDataBlock(pSupp, "", "", 0, pRes);

552✔

5734

    QUERY_CHECK_CODE(code, lino, _end);

552✔

5735

  } else if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {

1,074✔

5736

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

184✔

5737

    QUERY_CHECK_CODE(code, lino, _end);

184✔

5738

  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

890✔

5741

  } else if (strlen(pSupp->dbNameFilter) == 0) {

890✔

5742

    code = fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);

706✔

5743

    QUERY_CHECK_CODE(code, lino, _end);

706✔

5746

    code = fillTableCountScanDataBlock(pSupp, pSupp->dbNameFilter, pSupp->stbNameFilter, 0, pRes);

184✔

5747

    QUERY_CHECK_CODE(code, lino, _end);

184✔

5750

_end:

184✔

5751

  if (code != TSDB_CODE_SUCCESS) {

1,626✔

5756

  setOperatorCompleted(pOperator);

1,626✔

5757

1,626✔

5759

static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,900✔

5762

  int32_t        code = TSDB_CODE_SUCCESS;

9,900✔

5763

  int32_t        lino = 0;

9,900✔

5764

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,900✔

5765

  if (pInfo->currGrpIdx == 0) {

9,900✔

5766

    uint64_t groupId = 0;

3,300✔

5767

    if (pSupp->groupByDbName) {

3,300✔

5768

      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));

2,269✔

5770

      groupId = calcGroupId("", 0);

1,031✔

5773

    pRes->info.id.groupId = groupId;

3,300✔

5774

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

3,300✔

5775

    QUERY_CHECK_CODE(code, lino, _end);

3,300✔

5776

  } else if (pInfo->currGrpIdx == 1) {

6,600✔

5777

    uint64_t groupId = 0;

3,300✔

5778

    if (pSupp->groupByDbName) {

3,300✔

5779

      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));

2,269✔

5781

      groupId = calcGroupId("", 0);

1,031✔

5784

    pRes->info.id.groupId = groupId;

3,300✔

5785

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

3,300✔

5786

    QUERY_CHECK_CODE(code, lino, _end);

3,300✔

5788

    setOperatorCompleted(pOperator);

3,300✔

5790

  pInfo->currGrpIdx++;

9,900✔

5792

_end:

9,900✔

5793

  if (code != TSDB_CODE_SUCCESS) {

9,900✔

5798

9,900✔

5800

static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

73,389✔

5801

  int32_t                      code = TSDB_CODE_SUCCESS;

73,389✔

5802

  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;

73,389✔

5803

  STableCountScanOperatorInfo* pInfo = pOperator->info;

73,389✔

5804

  STableCountScanSupp*         pSupp = &pInfo->supp;

73,426✔

5805

  SSDataBlock*                 pRes = pInfo->pRes;

73,426✔

5807

  blockDataCleanup(pRes);

73,426✔

5808

  QRY_PARAM_CHECK(ppRes);

73,426✔

5810

  if (pOperator->status == OP_EXEC_DONE) {

73,426✔

5811

    return code;

19,610✔

5814

  if (pInfo->readHandle.mnd != NULL) {

53,816✔

5815

    (*ppRes) = buildSysDbTableCount(pOperator, pInfo);

11,526✔

5816

    return code;

11,526✔

5819

  code = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);

42,290✔

5820

  if (code != TSDB_CODE_SUCCESS) {

42,253✔

5825

  if (pRes->info.rows > 0) {

42,253✔

5826

    *ppRes = pRes;

33,883✔

5829

  return code;

42,253✔

5832

static int32_t buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

42,290✔

5834

  int32_t        code = TSDB_CODE_SUCCESS;

42,290✔

5835

  int32_t        lino = 0;

42,290✔

5836

  const char*    db = NULL;

42,290✔

5837

  int32_t        vgId = 0;

42,290✔

5838

  char           dbName[TSDB_DB_NAME_LEN] = {0};

42,290✔

5839

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

42,290✔

5840

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

42,290✔

5843

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);

42,290✔

5844

  SName sn = {0};

42,290✔

5846

  code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

42,290✔

5847

  QUERY_CHECK_CODE(code, lino, _end);

42,253✔

5849

  code = tNameGetDbName(&sn, dbName);

42,253✔

5850

  QUERY_CHECK_CODE(code, lino, _end);

42,253✔

5852

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

42,253✔

5853

    code = buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);

28,428✔

5855

    code = buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);

13,825✔

5858

_end:

42,253✔

5859

  if (code != TSDB_CODE_SUCCESS) {

42,253✔

5864

  return code;

42,253✔

5867

static int32_t buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

28,428✔

5869

  int32_t        code = TSDB_CODE_SUCCESS;

28,428✔

5870

  int32_t        lino = 0;

28,428✔

5871

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

28,428✔

5872

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

28,428✔

5874

  if (pSupp->groupByStbName) {

28,428✔

5875

    if (pInfo->stbUidList == NULL) {

24,306✔

5876

      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));

8,407✔

5877

      QUERY_CHECK_NULL(pInfo->stbUidList, code, lino, _end, terrno);

8,407✔

5878

      code = pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList);

8,407✔

5879

      QUERY_CHECK_CODE(code, lino, _end);

8,407✔

5882

    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {

24,306✔

5883

      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);

14,714✔

5884

      code = buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);

14,714✔

5885

      QUERY_CHECK_CODE(code, lino, _end);

14,714✔

5887

      pInfo->currGrpIdx++;

14,714✔

5888

    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {

9,592✔

5889

      code = buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);

8,407✔

5890

      QUERY_CHECK_CODE(code, lino, _end);

8,407✔

5892

      pInfo->currGrpIdx++;

8,407✔

5894

      setOperatorCompleted(pOperator);

1,185✔

5897

    uint64_t groupId = calcGroupId(dbName, strlen(dbName));

4,122✔

5898

    pRes->info.id.groupId = groupId;

4,122✔

5900

    int64_t dbTableCount = 0;

4,122✔

5901

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);

4,122✔

5902

    code = fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);

4,122✔

5903

    QUERY_CHECK_CODE(code, lino, _end);

4,122✔

5904

    setOperatorCompleted(pOperator);

4,122✔

5907

_end:

28,428✔

5908

  if (code != TSDB_CODE_SUCCESS) {

28,428✔

5912

  return code;

28,428✔

5915

static int32_t buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

13,825✔

5917

  int32_t        code = TSDB_CODE_SUCCESS;

13,825✔

5918

  int32_t        lino = 0;

13,825✔

5919

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

13,825✔

5920

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

13,825✔

5922

  if (strlen(pSupp->dbNameFilter) != 0 && strcmp(dbName, pSupp->dbNameFilter) != 0) {

13,825✔

5926

  } else if (strlen(pSupp->stbNameFilter) != 0) {

13,862✔

5927

    uint64_t uid = 0;

6,743✔

5928

    code = pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid);

6,743✔

5929

    if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

6,780✔

5931

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, 0, pRes);

1,472✔

5932

      QUERY_CHECK_CODE(code, lino, _end);

1,472✔

5934

      QUERY_CHECK_CODE(code, lino, _end);

5,308✔

5936

      int64_t numOfChildTables = 0;

5,308✔

5937

      code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL, NULL);

5,308✔

5938

      QUERY_CHECK_CODE(code, lino, _end);

5,308✔

5940

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes);

5,308✔

5941

      QUERY_CHECK_CODE(code, lino, _end);

5,308✔

5944

    int64_t tbNumVnode = 0;

7,082✔

5945

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);

7,082✔

5946

    code = fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);

7,082✔

5947

    QUERY_CHECK_CODE(code, lino, _end);

7,082✔

5950

_end:

13,862✔

5951

  if (code != TSDB_CODE_SUCCESS) {

13,862✔

5956

  setOperatorCompleted(pOperator);

13,862✔

5957

  return code;

13,825✔

5960

static int32_t buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

8,407✔

5962

  int32_t code = TSDB_CODE_SUCCESS;

8,407✔

5963

  int32_t lino = 0;

8,407✔

5964

  char    fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

8,407✔

5965

  if (pSupp->groupByDbName) {

8,407✔

5966

    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");

4,116✔

5969

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

8,407✔

5970

  pRes->info.id.groupId = groupId;

8,407✔

5972

  int64_t numOfTables = 0;

8,407✔

5973

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);

8,407✔

5975

  if (numOfTables != 0) {

8,407✔

5976

    code = fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);

1,185✔

5979

_end:

7,222✔

5980

  if (code != TSDB_CODE_SUCCESS) {

8,407✔

5983

  return code;

8,407✔

5986

static int32_t buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

14,714✔

5988

  int32_t code = TSDB_CODE_SUCCESS;

14,714✔

5989

  int32_t lino = 0;

14,714✔

5990

  char    stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

14,714✔

5991

  code = pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);

14,714✔

5992

  QUERY_CHECK_CODE(code, lino, _end);

14,714✔

5994

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

14,714✔

5995

  if (pSupp->groupByDbName) {

14,714✔

5996

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));

6,868✔

5998

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));

7,846✔

6001

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

14,714✔

6002

  pRes->info.id.groupId = groupId;

14,714✔

6004

  int64_t ctbNum = 0;

14,714✔

6005

  code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL, NULL);

14,714✔

6006

  QUERY_CHECK_CODE(code, lino, _end);

14,714✔

6007

  code = fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);

14,714✔

6008

  QUERY_CHECK_CODE(code, lino, _end);

14,714✔

6010

_end:

14,714✔

6011

  if (code != TSDB_CODE_SUCCESS) {

14,714✔

6014

  return code;

14,714✔

6017

static void destoryTableCountScanOperator(void* param) {

31,317✔

6018

  STableCountScanOperatorInfo* pTableCountScanInfo = param;

31,317✔

6019

  blockDataDestroy(pTableCountScanInfo->pRes);

31,317✔

6021

  taosArrayDestroy(pTableCountScanInfo->stbUidList);

31,317✔

6022

  taosMemoryFreeClear(param);

31,317✔

6023

31,317✔

taosdata / TDengine / #4952

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4952

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