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

76

bool processBlockWithProbability(const SSampleExecInfo* pInfo) {

534,584,527✔

85

  return true;

534,584,527✔

89

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

65,766✔

90

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

65,766✔

93

65,766✔

95

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

576,529,563✔

96

  int32_t     code = TSDB_CODE_SUCCESS;

576,529,563✔

97

  STimeWindow w = {0};

576,529,563✔

100

  if (pInterval->interval == 0) {

576,637,333✔

101

    *overlap = false;

540,004,547✔

102

    return code;

540,018,522✔

105

  if (order == TSDB_ORDER_ASC) {

36,608,073✔

106

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

34,566,845✔

107

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

34,558,828✔

112

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

34,558,531✔

113

      *overlap = true;

30,089,702✔

114

      return code;

30,090,055✔

118

      getNextTimeWindow(pInterval, &w, order);

4,473,794✔

119

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

4,478,917✔

120

        break;

4,213,059✔

123

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

264,899✔

127

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

264,382✔

128

        *overlap = true;

264,899✔

129

        return code;

264,382✔

133

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

2,041,228✔

134

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

2,040,608✔

139

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

2,040,608✔

140

      *overlap = true;

1,600,568✔

141

      return code;

1,600,568✔

145

      getNextTimeWindow(pInterval, &w, order);

440,660✔

146

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

440,660✔

147

        break;

438,098✔

150

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

2,562✔

154

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

2,562✔

155

        *overlap = true;

2,562✔

156

        return code;

2,562✔

161

  *overlap = false;

4,651,157✔

162

  return code;

4,651,416✔

166

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

125,794,049✔

167

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

125,794,049✔

171

  int64_t buf[2] = {0};

125,830,795✔

172

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

125,832,207✔

174

  STableScanInfo* pTableScanInfo = pOperator->info;

125,818,677✔

176

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

125,820,843✔

179

  if (p1 == NULL) {

125,868,946✔

180

    return NULL;

49,090,498✔

183

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

76,778,448✔

184

  if (NULL == *pPage) {

76,768,762✔

188

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

76,767,360✔

191

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

12,627,634✔

192

  if (NULL == pTableScanInfo->pIgnoreTables) {

12,627,634✔

193

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

1,821,134✔

194

    pTableScanInfo->pIgnoreTables =

1,820,539✔

195

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

1,821,134✔

196

    if (NULL == pTableScanInfo->pIgnoreTables) {

1,820,539✔

201

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

12,628,030✔

203

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

12,628,257✔

207

  return TSDB_CODE_SUCCESS;

12,628,257✔

210

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

634,546,886✔

211

  STableScanInfo* pTableScanInfo = pOperator->info;

634,546,886✔

212

  int32_t         code = TSDB_CODE_SUCCESS;

634,847,831✔

214

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

634,847,831✔

215

    return TSDB_CODE_SUCCESS;

508,994,280✔

218

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

125,814,510✔

220

  SFilePage*  pPage = NULL;

125,852,671✔

221

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

125,813,891✔

223

  if (pRow == NULL) {

125,857,630✔

224

    return TSDB_CODE_SUCCESS;

49,086,686✔

227

  bool notLoadBlock = true;

76,770,944✔

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;

64,139,145✔

236

      break;

64,139,145✔

241

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

76,078,365✔

243

  if (notLoadBlock) {

76,766,657✔

244

    *status = FUNC_DATA_REQUIRED_NOT_LOAD;

12,628,852✔

245

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

12,627,634✔

248

  return code;

76,724,546✔

251

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

9,477,319✔

253

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

9,477,319✔

258

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

9,477,319✔

261

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

138,536,518✔

263

  SStorageAPI* pAPI = &pTaskInfo->storageAPI;

138,536,518✔

264

  bool         allColumnsHaveAgg = true;

138,567,443✔

265

  bool         hasNullSMA = false;

138,569,733✔

266

  if (pLoad != NULL) {

138,546,916✔

267

    *pLoad = false;

138,571,789✔

270

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

138,549,656✔

272

  if (code != TSDB_CODE_SUCCESS) {

138,544,095✔

276

  if (!allColumnsHaveAgg || hasNullSMA) {

138,544,095✔

277

    *pLoad = false;

123,951,232✔

279

    *pLoad = true;

14,592,863✔

282

  return code;

138,512,650✔

285

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

664,627,921✔

287

  int32_t    code = 0;

664,627,921✔

288

  SExprSupp* pSup = &pTableScanInfo->pseudoSup;

664,627,921✔

289

  if (pSup->numOfExprs > 0) {

664,898,529✔

290

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

288,549,935✔

293

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

288,499,556✔

299

  return code;

664,738,422✔

302

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

1,130,750,847✔

303

  SLimit*     pLimit = &pLimitInfo->limit;

1,130,750,847✔

304

  const char* id = GET_TASKID(pTaskInfo);

1,130,854,671✔

306

  if (pLimitInfo->remainOffset > 0) {

1,130,952,093✔

307

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

5,198,663✔

308

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

2,623,600✔

309

      blockDataEmpty(pBlock);

2,623,600✔

310

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

2,623,600✔

311

      return false;

2,623,600✔

313

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

2,575,063✔

314

      if (code != TSDB_CODE_SUCCESS) {

2,575,063✔

319

      pLimitInfo->remainOffset = 0;

2,575,063✔

323

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

1,128,312,565✔

325

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

31,371,642✔

326

    blockDataKeepFirstNRows(pBlock, keep);

31,372,531✔

328

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

31,371,994✔

329

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

31,371,457✔

330

    return true;

31,372,250✔

333

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

1,096,705,910✔

334

  return false;

1,096,888,729✔

341

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

664,592,355✔

343

  int32_t        code = TSDB_CODE_SUCCESS;

664,592,355✔

344

  int32_t        lino = 0;

664,592,355✔

345

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

664,592,355✔

346

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

664,731,806✔

347

  bool           loadSMA = false;

664,710,460✔

349

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

664,710,460✔

351

  pCost->totalBlocks += 1;

664,611,497✔

352

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

664,678,424✔

353

  *status = pTableScanInfo->dataBlockLoadFlag;

664,635,546✔

355

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

664,695,232✔

356

    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

87,984,112✔

358

    bool overlap = false;

576,611,015✔

360

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

576,649,619✔

361

    if (ret != TSDB_CODE_SUCCESS) {

576,412,773✔

364

    if (overlap) {

576,412,773✔

365

      (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

31,953,376✔

369

  SDataBlockInfo* pBlockInfo = &pBlock->info;

664,596,339✔

370

  taosMemoryFreeClear(pBlock->pBlockAgg);

664,641,577✔

372

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {

664,601,538✔

379

  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

664,321,353✔

380

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

21,380,252✔

383

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

21,387,934✔

384

    pCost->skipBlocks += 1;

21,387,737✔

385

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

21,387,737✔

386

    return code;

21,386,695✔

387

  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {

642,891,920✔

388

    pCost->loadBlockStatis += 1;

50,573,490✔

389

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

50,575,007✔

390

    bool success = true;

50,575,007✔

391

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

50,578,455✔

392

    if (code) {

50,560,271✔

398

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

50,558,539✔

399

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

5,115,544✔

401

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

5,115,544✔

402

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

5,115,544✔

403

      return code;

5,115,544✔

405

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

45,442,995✔

406

      *status = FUNC_DATA_REQUIRED_DATA_LOAD;

45,467,581✔

410

  if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {

637,833,864✔

417

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

637,845,007✔

418

    bool success = true;

87,981,539✔

419

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

87,991,802✔

420

    if (code) {

87,946,649✔

423

      QUERY_CHECK_CODE(code, lino, _end);

20,525✔

426

    if (success) {

87,967,174✔

427

      size_t size = taosArrayGetSize(pBlock->pDataBlock);

9,477,319✔

428

      bool   keep = false;

9,477,319✔

429

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

9,477,319✔

430

      if (code) {

9,477,319✔

436

      if (!keep) {

9,477,319✔

437

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

3,250,405✔

439

        pCost->filterOutBlocks += 1;

3,250,405✔

440

        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

3,250,405✔

441

        taosMemoryFreeClear(pBlock->pBlockAgg);

3,250,405✔

443

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

3,250,405✔

444

        return TSDB_CODE_SUCCESS;

3,250,405✔

450

  taosMemoryFreeClear(pBlock->pBlockAgg);

634,752,669✔

453

  code = doDynamicPruneDataBlock(pOperator, pBlockInfo, status);

634,626,025✔

454

  if (code) {

634,674,955✔

459

  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

634,674,955✔

460

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

12,629,475✔

462

    pCost->skipBlocks += 1;

12,630,466✔

463

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

12,630,070✔

465

    STableScanInfo* p1 = pOperator->info;

12,629,578✔

466

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

12,629,578✔

467

      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;

3,761,070✔

469

      *status = FUNC_DATA_REQUIRED_FILTEROUT;

8,868,508✔

471

    return TSDB_CODE_SUCCESS;

12,630,070✔

474

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

621,998,226✔

475

  pCost->loadBlocks += 1;

622,111,409✔

477

  SSDataBlock* p = NULL;

622,137,976✔

478

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

622,139,345✔

479

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

622,233,063✔

483

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

622,233,063✔

484

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

494,119,109✔

488

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

592,363,618✔

489

    if (code) {

592,265,234✔

495

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

622,078,080✔

497

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

622,117,017✔

498

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

84,701,634✔

499

    QUERY_CHECK_CODE(code, lino, _end);

84,685,950✔

501

    int64_t st = taosGetTimestampUs();

84,575,964✔

502

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

84,591,304✔

503

    pTableScanInfo->readRecorder.filterTime += el;

84,591,304✔

505

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

84,580,698✔

506

      pCost->filterOutBlocks += 1;

15,125,745✔

507

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

15,120,575✔

510

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

69,433,988✔

514

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

622,043,214✔

515

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

622,061,520✔

516

    setOperatorCompleted(pOperator);

3,587,182✔

519

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

622,062,338✔

521

_end:

622,288,499✔

522

  if (code != TSDB_CODE_SUCCESS) {

622,193,121✔

523

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

128,166✔

525

  return code;

622,040,405✔

528

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

65,766✔

529

  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

65,766✔

531

  switchCtxOrder(pCtx, numOfOutput);

65,766✔

532

  pTableScanInfo->cond.order = TSDB_ORDER_DESC;

65,766✔

535

65,766✔

603

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

349,309,828✔

605

  int32_t          code = TSDB_CODE_SUCCESS;

349,309,828✔

606

  int32_t          lino = 0;

349,309,828✔

607

  bool             freeReader = false;

349,309,828✔

608

  LRUHandle*       h = NULL;

349,309,828✔

609

  STableCachedVal  val = {0};

349,309,828✔

610

  SMetaReader      mr = {0};

349,309,828✔

611

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

349,318,290✔

612

  int32_t          insertRet = TAOS_LRU_STATUS_OK;

349,364,548✔

613

  STableCachedVal* pVal = NULL;

349,364,548✔

616

  if (numOfExpr <= 0) {

349,329,334✔

617

    return TSDB_CODE_SUCCESS;

483,275✔

623

  int32_t backupRows = pBlock->info.rows;

348,846,059✔

624

  pBlock->info.rows = rows;

348,938,163✔

630

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

348,947,698✔

631

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

348,951,826✔

632

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

348,985,874✔

633

    if (code != TSDB_CODE_SUCCESS) {

348,750,237✔

648

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

348,750,237✔

650

    val.pName = mr.me.name;

348,899,988✔

651

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

348,899,988✔

653

    freeReader = true;

348,899,988✔

695

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

1,179,526,400✔

696

    const SExprInfo* pExpr1 = &pExpr[j];

830,505,397✔

697

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

830,565,520✔

699

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

830,510,003✔

700

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

830,740,259✔

702

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

830,709,608✔

705

    if (fmIsScanPseudoColumnFunc(functionId)) {

830,758,672✔

706

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

97,321,622✔

707

      if (fType == FUNCTION_TYPE_TBNAME) {

97,333,630✔

708

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

97,333,630✔

709

        QUERY_CHECK_CODE(code, lino, _end);

97,311,196✔

718

      STagVal tagVal = {0};

733,208,745✔

719

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

733,221,297✔

720

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

733,424,892✔

722

      char* data = NULL;

733,438,700✔

723

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

733,438,700✔

724

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

727,835,869✔

726

        data = (char*)p;

5,627,991✔

729

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

733,292,237✔

730

      if (isNullVal) {

733,301,145✔

731

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

5,144,889✔

732

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

728,156,256✔

733

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

727,742,201✔

734

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

727,772,485✔

735

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

292,522,645✔

736

          if (tmp != NULL) {

292,544,214✔

737

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

292,544,214✔

738

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

292,540,458✔

740

            taosMemoryFree(tmp);

292,580,438✔

742

          taosMemoryFree(data);

292,561,493✔

744

        QUERY_CHECK_CODE(code, lino, _end);

727,698,038✔

746

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

2,983,196✔

747

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

2,501,195✔

748

          QUERY_CHECK_CODE(code, lino, _end);

2,500,830✔

755

  pBlock->info.rows = backupRows;

349,021,003✔

757

_end:

348,905,071✔

759

  if (NULL != pVal) {

348,949,957✔

768

  if (freeReader) {

348,949,957✔

769

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

348,854,902✔

771

  if (code != TSDB_CODE_SUCCESS) {

348,934,624✔

774

  return code;

348,830,010✔

777

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

97,317,706✔

779

  int32_t                     code = TSDB_CODE_SUCCESS;

97,317,706✔

780

  int32_t                     lino = 0;

97,317,706✔

781

  struct SScalarFuncExecFuncs fpSet = {0};

97,317,706✔

782

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

97,328,557✔

783

  QUERY_CHECK_CODE(code, lino, _end);

97,328,108✔

785

  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;

97,328,108✔

786

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

97,328,108✔

787

  STR_TO_VARSTR(buf, name)

97,331,883✔

789

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

97,311,309✔

791

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

97,305,069✔

792

  QUERY_CHECK_CODE(code, lino, _end);

97,321,325✔

794

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

97,321,325✔

795

  QUERY_CHECK_CODE(code, lino, _end);

97,326,872✔

797

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

97,326,872✔

798

  SScalarParam param = {.columnData = pColInfoData};

97,328,595✔

800

  if (fpSet.process != NULL) {

97,319,114✔

801

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

97,319,114✔

802

    QUERY_CHECK_CODE(code, lino, _end);

97,338,422✔

807

  colDataDestroy(&infoData);

97,338,422✔

809

_end:

97,301,863✔

810

  if (code != TSDB_CODE_SUCCESS) {

97,324,361✔

813

  return code;

97,324,361✔

882

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

178,422,970✔

883

  int32_t code = TSDB_CODE_SUCCESS;

178,422,970✔

884

  int32_t lino = 0;

178,422,970✔

885

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

178,422,970✔

886

  QUERY_CHECK_CODE(code, lino, _end);

178,440,952✔

888

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

178,440,952✔

889

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

178,483,528✔

890

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

178,284,127✔

891

  taosMemoryFreeClear(pInfo->pResBlock->pBlockAgg);

178,447,674✔

893

  if (!pInfo->needCountEmptyTable) {

178,312,226✔

894

    pInfo->countState = TABLE_COUNT_STATE_END;

163,898,751✔

896

    pInfo->countState = TABLE_COUNT_STATE_SCAN;

14,523,988✔

899

_end:

178,408,202✔

900

  if (code != TSDB_CODE_SUCCESS) {

178,408,202✔

903

  return code;

178,417,325✔

906

void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {

552,773,883✔

907

  if (pInfo->countState == TABLE_COUNT_STATE_END) {

552,773,883✔

908

    return;

473,978,041✔

910

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

78,958,288✔

911

    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;

223,774✔

913

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

78,740,316✔

914

    if (code != TSDB_CODE_SUCCESS) {

78,742,269✔

915

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

15,250,976✔

920

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

45,964,602✔

922

  blockDataEmpty(pBlock);

45,964,602✔

923

  pBlock->info.rows = 1;

45,964,101✔

924

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

45,964,101✔

925

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

45,963,600✔

926

  pBlock->info.blankFill = true;

45,964,101✔

929

  int32_t col_num = blockDataGetNumOfCols(pBlock);

45,962,598✔

930

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

190,060,567✔

931

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

144,097,468✔

936

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

45,963,099✔

938

  return pBlock;

45,961,596✔

941

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

44,168,481✔

942

  STableScanInfo* pTableScanInfo = pOperator->info;

44,168,481✔

944

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

44,168,982✔

946

  pOperator->resultInfo.totalRows++;

44,165,475✔

947

  return pBlock;

44,164,974✔

954

static int32_t prepareTableScan(SOperatorInfo* pOperator) {

709,560,322✔

955

  int32_t code = TSDB_CODE_SUCCESS;

709,560,322✔

956

  int32_t lino = 0;

709,560,322✔

958

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

709,560,322✔

959

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

709,670,378✔

960

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

709,902,889✔

961

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

33,189,690✔

962

         NOTIFY_TYPE_SCAN_PARAM &&

2,975,371✔

963

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

2,975,371✔

964

    STableScanOperatorParam* pNotify =

2,975,371✔

967

    STableScanInfo* pTableScanInfo = pOperator->info;

2,975,371✔

968

    SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

2,975,371✔

969

    SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

2,975,371✔

970

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

2,975,371✔

972

    QUERY_CHECK_CODE(code, lino, _end);

2,975,371✔

974

    if (!pGetParam->reUse) {

2,975,371✔

975

      freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

90,894✔

981

      pGetParam->reUse = false;

2,884,477✔

986

    pOperator->pOperatorGetParam = NULL;

2,975,371✔

989

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

709,888,705✔

990

_end:

86,458,235✔

991

  if (code != TSDB_CODE_SUCCESS) {

709,892,929✔

995

  return code;

709,882,953✔

998

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

671,515,236✔

999

  int32_t         code = TSDB_CODE_SUCCESS;

671,515,236✔

1000

  int32_t         lino = 0;

671,515,236✔

1001

  STableScanInfo* pTableScanInfo = pOperator->info;

671,515,236✔

1002

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

671,636,286✔

1003

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

671,635,625✔

1004

  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;

671,532,109✔

1005

  bool            hasNext = false;

671,585,875✔

1006

  int64_t         st = taosGetTimestampUs();

671,577,452✔

1008

  QRY_PARAM_CHECK(ppRes);

671,577,452✔

1009

  pBlock->info.dataLoad = false;

671,607,831✔

1011

  while (true) {

22,043,943✔

1012

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

693,676,658✔

1013

    if (code != TSDB_CODE_SUCCESS) {

693,442,080✔

1015

      QUERY_CHECK_CODE(code, lino, _end);

135✔

1018

    if (!hasNext) {

693,413,329✔

1019

      break;

158,671,723✔

1022

    if (isTaskKilled(pTaskInfo)) {

534,741,606✔

1023

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

17,551✔

1024

      code = pTaskInfo->code;

17,551✔

1025

      QUERY_CHECK_CODE(code, lino, _end);

17,551✔

1028

    if (pOperator->status == OP_EXEC_DONE) {

535,053,830✔

1029

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

234,304✔

1030

      break;

234,304✔

1034

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

534,669,847✔

1035

    if (!processThisBlock) {

534,723,571✔

1039

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

534,723,571✔

1040

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

534,806,275✔

1043

    uint32_t status = 0;

534,831,761✔

1044

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

534,862,445✔

1045

    QUERY_CHECK_CODE(code, lino, _end);

534,664,216✔

1047

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

534,552,201✔

1048

      break;

3,761,070✔

1052

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

530,791,131✔

1053

      continue;

22,043,943✔

1056

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

508,817,183✔

1057

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

508,885,774✔

1059

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

508,912,223✔

1060

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

508,895,454✔

1062

    (*ppRes) = pBlock;

508,796,391✔

1063

    return code;

508,755,241✔

1066

_end:

162,799,959✔

1067

  if (code != TSDB_CODE_SUCCESS) {

162,808,058✔

1068

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

135,889✔

1069

    pTaskInfo->code = code;

135,889✔

1070

    T_LONG_JMP(pTaskInfo->env, code);

129,566✔

1072

  return code;

162,672,169✔

1075

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

718,562,629✔

1076

  int32_t         code = TSDB_CODE_SUCCESS;

718,562,629✔

1077

  int32_t         lino = 0;

718,562,629✔

1078

  STableScanInfo* pTableScanInfo = pOperator->info;

718,562,629✔

1079

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

718,623,998✔

1080

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

718,519,247✔

1082

  QRY_PARAM_CHECK(pBlock);

718,528,923✔

1085

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

718,542,436✔

1086

    return code;

2,921,757✔

1090

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

867,387,960✔

1091

    SSDataBlock* p = NULL;

623,205,240✔

1092

    code = doTableScanImplNext(pOperator, &p);

623,183,571✔

1093

    QUERY_CHECK_CODE(code, lino, _end);

622,950,297✔

1095

    if (p != NULL) {

622,950,297✔

1096

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

471,151,685✔

1097

      *pBlock = p;

471,124,127✔

1098

      return code;

471,105,181✔

1101

    pTableScanInfo->scanTimes += 1;

151,798,612✔

1102

    taosHashClear(pTableScanInfo->pIgnoreTables);

151,813,964✔

1104

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

151,780,708✔

1105

      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

84,857✔

1106

      pTableScanInfo->base.scanFlag = MAIN_SCAN;

84,857✔

1107

      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

84,857✔

1108

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

84,857✔

1111

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

84,857✔

1112

      QUERY_CHECK_CODE(code, lino, _end);

77,899✔

1116

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

244,249,845✔

1117

  if (pTableScanInfo->scanTimes < total) {

244,239,976✔

1118

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

48,417,486✔

1119

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

65,766✔

1120

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

65,766✔

1121

      QUERY_CHECK_CODE(code, lino, _end);

65,766✔

1122

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

65,766✔

1125

    while (pTableScanInfo->scanTimes < total) {

59,284,664✔

1126

      SSDataBlock* p = NULL;

48,418,164✔

1127

      code = doTableScanImplNext(pOperator, &p);

48,402,064✔

1128

      QUERY_CHECK_CODE(code, lino, _end);

48,340,091✔

1130

      if (p != NULL) {

48,340,091✔

1131

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

37,476,432✔

1132

        *pBlock = p;

37,469,175✔

1133

        return code;

37,461,968✔

1136

      pTableScanInfo->scanTimes += 1;

10,863,659✔

1137

      taosHashClear(pTableScanInfo->pIgnoreTables);

10,866,386✔

1139

      if (pTableScanInfo->scanTimes < total) {

10,864,679✔

1150

  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {

206,651,499✔

1151

    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;

57,941,887✔

1152

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

57,942,388✔

1153

      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {

126,360✔

1154

        pTableScanInfo->countState = TABLE_COUNT_STATE_PROCESSED;

6,848✔

1156

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

6,848✔

1158

        if (NULL == pStart) {

6,848✔

1162

        *pBlock = getBlockForEmptyTable(pOperator, pStart);

6,848✔

1163

        return code;

6,848✔

1166

      int32_t numOfTables = 0;

57,814,996✔

1167

      code = tableListGetSize(pTableListInfo, &numOfTables);

57,815,497✔

1168

      QUERY_CHECK_CODE(code, lino, _end);

57,816,028✔

1170

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

57,816,028✔

1172

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

57,452,346✔

1173

        if (pIte != NULL) {

57,452,346✔

1174

          size_t        keySize = 0;

44,161,633✔

1175

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

44,161,633✔

1176

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

44,162,134✔

1177

          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);

44,161,132✔

1178

          markGroupProcessed(pTableScanInfo, *pGroupId);

44,161,132✔

1179

          *pBlock = getBlockForEmptyTable(pOperator, &info);

44,161,132✔

1181

          return code;

44,156,623✔

1185

    pTableScanInfo->countState = TABLE_COUNT_STATE_END;

13,773,591✔

1188

_end:

148,759,241✔

1189

  if (code != TSDB_CODE_SUCCESS) {

162,517,232✔

1194

  return code;

162,517,232✔

1197

static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {

952,939✔

1198

  STableScanInfo*          pInfo = pOperator->info;

952,939✔

1199

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

952,939✔

1200

  int32_t                  code = 0;

952,939✔

1201

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

952,939✔

1202

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

952,939✔

1203

  int32_t                  num = taosArrayGetSize(pParam->pUidList);

952,939✔

1204

  if (num <= 0) {

952,939✔

1209

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

952,939✔

1213

  if (pParam->tableSeq) {

952,939✔

1214

    pListInfo->oneTableForEachGroup = true;

697,038✔

1215

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

697,038✔

1221

    pListInfo->oneTableForEachGroup = false;

255,901✔

1222

    pListInfo->numOfOuputGroups = 1;

255,901✔

1224

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

952,939✔

1227

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

692,658✔

1228

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

692,658✔

1231

  STableKeyInfo info = {.groupId = 0};

952,939✔

1232

  int32_t       tableIdx = 0;

952,939✔

1233

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

8,902,831✔

1234

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

7,949,892✔

1235

    if (!pUid) {

7,947,876✔

1240

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

7,947,876✔

1241

      if (TSDB_CODE_DUP_KEY == terrno) {

6,655,297✔

1242

        continue;

6,655,801✔

1247

    info.uid = *pUid;

1,295,099✔

1248

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

1,295,099✔

1249

    if (p == NULL) {

1,295,099✔

1253

    tableIdx++;

1,295,099✔

1254

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

1,295,099✔

1257

  return code;

952,939✔

1260

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

163,093,872✔

1262

  const char* idStr = GET_TASKID(pTaskInfo);

163,093,872✔

1263

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

163,131,895✔

1264

  if (code) {

163,050,871✔

1269

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

163,050,871✔

1274

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

326,123,715✔

1275

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

163,025,755✔

1276

  if (code) {

162,984,839✔

1277

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

2,175✔

1280

  return code;

162,940,095✔

1283

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

318,049,742✔

1284

  SColIdPair* node1 = (SColIdPair*)elem1;

318,049,742✔

1285

  SColIdPair* node2 = (SColIdPair*)elem2;

318,049,742✔

1287

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

318,049,742✔

1288

    return 0;

107,816✔

1291

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

317,973,375✔

1294

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

424,840,865✔

1295

  SColIdSlotIdPair* node1 = (SColIdSlotIdPair*)elem1;

424,840,865✔

1296

  SColIdSlotIdPair* node2 = (SColIdSlotIdPair*)elem2;

424,840,865✔

1298

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

424,840,865✔

1299

    return 0;

164,696✔

1302

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

424,685,376✔

1305

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

741,654,338✔

1306

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

741,654,338✔

1307

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

741,654,338✔

1309

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

741,654,338✔

1313

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

741,654,768✔

1316

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

97,313,266✔

1317

  if (pList == NULL) {

97,313,266✔

1321

  int32_t left = 0;

97,313,266✔

1322

  int32_t right = taosArrayGetSize(pList) - 1;

97,313,266✔

1323

  while (left <= right) {

302,152,187✔

1324

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

302,152,187✔

1325

    SColMatchItem* pItem = taosArrayGet(pList, mid);

302,152,187✔

1326

    if (pItem->colId == colId) {

302,152,187✔

1328

      while (mid > 0) {

97,313,266✔

1329

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

73,067,909✔

1330

        if (pPrev->colId != colId) {

73,067,909✔

1331

          break;

73,067,909✔

1335

      return taosArrayGet(pList, mid);

97,313,266✔

1338

    if (pItem->colId < colId) {

204,838,921✔

1339

      left = mid + 1;

96,185,414✔

1341

      right = mid - 1;

108,653,507✔

1348

static bool isNewScanParam(STableScanOperatorParam* pParam) {

41,474,609✔

1349

  return pParam->isNewParam;

41,474,609✔

1352

static int32_t createVTableScanInfoFromBatchParam(SOperatorInfo* pOperator) {

20,553,279✔

1353

  int32_t                  code = 0;

20,553,279✔

1354

  int32_t                  lino = 0;

20,553,279✔

1355

  STableScanInfo*          pInfo = pOperator->info;

20,553,279✔

1356

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

20,563,328✔

1357

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

20,539,956✔

1358

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

20,553,438✔

1359

  SMetaReader              orgTable = {0};

20,569,941✔

1360

  SMetaReader              superTable = {0};

20,578,430✔

1361

  SSchemaWrapper*          schema = NULL;

20,566,047✔

1362

  SArray*                  pColArray = NULL;

20,566,047✔

1363

  SArray*                  pBlockColArray = NULL;

20,566,047✔

1364

  int32_t                  num = 0;

20,566,047✔

1365

  STableKeyInfo*           pList = NULL;

20,562,354✔

1369

  cleanupQueryTableDataCond(&pInfo->base.cond);

20,563,406✔

1371

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

20,569,954✔

1372

  if (pGetParam &&

20,575,925✔

1373

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

5,012,906✔

1374

    STableScanOperatorParam* pParam =

5,018,341✔

1376

    pInfo->pBatchColMap = pParam->pBatchTbInfo;

5,025,480✔

1377

    pParam->pBatchTbInfo = NULL;

5,014,468✔

1378

    pInfo->cachedTimeWindow = pParam->window;

5,001,013✔

1379

    pInfo->currentBatchIdx = 0;

5,016,460✔

1380

    pInfo->lastBatchIdx = -1;

4,996,980✔

1381

    pInfo->cachedTagList = pParam->pTagList;

5,006,973✔

1382

    pParam->pTagList = NULL;

5,016,405✔

1383

    pInfo->cachedGroupId = pParam->groupid;

5,012,758✔

1385

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

5,020,905✔

1387

    taosHashClear(pInfo->pIgnoreTables);

5,020,905✔

1389

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

5,017,776✔

1390

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

4,827,136✔

1391

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

61,158,848✔

1392

        SqlFunctionCtx ctx = pSup->pCtx[i];

56,351,980✔

1393

        ctx.skipDynDataCheck = false;

56,333,200✔

1398

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

20,559,040✔

1399

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

20,580,437✔

1401

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

20,580,437✔

1402

    isNewTable = false;

12,578,413✔

1404

    isNewTable = true;

7,991,804✔

1405

    pInfo->lastBatchIdx = pInfo->currentBatchIdx;

7,991,804✔

1408

  if (isNewTable) {

20,576,322✔

1409

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

7,997,909✔

1410

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

7,998,556✔

1411

    pAPI->metaReaderFn.readerReleaseLock(&orgTable);

7,980,087✔

1412

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

7,976,583✔

1413

    QUERY_CHECK_CODE(code, lino, _return);

7,984,595✔

1414

    switch (orgTable.me.type) {

7,984,595✔

1415

      case TSDB_CHILD_TABLE:

279,224✔

1416

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

279,224✔

1417

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

279,224✔

1418

        pAPI->metaReaderFn.readerReleaseLock(&superTable);

279,224✔

1419

        QUERY_CHECK_CODE(code, lino, _return);

279,224✔

1420

        schema = &superTable.me.stbEntry.schemaRow;

279,224✔

1421

        break;

279,224✔

1422

      case TSDB_NORMAL_TABLE:

7,662,614✔

1423

        schema = &orgTable.me.ntbEntry.schemaRow;

7,662,614✔

1424

        break;

7,662,614✔

1425

      default:

42,757✔

1426

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

42,757✔

1431

    pListInfo->oneTableForEachGroup = true;

7,941,838✔

1432

    taosHashClear(pListInfo->map);

7,985,615✔

1433

    taosArrayClear(pListInfo->pTableList);

8,005,972✔

1435

    uint64_t      pUid = orgTable.me.uid;

8,005,972✔

1436

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

8,005,972✔

1437

    int32_t       tableIdx = 0;

8,003,992✔

1438

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

8,004,484✔

1439

    QUERY_CHECK_CODE(code, lino, _return);

8,002,528✔

1440

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

16,003,544✔

1441

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

8,001,016✔

1443

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

8,001,016✔

1444

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

7,992,524✔

1445

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

7,992,524✔

1446

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

7,993,600✔

1449

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

7,993,600✔

1450

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

8,003,512✔

1452

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

30,759,306✔

1453

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

22,749,282✔

1454

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

181,909,169✔

1455

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

181,578,454✔

1456

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

22,764,218✔

1457

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

22,768,214✔

1458

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

22,768,214✔

1460

          break;

22,757,798✔

1465

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

38,768,497✔

1466

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

30,761,630✔

1467

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

185,636,022✔

1468

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

185,450,805✔

1469

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

185,531,722✔

1470

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

30,765,627✔

1471

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

30,770,638✔

1472

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

30,770,638✔

1474

          break;

30,749,189✔

1479

    taosArraySort(pColArray, compareColIdPair);

8,000,939✔

1480

    taosArraySort(pBlockColArray, compareColIdSlotIdPair);

7,999,516✔

1481

    if (pInfo->pBlockColMap) {

8,000,420✔

1482

      taosArrayDestroy(pInfo->pBlockColMap);

5,531,868✔

1483

      pInfo->pBlockColMap = NULL;

5,526,888✔

1485

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

7,993,952✔

1486

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

7,997,020✔

1489

    taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

7,992,217✔

1490

    taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

7,989,908✔

1492

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

8,001,373✔

1493

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

7,283,279✔

1494

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

90,556,323✔

1495

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

83,299,551✔

1496

        bool           needScan = false;

83,281,585✔

1497

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

167,694,290✔

1498

          SFunctParam pParam = ctx->param[j];

84,382,806✔

1499

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

404,989,900✔

1500

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

340,409,106✔

1501

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

341,138,850✔

1502

              needScan |= true;

21,006,051✔

1503

              break;

21,006,051✔

1507

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

83,288,063✔

1508

        if (!needScan) {

83,336,334✔

1509

          ctx->skipDynDataCheck = true;

62,339,660✔

1511

          ctx->skipDynDataCheck = false;

20,996,674✔

1518

    pColArray = pInfo->lastColArray;

12,578,413✔

1519

    pBlockColArray = pInfo->lastBlockColArray;

12,578,917✔

1522

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

20,579,451✔

1523

  QUERY_CHECK_CODE(code, lino, _return);

20,582,429✔

1525

  if (pInfo->pResBlock) {

20,582,429✔

1526

    blockDataDestroy(pInfo->pResBlock);

15,553,030✔

1527

    pInfo->pResBlock = NULL;

15,551,385✔

1530

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

20,581,365✔

1538

  if (isNewTable) {

20,568,868✔

1539

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

7,990,455✔

1540

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

7,280,886✔

1541

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

7,247,488✔

1543

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

720,660✔

1544

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

708,561✔

1547

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

12,578,413✔

1548

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

1,700,136✔

1550

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

10,877,785✔

1554

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

20,563,565✔

1555

  pInfo->currentGroupId = 0;

20,556,533✔

1556

  pInfo->ignoreTag = true;

20,549,430✔

1558

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

20,494,987✔

1559

  QUERY_CHECK_CODE(code, lino, _return);

20,567,913✔

1561

  if (isNewTable) {

20,567,913✔

1562

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

7,988,996✔

1563

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

5,532,816✔

1565

    pInfo->base.dataReader = NULL;

7,953,694✔

1567

    taosRLockLatch(&pTaskInfo->lock);

7,956,157✔

1568

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

8,003,512✔

1569

    taosRUnLockLatch(&pTaskInfo->lock);

7,953,763✔

1570

    QUERY_CHECK_CODE(code, lino, _return);

7,980,553✔

1572

    pInfo->newReader = true;

7,980,553✔

1573

    if (pInfo->lastColArray) {

7,980,577✔

1574

      taosArrayDestroy(pInfo->lastColArray);

2,973,544✔

1575

      pInfo->lastColArray = NULL;

2,972,548✔

1577

    if (pInfo->lastBlockColArray) {

7,974,943✔

1578

      taosArrayDestroy(pInfo->lastBlockColArray);

2,972,560✔

1579

      pInfo->lastBlockColArray = NULL;

2,971,564✔

1581

    pInfo->lastColArray = pColArray;

7,948,615✔

1582

    pInfo->lastBlockColArray = pBlockColArray;

7,951,366✔

1584

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

12,578,917✔

1585

    QUERY_CHECK_CODE(code, lino, _return);

12,579,486✔

1587

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

12,579,486✔

1588

    pInfo->newReader = false;

12,579,486✔

1589

    pInfo->scanTimes = 0;

12,579,486✔

1592

  pOperator->status = OP_OPENED;

20,549,674✔

1594

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

20,546,128✔

1598

  pInfo->currentGroupId = -1;

20,554,361✔

1599

_return:

20,451,951✔

1600

  if (code) {

20,508,048✔

1603

  pAPI->metaReaderFn.clearReader(&superTable);

20,508,048✔

1604

  pAPI->metaReaderFn.clearReader(&orgTable);

20,527,808✔

1605

  return code;

20,548,980✔

1608

static int32_t createVTableScanInfoFromParam(SOperatorInfo* pOperator) {

24,245,142✔

1609

  int32_t                  code = 0;

24,245,142✔

1610

  int32_t                  lino = 0;

24,245,142✔

1611

  STableScanInfo*          pInfo = pOperator->info;

24,245,142✔

1612

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

24,245,357✔

1613

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

24,245,142✔

1614

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

24,245,357✔

1615

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

24,245,357✔

1616

  SMetaReader              orgTable = {0};

24,245,142✔

1617

  SMetaReader              superTable = {0};

24,245,142✔

1618

  SSchemaWrapper*          schema = NULL;

24,245,142✔

1619

  SArray*                  pColArray = NULL;

24,245,142✔

1620

  SArray*                  pBlockColArray = NULL;

24,245,142✔

1621

  int32_t                  num = 0;

24,245,142✔

1622

  STableKeyInfo*           pList = NULL;

24,245,357✔

1624

  SArray*                  pMatchList = NULL;

24,245,357✔

1626

  cleanupQueryTableDataCond(&pInfo->base.cond);

24,245,357✔

1628

  pOrgTbInfo = pParam->pOrgTbInfo;

24,245,357✔

1630

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

24,245,142✔

1632

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

24,245,142✔

1633

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

24,245,357✔

1634

  pAPI->metaReaderFn.readerReleaseLock(&orgTable);

24,245,142✔

1635

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

24,245,357✔

1636

  QUERY_CHECK_CODE(code, lino, _return);

24,245,142✔

1637

  switch (orgTable.me.type) {

24,245,142✔

1638

    case TSDB_CHILD_TABLE:

6,347,094✔

1639

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

6,347,094✔

1640

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

6,347,094✔

1641

      pAPI->metaReaderFn.readerReleaseLock(&superTable);

6,347,094✔

1642

      QUERY_CHECK_CODE(code, lino, _return);

6,347,094✔

1643

      schema = &superTable.me.stbEntry.schemaRow;

6,347,094✔

1644

      break;

6,347,094✔

1645

    case TSDB_NORMAL_TABLE:

17,898,263✔

1646

      schema = &orgTable.me.ntbEntry.schemaRow;

17,898,263✔

1647

      break;

17,898,263✔

1654

  pListInfo->oneTableForEachGroup = true;

24,245,357✔

1655

  taosHashClear(pListInfo->map);

24,245,142✔

1656

  taosArrayClear(pListInfo->pTableList);

24,245,357✔

1658

  uint64_t      pUid = orgTable.me.uid;

24,245,357✔

1659

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

24,245,357✔

1660

  int32_t       tableIdx = 0;

24,245,357✔

1661

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

24,245,142✔

1662

  QUERY_CHECK_CODE(code, lino, _return);

24,245,357✔

1663

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

48,490,714✔

1664

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

24,245,357✔

1666

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

24,245,357✔

1667

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

24,245,142✔

1668

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

24,245,142✔

1669

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

24,245,357✔

1670

  if (pInfo->pBlockColMap) {

24,245,357✔

1671

    taosArrayDestroy(pInfo->pBlockColMap);

21,809,188✔

1672

    pInfo->pBlockColMap = NULL;

21,809,188✔

1674

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

24,245,142✔

1675

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

24,245,142✔

1678

  SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID,

24,245,142✔

1682

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

24,245,357✔

1684

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

97,313,266✔

1685

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

73,067,909✔

1686

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

617,660,139✔

1687

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

617,660,139✔

1688

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

73,067,909✔

1689

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

73,067,694✔

1690

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

73,067,694✔

1691

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

73,067,694✔

1692

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

73,067,909✔

1693

        break;

73,067,909✔

1698

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

24,245,357✔

1699

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

24,245,142✔

1700

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

24,245,357✔

1701

    taosArraySort(pMatchList, compareColMatchItemByColId);

24,245,357✔

1705

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

121,557,783✔

1706

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

97,313,051✔

1707

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

97,313,266✔

1708

    if (pItem) {

97,313,266✔

1709

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

97,313,266✔

1710

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

1,055✔

1712

        code = TSDB_CODE_VTABLE_COLUMN_TYPE_MISMATCH;

625✔

1713

        goto _return;

625✔

1715

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

97,312,426✔

1716

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

97,312,641✔

1717

      if (i > 0) {

97,312,641✔

1718

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

146,134,568✔

1723

  taosArraySort(pColArray, compareColIdPair);

24,244,732✔

1724

  taosArraySort(pBlockColArray, compareColIdSlotIdPair);

24,244,517✔

1725

  taosArraySort(pInfo->pBlockColMap, compareColIdSlotIdPair);

24,244,732✔

1727

  taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

24,244,732✔

1728

  taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

24,244,517✔

1730

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

24,244,732✔

1731

  QUERY_CHECK_CODE(code, lino, _return);

24,244,732✔

1733

  if (pInfo->pResBlock) {

24,244,732✔

1734

    blockDataDestroy(pInfo->pResBlock);

17,232,344✔

1735

    pInfo->pResBlock = NULL;

17,232,344✔

1738

  if (isNewScanParam(pParam)) {

24,244,732✔

1739

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

7,019,458✔

1740

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

7,019,458✔

1741

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

7,019,458✔

1747

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

17,225,059✔

1749

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

24,244,517✔

1750

  pInfo->currentGroupId = 0;

24,244,517✔

1751

  pInfo->ignoreTag = true;

24,244,517✔

1753

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

24,244,517✔

1754

  QUERY_CHECK_CODE(code, lino, _return);

24,244,732✔

1756

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

24,244,732✔

1757

  if (reader) {

24,244,732✔

1758

    if (isNewScanParam(pParam)) {

17,230,092✔

1759

      pAPI->tsdReader.tsdReaderClose(*reader);

5,033✔

1760

      pInfo->base.dataReader = NULL;

5,033✔

1762

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

5,033✔

1763

      QUERY_CHECK_CODE(code, lino, _return);

5,033✔

1765

      taosRLockLatch(&pTaskInfo->lock);

5,033✔

1766

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

5,033✔

1767

      taosRUnLockLatch(&pTaskInfo->lock);

5,033✔

1768

      QUERY_CHECK_CODE(code, lino, _return);

5,033✔

1770

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

5,033✔

1771

      QUERY_CHECK_CODE(code, lino, _return);

5,033✔

1772

      pInfo->newReader = true;

5,033✔

1774

      pInfo->base.dataReader = *reader;

17,225,059✔

1775

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

17,225,059✔

1776

      QUERY_CHECK_CODE(code, lino, _return);

17,225,059✔

1778

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

17,225,059✔

1779

      pInfo->newReader = false;

17,225,059✔

1780

      pInfo->scanTimes = 0;

17,225,059✔

1783

    pInfo->base.dataReader = NULL;

7,014,640✔

1785

    taosRLockLatch(&pTaskInfo->lock);

7,014,640✔

1786

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

7,014,640✔

1787

    taosRUnLockLatch(&pTaskInfo->lock);

7,014,425✔

1788

    QUERY_CHECK_CODE(code, lino, _return);

7,014,640✔

1790

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

7,014,640✔

1791

    QUERY_CHECK_CODE(code, lino, _return);

7,014,640✔

1792

    pInfo->newReader = true;

7,014,640✔

1794

  pOperator->status = OP_OPENED;

24,244,517✔

1796

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

24,244,517✔

1800

  pInfo->currentGroupId = -1;

24,244,732✔

1801

_return:

24,245,142✔

1802

  if (code) {

24,245,357✔

1803

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

625✔

1805

  taosArrayDestroy(pMatchList);

24,245,357✔

1806

  taosArrayDestroy(pColArray);

24,245,357✔

1807

  taosArrayDestroy(pBlockColArray);

24,245,357✔

1808

  pAPI->metaReaderFn.clearReader(&superTable);

24,245,357✔

1809

  pAPI->metaReaderFn.clearReader(&orgTable);

24,245,357✔

1810

  return code;

24,245,357✔

1813

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

159,919,056✔

1814

  int32_t         code = TSDB_CODE_SUCCESS;

159,919,056✔

1815

  int32_t         lino = 0;

159,919,056✔

1816

  STableScanInfo* pInfo = pOperator->info;

159,919,056✔

1817

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

159,933,419✔

1818

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

159,898,315✔

1819

  int32_t         numOfTables = 0;

159,905,649✔

1821

  QRY_PARAM_CHECK(pResult);

159,878,269✔

1823

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

159,904,402✔

1824

  QUERY_CHECK_CODE(code, lino, _end);

159,938,767✔

1826

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

159,938,767✔

1827

    setOperatorCompleted(pOperator);

144,558,235✔

1828

    if (pOperator->dynamicTask) {

144,566,661✔

1829

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

951,108✔

1830

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

951,114✔

1832

    return code;

144,571,968✔

1836

  pOperator->status = OP_OPENED;

15,282,257✔

1837

  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);

15,351,210✔

1839

  int32_t        num = 0;

15,333,970✔

1840

  STableKeyInfo* pList = NULL;

15,334,677✔

1841

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

15,337,438✔

1842

  QUERY_CHECK_CODE(code, lino, _end);

15,359,189✔

1844

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

15,359,189✔

1845

  QUERY_CHECK_CODE(code, lino, _end);

15,357,759✔

1847

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

15,357,759✔

1848

  QUERY_CHECK_CODE(code, lino, _end);

15,330,809✔

1849

  pInfo->scanTimes = 0;

15,330,809✔

1851

  code = doGroupedTableScan(pOperator, pResult);

15,337,308✔

1852

  QUERY_CHECK_CODE(code, lino, _end);

15,384,322✔

1854

  if (*pResult != NULL) {

15,384,322✔

1855

    if (pOperator->dynamicTask) {

12,501,960✔

1856

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

12,245,880✔

1858

    return code;

12,502,744✔

1861

_end:

2,883,423✔

1862

  if (code != TSDB_CODE_SUCCESS) {

2,883,681✔

1866

  return code;

2,882,854✔

1869

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

662,149,374✔

1870

  int32_t         code = TSDB_CODE_SUCCESS;

662,149,374✔

1871

  int32_t         lino = 0;

662,149,374✔

1872

  STableScanInfo* pInfo = pOperator->info;

662,149,374✔

1873

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

662,170,893✔

1874

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

662,151,835✔

1875

  int32_t         num = 0;

662,172,218✔

1876

  STableKeyInfo*  pList = NULL;

662,157,687✔

1877

  SSDataBlock*    pResult = NULL;

662,088,050✔

1878

  const char*     idStr = GET_TASKID(pTaskInfo);

662,090,428✔

1880

  QRY_PARAM_CHECK(pResBlock);

662,094,290✔

1882

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

662,130,518✔

1883

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

148,057,340✔

1888

    taosRLockLatch(&pTaskInfo->lock);

148,075,946✔

1889

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

148,145,446✔

1890

    taosRUnLockLatch(&pTaskInfo->lock);

148,017,986✔

1891

    QUERY_CHECK_CODE(code, lino, _end);

148,092,342✔

1893

    if (pInfo->filesetDelimited) {

148,090,167✔

1894

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

256,984✔

1897

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

148,019,909✔

1902

  code = doGroupedTableScan(pOperator, &pResult);

662,159,287✔

1903

  QUERY_CHECK_CODE(code, lino, _end);

661,670,875✔

1905

  if (pResult != NULL) {

661,670,875✔

1906

    if (pOperator->dynamicTask) {

517,155,538✔

1907

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

1,996,303✔

1910

    *pResBlock = pResult;

517,155,595✔

1911

    return code;

516,838,844✔

1915

    code = startNextGroupScan(pOperator, &pResult);

144,595,885✔

1916

    QUERY_CHECK_CODE(code, lino, _end);

144,917,404✔

1918

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

144,917,404✔

1919

      *pResBlock = pResult;

144,843,477✔

1920

      return code;

144,826,687✔

1924

_end:

2,175✔

1925

  if (code != TSDB_CODE_SUCCESS) {

2,175✔

1926

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

2,175✔

1927

    pTaskInfo->code = code;

2,175✔

1930

  return code;

2,175✔

1933

static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {

24,244,517✔

1934

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

24,244,517✔

1937

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

952,939✔

1938

  int32_t                  code = TSDB_CODE_SUCCESS;

952,939✔

1939

  int32_t                  lino = 0;

952,939✔

1940

  STableScanInfo*          pInfo = pOperator->info;

952,939✔

1941

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

952,939✔

1942

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

952,939✔

1943

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

952,939✔

1945

  code = createTableListInfoFromParam(pOperator);

952,939✔

1946

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

952,939✔

1947

  pOperator->pOperatorGetParam = NULL;

952,939✔

1948

  QUERY_CHECK_CODE(code, lino, _end);

952,939✔

1950

  if (pOperator->status == OP_EXEC_DONE) {

952,939✔

1951

    pInfo->currentGroupId = -1;

22,764✔

1952

    pOperator->status = OP_OPENED;

22,764✔

1953

    SSDataBlock* result = NULL;

22,764✔

1956

      code = startNextGroupScan(pOperator, &result);

22,764✔

1957

      QUERY_CHECK_CODE(code, lino, _end);

22,764✔

1959

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

22,764✔

1960

        (*ppRes) = result;

22,764✔

1961

        return code;

22,764✔

1965

_end:

930,175✔

1966

  if (code) {

930,175✔

1969

  return code;

930,175✔

1972

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

17,233,852✔

1973

  int32_t      code = TSDB_CODE_SUCCESS;

17,233,852✔

1974

  int32_t      lino = 0;

17,233,852✔

1975

  bool         found = false;

17,233,852✔

1977

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

17,233,852✔

1978

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

17,233,852✔

1980

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

86,040,130✔

1981

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

68,806,493✔

1982

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

68,806,493✔

1983

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

197,279,567✔

1984

      SColIdSlotIdPair* pColPair = taosArrayGet(pColMap, j);

180,048,460✔

1985

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

180,048,675✔

1986

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

180,048,675✔

1987

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

51,575,386✔

1988

        found = true;

51,575,386✔

1989

        break;

51,575,386✔

1992

    if (!found) {

68,806,493✔

1993

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

17,231,107✔

1994

        pSrcCol->info.slotId = -1;

17,231,107✔

2003

  return code;

17,234,067✔

2009

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

24,245,142✔

2010

  int32_t                  code = TSDB_CODE_SUCCESS;

24,245,142✔

2011

  int32_t                  lino = 0;

24,245,142✔

2012

  STableScanInfo*          pInfo = pOperator->info;

24,245,142✔

2013

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

24,245,142✔

2014

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

24,245,142✔

2015

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

24,245,142✔

2017

  QRY_PARAM_CHECK(ppRes);

24,245,142✔

2018

  qTrace("%s call", __FUNCTION__);

24,245,142✔

2020

  code = createVTableScanInfoFromParam(pOperator);

24,245,142✔

2021

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

24,245,142✔

2022

  pOperator->pOperatorGetParam = NULL;

24,245,142✔

2024

  QUERY_CHECK_CODE(code, lino, _end);

24,245,142✔

2026

  SSDataBlock* result = NULL;

24,244,517✔

2027

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

24,244,732✔

2028

    (*ppRes) = result;

18,124✔

2029

    return code;

18,124✔

2032

  if (pInfo->newReader) {

24,226,608✔

2033

    code = startNextGroupScan(pOperator, &result);

7,019,458✔

2034

    QUERY_CHECK_CODE(code, lino, _end);

7,019,458✔

2036

    code = doGroupedTableScan(pOperator, &result);

17,206,935✔

2037

    QUERY_CHECK_CODE(code, lino, _end);

17,206,935✔

2040

  if (result) {

24,226,393✔

2041

    SSDataBlock* res = NULL;

17,233,852✔

2042

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

17,233,852✔

2043

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

17,233,852✔

2044

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

17,233,852✔

2046

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

17,234,067✔

2047

    QUERY_CHECK_CODE(code, lino, _end);

17,233,852✔

2053

    (*ppRes) = result;

17,233,852✔

2055

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

6,992,541✔

2056

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

6,992,541✔

2058

    blockDataDestroy(pInfo->pResBlock);

6,992,541✔

2059

    pInfo->pResBlock = NULL;

6,992,541✔

2061

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

6,992,541✔

2062

    if (reader) {

6,992,541✔

2063

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

6,992,541✔

2064

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

6,992,541✔

2065

        pInfo->base.dataReader = NULL;

6,992,541✔

2067

      if (pAPI->tsdReader.tsdReaderClose) {

6,992,541✔

2068

        pAPI->tsdReader.tsdReaderClose(*reader);

6,992,541✔

2072

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

6,992,541✔

2073

    QUERY_CHECK_CODE(code, lino, _end);

6,992,541✔

2075

_end:

24,227,018✔

2076

  if (code) {

24,227,018✔

2077

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

625✔

2079

  return code;

24,227,018✔

2082

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

3,642,360✔

2083

  int32_t                  code = TSDB_CODE_SUCCESS;

3,642,360✔

2084

  int32_t                  lino = 0;

3,642,360✔

2085

  STableScanInfo*          pInfo = pOperator->info;

3,642,360✔

2086

  int32_t                  index = 0;

3,642,360✔

2087

  char*                    tagVal = NULL;

3,642,360✔

2089

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

25,485,024✔

2090

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

21,842,172✔

2091

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

21,842,172✔

2093

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

21,842,664✔

2094

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

21,842,664✔

2096

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

2,147,483,647✔

2097

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

2,147,483,647✔

2098

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

2,147,483,647✔

2099

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

2,147,483,647✔

2101

        varDataSetLen(tagVal, pTagVal->nData);

2,147,483,647✔

2102

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

2,147,483,647✔

2103

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

2,147,483,647✔

2104

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2105

        taosMemoryFreeClear(tagVal);

2,147,483,647✔

2107

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

2,147,483,647✔

2108

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2111

    index++;

21,842,664✔

2113

  return code;

3,642,360✔

2120

static void clearVstbBatchDynamicTableScanInfo(STableScanInfo* pInfo) {

5,030,460✔

2121

  if (pInfo->pBatchColMap) {

5,030,460✔

2122

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

13,032,004✔

2123

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

8,003,512✔

2124

      taosArrayDestroy(pOrgTbInfo->colMap);

8,000,560✔

2126

    taosArrayDestroy(pInfo->pBatchColMap);

5,028,984✔

2127

    pInfo->pBatchColMap = NULL;

5,027,508✔

2129

  if (pInfo->cachedTagList) {

5,028,492✔

2130

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

15,777,828✔

2131

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

13,445,184✔

2132

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

13,448,628✔

2133

        taosMemoryFreeClear(pTagVal->pData);

5,905,848✔

2136

    taosArrayDestroy(pInfo->cachedTagList);

2,332,644✔

2137

    pInfo->cachedTagList = NULL;

2,331,168✔

2139

  if (pInfo->lastColArray) {

5,027,016✔

2140

    taosArrayDestroy(pInfo->lastColArray);

5,030,952✔

2141

    pInfo->lastColArray = NULL;

5,025,540✔

2143

  if (pInfo->lastBlockColArray) {

5,029,476✔

2144

    taosArrayDestroy(pInfo->lastBlockColArray);

5,029,968✔

2145

    pInfo->lastBlockColArray = NULL;

5,027,016✔

2148

  blockDataDestroy(pInfo->pResBlock);

5,028,984✔

2149

  pInfo->pResBlock = NULL;

5,028,492✔

2150

5,028,984✔

2152

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

17,590,889✔

2153

  int32_t                  code = TSDB_CODE_SUCCESS;

17,590,889✔

2154

  int32_t                  lino = 0;

17,590,889✔

2155

  STableScanInfo*          pInfo = pOperator->info;

17,590,889✔

2156

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

17,590,853✔

2157

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

17,592,377✔

2158

  SSDataBlock*             result = NULL;

17,582,950✔

2160

  QRY_PARAM_CHECK(ppRes);

17,588,870✔

2161

  qTrace("%s call", __FUNCTION__);

17,597,369✔

2163

  while (!result) {

33,148,025✔

2164

    code = createVTableScanInfoFromBatchParam(pOperator);

20,570,092✔

2165

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

20,547,158✔

2166

    pOperator->pOperatorGetParam = NULL;

20,517,642✔

2167

    QUERY_CHECK_CODE(code, lino, _end);

20,542,582✔

2169

    if (pInfo->newReader) {

20,542,582✔

2170

      code = startNextGroupScan(pOperator, &result);

7,948,061✔

2171

      QUERY_CHECK_CODE(code, lino, _end);

8,001,867✔

2173

      code = doGroupedTableScan(pOperator, &result);

12,596,513✔

2174

      QUERY_CHECK_CODE(code, lino, _end);

12,578,502✔

2176

    if (!result) {

20,580,369✔

2177

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

8,001,944✔

2179

      pInfo->currentBatchIdx++;

8,001,944✔

2180

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

8,001,452✔

2182

        break;

5,028,492✔

2187

  if (result) {

17,606,425✔

2188

    SSDataBlock* res = NULL;

12,577,441✔

2189

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

12,578,502✔

2190

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

12,576,949✔

2191

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

12,577,506✔

2193

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

12,577,441✔

2194

    QUERY_CHECK_CODE(code, lino, _end);

12,579,486✔

2196

    if (pInfo->cachedTagList) {

12,579,486✔

2197

      code = setTagValFromTagList(pOperator, res);

3,642,360✔

2198

      QUERY_CHECK_CODE(code, lino, _end);

3,642,360✔

2201

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

12,579,486✔

2202

    pInfo->pResBlock = res;

12,579,486✔

2203

    blockDataDestroy(result);

12,579,486✔

2204

    (*ppRes) = res;

12,579,486✔

2206

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

12,579,486✔

2208

    clearVstbBatchDynamicTableScanInfo(pInfo);

5,028,984✔

2209

    setOperatorCompleted(pOperator);

5,029,476✔

2212

_end:

17,608,470✔

2213

  return code;

17,608,470✔

2216

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

42,801,482✔

2217

  int32_t                  code = TSDB_CODE_SUCCESS;

42,801,482✔

2218

  int32_t                  lino = 0;

42,801,482✔

2219

  STableScanInfo*          pInfo = pOperator->info;

42,801,482✔

2220

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

42,805,946✔

2221

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

42,800,474✔

2223

  QRY_PARAM_CHECK(ppRes);

42,801,193✔

2224

  qTrace("%s call", __FUNCTION__);

42,794,474✔

2226

  pOperator->dynamicTask = true;

42,794,474✔

2228

  switch (type) {

42,800,966✔

2229

    case DYN_TYPE_VSTB_WIN_SCAN:

952,939✔

2231

      code = doStbJoinDynamicTableScanNext(pOperator, ppRes);

952,939✔

2232

      QUERY_CHECK_CODE(code, lino, _end);

952,939✔

2233

      break;

952,939✔

2235

    case DYN_TYPE_VSTB_SINGLE_SCAN: {

24,245,142✔

2236

      code = doVstbSingleDynamicTableScanNext(pOperator, ppRes);

24,245,142✔

2237

      QUERY_CHECK_CODE(code, lino, _end);

24,245,142✔

2238

      break;

24,244,517✔

2240

    case DYN_TYPE_VSTB_BATCH_SCAN:

17,602,885✔

2241

      code = doVstbBatchDynamicTableScanNext(pOperator, ppRes);

17,602,885✔

2242

      QUERY_CHECK_CODE(code, lino, _end);

17,608,962✔

2243

      break;

17,608,962✔

2250

_end:

42,807,043✔

2251

  return code;

42,807,043✔

2254

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

709,617,172✔

2255

  int32_t         code = TSDB_CODE_SUCCESS;

709,617,172✔

2256

  int32_t         lino = 0;

709,617,172✔

2257

  STableScanInfo* pInfo = pOperator->info;

709,617,172✔

2258

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

709,821,028✔

2259

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

709,722,740✔

2260

  QRY_PARAM_CHECK(ppRes);

709,832,227✔

2261

  qTrace("%s call", __FUNCTION__);

709,834,151✔

2263

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

709,835,988✔

2264

  QUERY_CHECK_CODE(code, lino, _end);

709,879,045✔

2266

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

709,879,045✔

2267

  if (pGetParam || pInfo->pBatchColMap) {

709,888,255✔

2268

    ETableScanDynType type = pGetParam ?

42,795,081✔

2269

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

42,801,974✔

2272

    code = doDynamicTableScanNext(pOperator, type, ppRes);

42,795,081✔

2273

    QUERY_CHECK_CODE(code, lino, _end);

42,807,043✔

2275

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

42,806,418✔

2276

      return code;

41,877,719✔

2281

  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {

667,992,378✔

2282

    int32_t       numOfTables = 0;

5,875,135✔

2283

    STableKeyInfo tInfo = {0};

5,875,135✔

2284

    pInfo->countState = TABLE_COUNT_STATE_END;

5,875,135✔

2286

    while (1) {

5,491,263✔

2287

      SSDataBlock* result = NULL;

11,366,398✔

2288

      code = doGroupedTableScan(pOperator, &result);

11,363,935✔

2289

      QUERY_CHECK_CODE(code, lino, _end);

11,356,536✔

2291

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

11,356,536✔

2292

        (*ppRes) = result;

5,821,843✔

2293

        return code;

5,833,561✔

2297

      pInfo->currentTable++;

5,532,221✔

2299

      taosRLockLatch(&pTaskInfo->lock);

5,524,215✔

2300

      numOfTables = 0;

5,534,385✔

2301

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

5,534,385✔

2302

      if (code != TSDB_CODE_SUCCESS) {

5,533,149✔

2307

      if (pInfo->currentTable >= numOfTables) {

5,533,149✔

2308

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

41,265✔

2309

        taosRUnLockLatch(&pTaskInfo->lock);

41,265✔

2310

        (*ppRes) = NULL;

41,265✔

2311

        return code;

41,265✔

2314

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

5,477,386✔

2315

      if (!tmp) {

5,489,414✔

2321

      tInfo = *tmp;

5,489,414✔

2322

      taosRUnLockLatch(&pTaskInfo->lock);

5,482,632✔

2324

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

5,493,113✔

2325

      QUERY_CHECK_CODE(code, lino, _end);

5,486,661✔

2326

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

5,486,661✔

2329

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

5,495,586✔

2330

      QUERY_CHECK_CODE(code, lino, _end);

5,488,176✔

2331

      pInfo->scanTimes = 0;

5,488,176✔

2334

    code = groupSeqTableScan(pOperator, ppRes);

662,068,748✔

2335

    QUERY_CHECK_CODE(code, lino, _end);

661,690,270✔

2338

_end:

661,688,095✔

2339

  if (code != TSDB_CODE_SUCCESS) {

661,732,853✔

2340

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

2,800✔

2341

    pTaskInfo->code = code;

2,800✔

2342

    T_LONG_JMP(pTaskInfo->env, code);

2,800✔

2345

  return code;

661,730,053✔

2348

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

1,823,330✔

2349

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

1,823,330✔

2350

  if (!pRecorder) {

1,823,330✔

2353

  STableScanInfo* pTableScanInfo = pOptr->info;

1,823,330✔

2354

  *pRecorder = pTableScanInfo->base.readRecorder;

1,823,330✔

2355

  *pOptrExplain = pRecorder;

1,823,330✔

2356

  *len = sizeof(SFileBlockLoadRecorder);

1,823,330✔

2357

  return 0;

1,823,330✔

2360

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

184,002,781✔

2361

  cleanupQueryTableDataCond(&pBase->cond);

184,002,781✔

2362

  cleanupQueryTableDataCond(&pBase->orgCond);

183,996,904✔

2364

  if (pAPI->tsdReaderClose) {

184,011,274✔

2365

    pAPI->tsdReaderClose(pBase->dataReader);

184,015,995✔

2367

  pBase->dataReader = NULL;

183,948,001✔

2369

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

183,955,045✔

2370

    taosArrayDestroy(pBase->matchInfo.pList);

183,961,803✔

2373

  tableListDestroy(pBase->pTableListInfo);

183,931,781✔

2374

  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);

183,994,117✔

2375

  cleanupExprSupp(&pBase->pseudoSup);

183,967,935✔

2376

184,010,391✔

2378

static void cleanReaderForVTable(STableScanInfo* pInfo){

6,075,671✔

2379

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

6,075,671✔

2380

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

6,075,749✔

2381

    while (pIter != NULL) {

6,099,246✔

2382

      void **reader = pIter;

22,099✔

2383

      if (*reader) {

22,099✔

2384

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

22,099✔

2385

          pInfo->base.dataReader = NULL;

19,847✔

2387

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

22,099✔

2389

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

22,099✔

2392

6,078,053✔

2394

static void destroyTableScanOperatorInfo(void* param) {

159,114,365✔

2395

  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;

159,114,365✔

2396

  blockDataDestroy(pTableScanInfo->pResBlock);

159,114,365✔

2397

  blockDataDestroy(pTableScanInfo->pOrgBlock);

159,096,075✔

2398

  taosHashCleanup(pTableScanInfo->pIgnoreTables);

159,099,148✔

2399

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

159,092,738✔

2400

    cleanReaderForVTable(pTableScanInfo);

6,072,627✔

2401

    taosHashCleanup(pTableScanInfo->readerCache);

6,077,726✔

2403

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

159,096,452✔

2404

  taosArrayDestroy(pTableScanInfo->pBlockColMap);

159,099,202✔

2405

  taosMemoryFreeClear(param);

159,101,675✔

2406

159,068,042✔

2408

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

158,984,575✔

2409

  if (pBlock && dataRequireFlag == FUNC_DATA_REQUIRED_NOT_LOAD) {

158,984,575✔

2410

    int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

7,441,713✔

2411

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

18,888,040✔

2412

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

11,449,782✔

2413

      if (pCol) {

11,453,717✔

2414

        pCol->info.noData = true;

11,452,647✔

2418

158,981,120✔

2420

static int32_t resetTableScanOperatorState(SOperatorInfo* pOper) {

276,515✔

2421

  int32_t         code = TSDB_CODE_SUCCESS;

276,515✔

2422

  STableScanInfo*   pInfo = pOper->info;

276,515✔

2423

  pOper->status = OP_NOT_OPENED;

276,515✔

2425

  pInfo->scanTimes = 0;

276,515✔

2426

  pInfo->currentGroupId = -1;

276,515✔

2427

  pInfo->tableEndIndex = -1;

276,515✔

2428

  pInfo->tableStartIndex = 0;

276,515✔

2429

  pInfo->currentTable = 0;

276,515✔

2430

  pInfo->scanMode = 0;

276,515✔

2431

  pInfo->countState = 0;

276,515✔

2432

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

276,515✔

2434

  if (!pInfo->virtualStableScan) {

276,515✔

2440

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

140,227✔

2441

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

140,227✔

2443

    pInfo->base.dataReader = NULL;

140,227✔

2445

  tableListDestroy(pInfo->base.pTableListInfo);

276,515✔

2446

  pInfo->base.pTableListInfo = tableListCreate();

276,515✔

2447

  if (!pInfo->base.pTableListInfo) {

276,515✔

2451

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

276,515✔

2453

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

276,515✔

2454

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

276,515✔

2455

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

277,552✔

2457

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

277,552✔

2458

    if (code) {

139,008✔

2464

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

276,300✔

2465

  cleanupQueryTableDataCond(&pInfo->base.cond);

276,085✔

2466

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

276,300✔

2467

  if (code) {

276,085✔

2471

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

276,085✔

2472

    cleanupQueryTableDataCond(&pInfo->base.orgCond);

136,288✔

2473

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

136,288✔

2474

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

136,288✔

2477

  pOper->resultInfo.totalRows = 0;

276,300✔

2478

  blockDataEmpty(pInfo->pResBlock);

276,300✔

2479

  blockDataEmpty(pInfo->pOrgBlock);

276,300✔

2480

  taosHashClear(pInfo->pIgnoreTables);

276,300✔

2481

  taosArrayDestroy(pInfo->pBlockColMap);

276,300✔

2482

  pInfo->pBlockColMap = NULL;

276,515✔

2483

  return code;

276,515✔

2486

int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

159,101,473✔

2489

  QRY_PARAM_CHECK(pOptrInfo);

159,101,473✔

2491

  int32_t         code = TSDB_CODE_SUCCESS;

159,113,523✔

2492

  int32_t         lino = 0;

159,113,523✔

2493

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

159,113,523✔

2494

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

159,073,729✔

2495

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

159,064,814✔

2496

    code = terrno;

5✔

2500

  pOperator->pPhyNode = pTableScanNode;

159,066,672✔

2501

  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;

159,068,993✔

2502

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

159,071,791✔

2504

  int32_t numOfCols = 0;

159,093,509✔

2506

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

159,101,252✔

2507

  QUERY_CHECK_CODE(code, lino, _error);

159,039,690✔

2509

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

159,039,690✔

2510

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

159,021,903✔

2511

  QUERY_CHECK_CODE(code, lino, _error);

158,930,193✔

2513

  if (pScanNode->pScanPseudoCols != NULL) {

158,930,193✔

2514

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

47,875,589✔

2515

    pSup->pExprInfo = NULL;

47,916,448✔

2516

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

47,929,939✔

2517

    QUERY_CHECK_CODE(code, lino, _error);

47,876,738✔

2519

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

47,876,738✔

2521

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

47,931,696✔

2524

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

158,944,002✔

2525

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

158,821,325✔

2527

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

158,946,088✔

2528

  pInfo->base.readHandle = *readHandle;

158,789,809✔

2529

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

158,999,974✔

2531

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

158,934,388✔

2532

  pInfo->sample.seed = taosGetTimestampSec();

159,001,662✔

2534

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

158,994,750✔

2535

  initResultSizeInfo(&pOperator->resultInfo, 4096);

158,992,464✔

2536

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

158,794,377✔

2537

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

158,994,060✔

2538

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

158,780,508✔

2540

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

158,757,714✔

2541

  QUERY_CHECK_CODE(code, lino, _error);

158,794,053✔

2543

  pInfo->virtualStableScan = pScanNode->virtualStableScan;

158,794,053✔

2544

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

158,827,837✔

2545

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

6,074,516✔

2546

    pInfo->pResBlock = NULL;

6,066,699✔

2547

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

6,071,632✔

2548

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

6,065,325✔

2551

  if (pInfo->virtualStableScan) {

158,841,655✔

2552

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

6,071,896✔

2553

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

6,078,790✔

2556

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

159,000,133✔

2557

                            pTaskInfo->pStreamRuntimeInfo);

158,986,668✔

2558

  QUERY_CHECK_CODE(code, lino, _error);

158,863,695✔

2560

  pInfo->currentGroupId = -1;

158,841,223✔

2562

  pInfo->tableEndIndex = -1;

158,936,597✔

2563

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

158,934,897✔

2564

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

158,857,113✔

2566

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

158,958,160✔

2568

  pOperator->exprSupp.numOfExprs = numOfCols;

158,828,172✔

2570

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

158,931,488✔

2571

  pInfo->ignoreTag = false;

158,998,648✔

2573

  pInfo->lastBatchIdx = -1;

158,874,037✔

2574

  pInfo->currentBatchIdx = 0;

158,815,402✔

2575

  pInfo->base.pTableListInfo = pTableListInfo;

159,001,791✔

2576

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

158,909,997✔

2586

  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

158,909,997✔

2588

  pOperator->fpSet = createOperatorFpSet(prepareTableScan, doTableScanNext,

158,894,730✔

2594

  setOperatorResetStateFn(pOperator, resetTableScanOperatorState);

158,805,305✔

2596

  pOperator->cost.openCost = 0;

158,798,716✔

2597

  *pOptrInfo = pOperator;

158,815,067✔

2598

  return TSDB_CODE_SUCCESS;

158,752,163✔

2600

_error:

22,472✔

2601

  if (pInfo != NULL) {

22,472✔

2602

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

22,472✔

2603

    destroyTableScanOperatorInfo(pInfo);

22,472✔

2606

  if (pOperator != NULL) {

22,472✔

2607

    pOperator->info = NULL;

22,472✔

2608

    destroyOperator(pOperator);

22,472✔

2610

  pTaskInfo->code = code;

22,472✔

2611

  return code;

22,472✔

2648

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

13,236✔

2649

  int32_t code = TSDB_CODE_SUCCESS;

13,236✔

2650

  int32_t lino = 0;

13,236✔

2651

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

13,236✔

2652

    return code;

1,625✔

2654

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

11,611✔

2655

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

11,611✔

2656

  bool hasUnqualified = false;

11,611✔

2658

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

11,611✔

2659

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

11,611✔

2661

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

11,611✔

2663

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

11,611✔

2666

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

11,611✔

2667

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

11,611✔

2668

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

41,631,619✔

2669

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

41,620,008✔

2670

    void*    data = colDataGetData(pColPk, i);

41,620,008✔

2671

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

41,622,256✔

2672

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

2,248✔

2675

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

2,248✔

2676

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

2,248✔

2677

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

2,248✔

2678

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

2,248✔

2679

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

2,248✔

2680

      taosMemoryFree(tmq);

2,248✔

2682

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

41,617,760✔

2685

    if (!p[i]) {

41,620,008✔

2686

      hasUnqualified = true;

14,803,506✔

2690

  if (hasUnqualified) {

11,611✔

2691

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

3,895✔

2692

    QUERY_CHECK_CODE(code, lino, _end);

3,895✔

2695

_end:

11,611✔

2696

  taosMemoryFree(p);

11,611✔

2697

  if (code != TSDB_CODE_SUCCESS) {

11,611✔

2700

  return code;

11,611✔

2703

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

1,320,817,441✔

2704

  int32_t p1 = *(int32_t*)param1;

1,320,817,441✔

2705

  int32_t p2 = *(int32_t*)param2;

1,320,962,631✔

2707

  if (p1 == p2) {

1,320,964,131✔

2710

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

1,320,964,131✔

2714

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

95,508,178✔

2715

  int32_t         code = TSDB_CODE_SUCCESS;

95,508,178✔

2716

  int32_t         lino = 0;

95,508,178✔

2717

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

95,508,178✔

2718

  SOperatorInfo*  pOperator = pInfo->pStreamScanOp;

95,508,178✔

2719

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

95,508,518✔

2720

  const char*     id = GET_TASKID(pTaskInfo);

95,508,237✔

2721

  SArray* pColList = NULL;

95,510,568✔

2723

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

95,510,568✔

2724

  QUERY_CHECK_CODE(code, lino, _end);

95,508,518✔

2726

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

95,508,518✔

2727

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

95,508,518✔

2728

  pBlockInfo->type = STREAM_NORMAL;

95,508,518✔

2729

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

95,508,858✔

2731

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

95,507,838✔

2732

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

95,508,868✔

2734

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

95,509,888✔

2735

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

95,510,258✔

2738

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

506,895,844✔

2739

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

411,383,425✔

2740

    if (!pColMatchInfo->needOutput) {

411,377,662✔

2741

      continue;

634✔

2744

    bool colExists = false;

411,385,163✔

2745

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

1,281,083,351✔

2746

      SColumnInfoData* pResCol = NULL;

1,280,463,583✔

2747

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

1,280,362,301✔

2748

      QUERY_CHECK_CODE(code, lino, _end);

1,280,448,102✔

2750

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

1,280,448,102✔

2751

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

411,397,314✔

2752

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

411,394,666✔

2753

        QUERY_CHECK_CODE(code, lino, _end);

411,274,805✔

2755

        colExists = true;

411,274,805✔

2756

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

411,274,805✔

2757

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

411,391,803✔

2758

        break;

411,391,803✔

2763

    if (!colExists) {

411,384,952✔

2772

  if (pInfo->numOfPseudoExpr > 0) {

95,506,923✔

2773

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

120,482,649✔

2774

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

60,241,404✔

2776

    if (code) {

60,243,033✔

2781

    terrno = 0;

60,243,033✔

2783

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

244,208,582✔

2784

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

183,957,908✔

2785

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

183,966,569✔

2789

  taosArraySort(pColList, colIdComparFn);

95,498,513✔

2791

  int32_t i = 0, j = 0;

95,507,894✔

2792

  while (i < taosArrayGetSize(pColList)) {

690,757,861✔

2793

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

595,239,613✔

2794

    if (slot1 > j) {

595,249,967✔

2799

      i += 1;

595,249,967✔

2800

      j += 1;

595,249,967✔

2804

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

95,507,239✔

2805

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

634✔

2806

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

634✔

2807

    j += 1;

634✔

2810

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

95,507,211✔

2811

  QUERY_CHECK_CODE(code, lino, _end);

95,506,354✔

2813

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

95,506,354✔

2814

  QUERY_CHECK_CODE(code, lino, _end);

95,507,241✔

2816

_end:

95,507,241✔

2817

  taosArrayDestroy(pColList);

94,927,370✔

2818

  if (code != TSDB_CODE_SUCCESS) {

95,506,278✔

2821

  return code;

95,504,291✔

2824

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

6,557,232✔

2825

  int32_t code = TSDB_CODE_SUCCESS;

6,557,232✔

2826

  int32_t lino = 0;

6,557,232✔

2827

  SValue  val = {0};

6,557,232✔

2828

  if (hasPrimaryKey) {

6,557,232✔

2829

    code = doBlockDataPrimaryKeyFilter(pBlock, offset);

13,236✔

2830

    if (code != TSDB_CODE_SUCCESS) {

13,236✔

2834

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

13,236✔

2836

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

13,236✔

2837

      return code;

3,614✔

2839

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

9,622✔

2840

    val.type = pColPk->info.type;

9,622✔

2841

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

9,622✔

2842

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

617✔

2845

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

617✔

2846

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

617✔

2847

      val.nData = varDataLen(tmp);

617✔

2848

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

617✔

2850

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

9,005✔

2853

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

6,553,618✔

2855

_end:

6,552,736✔

2856

  if (code != TSDB_CODE_SUCCESS) {

6,552,736✔

2859

  return code;

6,553,354✔

2862

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

121,749,609✔

2863

  int32_t        code = TSDB_CODE_SUCCESS;

121,749,609✔

2864

  int32_t        lino = 0;

121,749,609✔

2865

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

121,749,609✔

2866

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

121,755,639✔

2868

  SStreamScanInfo* pInfo = pOperator->info;

121,757,302✔

2869

  const char*      id = GET_TASKID(pTaskInfo);

121,757,002✔

2871

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

121,754,631✔

2873

  if (isTaskKilled(pTaskInfo)) {

121,757,608✔

2878

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

121,757,271✔

2879

    while (1) {

3,614✔

2880

      SSDataBlock* pResult = NULL;

5,875,135✔

2881

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

5,875,135✔

2882

      QUERY_CHECK_CODE(code, lino, _end);

5,874,208✔

2884

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

5,874,208✔

2885

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

5,833,870✔

2886

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

5,833,252✔

2887

        QUERY_CHECK_CODE(code, lino, _end);

5,831,089✔

2888

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

5,831,089✔

2889

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

5,833,252✔

2890

          (*ppRes) = pResult;

5,830,256✔

2891

          return code;

5,830,256✔

2898

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

41,265✔

2899

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

41,265✔

2901

    pTSInfo->base.dataReader = NULL;

41,265✔

2902

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

41,265✔

2903

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

41,265✔

2904

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

41,265✔

2905

      (*ppRes) = NULL;

10,735✔

2906

      return code;

10,735✔

2909

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

30,530✔

2912

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

115,915,944✔

2913

    while (1) {

754,900✔

2915

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

116,671,860✔

2917

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

116,641,290✔

2918

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

116,662,071✔

2921

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

116,665,711✔

2924

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

116,661,653✔

2926

      if (hasResult) {

116,664,147✔

2927

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

95,502,114✔

2929

        blockDataCleanup(pInfo->pRes);

95,512,606✔

2930

        code = setBlockIntoRes(pInfo, pRes);

95,505,458✔

2931

        QUERY_CHECK_CODE(code, lino, _end);

95,504,651✔

2932

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

95,504,651✔

2934

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

95,507,961✔

2935

          (*ppRes) = pInfo->pRes;

94,754,648✔

2936

          return code;

94,752,948✔

2939

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

21,162,033✔

2940

        (*ppRes) = NULL;

21,162,346✔

2941

        return code;

21,162,652✔

2966

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

323,398✔

2967

  int32_t code = TSDB_CODE_SUCCESS;

323,398✔

2968

  int32_t lino = 0;

323,398✔

2969

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

323,398✔

2970

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

324,995✔

2973

  int32_t size = 0;

324,995✔

2974

  code = tableListGetSize(pTableListInfo, &size);

324,995✔

2975

  QUERY_CHECK_CODE(code, lino, _end);

324,723✔

2976

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

9,585,625✔

2977

    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);

9,259,972✔

2978

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

9,236,821✔

2979

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

9,236,821✔

2980

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

9,260,902✔

2983

  (*ppArrayRes) = tableIdList;

325,653✔

2985

_end:

324,696✔

2986

  if (code != TSDB_CODE_SUCCESS) {

324,995✔

2989

  return code;

324,995✔

2992

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

1,497,047✔

2993

  int32_t        code = TSDB_CODE_SUCCESS;

1,497,047✔

2994

  int32_t        lino = 0;

1,497,047✔

2995

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,497,047✔

2996

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,497,047✔

2998

  SStreamRawScanInfo* pInfo = pOperator->info;

1,497,047✔

2999

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

1,497,047✔

3000

  pTaskInfo->streamInfo.btMetaRsp.batchMetaLen = NULL;

1,497,047✔

3002

  qDebug("tmqsnap doRawScan called");

1,497,047✔

3003

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

1,497,047✔

3004

    bool hasNext = false;

1,492,220✔

3005

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

1,492,220✔

3006

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

1,491,373✔

3007

      if (code != TSDB_CODE_SUCCESS) {

1,490,491✔

3013

    if (pInfo->dataReader && hasNext) {

1,491,338✔

3014

      if (isTaskKilled(pTaskInfo)) {

725,231✔

3019

      SSDataBlock* pBlock = NULL;

725,525✔

3020

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

725,525✔

3021

      QUERY_CHECK_CODE(code, lino, _end);

724,643✔

3023

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

724,643✔

3024

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

724,937✔

3025

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

725,525✔

3026

        QUERY_CHECK_CODE(code, lino, _end);

724,643✔

3027

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

724,643✔

3028

        (*ppRes) = pBlock;

725,525✔

3029

        return code;

725,525✔

3033

    SMetaTableInfo mtInfo = {0};

766,401✔

3034

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

766,695✔

3035

    QUERY_CHECK_CODE(code, lino, _end);

766,695✔

3036

    if (code != 0) {

766,695✔

3040

    STqOffsetVal offset = {0};

766,695✔

3041

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

766,107✔

3042

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

10,880✔

3043

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

10,880✔

3045

      SValue val = {0};

755,521✔

3046

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

755,521✔

3047

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

756,109✔

3050

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

766,695✔

3051

    QUERY_CHECK_CODE(code, lino, _end);

766,401✔

3052

    (*ppRes) = NULL;

766,401✔

3053

    return code;

766,695✔

3054

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

4,827✔

3055

    SSnapContext* sContext = pInfo->sContext;

4,827✔

3056

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

40,480✔

3057

      void*   data = NULL;

40,480✔

3058

      int32_t dataLen = 0;

40,480✔

3059

      int16_t type = 0;

40,480✔

3060

      int64_t uid = 0;

40,480✔

3061

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

40,480✔

3067

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

40,480✔

3068

        STqOffsetVal offset = {0};

4,827✔

3069

        SValue       val = {0};

4,827✔

3071

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

4,827✔

3072

        QUERY_CHECK_CODE(code, lino, _end);

4,827✔

3073

        break;

4,827✔

3075

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

35,653✔

3076

        SMqMetaRsp tmpMetaRsp = {0};

35,653✔

3077

        tmpMetaRsp.resMsgType = type;

35,653✔

3078

        tmpMetaRsp.metaRspLen = dataLen;

35,653✔

3079

        tmpMetaRsp.metaRsp = data;

35,653✔

3080

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

35,653✔

3081

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

4,289✔

3082

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

4,289✔

3084

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

4,289✔

3085

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

4,289✔

3087

        int32_t  tempRes = TSDB_CODE_SUCCESS;

35,653✔

3088

        uint32_t len = 0;

35,653✔

3089

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

35,653✔

3090

        if (TSDB_CODE_SUCCESS != tempRes) {

35,653✔

3096

        int32_t tLen = sizeof(SMqRspHead) + len;

35,653✔

3097

        void*   tBuf = taosMemoryCalloc(1, tLen);

35,653✔

3098

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

35,384✔

3100

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

35,384✔

3101

        SEncoder encoder = {0};

35,384✔

3102

        tEncoderInit(&encoder, metaBuff, len);

35,384✔

3103

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

35,384✔

3104

        if (tempLen < 0) {

35,653✔

3111

        taosMemoryFreeClear(data);

35,653✔

3112

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

35,653✔

3113

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

35,384✔

3115

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

35,384✔

3116

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

35,653✔

3120

    (*ppRes) = NULL;

4,827✔

3121

    return code;

4,827✔

3135

static void destroyTmqRawScanOperatorInfo(void* param) {

118,471✔

3136

  SStreamRawScanInfo* pRawScan = (SStreamRawScanInfo*)param;

118,471✔

3137

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

118,471✔

3138

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

118,471✔

3139

  tableListDestroy(pRawScan->pTableListInfo);

118,803✔

3140

  taosMemoryFree(pRawScan);

118,803✔

3141

118,803✔

3146

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

118,803✔

3152

  QRY_PARAM_CHECK(pOptrInfo);

118,803✔

3153

  int32_t code = TSDB_CODE_SUCCESS;

118,803✔

3154

  int32_t lino = 0;

118,803✔

3156

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

118,803✔

3157

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

118,803✔

3158

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

118,803✔

3164

  pInfo->pTableListInfo = tableListCreate();

118,803✔

3165

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

118,803✔

3166

  pInfo->vnode = pHandle->vnode;

118,803✔

3167

  pInfo->pAPI = &pTaskInfo->storageAPI;

118,803✔

3169

  pInfo->sContext = pHandle->sContext;

118,803✔

3170

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

118,803✔

3171

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

118,803✔

3174

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

118,803✔

3176

  *pOptrInfo = pOperator;

118,803✔

3177

  return code;

118,803✔

3189

void destroyTmqScanOperatorInfo(void* param) {

324,995✔

3190

  if (param == NULL) {

324,995✔

3194

  SStreamScanInfo* pStreamScan = (SStreamScanInfo*)param;

324,995✔

3195

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

324,995✔

3196

    destroyOperator(pStreamScan->pTableScanOp);

324,995✔

3199

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

324,995✔

3200

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

324,995✔

3202

  if (pStreamScan->pVtableMergeHandles) {

324,995✔

3206

  if (pStreamScan->pVtableMergeBuf) {

324,995✔

3210

  if (pStreamScan->pVtableReadyHandles) {

324,995✔

3214

  if (pStreamScan->pTableListInfo) {

324,995✔

3218

  if (pStreamScan->matchInfo.pList) {

324,995✔

3219

    taosArrayDestroy(pStreamScan->matchInfo.pList);

324,995✔

3221

  if (pStreamScan->pPseudoExpr) {

324,995✔

3222

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

188,568✔

3223

    taosMemoryFree(pStreamScan->pPseudoExpr);

188,568✔

3226

  cleanupExprSupp(&pStreamScan->tbnameCalSup);

324,995✔

3227

  cleanupExprSupp(&pStreamScan->tagCalSup);

324,995✔

3229

  blockDataDestroy(pStreamScan->pRes);

324,995✔

3230

  blockDataDestroy(pStreamScan->pUpdateRes);

324,995✔

3231

  blockDataDestroy(pStreamScan->pDeleteDataRes);

324,995✔

3232

  blockDataDestroy(pStreamScan->pUpdateDataRes);

324,726✔

3233

  blockDataDestroy(pStreamScan->pCreateTbRes);

324,686✔

3234

  taosArrayDestroy(pStreamScan->pBlockLists);

324,995✔

3235

  blockDataDestroy(pStreamScan->pCheckpointRes);

324,701✔

3237

  taosMemoryFree(pStreamScan);

324,686✔

3257

 int32_t createTmqScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,

324,696✔

3260

  QRY_PARAM_CHECK(pOptrInfo);

324,696✔

3262

  int32_t          code = TSDB_CODE_SUCCESS;

324,995✔

3263

  int32_t          lino = 0;

324,995✔

3264

  SArray*          pColIds = NULL;

324,995✔

3265

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

324,995✔

3266

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

324,995✔

3267

  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;

324,995✔

3268

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

324,995✔

3270

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

324,995✔

3275

  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;

324,995✔

3276

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;

324,995✔

3278

  pInfo->pTagCond = pTagCond;

324,995✔

3279

  pInfo->pGroupTags = pTableScanNode->pGroupTags;

324,995✔

3281

  int32_t numOfCols = 0;

324,995✔

3282

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

324,995✔

3283

  if (code != TSDB_CODE_SUCCESS) {

324,995✔

3287

  SDataType pkType = {0};

324,995✔

3288

  pInfo->primaryKeyIndex = -1;

324,995✔

3289

  pInfo->basic.primaryPkIndex = -1;

324,995✔

3290

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

324,665✔

3291

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

324,995✔

3292

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

324,665✔

3294

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

2,041,288✔

3295

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

1,716,623✔

3296

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

1,716,304✔

3298

    int16_t colId = id->colId;

1,716,304✔

3299

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

1,716,623✔

3300

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

1,716,623✔

3302

    if (id->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

1,716,623✔

3303

      pInfo->primaryTsIndex = id->dstSlotId;

324,665✔

3305

    if (id->isPk) {

1,716,623✔

3306

      pInfo->primaryKeyIndex = id->dstSlotId;

1,454✔

3307

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

1,454✔

3308

      pkType = id->dataType;

1,454✔

3312

  pInfo->pPartTbnameSup = NULL;

324,665✔

3313

  if (pTableScanNode->pSubtable != NULL) {

324,995✔

3329

  if (pTableScanNode->pTags != NULL) {

324,346✔

3341

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

324,995✔

3342

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

324,665✔

3344

  if (pHandle->vnode) {

324,047✔

3345

    SOperatorInfo* pTableScanOp = NULL;

324,377✔

3346

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

324,676✔

3347

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

324,377✔

3352

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

324,384✔

3353

    if (pHandle->version > 0) {

324,387✔

3354

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

324,995✔

3357

    STableKeyInfo* pList = NULL;

324,457✔

3358

    int32_t        num = 0;

324,726✔

3359

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

324,414✔

3360

    QUERY_CHECK_CODE(code, lino, _error);

324,683✔

3362

    if (pHandle->initTableReader) {

324,683✔

3363

      pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;

324,683✔

3364

      pTSInfo->base.dataReader = NULL;

324,384✔

3367

    if (pHandle->initTqReader) {

324,075✔

3368

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

324,384✔

3369

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

324,995✔

3375

    pInfo->pUpdateInfo = NULL;

324,995✔

3376

    pInfo->pTableScanOp = pTableScanOp;

324,995✔

3377

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

324,995✔

3382

    pInfo->readHandle = *pHandle;

324,995✔

3383

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

324,995✔

3384

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

324,995✔

3385

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

324,995✔

3386

    pInfo->hasPart = false;

324,995✔

3388

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

324,995✔

3389

    QUERY_CHECK_CODE(code, lino, _error);

324,384✔

3392

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

324,384✔

3393

    QUERY_CHECK_CODE(code, lino, _error);

324,995✔

3395

    SArray* tableIdList = NULL;

324,995✔

3396

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

324,995✔

3397

    QUERY_CHECK_CODE(code, lino, _error);

324,696✔

3398

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

324,696✔

3399

    QUERY_CHECK_CODE(code, lino, _error);

324,995✔

3400

    taosArrayDestroy(tableIdList);

324,995✔

3401

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

324,995✔

3408

  pColIds = NULL;

324,995✔

3411

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

324,995✔

3412

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

188,568✔

3413

    QUERY_CHECK_CODE(code, lino, _error);

188,568✔

3416

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

324,995✔

3417

  QUERY_CHECK_CODE(code, lino, _error);

324,995✔

3419

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

324,995✔

3420

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

324,995✔

3421

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

324,995✔

3422

  QUERY_CHECK_CODE(code, lino, _error);

324,995✔

3424

  pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;

324,995✔

3425

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

324,995✔

3426

  pInfo->groupId = 0;

324,995✔

3427

  pInfo->igCheckGroupId = false;

324,686✔

3428

  pInfo->pStreamScanOp = pOperator;

324,995✔

3429

  pInfo->deleteDataIndex = 0;

324,686✔

3430

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

324,686✔

3431

  QUERY_CHECK_CODE(code, lino, _error);

324,995✔

3433

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

324,995✔

3434

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

324,995✔

3435

  QUERY_CHECK_CODE(code, lino, _error);

324,995✔

3437

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

324,995✔

3438

  pInfo->partitionSup.needCalc = false;

324,995✔

3439

  pInfo->igCheckUpdate = pTableScanNode->igCheckUpdate;

324,995✔

3440

  pInfo->igExpired = pTableScanNode->igExpired;

324,995✔

3441

  pInfo->twAggSup.maxTs = INT64_MIN;

324,995✔

3442

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

324,995✔

3443

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

324,995✔

3444

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

324,686✔

3445

  pInfo->pFillSup = NULL;

324,686✔

3446

  pInfo->useGetResultRange = false;

324,686✔

3447

  pInfo->pRangeScanRes = NULL;

324,995✔

3449

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

324,995✔

3450

  QUERY_CHECK_CODE(code, lino, _error);

324,995✔

3452

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

324,995✔

3454

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

324,995✔

3456

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

324,686✔

3459

  *pOptrInfo = pOperator;

324,995✔

3460

  return code;

324,995✔

3483

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

10,463,462✔

3484

  int32_t        code = TSDB_CODE_SUCCESS;

10,463,462✔

3485

  int32_t        lino = 0;

10,463,462✔

3486

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

10,463,462✔

3487

  STagScanInfo*  pInfo = pOperator->info;

10,464,691✔

3488

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

10,467,703✔

3489

  int32_t        count = pRes->info.rows;

10,468,207✔

3491

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

10,465,723✔

3492

  if (!item) {

10,460,719✔

3499

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

10,460,719✔

3500

  tDecoderClear(&(*mr).coder);

10,466,701✔

3501

  if (code != TSDB_CODE_SUCCESS) {

10,467,715✔

3507

  char str[512];

10,467,611✔

3508

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

47,915,794✔

3509

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

37,452,555✔

3512

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

37,456,550✔

3513

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

4,434,693✔

3514

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

4,433,766✔

3515

      QUERY_CHECK_CODE(code, lino, _end);

4,434,755✔

3516

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

33,024,285✔

3517

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

6,342,075✔

3518

      QUERY_CHECK_CODE(code, lino, _end);

6,343,071✔

3519

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

26,681,214✔

3520

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

15,019✔

3521

      QUERY_CHECK_CODE(code, lino, _end);

15,019✔

3523

      STagVal val = {0};

26,668,187✔

3524

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

26,664,695✔

3525

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

26,667,215✔

3527

      char* data = NULL;

26,665,667✔

3528

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

26,665,667✔

3529

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

26,665,298✔

3531

        data = (char*)p;

1,377✔

3534

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

26,661,690✔

3535

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

26,661,186✔

3536

      QUERY_CHECK_CODE(code, lino, _end);

26,664,678✔

3538

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

26,664,678✔

3540

        taosMemoryFree(data);

9,591,505✔

3545

_end:

10,474,669✔

3546

  if (code != TSDB_CODE_SUCCESS) {

10,468,231✔

3550

    pRes->info.rows++;

10,468,231✔

3553

  return code;

10,467,138✔

3556

static void tagScanFreeUidTag(void* p) {

7,024,921✔

3557

  STUidTagInfo* pInfo = p;

7,024,921✔

3558

  if (pInfo->pTagVal != NULL) {

7,024,921✔

3559

    taosMemoryFree(pInfo->pTagVal);

7,025,444✔

3561

7,021,597✔

3563

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

1,672,843✔

3564

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

1,672,843✔

3565

  if (pColumnData == NULL) {

1,671,869✔

3569

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

1,671,869✔

3570

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

1,672,356✔

3571

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

1,672,014✔

3572

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

1,670,379✔

3574

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

1,671,674✔

3575

  if (code != TSDB_CODE_SUCCESS) {

1,669,395✔

3581

  pParam->columnData = pColumnData;

1,669,395✔

3582

  pParam->colAlloced = true;

1,672,014✔

3583

  return TSDB_CODE_SUCCESS;

1,672,014✔

3586

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

10,305,634✔

3587

  int32_t                code = TSDB_CODE_SUCCESS;

10,305,634✔

3588

  int32_t                lino = 0;

10,305,634✔

3589

  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;

10,305,634✔

3590

  SColumnNode*           pSColumnNode = NULL;

10,305,634✔

3591

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

10,304,660✔

3592

    pSColumnNode = *(SColumnNode**)pNode;

2,887,418✔

3593

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

7,429,050✔

3594

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

502✔

3595

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

502✔

3608

      return DEAL_RES_CONTINUE;

502✔

3611

    return DEAL_RES_CONTINUE;

7,423,479✔

3614

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

2,886,929✔

3615

  if (!data) {

2,885,271✔

3616

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

1,673,980✔

3617

    if (code == TSDB_CODE_DUP_KEY) {

1,675,288✔

3620

    QUERY_CHECK_CODE(code, lino, _end);

1,675,288✔

3621

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

1,675,288✔

3622

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

1,674,001✔

3623

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

1,675,297✔

3624

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

1,673,018✔

3625

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

1,673,494✔

3626

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

1,676,272✔

3628

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

1,211,291✔

3629

    pSColumnNode->slotId = col->slotId;

1,211,291✔

3632

_end:

2,885,604✔

3633

  if (code != TSDB_CODE_SUCCESS) {

2,884,796✔

3634

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

1,461✔

3637

  return DEAL_RES_CONTINUE;

2,883,335✔

3640

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

1,673,185✔

3642

  int32_t code = TSDB_CODE_SUCCESS;

1,673,185✔

3643

  int32_t lino = 0;

1,673,185✔

3644

  int32_t numOfTables = taosArrayGetSize(aUidTags);

1,673,185✔

3645

  SArray* pBlockList = NULL;

1,672,211✔

3647

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

1,672,211✔

3648

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

1,671,040✔

3650

  pBlockList = taosArrayInit(1, POINTER_BYTES);

1,671,040✔

3651

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

1,671,693✔

3653

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

1,672,501✔

3654

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

1,672,501✔

3656

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

1,672,501✔

3658

  SScalarParam output = {0};

1,672,501✔

3659

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

1,672,843✔

3660

  QUERY_CHECK_CODE(code, lino, _end);

1,671,527✔

3662

  gTaskScalarExtra.pStreamInfo = (void*)pStreamRuntimeInfo;

1,671,527✔

3663

  gTaskScalarExtra.pStreamRange = NULL;

1,671,527✔

3664

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

1,669,372✔

3665

  QUERY_CHECK_CODE(code, lino, _end);

1,669,279✔

3667

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

1,669,279✔

3668

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

4,173,787✔

3669

    if (result[i]) {

2,505,645✔

3670

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

2,359,549✔

3671

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

2,359,549✔

3675

  colDataDestroy(output.columnData);

1,671,175✔

3676

  taosMemoryFreeClear(output.columnData);

1,666,183✔

3678

_end:

1,664,745✔

3679

  blockDataDestroy(pResBlock);

1,665,097✔

3680

  taosArrayDestroy(pBlockList);

1,664,330✔

3681

  if (code != TSDB_CODE_SUCCESS) {

1,663,687✔

3684

  return code;

1,663,687✔

3687

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

28,625,040✔

3690

  int32_t code = TSDB_CODE_SUCCESS;

28,625,040✔

3691

  int32_t lino = 0;

28,625,040✔

3692

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

28,625,040✔

3693

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

13,672,494✔

3699

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

13,663,481✔

3700

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

6,833,070✔

3701

      QUERY_CHECK_CODE(code, lino, _end);

6,836,016✔

3702

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

6,836,522✔

3703

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

6,833,341✔

3704

      QUERY_CHECK_CODE(code, lino, _end);

6,836,967✔

3707

    STagVal tagVal = {0};

14,949,898✔

3708

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

14,950,580✔

3709

    if (pUidTagInfo->pTagVal == NULL) {

14,953,262✔

3712

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

14,947,770✔

3714

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

14,954,549✔

3715

        colDataSetNULL(pColInfo, rowIndex);

6,696✔

3716

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

14,946,921✔

3719

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

18,659,368✔

3720

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

3,723,639✔

3723

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

3,721,261✔

3724

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

3,722,120✔

3726

        varDataSetLen(tmp, tagVal.nData);

3,722,120✔

3727

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

3,719,187✔

3728

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

3,724,877✔

3729

        taosMemoryFree(tmp);

3,724,741✔

3730

        QUERY_CHECK_CODE(code, lino, _end);

3,719,729✔

3732

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

11,214,789✔

3733

        QUERY_CHECK_CODE(code, lino, _end);

11,219,364✔

3738

_end:

28,623,396✔

3739

  if (code != TSDB_CODE_SUCCESS) {

28,623,396✔

3742

  return code;

28,612,235✔

3745

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

4,495,083✔

3747

  int32_t       code = TSDB_CODE_SUCCESS;

4,495,083✔

3748

  int32_t       lino = 0;

4,495,083✔

3749

  STagScanInfo* pInfo = pOperator->info;

4,495,083✔

3750

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

4,496,772✔

3751

  if (!ignoreFilterIdx) {

4,495,085✔

3752

    size_t szTables = taosArrayGetSize(aFilterIdxs);

1,669,086✔

3753

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

4,032,104✔

3754

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

2,358,069✔

3755

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);

2,358,420✔

3756

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

2,359,651✔

3757

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

13,468,622✔

3758

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

11,101,340✔

3759

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

11,112,317✔

3760

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

11,112,317✔

3761

        QUERY_CHECK_CODE(code, lino, _end);

11,108,971✔

3765

    size_t szTables = taosArrayGetSize(aUidTags);

2,825,999✔

3766

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

7,354,153✔

3767

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);

4,528,276✔

3768

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

4,527,947✔

3769

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

22,026,218✔

3770

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

17,500,707✔

3771

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

17,507,139✔

3772

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

17,507,139✔

3773

        QUERY_CHECK_CODE(code, lino, _end);

17,498,271✔

3778

_end:

2,825,877✔

3779

  if (code != TSDB_CODE_SUCCESS) {

4,499,912✔

3782

  return code;

4,499,891✔

3785

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

8,954,514✔

3786

  int32_t        code = TSDB_CODE_SUCCESS;

8,954,514✔

3787

  int32_t        lino = 0;

8,954,514✔

3788

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

8,954,514✔

3789

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

8,954,458✔

3791

  STagScanInfo* pInfo = pOperator->info;

8,952,433✔

3792

  SSDataBlock*  pRes = pInfo->pRes;

8,957,789✔

3794

  QRY_PARAM_CHECK(ppRes);

8,956,173✔

3796

  if (pOperator->status == OP_EXEC_DONE) {

8,957,584✔

3797

    return TSDB_CODE_SUCCESS;

4,392,975✔

3799

  blockDataCleanup(pRes);

4,566,056✔

3801

  if (pInfo->pCtbCursor == NULL) {

4,569,422✔

3802

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

4,567,794✔

3803

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

4,571,567✔

3809

  SArray* aUidTags = pInfo->aUidTags;

4,571,202✔

3810

  SArray* aFilterIdxs = pInfo->aFilterIdxs;

4,571,202✔

3811

  int32_t count = 0;

4,571,080✔

3812

  bool    ctbCursorFinished = false;

4,571,080✔

3813

  while (1) {

93,717✔

3814

    taosArrayClearEx(aUidTags, tagScanFreeUidTag);

4,664,797✔

3815

    taosArrayClear(aFilterIdxs);

4,660,580✔

3817

    int32_t numTables = 0;

4,661,101✔

3818

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

11,688,406✔

3819

      SMCtbCursor* pCur = pInfo->pCtbCursor;

11,695,414✔

3820

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

11,695,768✔

3821

      if (uid == 0) {

11,699,683✔

3822

        ctbCursorFinished = true;

4,666,001✔

3823

        break;

4,666,001✔

3825

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

7,033,682✔

3826

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

7,029,167✔

3827

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

7,025,626✔

3829

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

7,025,626✔

3830

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

7,033,538✔

3831

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

7,033,538✔

3832

      ++numTables;

7,033,538✔

3835

    if (numTables == 0) {

4,666,001✔

3836

      break;

166,110✔

3838

    bool ignoreFilterIdx = true;

4,499,891✔

3839

    if (pInfo->pTagCond != NULL) {

4,499,891✔

3840

      ignoreFilterIdx = false;

1,673,185✔

3841

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

1,673,185✔

3842

      QUERY_CHECK_CODE(code, lino, _end);

1,666,165✔

3844

      ignoreFilterIdx = true;

2,826,341✔

3847

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

4,492,506✔

3848

    QUERY_CHECK_CODE(code, lino, _end);

4,499,526✔

3850

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

4,499,526✔

3852

    if (count != 0) {

4,498,686✔

3853

      break;

4,404,969✔

3857

  if (count > 0) {

4,571,079✔

3858

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

4,404,969✔

3860

  if (count == 0 || ctbCursorFinished) {

4,569,288✔

3861

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

4,569,754✔

3862

    pInfo->pCtbCursor = NULL;

4,563,938✔

3863

    setOperatorCompleted(pOperator);

4,565,477✔

3866

  pRes->info.rows = count;

4,565,930✔

3867

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

4,567,226✔

3868

  if (bLimitReached) {

4,566,298✔

3872

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

4,566,298✔

3873

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

4,566,783✔

3875

_end:

4,567,270✔

3876

  if (code != TSDB_CODE_SUCCESS) {

4,567,270✔

3882

  return code;

4,567,270✔

3885

static int32_t createTagScanTableListInfoFromParam(SOperatorInfo* pOperator) {

3,609,382✔

3886

  STagScanInfo*          pInfo = pOperator->info;

3,609,382✔

3887

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

3,609,382✔

3888

  int32_t                code = 0;

3,609,382✔

3889

  STableListInfo*        pListInfo = pInfo->pTableListInfo;

3,609,382✔

3890

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

3,609,382✔

3891

  tb_uid_t               pUid = pParam->vcUid;

3,609,382✔

3897

  STableKeyInfo info = {.groupId = 0};

3,609,382✔

3899

  int32_t tableIdx = 0;

3,609,382✔

3900

  taosHashClear(pListInfo->map);

3,609,382✔

3901

  taosArrayClear(pListInfo->pTableList);

3,609,382✔

3903

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

3,609,382✔

3910

  info.uid = pUid;

3,609,382✔

3911

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

3,609,382✔

3912

  if (p == NULL) {

3,609,382✔

3916

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

3,609,382✔

3918

  return code;

3,609,382✔

3921

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

7,299,159✔

3922

  int32_t       code = TSDB_CODE_SUCCESS;

7,299,159✔

3923

  int32_t       lino = 0;

7,299,159✔

3924

  STagScanInfo* pInfo = pOperator->info;

7,299,159✔

3925

  if (pOperator->pOperatorGetParam) {

7,300,155✔

3926

    pOperator->resultInfo.totalRows = 0;

3,609,382✔

3927

    pOperator->dynamicTask = true;

3,609,382✔

3928

    pInfo->curPos = 0;

3,609,382✔

3929

    code = createTagScanTableListInfoFromParam(pOperator);

3,609,382✔

3930

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

3,609,382✔

3931

    pOperator->pOperatorGetParam = NULL;

3,609,382✔

3932

    QUERY_CHECK_CODE(code, lino, _end);

3,609,382✔

3934

    if (pOperator->status == OP_EXEC_DONE) {

3,609,382✔

3935

      pOperator->status = OP_OPENED;

1,989,756✔

3939

  if (pOperator->status == OP_EXEC_DONE) {

7,298,643✔

3940

    (*ppRes) = NULL;

1,829,340✔

3941

    return code;

1,828,836✔

3944

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

5,466,795✔

3945

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

5,468,823✔

3947

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

5,467,827✔

3948

  SSDataBlock* pRes = pInfo->pRes;

5,467,323✔

3949

  blockDataCleanup(pRes);

5,462,811✔

3951

  int32_t size = 0;

5,468,811✔

3952

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

5,469,819✔

3953

  QUERY_CHECK_CODE(code, lino, _end);

5,466,807✔

3955

  if (size == 0) {

5,466,807✔

3956

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

18,690✔

3957

    (*ppRes) = NULL;

18,690✔

3958

    return code;

18,690✔

3961

  SMetaReader mr = {0};

5,448,117✔

3962

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

5,450,121✔

3963

  pRes->info.rows = 0;

5,449,617✔

3965

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

15,916,336✔

3966

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

10,467,199✔

3967

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

10,466,652✔

3970

      code = TSDB_CODE_SUCCESS;

10,468,157✔

3972

    QUERY_CHECK_CODE(code, lino, _end);

10,466,652✔

3974

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

10,466,652✔

3975

      setOperatorCompleted(pOperator);

5,451,129✔

3979

  pAPI->metaReaderFn.clearReader(&mr);

5,452,646✔

3980

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

5,450,637✔

3981

  if (bLimitReached) {

5,450,071✔

3986

  if (pOperator->status == OP_EXEC_DONE) {

5,450,071✔

3987

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

5,451,129✔

3990

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

5,449,505✔

3992

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

5,450,637✔

3993

_end:

5,450,059✔

3994

  if (code != TSDB_CODE_SUCCESS) {

5,450,625✔

3999

  return code;

5,450,625✔

4002

static void destroyTagScanOperatorInfo(void* param) {

8,370,718✔

4003

  STagScanInfo* pInfo = (STagScanInfo*)param;

8,370,718✔

4004

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

8,370,718✔

4007

  taosHashCleanup(pInfo->filterCtx.colHash);

8,377,277✔

4008

  taosArrayDestroy(pInfo->filterCtx.cInfoList);

8,363,259✔

4009

  taosArrayDestroy(pInfo->aFilterIdxs);

8,363,639✔

4010

  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);

8,353,646✔

4012

  blockDataDestroy(pInfo->pRes);

8,362,269✔

4013

  taosArrayDestroy(pInfo->matchInfo.pList);

8,369,925✔

4014

  tableListDestroy(pInfo->pTableListInfo);

8,365,446✔

4016

  pInfo->pRes = NULL;

8,373,253✔

4017

  pInfo->pTableListInfo = NULL;

8,372,901✔

4018

  taosMemoryFreeClear(param);

8,372,966✔

4019

8,372,329✔

4021

static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {

4,588✔

4022

  int32_t         code = TSDB_CODE_SUCCESS;

4,588✔

4023

  STagScanInfo*   pInfo = pOper->info;

4,588✔

4025

  pOper->status = OP_NOT_OPENED;

4,588✔

4026

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

4,588✔

4027

  if (pTagScanNode->onlyMetaCtbIdx) {

4,588✔

4028

      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

2,008✔

4029

      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,008✔

4030

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

2,008✔

4031

      pInfo->pCtbCursor = NULL;

2,008✔

4033

      pInfo->curPos = 0;

2,580✔

4036

  return code;

4,588✔

4039

int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,

8,374,360✔

4042

  QRY_PARAM_CHECK(pOptrInfo);

8,374,360✔

4044

  int32_t         code = TSDB_CODE_SUCCESS;

8,378,752✔

4045

  int32_t         lino = 0;

8,378,752✔

4046

  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;

8,378,752✔

4047

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

8,378,752✔

4048

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

8,370,205✔

4049

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

8,364,578✔

4053

  pOperator->pPhyNode = pTagScanNode;

8,365,417✔

4054

  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

8,365,893✔

4056

  int32_t    numOfExprs = 0;

8,369,802✔

4057

  SExprInfo* pExprInfo = NULL;

8,373,269✔

4059

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

8,370,616✔

4060

  QUERY_CHECK_CODE(code, lino, _error);

8,372,615✔

4062

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

8,372,615✔

4063

  QUERY_CHECK_CODE(code, lino, _error);

8,365,971✔

4065

  int32_t num = 0;

8,365,971✔

4066

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

8,367,632✔

4067

  QUERY_CHECK_CODE(code, lino, _error);

8,367,316✔

4069

  pInfo->pTagCond = pTagCond;

8,367,316✔

4070

  pInfo->pTagIndexCond = pTagIndexCond;

8,369,265✔

4071

  pInfo->suid = pPhyNode->suid;

8,365,733✔

4072

  pInfo->pStorageAPI = &pTaskInfo->storageAPI;

8,366,526✔

4074

  pInfo->pTableListInfo = pTableListInfo;

8,355,753✔

4075

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

8,372,079✔

4076

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

8,371,444✔

4078

  pInfo->readHandle = *pReadHandle;

8,346,143✔

4079

  pInfo->curPos = 0;

8,373,384✔

4081

  initLimitInfo(pPhyNode->node.pLimit, pPhyNode->node.pSlimit, &pInfo->limitInfo);

8,352,184✔

4082

  setOperatorInfo(pOperator, "TagScanOperator", QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN, false, OP_NOT_OPENED, pInfo,

8,358,697✔

4084

  initResultSizeInfo(&pOperator->resultInfo, 4096);

8,354,724✔

4085

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

8,344,362✔

4086

  QUERY_CHECK_CODE(code, lino, _error);

8,378,617✔

4088

  if (pTagScanNode->onlyMetaCtbIdx) {

8,378,617✔

4089

    pInfo->aUidTags = taosArrayInit(pOperator->resultInfo.capacity, sizeof(STUidTagInfo));

4,576,545✔

4090

    QUERY_CHECK_NULL(pInfo->aUidTags, code, lino, _error, terrno);

4,577,077✔

4092

    pInfo->aFilterIdxs = taosArrayInit(pOperator->resultInfo.capacity, sizeof(int32_t));

4,576,242✔

4093

    QUERY_CHECK_NULL(pInfo->aFilterIdxs, code, lino, _error, terrno);

4,577,410✔

4095

    pInfo->filterCtx.colHash =

4,575,258✔

4096

        taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);

4,577,895✔

4097

    QUERY_CHECK_NULL(pInfo->filterCtx.colHash, code, lino, _error, terrno);

4,577,429✔

4099

    pInfo->filterCtx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));

4,574,428✔

4100

    QUERY_CHECK_NULL(pInfo->filterCtx.cInfoList, code, lino, _error, terrno);

4,573,448✔

4102

    if (pInfo->pTagCond != NULL) {

4,573,071✔

4103

      nodesRewriteExprPostOrder(&pTagCond, tagScanRewriteTagColumn, (void*)&pInfo->filterCtx);

1,675,796✔

4106

  __optr_fn_t tagScanNextFn = (pTagScanNode->onlyMetaCtbIdx) ? doTagScanFromCtbIdxNext : doTagScanFromMetaEntryNext;

8,377,330✔

4107

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, tagScanNextFn, NULL, destroyTagScanOperatorInfo,

8,378,854✔

4109

  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);

8,372,224✔

4111

  *pOptrInfo = pOperator;

8,371,702✔

4112

  return code;

8,374,295✔

4129

static int32_t subTblRowCompareTsFn(const void* pLeft, const void* pRight, void* param) {

231,918,752✔

4130

  int32_t                 left = *(int32_t*)pLeft;

231,918,752✔

4131

  int32_t                 right = *(int32_t*)pRight;

231,919,104✔

4132

  STmsSubTablesMergeInfo* pInfo = (STmsSubTablesMergeInfo*)param;

231,919,456✔

4134

  int32_t leftIdx = pInfo->aInputs[left].rowIdx;

231,919,456✔

4135

  int32_t rightIdx = pInfo->aInputs[right].rowIdx;

231,918,752✔

4137

  if (leftIdx == -1) {

231,884,256✔

4138

    return 1;

198,723,936✔

4139

  } else if (rightIdx == -1) {

33,160,320✔

4140

    return -1;

328,320✔

4143

  int64_t leftTs = pInfo->aInputs[left].aTs[leftIdx];

32,832,000✔

4144

  int64_t rightTs = pInfo->aInputs[right].aTs[rightIdx];

32,832,000✔

4145

  int32_t ret = leftTs > rightTs ? 1 : ((leftTs < rightTs) ? -1 : 0);

32,832,000✔

4146

  if (pInfo->pTsOrderInfo->order == TSDB_ORDER_DESC) {

32,832,000✔

4149

  return ret;

32,832,000✔

4179

int32_t dumpQueryTableCond(const SQueryTableDataCond* src, SQueryTableDataCond* dst) {

2,937,338✔

4180

  int32_t code = TSDB_CODE_SUCCESS;

2,937,338✔

4181

  int32_t lino = 0;

2,937,338✔

4182

  memcpy((void*)dst, (void*)src, sizeof(SQueryTableDataCond));

2,937,338✔

4183

  dst->colList = taosMemoryCalloc(src->numOfCols, sizeof(SColumnInfo));

2,937,338✔

4184

  QUERY_CHECK_NULL(dst->colList, code, lino, _end, terrno);

2,938,394✔

4185

  for (int i = 0; i < src->numOfCols; i++) {

9,858,034✔

4186

    dst->colList[i] = src->colList[i];

6,933,058✔

4189

_end:

2,934,622✔

4190

  if (code != TSDB_CODE_SUCCESS) {

2,934,622✔

4193

  return code;

2,932,158✔

4196

static int32_t fetchNextSubTableBlockFromReader(SOperatorInfo* pOperator, STmsSubTableInput* pInput,

4,414,844✔

4198

  int32_t code = 0;

4,414,844✔

4200

  STableMergeScanInfo*    pInfo = pOperator->info;

4,414,844✔

4201

  SReadHandle*            pHandle = &pInfo->base.readHandle;

4,415,538✔

4202

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

4,415,578✔

4203

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

4,415,216✔

4204

  const SStorageAPI*      pAPI = &pTaskInfo->storageAPI;

4,413,164✔

4206

  blockDataCleanup(pInput->pReaderBlock);

4,418,344✔

4207

  if (!pInput->bInMemReader) {

4,416,272✔

4208

    code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, pInput->pKeyInfo, 1, pInput->pReaderBlock,

4,421,404✔

4209

                                         (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

2,213,106✔

4210

    if (code != 0) {

2,210,008✔

4215

  pInfo->base.dataReader = pInput->pReader;

4,414,904✔

4217

  while (true) {

5,984✔

4218

    bool hasNext = false;

4,419,822✔

4219

    code = pAPI->tsdReader.tsdNextDataBlock(pInfo->base.dataReader, &hasNext);

4,419,128✔

4220

    if (code != 0) {

4,425,706✔

4226

    if (!hasNext || isTaskKilled(pTaskInfo)) {

4,425,706✔

4227

      if (isTaskKilled(pTaskInfo)) {

2,950,122✔

4233

      *pSubTableHasBlock = false;

2,950,122✔

4234

      break;

2,949,780✔

4237

    if (pInput->tblCond.order == TSDB_ORDER_ASC) {

1,475,584✔

4238

      pInput->tblCond.twindows.skey = pInput->pReaderBlock->info.window.ekey + 1;

1,450,240✔

4240

      pInput->tblCond.twindows.ekey = pInput->pReaderBlock->info.window.skey - 1;

24,992✔

4243

    uint32_t status = 0;

1,475,584✔

4244

    code = loadDataBlock(pOperator, &pInfo->base, pInput->pReaderBlock, &status);

1,475,584✔

4245

    if (code != 0) {

1,475,242✔

4250

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

1,475,242✔

4254

    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pInput->pReaderBlock->info.rows == 0) {

1,475,242✔

4255

      continue;

5,984✔

4258

    *pSubTableHasBlock = true;

1,469,248✔

4259

    break;

1,469,600✔

4262

  if (*pSubTableHasBlock) {

4,419,390✔

4263

    pInput->pReaderBlock->info.id.groupId =

2,938,506✔

4264

        tableListGetTableGroupId(pInfo->base.pTableListInfo, pInput->pReaderBlock->info.id.uid);

2,938,154✔

4265

    pOperator->resultInfo.totalRows += pInput->pReaderBlock->info.rows;

1,469,600✔

4267

  if (!pInput->bInMemReader || !*pSubTableHasBlock) {

4,419,400✔

4268

    pAPI->tsdReader.tsdReaderClose(pInput->pReader);

3,319,220✔

4269

    pInput->pReader = NULL;

3,318,184✔

4272

  pInfo->base.dataReader = NULL;

4,418,354✔

4273

  return TSDB_CODE_SUCCESS;

4,418,706✔

4276

static int32_t setGroupStartEndIndex(STableMergeScanInfo* pInfo) {

1,831,938✔

4277

  pInfo->bGroupProcessed = false;

1,831,938✔

4279

  int32_t numOfTables = 0;

1,832,984✔

4280

  int32_t code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

1,820,492✔

4281

  if (code != TSDB_CODE_SUCCESS) {

1,808,170✔

4286

  int32_t i = pInfo->tableStartIndex + 1;

1,808,170✔

4287

  for (; i < numOfTables; ++i) {

2,913,228✔

4288

    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

1,156,336✔

4289

    if (!tableKeyInfo) {

1,154,928✔

4293

    if (tableKeyInfo->groupId != pInfo->groupId) {

1,154,928✔

4294

      break;

55,120✔

4297

  pInfo->tableEndIndex = i - 1;

1,812,012✔

4298

  return TSDB_CODE_SUCCESS;

1,838,446✔

4301

static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {

1,844,018✔

4302

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,792,048✔

4303

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

2,949,770✔

4304

    if (pInput->rowIdx == -1) {

2,946,692✔

4305

      continue;

1,479,486✔

4308

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,469,248✔

4309

      pInput->rowIdx = 0;

1,468,896✔

4310

      pInput->pageIdx = -1;

1,468,192✔

4313

    pInput->pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,469,600✔

4314

    SColumnInfoData* col = taosArrayGet(pInput->pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

1,468,896✔

4315

    if (!col) {

1,468,896✔

4319

    pInput->aTs = (int64_t*)col->pData;

1,468,896✔

4322

  __merge_compare_fn_t mergeCompareFn = (!pSubTblsInfo->pPkOrderInfo) ? subTblRowCompareTsFn : subTblRowCompareTsPkFn;

1,844,682✔

4323

  return tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, mergeCompareFn);

1,844,682✔

4326

static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {

1,832,260✔

4327

  int32_t code = setGroupStartEndIndex(pInfo);

1,832,260✔

4328

  if (code != TSDB_CODE_SUCCESS) {

1,833,880✔

4332

  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));

1,833,880✔

4333

  if (pSubTblsInfo == NULL) {

1,826,004✔

4336

  pSubTblsInfo->pTsOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);

1,826,004✔

4337

  if (!pSubTblsInfo->pTsOrderInfo) {

1,834,664✔

4341

  if (taosArrayGetSize(pInfo->pSortInfo) == 2) {

1,834,664✔

4348

    pSubTblsInfo->pPkOrderInfo = NULL;

1,823,278✔

4350

  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

1,837,470✔

4351

  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));

1,819,476✔

4352

  if (pSubTblsInfo->aInputs == NULL) {

1,811,630✔

4356

  int32_t bufPageSize = pInfo->bufPageSize;

1,811,952✔

4357

  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;

1,832,974✔

4358

  code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);

1,833,326✔

4359

  if (code != TSDB_CODE_SUCCESS) {

1,819,848✔

4364

  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;

1,819,848✔

4365

  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;

1,835,720✔

4367

  pInfo->pSubTablesMergeInfo = pSubTblsInfo;

1,826,758✔

4368

  return TSDB_CODE_SUCCESS;

1,823,962✔

4371

static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {

1,833,638✔

4372

  int32_t        code = TSDB_CODE_SUCCESS;

1,833,638✔

4373

  int32_t        lino = 0;

1,833,638✔

4374

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,833,638✔

4375

  SReadHandle*   pHandle = &pInfo->base.readHandle;

1,832,964✔

4376

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,832,260✔

4378

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

1,837,128✔

4380

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,762,406✔

4381

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

2,942,196✔

4382

    pInput->type = SUB_TABLE_MEM_BLOCK;

2,926,334✔

4384

    code = dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);

2,937,036✔

4385

    QUERY_CHECK_CODE(code, lino, _end);

2,929,784✔

4387

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pReaderBlock);

2,929,784✔

4388

    QUERY_CHECK_CODE(code, lino, _end);

2,949,096✔

4390

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pPageBlock);

2,949,096✔

4391

    QUERY_CHECK_CODE(code, lino, _end);

2,950,474✔

4393

    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);

2,950,474✔

4394

    pInput->pKeyInfo = keyInfo;

2,948,070✔

4396

    if (isTaskKilled(pTaskInfo)) {

2,944,962✔

4400

    if (i + 1 < pSubTblsInfo->numInMemReaders) {

2,938,746✔

4401

      code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, keyInfo, 1, pInput->pReaderBlock,

2,204,896✔

4402

                                           (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

1,104,032✔

4403

      QUERY_CHECK_CODE(code, lino, _end);

1,101,920✔

4404

      pInput->bInMemReader = true;

1,101,920✔

4406

      pInput->pReader = NULL;

1,842,630✔

4407

      pInput->bInMemReader = false;

1,841,604✔

4409

    bool hasNext = true;

2,943,202✔

4410

    code = fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);

2,943,896✔

4411

    QUERY_CHECK_CODE(code, lino, _end);

2,949,770✔

4412

    if (!hasNext) {

2,949,770✔

4413

      pInput->rowIdx = -1;

1,480,170✔

4414

      ++pSubTblsInfo->numSubTablesCompleted;

1,480,874✔

4415

      continue;

1,480,190✔

4417

      pInput->rowIdx = 0;

1,469,600✔

4418

      pInput->pageIdx = -1;

1,469,600✔

4422

_end:

1,844,360✔

4423

  if (code != TSDB_CODE_SUCCESS) {

1,844,008✔

4426

  return code;

1,844,008✔

4429

static void adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

1,469,248✔

4430

  int32_t              code = TSDB_CODE_SUCCESS;

1,469,248✔

4431

  int32_t              lino = 0;

1,469,248✔

4432

  SExecTaskInfo*       pTaskInfo = pOperatorInfo->pTaskInfo;

1,469,248✔

4433

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

1,469,600✔

4434

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,469,248✔

4435

  bool                 hasNext = true;

1,469,248✔

4436

  code = fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);

1,469,248✔

4437

  QUERY_CHECK_CODE(code, lino, _end);

1,469,248✔

4439

  if (!hasNext) {

1,469,248✔

4440

    pInput->rowIdx = -1;

1,469,248✔

4441

    ++pSubTblsInfo->numSubTablesCompleted;

1,469,248✔

4446

_end:

1,469,248✔

4447

  if (code != TSDB_CODE_SUCCESS) {

1,469,248✔

4452

1,469,248✔

4454

static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

267,626,896✔

4455

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

267,626,896✔

4456

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

267,629,340✔

4458

  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

267,628,676✔

4459

  if (pInput->rowIdx < pInputBlock->info.rows - 1) {

267,666,212✔

4460

    ++pInput->rowIdx;

266,192,096✔

4461

  } else if (pInput->rowIdx == pInputBlock->info.rows - 1) {

1,469,600✔

4462

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,469,600✔

4463

      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);

1,469,600✔

4465

    if (pInput->rowIdx != -1) {

1,469,600✔

4476

  return tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));

267,662,078✔

4479

static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {

267,498,678✔

4480

  int32_t            code = TSDB_CODE_SUCCESS;

267,498,678✔

4481

  int32_t            lino = 0;

267,498,678✔

4482

  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

267,498,678✔

4483

  SSDataBlock*       pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

267,567,702✔

4485

  for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {

1,108,213,302✔

4486

    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);

839,881,560✔

4487

    QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);

839,980,866✔

4489

    SColumnInfoData* pSrcColInfo = taosArrayGet(pInputBlock->pDataBlock, i);

839,980,866✔

4490

    QUERY_CHECK_NULL(pSrcColInfo, code, lino, _end, terrno);

839,618,074✔

4491

    bool isNull = colDataIsNull(pSrcColInfo, pInputBlock->info.rows, pInput->rowIdx, NULL);

839,618,074✔

4493

    if (isNull) {

840,050,806✔

4494

      code = colDataSetVal(pColInfo, pBlock->info.rows, NULL, true);

107,972,512✔

4495

      QUERY_CHECK_CODE(code, lino, _end);

107,971,104✔

4497

      if (pSrcColInfo->pData != NULL) {

732,078,294✔

4498

        char* pData = colDataGetData(pSrcColInfo, pInput->rowIdx);

732,155,068✔

4499

        code = colDataSetVal(pColInfo, pBlock->info.rows, pData, false);

732,919,714✔

4500

        QUERY_CHECK_CODE(code, lino, _end);

732,799,492✔

4504

  pBlock->info.dataLoad = 1;

267,625,438✔

4505

  pBlock->info.scanFlag = pInputBlock->info.scanFlag;

267,626,132✔

4506

  pBlock->info.rows += 1;

267,631,020✔

4508

_end:

267,630,718✔

4509

  if (code != TSDB_CODE_SUCCESS) {

267,630,718✔

4512

  return code;

267,628,626✔

4515

static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,

2,985,268✔

4517

  int32_t                 code = TSDB_CODE_SUCCESS;

2,985,268✔

4518

  int32_t                 lino = 0;

2,985,268✔

4519

  STableMergeScanInfo*    pInfo = pOperator->info;

2,985,268✔

4520

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

2,986,666✔

4521

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

2,985,972✔

4522

  bool                    finished = false;

2,985,298✔

4524

  QRY_PARAM_CHECK(pResBlock);

2,985,298✔

4526

  blockDataCleanup(pBlock);

2,985,288✔

4530

      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {

270,535,982✔

4531

        finished = true;

2,987,008✔

4532

        break;

2,987,008✔

4535

      code = appendChosenRowToDataBlock(pSubTblsInfo, pBlock);

267,546,872✔

4536

      QUERY_CHECK_CODE(code, lino, _end);

267,628,998✔

4538

      code = adjustSubTableForNextRow(pOperator, pSubTblsInfo);

267,628,998✔

4539

      QUERY_CHECK_CODE(code, lino, _end);

267,548,290✔

4541

      if (pBlock->info.rows >= capacity) {

267,548,290✔

4546

    if (isTaskKilled(pTaskInfo)) {

2,987,008✔

4550

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pBlock, pTaskInfo);

2,987,018✔

4551

    if (finished || limitReached || pBlock->info.rows > 0) {

2,986,324✔

4556

  if (pBlock->info.rows > 0) {

2,986,324✔

4557

    *pResBlock = pBlock;

1,140,928✔

4560

_end:

1,845,396✔

4561

  if (code != TSDB_CODE_SUCCESS) {

2,986,324✔

4565

  return code;

2,986,324✔

4568

static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {

1,833,648✔

4569

  int32_t              code = TSDB_CODE_SUCCESS;

1,833,648✔

4570

  int32_t              lino = 0;

1,833,648✔

4571

  STableMergeScanInfo* pInfo = pOperator->info;

1,833,648✔

4573

  code = initSubTablesMergeInfo(pInfo);

1,835,438✔

4574

  QUERY_CHECK_CODE(code, lino, _end);

1,815,040✔

4576

  code = initSubTableInputs(pOperator, pInfo);

1,815,040✔

4577

  QUERY_CHECK_CODE(code, lino, _end);

1,843,666✔

4579

  code = openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);

1,843,666✔

4580

  QUERY_CHECK_CODE(code, lino, _end);

1,844,702✔

4582

_end:

1,844,702✔

4583

  if (code != TSDB_CODE_SUCCESS) {

1,844,702✔

4586

  return code;

1,844,692✔

4589

static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {

28,716,810✔

4590

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

28,716,810✔

4591

  if (pSubTblsInfo != NULL) {

28,723,546✔

4592

    tMergeTreeDestroy(&pSubTblsInfo->pTree);

1,844,702✔

4594

    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,794,824✔

4595

      STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

2,950,474✔

4596

      taosMemoryFree(pInput->tblCond.colList);

2,949,096✔

4597

      blockDataDestroy(pInput->pReaderBlock);

2,949,448✔

4598

      blockDataDestroy(pInput->pPageBlock);

2,950,474✔

4599

      taosArrayDestroy(pInput->aBlockPages);

2,950,122✔

4600

      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);

2,950,816✔

4601

      pInput->pReader = NULL;

2,950,474✔

4604

    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);

1,845,738✔

4605

    taosMemoryFree(pSubTblsInfo->aInputs);

1,843,676✔

4607

    taosMemoryFree(pSubTblsInfo);

1,841,956✔

4608

    pInfo->pSubTablesMergeInfo = NULL;

1,845,396✔

4612

28,724,240✔

4614

int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

2,969,420✔

4615

  QRY_PARAM_CHECK(ppRes);

2,969,420✔

4617

  int32_t lino = 0;

2,972,548✔

4618

  int32_t tableListSize = 0;

2,972,548✔

4619

  int64_t st = taosGetTimestampUs();

2,963,888✔

4621

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

2,963,888✔

4622

  STableMergeScanInfo* pInfo = pOperator->info;

2,969,752✔

4624

  if (pOperator->status == OP_EXEC_DONE) {

2,975,656✔

4628

  int32_t code = pOperator->fpSet._openFn(pOperator);

2,948,710✔

4629

  QUERY_CHECK_CODE(code, lino, _end);

2,964,582✔

4631

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

2,964,582✔

4632

  QUERY_CHECK_CODE(code, lino, _end);

2,967,398✔

4634

  if (!pInfo->hasGroupId) {

2,967,398✔

4635

    pInfo->hasGroupId = true;

1,827,456✔

4637

    if (tableListSize == 0) {

1,806,062✔

4638

      setOperatorCompleted(pOperator);

49,280✔

4639

      (*ppRes) = NULL;

48,928✔

4640

      return code;

48,928✔

4643

    pInfo->tableStartIndex = 0;

1,756,782✔

4644

    STableKeyInfo* pTmpGpId = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,763,652✔

4645

    QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

1,749,218✔

4646

    pInfo->groupId = pTmpGpId->groupId;

1,749,218✔

4647

    code = startSubTablesTableMergeScan(pOperator);

1,781,646✔

4648

    QUERY_CHECK_CODE(code, lino, _end);

1,789,230✔

4651

  SSDataBlock* pBlock = NULL;

2,929,172✔

4652

  while (pInfo->tableStartIndex < tableListSize) {

2,985,982✔

4653

    if (isTaskKilled(pTaskInfo)) {

2,984,946✔

4657

    code = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity, &pBlock);

2,984,956✔

4658

    QUERY_CHECK_CODE(code, lino, _end);

2,986,676✔

4660

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

2,986,676✔

4667

    if (pBlock != NULL) {

2,985,972✔

4668

      pBlock->info.id.groupId = pInfo->groupId;

1,140,928✔

4669

      pOperator->resultInfo.totalRows += pBlock->info.rows;

1,141,280✔

4670

      pInfo->bGroupProcessed = true;

1,141,280✔

4671

      break;

1,140,928✔

4674

      stopSubTablesTableMergeScan(pInfo);

1,845,044✔

4675

      if (pInfo->tableEndIndex >= tableListSize - 1) {

1,844,370✔

4676

        setOperatorCompleted(pOperator);

1,788,214✔

4677

        break;

1,788,898✔

4680

      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

55,120✔

4681

      STableKeyInfo* pTmpGpId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

55,120✔

4682

      QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

55,120✔

4684

      pInfo->groupId = pTmpGpId->groupId;

55,120✔

4685

      code = startSubTablesTableMergeScan(pOperator);

55,120✔

4686

      QUERY_CHECK_CODE(code, lino, _end);

55,120✔

4687

      resetLimitInfoForNextGroup(&pInfo->limitInfo);

55,120✔

4691

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

2,931,556✔

4693

_end:

2,928,820✔

4694

  if (code != TSDB_CODE_SUCCESS) {

2,927,784✔

4699

    (*ppRes) = pBlock;

2,927,784✔

4702

  return code;

2,928,810✔

4705

static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {

3,438,773✔

4706

  int32_t              code = TSDB_CODE_SUCCESS;

3,438,773✔

4707

  int32_t              lino = 0;

3,438,773✔

4708

  SOperatorInfo*       pOperator = (SOperatorInfo*)pTableMergeOpInfo;

3,438,773✔

4709

  STableMergeScanInfo* pInfo = pOperator->info;

3,438,773✔

4710

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,439,296✔

4712

  if (pInfo->mSkipTables == NULL) {

3,439,296✔

4713

    pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,

456,599✔

4715

    QUERY_CHECK_NULL(pInfo->mSkipTables, code, lino, _end, terrno);

456,599✔

4717

  int bSkip = 1;

3,439,296✔

4718

  if (pInfo->mSkipTables != NULL) {

3,439,296✔

4719

    code = taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));

3,439,296✔

4720

    if (code == TSDB_CODE_DUP_KEY) {

3,439,296✔

4723

    QUERY_CHECK_CODE(code, lino, _end);

3,439,296✔

4726

_end:

3,439,296✔

4727

  if (code != TSDB_CODE_SUCCESS) {

3,439,296✔

4732

3,439,296✔

4734

static int32_t doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {

156,648,036✔

4735

  STableMergeScanInfo* pInfo = pOperator->info;

156,648,036✔

4736

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

156,661,573✔

4737

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

156,652,182✔

4738

  SSDataBlock*         pBlock = pInfo->pReaderBlock;

156,659,703✔

4739

  int32_t              code = 0;

156,630,216✔

4740

  bool                 hasNext = false;

156,630,216✔

4741

  STsdbReader*         reader = pInfo->base.dataReader;

156,638,754✔

4743

  code = pAPI->tsdReader.tsdNextDataBlock(reader, &hasNext);

156,641,401✔

4744

  if (code != 0) {

156,621,163✔

4751

  if (!hasNext || isTaskKilled(pTaskInfo)) {

156,621,163✔

4752

    if (isTaskKilled(pTaskInfo)) {

28,262,903✔

4756

    *pFinished = true;

28,258,603✔

4757

    return code;

28,248,726✔

4760

  uint32_t status = 0;

128,374,049✔

4761

  code = loadDataBlock(pOperator, &pInfo->base, pBlock, &status);

128,379,865✔

4763

  if (code != TSDB_CODE_SUCCESS) {

128,378,727✔

4764

    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));

16,151✔

4765

    pTaskInfo->code = code;

16,151✔

4766

    return code;

16,151✔

4769

  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

128,362,576✔

4775

  if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {

128,362,576✔

4776

    *pSkipped = true;

6,200,304✔

4777

    return code;

6,210,418✔

4780

  return code;

122,147,342✔

4783

static int32_t getBlockForTableMergeScan(void* param, SSDataBlock** ppBlock) {

156,346,947✔

4784

  STableMergeScanSortSourceParam* source = param;

156,346,947✔

4786

  SOperatorInfo*       pOperator = source->pOperator;

156,346,947✔

4787

  STableMergeScanInfo* pInfo = pOperator->info;

156,352,096✔

4788

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

156,349,727✔

4789

  SSDataBlock*         pBlock = NULL;

156,331,109✔

4790

  int64_t              st = taosGetTimestampUs();

156,330,677✔

4791

  int32_t              code = TSDB_CODE_SUCCESS;

156,330,677✔

4794

    if (pInfo->rtnNextDurationBlocks) {

167,699,062✔

4795

      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d",

11,063,258✔

4798

      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {

11,071,208✔

4799

        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];

5,162,288✔

4800

        ++pInfo->nextDurationBlocksIdx;

5,162,288✔

4802

        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

11,063,507✔

4803

          blockDataDestroy(pInfo->nextDurationBlocks[i]);

5,161,617✔

4804

          pInfo->nextDurationBlocks[i] = NULL;

5,158,332✔

4807

        pInfo->rtnNextDurationBlocks = false;

5,901,890✔

4808

        pInfo->nextDurationBlocksIdx = 0;

5,902,620✔

4809

        pInfo->numNextDurationBlocks = 0;

5,902,985✔

4810

        continue;

5,903,715✔

4813

      bool bFinished = false;

156,654,199✔

4814

      bool bSkipped = false;

156,649,784✔

4816

      code = doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);

156,667,404✔

4817

      if (code != 0) {

156,621,297✔

4818

        return code;

16,151✔

4821

      pBlock = pInfo->pReaderBlock;

156,605,146✔

4822

      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d",

156,614,268✔

4824

      if (bFinished) {

156,658,512✔

4825

        pInfo->bNewFilesetEvent = false;

28,262,726✔

4826

        break;

28,259,804✔

4829

      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {

128,395,786✔

4830

        if (!bSkipped) {

5,907,270✔

4831

          code = createOneDataBlock(pBlock, true, &pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks]);

5,162,288✔

4832

          if (code) {

5,155,718✔

4837

          ++pInfo->numNextDurationBlocks;

5,155,718✔

4838

          if (pInfo->numNextDurationBlocks > 2) {

5,154,028✔

4846

        if (pInfo->bNewFilesetEvent) {

5,905,080✔

4847

          pInfo->rtnNextDurationBlocks = true;

5,899,411✔

4848

          *ppBlock = NULL;

5,900,736✔

4849

          return code;

5,891,746✔

4852

        if (pInfo->bNextDurationBlockEvent) {

1,095✔

4857

      if (bSkipped) continue;

122,493,706✔

4860

    pBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pBlock->info.id.uid);

122,190,203✔

4862

    pOperator->resultInfo.totalRows += pBlock->info.rows;

122,193,479✔

4863

    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

122,186,650✔

4864

    *ppBlock = pBlock;

122,189,712✔

4866

    return code;

122,184,984✔

4869

  *ppBlock = NULL;

28,259,804✔

4870

  return code;

28,263,247✔

4873

int32_t generateSortByTsPkInfo(SArray* colMatchInfo, int32_t order, SArray** ppSortArray) {

24,873,353✔

4874

  int32_t code = TSDB_CODE_SUCCESS;

24,873,353✔

4875

  int32_t lino = 0;

24,873,353✔

4876

  SArray* pSortInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));

24,873,353✔

4877

  QUERY_CHECK_NULL(pSortInfo, code, lino, _end, terrno);

24,825,462✔

4878

  SBlockOrderInfo biTs = {0};

24,825,462✔

4879

  SBlockOrderInfo biPk = {0};

24,833,168✔

4881

  int32_t tsTargetSlotId = 0;

24,837,402✔

4882

  int32_t pkTargetSlotId = -1;

24,837,402✔

4883

  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {

89,286,797✔

4884

    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);

64,536,936✔

4885

    QUERY_CHECK_NULL(colInfo, code, lino, _end, terrno);

64,551,589✔

4886

    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

64,551,589✔

4887

      tsTargetSlotId = colInfo->dstSlotId;

24,867,703✔

4888

      biTs.order = order;

24,878,770✔

4889

      biTs.slotId = tsTargetSlotId;

24,878,770✔

4890

      biTs.nullFirst = (order == TSDB_ORDER_ASC);

24,878,770✔

4891

      biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, order);

24,878,770✔

4894

    if (colInfo->isPk) {

64,490,704✔

4895

      pkTargetSlotId = colInfo->dstSlotId;

1,177,838✔

4896

      biPk.order = order;

1,177,838✔

4897

      biPk.slotId = pkTargetSlotId;

1,177,838✔

4898

      biPk.nullFirst = (order == TSDB_ORDER_ASC);

1,177,838✔

4899

      biPk.compFn = getKeyComparFunc(colInfo->dataType.type, order);

1,177,838✔

4903

  void* tmp = taosArrayPush(pSortInfo, &biTs);

24,883,731✔

4904

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

24,883,731✔

4905

  if (pkTargetSlotId != -1) {

24,883,731✔

4906

    tmp = taosArrayPush(pSortInfo, &biPk);

1,179,673✔

4907

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

1,179,673✔

4910

  (*ppSortArray) = pSortInfo;

24,884,971✔

4912

_end:

24,849,069✔

4913

  if (code != TSDB_CODE_SUCCESS) {

24,856,278✔

4916

  return code;

24,856,278✔

4919

void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {

16,586,842✔

4920

  STableMergeScanInfo* pTmsInfo = param;

16,586,842✔

4921

  if (type == TSD_READER_NOTIFY_DURATION_START) {

16,586,842✔

4922

    pTmsInfo->bNewFilesetEvent = true;

16,591,543✔

4926

  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);

16,581,021✔

4928

  return;

16,610,629✔

4931

int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

34,356,949✔

4932

  STableMergeScanInfo* pInfo = pOperator->info;

34,356,949✔

4933

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

34,379,023✔

4935

  int32_t code = TSDB_CODE_SUCCESS;

34,368,682✔

4936

  int32_t lino = 0;

34,368,682✔

4937

  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

34,368,682✔

4939

  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));

34,343,284✔

4940

  pInfo->bNewFilesetEvent = false;

34,378,051✔

4941

  pInfo->bNextDurationBlockEvent = false;

34,396,077✔

4943

  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;

34,385,819✔

4944

  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;

34,385,366✔

4946

  pInfo->pSortHandle = NULL;

34,382,418✔

4947

  code = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,

68,779,990✔

4948

                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0, &pInfo->pSortHandle);

34,377,729✔

4949

  if (code) {

34,349,547✔

4953

  if (pInfo->bSortRowId && numOfTable != 1) {

34,349,547✔

4954

    int32_t memSize = 512 * 1024 * 1024;

2,657,319✔

4955

    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);

2,657,319✔

4956

    QUERY_CHECK_CODE(code, lino, _end);

2,656,724✔

4959

  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);

34,354,463✔

4960

  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pOperator);

34,355,858✔

4961

  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);

34,359,806✔

4963

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);

34,359,952✔

4964

  QUERY_CHECK_CODE(code, lino, _end);

34,331,757✔

4966

  STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));

34,331,757✔

4967

  QUERY_CHECK_NULL(param, code, lino, _end, terrno);

34,333,945✔

4968

  param->pOperator = pOperator;

34,333,945✔

4970

  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));

34,337,994✔

4971

  if (ps == NULL) {

34,311,880✔

4976

  ps->param = param;

34,311,880✔

4977

  ps->onlyRef = false;

34,343,479✔

4978

  code = tsortAddSource(pInfo->pSortHandle, ps);

34,361,250✔

4979

  QUERY_CHECK_CODE(code, lino, _end);

34,365,748✔

4981

  if (numOfTable == 1) {

34,365,748✔

4982

    tsortSetSingleTableMerge(pInfo->pSortHandle);

23,534,307✔

4984

    code = tsortOpen(pInfo->pSortHandle);

10,831,441✔

4985

    QUERY_CHECK_CODE(code, lino, _end);

10,837,015✔

4988

_end:

10,829,200✔

4989

  if (code != TSDB_CODE_SUCCESS) {

34,368,421✔

4990

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

7,815✔

4992

  return code;

34,376,003✔

4995

void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

34,386,155✔

4996

  STableMergeScanInfo* pInfo = pOperator->info;

34,386,155✔

4997

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

34,384,014✔

4998

  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));

34,384,041✔

5000

  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

34,388,303✔

5001

  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;

34,381,305✔

5002

  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;

34,380,782✔

5003

  pInfo->sortExecInfo.loops += sortExecInfo.loops;

34,380,269✔

5004

  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;

34,379,315✔

5005

  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

34,373,589✔

5007

  tsortDestroySortHandle(pInfo->pSortHandle);

34,379,776✔

5008

  pInfo->pSortHandle = NULL;

34,376,002✔

5009

34,373,711✔

5011

void startGroupTableMergeScan(SOperatorInfo* pOperator) {

28,470,953✔

5012

  int32_t              code = TSDB_CODE_SUCCESS;

28,470,953✔

5013

  int32_t              lino = 0;

28,470,953✔

5014

  STableMergeScanInfo* pInfo = pOperator->info;

28,470,953✔

5015

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,476,853✔

5016

  SReadHandle*         pHandle = &pInfo->base.readHandle;

28,450,289✔

5017

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,425,541✔

5018

  qDebug("%s table merge scan start group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,464,498✔

5021

    int32_t numOfTables = 0;

28,470,306✔

5022

    code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

28,440,028✔

5023

    QUERY_CHECK_CODE(code, lino, _end);

28,463,804✔

5025

    int32_t i = pInfo->tableStartIndex + 1;

28,463,804✔

5026

    for (; i < numOfTables; ++i) {

131,206,025✔

5027

      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

108,903,044✔

5028

      QUERY_CHECK_NULL(tableKeyInfo, code, lino, _end, terrno);

108,903,510✔

5029

      if (tableKeyInfo->groupId != pInfo->groupId) {

108,903,510✔

5030

        break;

6,158,048✔

5033

    pInfo->tableEndIndex = i - 1;

28,461,029✔

5035

  pInfo->bGroupProcessed = false;

28,487,957✔

5036

  int32_t tableStartIdx = pInfo->tableStartIndex;

28,488,482✔

5037

  int32_t tableEndIdx = pInfo->tableEndIndex;

28,475,726✔

5039

  int32_t        numOfTable = tableEndIdx - tableStartIdx + 1;

28,489,878✔

5040

  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);

28,489,878✔

5041

  code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,

56,940,524✔

5042

                                       (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->mSkipTables);

28,474,007✔

5043

  QUERY_CHECK_CODE(code, lino, _end);

28,456,914✔

5044

  if (pInfo->filesetDelimited) {

28,456,914✔

5045

    pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);

28,455,801✔

5047

  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);

28,478,244✔

5049

  code = startDurationForGroupTableMergeScan(pOperator);

28,423,614✔

5050

  QUERY_CHECK_CODE(code, lino, _end);

28,444,802✔

5052

_end:

28,436,987✔

5053

  if (code != TSDB_CODE_SUCCESS) {

28,443,794✔

5054

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

7,815✔

5055

    pTaskInfo->code = code;

7,815✔

5056

    T_LONG_JMP(pTaskInfo->env, code);

7,815✔

5058

28,435,979✔

5060

void stopGroupTableMergeScan(SOperatorInfo* pOperator) {

28,481,075✔

5061

  STableMergeScanInfo* pInfo = pOperator->info;

28,481,075✔

5062

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,481,933✔

5063

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,478,085✔

5065

  stopDurationForGroupTableMergeScan(pOperator);

28,477,748✔

5067

  if (pInfo->base.dataReader != NULL) {

28,472,965✔

5068

    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);

28,474,435✔

5069

    pInfo->base.dataReader = NULL;

28,434,933✔

5071

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

28,439,224✔

5072

    if (pInfo->nextDurationBlocks[i]) {

1,036✔

5073

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

1,036✔

5074

      pInfo->nextDurationBlocks[i] = NULL;

1,036✔

5076

    pInfo->numNextDurationBlocks = 0;

1,036✔

5077

    pInfo->nextDurationBlocksIdx = 0;

1,036✔

5079

  resetLimitInfoForNextGroup(&pInfo->limitInfo);

28,457,376✔

5080

  taosHashCleanup(pInfo->mSkipTables);

28,445,626✔

5081

  pInfo->mSkipTables = NULL;

28,427,458✔

5082

  qDebug("%s table merge scan stop group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,431,228✔

5083

28,478,199✔

5087

SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,

70,642,096✔

5089

  STableMergeScanInfo* pInfo = pOperator->info;

70,642,096✔

5090

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

70,655,694✔

5091

  STupleHandle*        pTupleHandle = NULL;

70,647,900✔

5093

  blockDataCleanup(pResBlock);

70,664,925✔

5096

    while (1) {

2,147,483,647✔

5097

      pTupleHandle = NULL;

2,147,483,647✔

5098

      int32_t code = tsortNextTuple(pHandle, &pTupleHandle);

2,147,483,647✔

5099

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5100

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

146✔

5101

        T_LONG_JMP(pOperator->pTaskInfo->env, code);

146✔

5103

      if (pTupleHandle == NULL) {

2,147,483,647✔

5104

        break;

55,463,224✔

5107

      code = tsortAppendTupleToBlock(pInfo->pSortHandle, pResBlock, pTupleHandle);

2,147,483,647✔

5108

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5113

      if (pResBlock->info.rows >= capacity) {

2,147,483,647✔

5114

        break;

15,191,769✔

5118

    if (tsortIsClosed(pHandle)) {

70,654,993✔

5123

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);

70,667,562✔

5124

    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,

70,662,925✔

5126

    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {

70,675,773✔

5130

  return (pResBlock->info.rows > 0) ? pResBlock : NULL;

70,675,773✔

5133

int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

61,335,374✔

5134

  if (pOperator->status == OP_EXEC_DONE) {

61,335,374✔

5135

    (*ppRes) = NULL;

41,493✔

5136

    return TSDB_CODE_SUCCESS;

41,493✔

5139

  int32_t              code = TSDB_CODE_SUCCESS;

61,273,584✔

5140

  int32_t              lino = 0;

61,273,584✔

5141

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

61,273,584✔

5142

  STableMergeScanInfo* pInfo = pOperator->info;

61,294,819✔

5144

  code = pOperator->fpSet._openFn(pOperator);

61,287,255✔

5145

  QUERY_CHECK_CODE(code, lino, _end);

61,269,229✔

5147

  int64_t st = taosGetTimestampUs();

61,290,065✔

5149

  int32_t tableListSize = 0;

61,290,065✔

5150

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

61,282,075✔

5151

  QUERY_CHECK_CODE(code, lino, _end);

61,290,988✔

5153

  if (!pInfo->hasGroupId) {

61,290,988✔

5154

    pInfo->hasGroupId = true;

25,012,607✔

5156

    if (tableListSize == 0) {

24,972,950✔

5157

      setOperatorCompleted(pOperator);

2,703,358✔

5158

      (*ppRes) = NULL;

2,703,729✔

5159

      return code;

2,702,999✔

5161

    pInfo->tableStartIndex = 0;

22,269,592✔

5162

    STableKeyInfo* tmp = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

22,308,888✔

5163

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

22,283,382✔

5164

    pInfo->groupId = tmp->groupId;

22,283,382✔

5165

    startGroupTableMergeScan(pOperator);

22,307,241✔

5168

  SSDataBlock* pBlock = NULL;

58,582,895✔

5169

  while (pInfo->tableStartIndex < tableListSize) {

70,642,780✔

5170

    if (isTaskKilled(pTaskInfo)) {

70,654,111✔

5174

    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,

70,646,423✔

5176

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

70,672,330✔

5177

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,795,620✔

5178

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

1,795,620✔

5179

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

1,795,620✔

5181

    if (pBlock != NULL) {

70,673,690✔

5182

      pBlock->info.id.groupId = pInfo->groupId;

36,285,546✔

5183

      pOperator->resultInfo.totalRows += pBlock->info.rows;

36,285,526✔

5184

      pInfo->bGroupProcessed = true;

36,282,856✔

5185

      break;

36,283,746✔

5187

      if (pInfo->bNewFilesetEvent) {

34,388,144✔

5188

        stopDurationForGroupTableMergeScan(pOperator);

5,905,481✔

5189

        code = startDurationForGroupTableMergeScan(pOperator);

5,901,101✔

5190

        QUERY_CHECK_CODE(code, lino, _end);

5,901,371✔

5193

        stopGroupTableMergeScan(pOperator);

28,482,298✔

5194

        if (pInfo->tableEndIndex >= tableListSize - 1) {

28,480,980✔

5195

          setOperatorCompleted(pOperator);

22,318,366✔

5196

          break;

22,318,706✔

5199

        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

6,158,514✔

5200

        STableKeyInfo* tmp = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

6,158,514✔

5201

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

6,158,514✔

5202

        pInfo->groupId = tmp->groupId;

6,158,514✔

5203

        startGroupTableMergeScan(pOperator);

6,158,514✔

5204

        resetLimitInfoForNextGroup(&pInfo->limitInfo);

6,157,725✔

5209

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

58,598,948✔

5211

_end:

58,598,653✔

5212

  if (code != TSDB_CODE_SUCCESS) {

58,600,220✔

5217

    (*ppRes) = pBlock;

58,600,220✔

5220

  return code;

58,601,318✔

5229

void destroyTableMergeScanOperatorInfo(void* param) {

24,904,628✔

5230

  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;

24,904,628✔

5233

  if (pTableScanInfo->base.readerAPI.tsdReaderClose != NULL) {

24,904,628✔

5234

    pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);

24,906,461✔

5235

    pTableScanInfo->base.dataReader = NULL;

24,898,932✔

5238

  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {

24,900,219✔

5245

  tsortDestroySortHandle(pTableScanInfo->pSortHandle);

24,901,713✔

5246

  pTableScanInfo->pSortHandle = NULL;

24,896,507✔

5247

  taosHashCleanup(pTableScanInfo->mSkipTables);

24,904,552✔

5248

  pTableScanInfo->mSkipTables = NULL;

24,889,869✔

5249

  blockDataDestroy(pTableScanInfo->pSortInputBlock);

24,892,375✔

5250

  pTableScanInfo->pSortInputBlock = NULL;

24,902,176✔

5253

  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);

24,905,438✔

5255

  blockDataDestroy(pTableScanInfo->pResBlock);

24,897,734✔

5256

  pTableScanInfo->pResBlock = NULL;

24,901,524✔

5259

  blockDataDestroy(pTableScanInfo->pReaderBlock);

24,902,631✔

5260

  pTableScanInfo->pReaderBlock = NULL;

24,900,151✔

5261

  taosArrayDestroy(pTableScanInfo->pSortInfo);

24,904,158✔

5263

  stopSubTablesTableMergeScan(pTableScanInfo);

24,900,064✔

5265

  taosMemoryFreeClear(param);

24,890,768✔

5266

24,887,138✔

5268

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {

1,194,059✔

5269

  if (pOptr == NULL) {

1,194,059✔

5274

  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));

1,194,059✔

5275

  if (!execInfo) {

1,193,323✔

5278

  STableMergeScanInfo* pInfo = pOptr->info;

1,193,323✔

5279

  execInfo->blockRecorder = pInfo->base.readRecorder;

1,193,688✔

5280

  execInfo->sortExecInfo = pInfo->sortExecInfo;

1,194,053✔

5282

  *pOptrExplain = execInfo;

1,194,436✔

5283

  *len = sizeof(STableMergeScanExecInfo);

1,194,059✔

5285

  return TSDB_CODE_SUCCESS;

1,194,795✔

5288

static int32_t resetTableMergeScanOperatorState(SOperatorInfo* pOper) {

1,972,866✔

5289

  int32_t         code = TSDB_CODE_SUCCESS;

1,972,866✔

5290

  STableMergeScanInfo*   pInfo = pOper->info;

1,972,866✔

5291

  pOper->status = OP_NOT_OPENED;

1,972,866✔

5293

  pInfo->tableEndIndex = 0;

1,972,866✔

5294

  pInfo->tableStartIndex = 0;

1,972,866✔

5295

  pInfo->hasGroupId = false;

1,972,866✔

5296

  if (pInfo->base.readerAPI.tsdReaderClose) {

1,972,866✔

5297

    pInfo->base.readerAPI.tsdReaderClose(pInfo->base.dataReader);

1,972,866✔

5299

  pInfo->base.dataReader = NULL;

1,972,866✔

5300

  pInfo->base.scanFlag = MAIN_SCAN;

1,972,866✔

5302

  pInfo->base.limitInfo = (SLimitInfo){0};

1,972,866✔

5303

  pInfo->base.limitInfo.limit.limit = -1;

1,972,866✔

5304

  pInfo->base.limitInfo.slimit.limit = -1;

1,972,866✔

5306

  tableListDestroy(pInfo->base.pTableListInfo);

1,972,866✔

5307

  pInfo->base.pTableListInfo = tableListCreate();

1,972,615✔

5308

  if (!pInfo->base.pTableListInfo) {

1,972,866✔

5312

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

1,972,866✔

5314

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

1,972,866✔

5315

  code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

3,945,732✔

5317

                                  pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

3,945,732✔

5318

  if (code) {

1,972,866✔

5323

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

1,972,866✔

5324

  cleanupQueryTableDataCond(&pInfo->base.cond);

1,972,866✔

5325

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, &pInfo->base.readHandle, false);

1,972,866✔

5326

  if (code) {

1,972,866✔

5331

  tsortDestroySortHandle(pInfo->pSortHandle);

1,972,866✔

5332

  pInfo->pSortHandle = NULL;

1,972,866✔

5334

  taosHashCleanup(pInfo->mSkipTables);

1,972,866✔

5335

  pInfo->mSkipTables = NULL;

1,972,866✔

5336

  pOper->resultInfo.totalRows = 0;

1,972,866✔

5338

  pInfo->bGroupProcessed = false;

1,972,866✔

5339

  pInfo->bNewFilesetEvent = false;

1,972,866✔

5340

  pInfo->bNextDurationBlockEvent = false;

1,972,866✔

5341

  pInfo->rtnNextDurationBlocks = false;

1,972,866✔

5342

  pInfo->nextDurationBlocksIdx = 0;

1,972,866✔

5343

  pInfo->bSortRowId = false;

1,972,866✔

5345

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

1,972,866✔

5351

  pInfo->numNextDurationBlocks = 0;

1,972,866✔

5353

  stopSubTablesTableMergeScan(pInfo);

1,972,866✔

5354

  return code;

1,972,866✔

5357

int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

24,851,695✔

5360

  QRY_PARAM_CHECK(pOptrInfo);

24,851,695✔

5362

  int32_t              code = TSDB_CODE_SUCCESS;

24,857,659✔

5363

  int32_t              lino = 0;

24,857,659✔

5364

  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));

24,857,659✔

5365

  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

24,833,926✔

5366

  if (pInfo == NULL || pOperator == NULL) {

24,790,471✔

5371

  pOperator->pPhyNode = pTableScanNode;

24,791,181✔

5372

  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

24,821,146✔

5374

  int32_t numOfCols = 0;

24,831,885✔

5375

  code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,

24,835,555✔

5377

  QUERY_CHECK_CODE(code, lino, _error);

24,829,358✔

5379

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle, false);

24,829,358✔

5380

  QUERY_CHECK_CODE(code, lino, _error);

24,790,541✔

5382

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

24,790,541✔

5383

    SExprSupp* pSup = &pInfo->base.pseudoSup;

10,314,057✔

5384

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->pExprInfo, &pSup->numOfExprs);

10,300,654✔

5385

    QUERY_CHECK_CODE(code, lino, _error);

10,284,204✔

5387

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

10,284,204✔

5389

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

10,293,687✔

5392

  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};

24,779,047✔

5394

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

24,738,056✔

5399

  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;

24,738,056✔

5400

  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

24,836,064✔

5401

  pInfo->base.scanFlag = MAIN_SCAN;

24,722,968✔

5402

  pInfo->base.readHandle = *readHandle;

24,773,246✔

5404

  pInfo->readIdx = -1;

24,847,288✔

5406

  pInfo->base.limitInfo.limit.limit = -1;

24,721,772✔

5407

  pInfo->base.limitInfo.slimit.limit = -1;

24,737,836✔

5408

  pInfo->base.pTableListInfo = pTableListInfo;

24,791,008✔

5410

  pInfo->sample.sampleRatio = pTableScanNode->ratio;

24,854,290✔

5411

  pInfo->sample.seed = taosGetTimestampSec();

24,738,809✔

5413

  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,

24,841,922✔

5414

                            pTaskInfo->pStreamRuntimeInfo);

24,823,748✔

5415

  QUERY_CHECK_CODE(code, lino, _error);

24,782,980✔

5417

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

24,782,980✔

5419

  pInfo->mergeLimit = -1;

24,796,854✔

5420

  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;

24,838,462✔

5421

  if (hasLimit) {

24,754,703✔

5422

    pInfo->mergeLimit = pInfo->limitInfo.limit.offset != -1

4,524,021✔

5423

                            ? pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset

847,831✔

5424

                            : pInfo->limitInfo.limit.limit;

3,107,395✔

5425

    pInfo->mSkipTables = NULL;

2,262,184✔

5428

  initResultSizeInfo(&pOperator->resultInfo, 1024);

24,752,080✔

5429

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

24,804,371✔

5430

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

24,851,623✔

5431

  code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

24,789,714✔

5432

  QUERY_CHECK_CODE(code, lino, _error);

24,894,488✔

5433

  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {

24,894,488✔

5434

    pInfo->bSortRowId = true;

4,813,470✔

5436

    pInfo->bSortRowId = false;

20,065,757✔

5439

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

24,862,040✔

5440

  QUERY_CHECK_CODE(code, lino, _error);

24,843,947✔

5442

  code = generateSortByTsPkInfo(pInfo->base.matchInfo.pList, pInfo->base.cond.order, &pInfo->pSortInfo);

24,843,947✔

5443

  QUERY_CHECK_CODE(code, lino, _error);

24,845,856✔

5445

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pReaderBlock);

24,845,856✔

5446

  QUERY_CHECK_CODE(code, lino, _error);

24,851,679✔

5448

  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;

24,851,679✔

5450

  int32_t  rowSize = pInfo->pResBlock->info.rowSize;

24,878,705✔

5451

  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);

24,840,632✔

5453

  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);

24,778,547✔

5456

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pSortInputBlock);

24,863,254✔

5457

  QUERY_CHECK_CODE(code, lino, _error);

24,867,567✔

5459

  if (!tsExperimental) {

24,867,567✔

5462

    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

24,867,567✔

5465

  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,

24,860,497✔

5467

  pOperator->exprSupp.numOfExprs = numOfCols;

24,837,390✔

5469

  pOperator->fpSet = createOperatorFpSet(

24,830,567✔

5470

      optrDummyOpenFn, pTableScanNode->paraTablesSort ? doTableMergeScanParaSubTablesNext : doTableMergeScanNext, NULL,

24,873,052✔

5473

  setOperatorResetStateFn(pOperator, resetTableMergeScanOperatorState);

24,810,325✔

5474

  pOperator->cost.openCost = 0;

24,848,405✔

5476

  *pOptrInfo = pOperator;

24,860,668✔

5477

  return code;

24,771,385✔

5516

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {

25,094✔

5517

  if (scanCols != NULL) {

25,094✔

5518

    SNode* pNode = NULL;

14,277✔

5519

    FOREACH(pNode, scanCols) {

33,340✔

5520

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

19,063✔

5523

      STargetNode* targetNode = (STargetNode*)pNode;

19,063✔

5524

      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {

19,063✔

5527

      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);

19,063✔

5528

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

19,063✔

5529

        supp->dbNameSlotId = targetNode->slotId;

9,404✔

5530

      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

9,659✔

5531

        supp->stbNameSlotId = targetNode->slotId;

9,659✔

5535

  return TSDB_CODE_SUCCESS;

25,094✔

5538

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {

25,094✔

5539

  if (pseudoCols != NULL) {

25,094✔

5540

    SNode* pNode = NULL;

25,094✔

5541

    FOREACH(pNode, pseudoCols) {

50,188✔

5542

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

25,094✔

5545

      STargetNode* targetNode = (STargetNode*)pNode;

25,094✔

5546

      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {

25,094✔

5549

      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);

25,094✔

5550

      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {

25,094✔

5551

        supp->tbCountSlotId = targetNode->slotId;

25,094✔

5555

  return TSDB_CODE_SUCCESS;

25,094✔

5558

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {

25,094✔

5559

  if (groupTags != NULL) {

25,094✔

5560

    SNode* pNode = NULL;

14,277✔

5561

    FOREACH(pNode, groupTags) {

33,340✔

5562

      if (nodeType(pNode) != QUERY_NODE_COLUMN) {

19,063✔

5565

      SColumnNode* colNode = (SColumnNode*)pNode;

19,063✔

5566

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

19,063✔

5567

        supp->groupByDbName = true;

9,404✔

5569

      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

19,063✔

5570

        supp->groupByStbName = true;

9,659✔

5574

    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);

10,817✔

5575

    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);

10,817✔

5577

  return TSDB_CODE_SUCCESS;

25,094✔

5580

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,

25,094✔

5582

  int32_t code = 0;

25,094✔

5583

  code = tblCountScanGetInputs(groupTags, tableName, supp);

25,094✔

5584

  if (code != TSDB_CODE_SUCCESS) {

25,094✔

5589

  supp->dbNameSlotId = -1;

25,094✔

5590

  supp->stbNameSlotId = -1;

25,094✔

5591

  supp->tbCountSlotId = -1;

25,094✔

5593

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);

25,094✔

5594

  if (code != TSDB_CODE_SUCCESS) {

25,094✔

5599

  code = tblCountScanGetCountSlotId(pseudoCols, supp);

25,094✔

5600

  if (code != TSDB_CODE_SUCCESS) {

25,094✔

5604

  return code;

25,094✔

5607

int32_t createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,

25,094✔

5609

  QRY_PARAM_CHECK(pOptrInfo);

25,094✔

5611

  int32_t                      code = TSDB_CODE_SUCCESS;

25,094✔

5612

  int32_t                      lino = 0;

25,094✔

5613

  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;

25,094✔

5614

  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));

25,094✔

5615

  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

25,094✔

5616

  if (!pInfo || !pOperator) {

25,094✔

5621

  pInfo->readHandle = *readHandle;

25,094✔

5623

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

25,094✔

5624

  initResultSizeInfo(&pOperator->resultInfo, 1);

25,094✔

5625

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

25,094✔

5626

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

25,094✔

5628

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

25,094✔

5629

  QUERY_CHECK_CODE(code, lino, _error);

25,094✔

5631

  code = getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,

25,094✔

5634

  QUERY_CHECK_CODE(code, lino, _error);

25,094✔

5636

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,

25,094✔

5638

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScanNext, NULL, destoryTableCountScanOperator,

25,094✔

5640

  *pOptrInfo = pOperator;

25,094✔

5641

  return code;

25,094✔

5655

int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,

34,838✔

5657

  int32_t code = TSDB_CODE_SUCCESS;

34,838✔

5658

  int32_t lino = 0;

34,838✔

5659

  if (pSupp->dbNameSlotId != -1) {

34,838✔

5660

    QUERY_CHECK_CONDITION((strlen(dbName) > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

14,172✔

5661

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);

14,172✔

5662

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

14,172✔

5664

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

14,172✔

5665

    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

14,172✔

5667

    varDataSetLen(varDbName, strlen(dbName));

14,172✔

5668

    code = colDataSetVal(colInfoData, 0, varDbName, false);

14,172✔

5669

    QUERY_CHECK_CODE(code, lino, _end);

14,172✔

5672

  if (pSupp->stbNameSlotId != -1) {

34,838✔

5673

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);

18,459✔

5674

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

18,459✔

5675

    if (strlen(stbName) != 0) {

18,459✔

5676

      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

13,238✔

5677

      tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);

13,238✔

5678

      varDataSetLen(varStbName, strlen(stbName));

13,238✔

5679

      code = colDataSetVal(colInfoData, 0, varStbName, false);

13,238✔

5680

      QUERY_CHECK_CODE(code, lino, _end);

13,238✔

5686

  if (pSupp->tbCountSlotId != -1) {

34,838✔

5687

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);

34,838✔

5688

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

34,838✔

5689

    code = colDataSetVal(colInfoData, 0, (char*)&count, false);

34,838✔

5690

    QUERY_CHECK_CODE(code, lino, _end);

34,838✔

5692

  pRes->info.rows = 1;

34,838✔

5694

_end:

34,802✔

5695

  if (code != TSDB_CODE_SUCCESS) {

34,802✔

5698

  return code;

34,838✔

5701

static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {

10,557✔

5702

  STableCountScanSupp* pSupp = &pInfo->supp;

10,557✔

5703

  SSDataBlock*         pRes = pInfo->pRes;

10,557✔

5705

  size_t infodbTableNum;

10,557✔

5706

  getInfosDbMeta(NULL, &infodbTableNum);

10,557✔

5707

  infodbTableNum -= 1;

10,557✔

5708

  size_t perfdbTableNum;

10,557✔

5709

  getPerfDbMeta(NULL, &perfdbTableNum);

10,557✔

5711

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

10,557✔

5712

    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);

9,048✔

5713

    return (pRes->info.rows > 0) ? pRes : NULL;

9,048✔

5715

    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);

1,509✔

5716

    return (pRes->info.rows > 0) ? pRes : NULL;

1,509✔

5720

static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,

1,509✔

5722

  int32_t        code = TSDB_CODE_SUCCESS;

1,509✔

5723

  int32_t        lino = 0;

1,509✔

5724

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,509✔

5725

  if (strlen(pSupp->stbNameFilter) != 0) {

1,509✔

5727

    code = fillTableCountScanDataBlock(pSupp, "", "", 0, pRes);

513✔

5728

    QUERY_CHECK_CODE(code, lino, _end);

513✔

5729

  } else if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {

996✔

5730

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

171✔

5731

    QUERY_CHECK_CODE(code, lino, _end);

171✔

5732

  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

825✔

5735

  } else if (strlen(pSupp->dbNameFilter) == 0) {

825✔

5736

    code = fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);

654✔

5737

    QUERY_CHECK_CODE(code, lino, _end);

654✔

5740

    code = fillTableCountScanDataBlock(pSupp, pSupp->dbNameFilter, pSupp->stbNameFilter, 0, pRes);

171✔

5741

    QUERY_CHECK_CODE(code, lino, _end);

171✔

5744

_end:

171✔

5745

  if (code != TSDB_CODE_SUCCESS) {

1,509✔

5750

  setOperatorCompleted(pOperator);

1,509✔

5751

1,509✔

5753

static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,048✔

5756

  int32_t        code = TSDB_CODE_SUCCESS;

9,048✔

5757

  int32_t        lino = 0;

9,048✔

5758

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,048✔

5759

  if (pInfo->currGrpIdx == 0) {

9,048✔

5760

    uint64_t groupId = 0;

3,016✔

5761

    if (pSupp->groupByDbName) {

3,016✔

5762

      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));

2,058✔

5764

      groupId = calcGroupId("", 0);

958✔

5767

    pRes->info.id.groupId = groupId;

3,016✔

5768

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

3,016✔

5769

    QUERY_CHECK_CODE(code, lino, _end);

3,016✔

5770

  } else if (pInfo->currGrpIdx == 1) {

6,032✔

5771

    uint64_t groupId = 0;

3,016✔

5772

    if (pSupp->groupByDbName) {

3,016✔

5773

      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));

2,058✔

5775

      groupId = calcGroupId("", 0);

958✔

5778

    pRes->info.id.groupId = groupId;

3,016✔

5779

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

3,016✔

5780

    QUERY_CHECK_CODE(code, lino, _end);

3,016✔

5782

    setOperatorCompleted(pOperator);

3,016✔

5784

  pInfo->currGrpIdx++;

9,048✔

5786

_end:

9,048✔

5787

  if (code != TSDB_CODE_SUCCESS) {

9,048✔

5792

9,048✔

5794

static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

59,932✔

5795

  int32_t                      code = TSDB_CODE_SUCCESS;

59,932✔

5796

  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;

59,932✔

5797

  STableCountScanOperatorInfo* pInfo = pOperator->info;

59,932✔

5798

  STableCountScanSupp*         pSupp = &pInfo->supp;

59,932✔

5799

  SSDataBlock*                 pRes = pInfo->pRes;

59,932✔

5801

  blockDataCleanup(pRes);

59,932✔

5802

  QRY_PARAM_CHECK(ppRes);

59,932✔

5804

  if (pOperator->status == OP_EXEC_DONE) {

59,932✔

5805

    return code;

14,491✔

5808

  if (pInfo->readHandle.mnd != NULL) {

45,441✔

5809

    (*ppRes) = buildSysDbTableCount(pOperator, pInfo);

10,557✔

5810

    return code;

10,557✔

5813

  code = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);

34,884✔

5814

  if (code != TSDB_CODE_SUCCESS) {

34,884✔

5819

  if (pRes->info.rows > 0) {

34,884✔

5820

    *ppRes = pRes;

27,297✔

5823

  return code;

34,884✔

5826

static int32_t buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

34,884✔

5828

  int32_t        code = TSDB_CODE_SUCCESS;

34,884✔

5829

  int32_t        lino = 0;

34,884✔

5830

  const char*    db = NULL;

34,884✔

5831

  int32_t        vgId = 0;

34,884✔

5832

  char           dbName[TSDB_DB_NAME_LEN] = {0};

34,884✔

5833

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

34,884✔

5834

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

34,884✔

5837

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);

34,884✔

5838

  SName sn = {0};

34,884✔

5840

  code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

34,884✔

5841

  QUERY_CHECK_CODE(code, lino, _end);

34,884✔

5843

  code = tNameGetDbName(&sn, dbName);

34,884✔

5844

  QUERY_CHECK_CODE(code, lino, _end);

34,884✔

5846

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

34,884✔

5847

    code = buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);

25,576✔

5849

    code = buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);

9,308✔

5852

_end:

34,848✔

5853

  if (code != TSDB_CODE_SUCCESS) {

34,848✔

5858

  return code;

34,848✔

5861

static int32_t buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

25,576✔

5863

  int32_t        code = TSDB_CODE_SUCCESS;

25,576✔

5864

  int32_t        lino = 0;

25,576✔

5865

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

25,576✔

5866

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

25,576✔

5868

  if (pSupp->groupByStbName) {

25,576✔

5869

    if (pInfo->stbUidList == NULL) {

21,902✔

5870

      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));

7,587✔

5871

      QUERY_CHECK_NULL(pInfo->stbUidList, code, lino, _end, terrno);

7,587✔

5872

      code = pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList);

7,587✔

5873

      QUERY_CHECK_CODE(code, lino, _end);

7,587✔

5876

    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {

21,902✔

5877

      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);

13,238✔

5878

      code = buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);

13,068✔

5879

      QUERY_CHECK_CODE(code, lino, _end);

13,238✔

5881

      pInfo->currGrpIdx++;

13,238✔

5882

    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {

8,664✔

5883

      code = buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);

7,587✔

5884

      QUERY_CHECK_CODE(code, lino, _end);

7,587✔

5886

      pInfo->currGrpIdx++;

7,587✔

5888

      setOperatorCompleted(pOperator);

1,077✔

5891

    uint64_t groupId = calcGroupId(dbName, strlen(dbName));

3,674✔

5892

    pRes->info.id.groupId = groupId;

3,674✔

5894

    int64_t dbTableCount = 0;

3,674✔

5895

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);

3,674✔

5896

    code = fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);

3,674✔

5897

    QUERY_CHECK_CODE(code, lino, _end);

3,674✔

5898

    setOperatorCompleted(pOperator);

3,674✔

5901

_end:

25,576✔

5902

  if (code != TSDB_CODE_SUCCESS) {

25,576✔

5906

  return code;

25,576✔

5909

static int32_t buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,308✔

5911

  int32_t        code = TSDB_CODE_SUCCESS;

9,308✔

5912

  int32_t        lino = 0;

9,308✔

5913

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,308✔

5914

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

9,308✔

5916

  if (strlen(pSupp->dbNameFilter) != 0 && strcmp(dbName, pSupp->dbNameFilter) != 0) {

9,308✔

5920

  } else if (strlen(pSupp->stbNameFilter) != 0) {

9,308✔

5921

    uint64_t uid = 0;

3,018✔

5922

    code = pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid);

3,018✔

5923

    if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

3,018✔

5925

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, 0, pRes);

1,368✔

5926

      QUERY_CHECK_CODE(code, lino, _end);

1,368✔

5928

      QUERY_CHECK_CODE(code, lino, _end);

1,650✔

5930

      int64_t numOfChildTables = 0;

1,650✔

5931

      code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL, NULL);

1,650✔

5932

      QUERY_CHECK_CODE(code, lino, _end);

1,650✔

5934

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes);

1,650✔

5935

      QUERY_CHECK_CODE(code, lino, _end);

1,650✔

5938

    int64_t tbNumVnode = 0;

6,290✔

5939

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);

6,290✔

5940

    code = fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);

6,290✔

5941

    QUERY_CHECK_CODE(code, lino, _end);

6,290✔

5944

_end:

9,308✔

5945

  if (code != TSDB_CODE_SUCCESS) {

9,308✔

5950

  setOperatorCompleted(pOperator);

9,308✔

5951

  return code;

9,272✔

5954

static int32_t buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

7,587✔

5956

  int32_t code = TSDB_CODE_SUCCESS;

7,587✔

5957

  int32_t lino = 0;

7,587✔

5958

  char    fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

7,587✔

5959

  if (pSupp->groupByDbName) {

7,587✔

5960

    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");

3,672✔

5963

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

7,587✔

5964

  pRes->info.id.groupId = groupId;

7,587✔

5966

  int64_t numOfTables = 0;

7,587✔

5967

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);

7,587✔

5969

  if (numOfTables != 0) {

7,587✔

5970

    code = fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);

1,077✔

5973

_end:

6,510✔

5974

  if (code != TSDB_CODE_SUCCESS) {

7,587✔

5977

  return code;

7,587✔

5980

static int32_t buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

13,068✔

5982

  int32_t code = TSDB_CODE_SUCCESS;

13,068✔

5983

  int32_t lino = 0;

13,068✔

5984

  char    stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

13,068✔

5985

  code = pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);

13,238✔

5986

  QUERY_CHECK_CODE(code, lino, _end);

13,238✔

5988

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

13,238✔

5989

  if (pSupp->groupByDbName) {

13,238✔

5990

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));

6,092✔

5992

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));

7,146✔

5995

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

13,238✔

5996

  pRes->info.id.groupId = groupId;

13,238✔

5998

  int64_t ctbNum = 0;

13,238✔

5999

  code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL, NULL);

13,238✔

6000

  QUERY_CHECK_CODE(code, lino, _end);

13,238✔

6001

  code = fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);

13,238✔

6002

  QUERY_CHECK_CODE(code, lino, _end);

13,238✔

6004

_end:

13,238✔

6005

  if (code != TSDB_CODE_SUCCESS) {

13,238✔

6008

  return code;

13,238✔

6011

static void destoryTableCountScanOperator(void* param) {

25,094✔

6012

  STableCountScanOperatorInfo* pTableCountScanInfo = param;

25,094✔

6013

  blockDataDestroy(pTableCountScanInfo->pRes);

25,094✔

6015

  taosArrayDestroy(pTableCountScanInfo->stbUidList);

25,094✔

6016

  taosMemoryFreeClear(param);

25,094✔

6017

25,094✔

taosdata / TDengine / #4946

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4946

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