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

76

bool processBlockWithProbability(const SSampleExecInfo* pInfo) {

482,807,789✔

85

  return true;

482,807,789✔

89

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

65,922✔

90

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

65,922✔

93

65,922✔

95

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

524,204,114✔

96

  int32_t     code = TSDB_CODE_SUCCESS;

524,204,114✔

97

  STimeWindow w = {0};

524,204,114✔

100

  if (pInterval->interval == 0) {

524,279,807✔

101

    *overlap = false;

488,711,901✔

102

    return code;

488,732,587✔

105

  if (order == TSDB_ORDER_ASC) {

35,508,506✔

106

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

34,321,960✔

107

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

34,319,111✔

112

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

34,316,628✔

113

      *overlap = true;

30,010,404✔

114

      return code;

30,009,530✔

118

      getNextTimeWindow(pInterval, &w, order);

4,309,775✔

119

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

4,313,957✔

120

        break;

4,049,665✔

123

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

263,944✔

127

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

263,944✔

128

        *overlap = true;

263,944✔

129

        return code;

263,944✔

133

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

1,186,546✔

134

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

1,188,540✔

139

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

1,186,674✔

140

      *overlap = true;

1,022,616✔

141

      return code;

1,021,812✔

145

      getNextTimeWindow(pInterval, &w, order);

164,278✔

146

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

165,704✔

147

        break;

163,746✔

150

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

2,580✔

154

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

2,580✔

155

        *overlap = true;

2,580✔

156

        return code;

2,580✔

161

  *overlap = false;

4,213,411✔

162

  return code;

4,211,019✔

166

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

104,304,099✔

167

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

104,304,099✔

171

  int64_t buf[2] = {0};

104,326,924✔

172

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

104,319,906✔

174

  STableScanInfo* pTableScanInfo = pOperator->info;

104,307,591✔

176

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

104,307,579✔

179

  if (p1 == NULL) {

104,353,530✔

180

    return NULL;

47,556,838✔

183

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

56,796,692✔

184

  if (NULL == *pPage) {

56,779,914✔

188

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

56,778,251✔

191

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

11,569,693✔

192

  if (NULL == pTableScanInfo->pIgnoreTables) {

11,569,693✔

193

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

1,162,056✔

194

    pTableScanInfo->pIgnoreTables =

1,162,056✔

195

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

1,162,056✔

196

    if (NULL == pTableScanInfo->pIgnoreTables) {

1,162,056✔

201

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

11,569,693✔

203

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

11,570,884✔

207

  return TSDB_CODE_SUCCESS;

11,570,884✔

210

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

581,299,898✔

211

  STableScanInfo* pTableScanInfo = pOperator->info;

581,299,898✔

212

  int32_t         code = TSDB_CODE_SUCCESS;

581,519,011✔

214

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

581,519,011✔

215

    return TSDB_CODE_SUCCESS;

477,157,559✔

218

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

104,319,807✔

220

  SFilePage*  pPage = NULL;

104,333,817✔

221

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

104,295,854✔

223

  if (pRow == NULL) {

104,334,671✔

224

    return TSDB_CODE_SUCCESS;

47,547,819✔

227

  bool notLoadBlock = true;

56,786,852✔

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;

45,217,269✔

236

      break;

45,217,269✔

241

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

54,876,822✔

243

  if (notLoadBlock) {

56,784,076✔

244

    *status = FUNC_DATA_REQUIRED_NOT_LOAD;

11,571,281✔

245

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

11,569,693✔

248

  return code;

56,753,223✔

251

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

8,826,272✔

253

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

8,826,272✔

258

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

8,827,572✔

261

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

118,934,437✔

263

  SStorageAPI* pAPI = &pTaskInfo->storageAPI;

118,934,437✔

264

  bool         allColumnsHaveAgg = true;

118,951,903✔

265

  bool         hasNullSMA = false;

118,956,250✔

266

  if (pLoad != NULL) {

118,944,357✔

267

    *pLoad = false;

118,958,713✔

270

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

118,947,659✔

272

  if (code != TSDB_CODE_SUCCESS) {

118,928,948✔

276

  if (!allColumnsHaveAgg || hasNullSMA) {

118,928,948✔

277

    *pLoad = false;

104,628,846✔

279

    *pLoad = true;

14,300,102✔

282

  return code;

118,909,767✔

285

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

614,323,238✔

287

  int32_t    code = 0;

614,323,238✔

288

  SExprSupp* pSup = &pTableScanInfo->pseudoSup;

614,323,238✔

289

  if (pSup->numOfExprs > 0) {

614,503,145✔

290

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

272,756,791✔

293

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

272,733,960✔

299

  return code;

614,373,070✔

302

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

1,046,703,542✔

303

  SLimit*     pLimit = &pLimitInfo->limit;

1,046,703,542✔

304

  const char* id = GET_TASKID(pTaskInfo);

1,046,802,030✔

306

  if (pLimitInfo->remainOffset > 0) {

1,046,868,749✔

307

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

5,347,240✔

308

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

2,497,187✔

309

      blockDataEmpty(pBlock);

2,497,187✔

310

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

2,497,187✔

311

      return false;

2,497,187✔

313

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

2,850,053✔

314

      if (code != TSDB_CODE_SUCCESS) {

2,850,053✔

319

      pLimitInfo->remainOffset = 0;

2,850,053✔

323

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

1,044,353,808✔

325

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

31,374,396✔

326

    blockDataKeepFirstNRows(pBlock, keep);

31,374,360✔

328

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

31,373,356✔

329

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

31,373,998✔

330

    return true;

31,373,960✔

333

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

1,012,821,872✔

334

  return false;

1,012,900,328✔

341

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

610,092,880✔

343

  int32_t        code = TSDB_CODE_SUCCESS;

610,092,880✔

344

  int32_t        lino = 0;

610,092,880✔

345

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

610,092,880✔

346

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

610,222,220✔

347

  bool           loadSMA = false;

610,214,983✔

349

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

610,214,983✔

351

  pCost->totalBlocks += 1;

610,102,400✔

352

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

610,178,797✔

353

  *status = pTableScanInfo->dataBlockLoadFlag;

610,139,617✔

355

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

610,197,010✔

356

    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

85,866,492✔

358

    bool overlap = false;

524,202,381✔

360

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

524,247,503✔

361

    if (ret != TSDB_CODE_SUCCESS) {

524,077,143✔

364

    if (overlap) {

524,077,143✔

365

      (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

31,294,403✔

369

  SDataBlockInfo* pBlockInfo = &pBlock->info;

610,113,722✔

370

  taosMemoryFreeClear(pBlock->pBlockAgg);

610,163,266✔

372

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {

610,122,501✔

379

  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

609,924,428✔

380

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

19,837,749✔

383

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

19,847,243✔

384

    pCost->skipBlocks += 1;

19,847,843✔

385

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

19,846,283✔

386

    return code;

19,845,373✔

387

  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {

590,067,310✔

388

    pCost->loadBlockStatis += 1;

33,071,221✔

389

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

33,071,433✔

390

    bool success = true;

33,071,433✔

391

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

33,074,674✔

392

    if (code) {

33,069,962✔

395

      QUERY_CHECK_CODE(code, lino, _end);

14✔

398

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

33,069,260✔

399

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

5,473,050✔

401

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

5,473,050✔

402

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

5,472,920✔

403

      return code;

5,474,610✔

405

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

27,596,210✔

406

      *status = FUNC_DATA_REQUIRED_DATA_LOAD;

27,609,092✔

410

  if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {

584,591,453✔

417

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

584,604,203✔

418

    bool success = true;

85,868,488✔

419

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

85,873,909✔

420

    if (code) {

85,837,788✔

423

      QUERY_CHECK_CODE(code, lino, _end);

26,318✔

426

    if (success) {

85,864,106✔

427

      size_t size = taosArrayGetSize(pBlock->pDataBlock);

8,824,712✔

428

      bool   keep = false;

8,826,402✔

429

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

8,827,052✔

430

      if (code) {

8,827,702✔

436

      if (!keep) {

8,825,102✔

437

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

3,259,292✔

439

        pCost->filterOutBlocks += 1;

3,259,292✔

440

        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

3,259,292✔

441

        taosMemoryFreeClear(pBlock->pBlockAgg);

3,259,292✔

443

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

3,259,292✔

444

        return TSDB_CODE_SUCCESS;

3,259,292✔

450

  taosMemoryFreeClear(pBlock->pBlockAgg);

581,492,155✔

453

  code = doDynamicPruneDataBlock(pOperator, pBlockInfo, status);

581,351,264✔

454

  if (code) {

581,414,702✔

459

  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

581,414,702✔

460

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

11,571,678✔

462

    pCost->skipBlocks += 1;

11,571,678✔

463

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

11,571,678✔

465

    STableScanInfo* p1 = pOperator->info;

11,571,678✔

466

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

11,571,678✔

467

      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;

2,724,720✔

469

      *status = FUNC_DATA_REQUIRED_FILTEROUT;

8,846,958✔

471

    return TSDB_CODE_SUCCESS;

11,571,678✔

474

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

569,823,658✔

475

  pCost->loadBlocks += 1;

569,890,948✔

477

  SSDataBlock* p = NULL;

569,895,878✔

478

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

569,850,362✔

479

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

569,987,937✔

483

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

569,987,937✔

484

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

444,504,982✔

488

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

543,027,158✔

489

    if (code) {

542,959,933✔

495

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

569,909,656✔

497

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

569,903,258✔

498

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

82,581,338✔

499

    QUERY_CHECK_CODE(code, lino, _end);

82,565,574✔

501

    int64_t st = taosGetTimestampUs();

82,448,076✔

502

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

82,463,030✔

503

    pTableScanInfo->readRecorder.filterTime += el;

82,463,030✔

505

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

82,451,050✔

506

      pCost->filterOutBlocks += 1;

13,790,089✔

507

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

13,779,834✔

510

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

68,652,498✔

514

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

569,802,172✔

515

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

569,798,945✔

516

    setOperatorCompleted(pOperator);

3,691,571✔

519

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

569,799,063✔

521

_end:

570,035,792✔

522

  if (code != TSDB_CODE_SUCCESS) {

569,945,638✔

523

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

127,400✔

525

  return code;

569,871,062✔

528

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

65,922✔

529

  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

65,922✔

531

  switchCtxOrder(pCtx, numOfOutput);

65,922✔

532

  pTableScanInfo->cond.order = TSDB_ORDER_DESC;

65,922✔

535

65,922✔

603

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

331,104,245✔

605

  int32_t          code = TSDB_CODE_SUCCESS;

331,104,245✔

606

  int32_t          lino = 0;

331,104,245✔

607

  bool             freeReader = false;

331,104,245✔

608

  LRUHandle*       h = NULL;

331,104,245✔

609

  STableCachedVal  val = {0};

331,104,245✔

610

  SMetaReader      mr = {0};

331,104,245✔

611

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

331,095,509✔

612

  int32_t          insertRet = TAOS_LRU_STATUS_OK;

331,140,145✔

613

  STableCachedVal* pVal = NULL;

331,140,145✔

616

  if (numOfExpr <= 0) {

331,088,798✔

617

    return TSDB_CODE_SUCCESS;

168,082✔

623

  int32_t backupRows = pBlock->info.rows;

330,920,716✔

624

  pBlock->info.rows = rows;

330,989,704✔

630

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

331,017,320✔

631

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

331,015,817✔

632

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

331,039,005✔

633

    if (code != TSDB_CODE_SUCCESS) {

330,864,559✔

648

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

330,864,559✔

650

    val.pName = mr.me.name;

331,007,094✔

651

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

331,007,094✔

653

    freeReader = true;

331,007,094✔

695

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

1,094,642,894✔

696

    const SExprInfo* pExpr1 = &pExpr[j];

763,587,359✔

697

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

763,572,290✔

699

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

763,520,049✔

700

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

763,755,784✔

702

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

763,720,672✔

705

    if (fmIsScanPseudoColumnFunc(functionId)) {

763,770,124✔

706

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

90,905,181✔

707

      if (fType == FUNCTION_TYPE_TBNAME) {

90,913,221✔

708

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

90,913,221✔

709

        QUERY_CHECK_CODE(code, lino, _end);

90,891,647✔

718

      STagVal tagVal = {0};

672,667,030✔

719

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

672,699,974✔

720

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

672,851,244✔

722

      char* data = NULL;

672,829,889✔

723

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

672,829,889✔

724

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

666,674,085✔

726

        data = (char*)p;

6,195,658✔

729

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

672,741,063✔

730

      if (isNullVal) {

672,747,690✔

731

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

5,712,260✔

732

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

667,035,430✔

733

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

666,603,862✔

734

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

666,627,482✔

735

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

274,876,760✔

736

          if (tmp != NULL) {

274,925,647✔

737

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

274,925,647✔

738

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

274,911,559✔

740

            taosMemoryFree(tmp);

274,958,374✔

742

          taosMemoryFree(data);

274,954,602✔

744

        QUERY_CHECK_CODE(code, lino, _end);

666,552,471✔

746

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

2,990,175✔

747

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

2,506,778✔

748

          QUERY_CHECK_CODE(code, lino, _end);

2,507,510✔

755

  pBlock->info.rows = backupRows;

331,055,535✔

757

_end:

330,993,977✔

759

  if (NULL != pVal) {

331,027,807✔

768

  if (freeReader) {

331,027,807✔

769

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

330,981,176✔

771

  if (code != TSDB_CODE_SUCCESS) {

331,013,440✔

774

  return code;

330,982,684✔

777

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

90,897,182✔

779

  int32_t                     code = TSDB_CODE_SUCCESS;

90,897,182✔

780

  int32_t                     lino = 0;

90,897,182✔

781

  struct SScalarFuncExecFuncs fpSet = {0};

90,897,182✔

782

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

90,903,763✔

783

  QUERY_CHECK_CODE(code, lino, _end);

90,907,027✔

785

  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;

90,907,027✔

786

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

90,907,027✔

787

  STR_TO_VARSTR(buf, name)

90,910,442✔

789

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

90,896,795✔

791

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

90,888,439✔

792

  QUERY_CHECK_CODE(code, lino, _end);

90,909,077✔

794

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

90,909,077✔

795

  QUERY_CHECK_CODE(code, lino, _end);

90,906,104✔

797

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

90,906,104✔

798

  SScalarParam param = {.columnData = pColInfoData};

90,908,423✔

800

  if (fpSet.process != NULL) {

90,898,669✔

801

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

90,898,669✔

802

    QUERY_CHECK_CODE(code, lino, _end);

90,915,152✔

807

  colDataDestroy(&infoData);

90,915,152✔

809

_end:

90,887,364✔

810

  if (code != TSDB_CODE_SUCCESS) {

90,906,334✔

813

  return code;

90,906,334✔

882

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

177,762,913✔

883

  int32_t code = TSDB_CODE_SUCCESS;

177,762,913✔

884

  int32_t lino = 0;

177,762,913✔

885

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

177,762,913✔

886

  QUERY_CHECK_CODE(code, lino, _end);

177,753,645✔

888

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

177,753,645✔

889

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

177,802,062✔

890

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

177,604,243✔

891

  taosMemoryFreeClear(pInfo->pResBlock->pBlockAgg);

177,765,171✔

893

  if (!pInfo->needCountEmptyTable) {

177,609,329✔

894

    pInfo->countState = TABLE_COUNT_STATE_END;

158,505,145✔

896

    pInfo->countState = TABLE_COUNT_STATE_SCAN;

19,244,201✔

899

_end:

177,708,166✔

900

  if (code != TSDB_CODE_SUCCESS) {

177,708,166✔

903

  return code;

177,736,831✔

906

void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {

502,696,572✔

907

  if (pInfo->countState == TABLE_COUNT_STATE_END) {

502,696,572✔

908

    return;

433,183,497✔

910

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

69,558,283✔

911

    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;

218,768✔

913

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

69,335,358✔

914

    if (code != TSDB_CODE_SUCCESS) {

69,338,245✔

915

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

5,999,888✔

920

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

46,136,910✔

922

  blockDataEmpty(pBlock);

46,136,910✔

923

  pBlock->info.rows = 1;

46,138,988✔

924

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

46,138,988✔

925

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

46,138,485✔

926

  pBlock->info.blankFill = true;

46,140,497✔

929

  int32_t col_num = blockDataGetNumOfCols(pBlock);

46,138,419✔

930

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

190,784,091✔

931

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

144,644,163✔

936

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

46,139,928✔

938

  return pBlock;

46,138,485✔

941

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

44,333,113✔

942

  STableScanInfo* pTableScanInfo = pOperator->info;

44,333,113✔

944

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

44,335,125✔

946

  pOperator->resultInfo.totalRows++;

44,330,598✔

947

  return pBlock;

44,333,616✔

954

static int32_t prepareTableScan(SOperatorInfo* pOperator) {

660,229,766✔

955

  int32_t code = TSDB_CODE_SUCCESS;

660,229,766✔

956

  int32_t lino = 0;

660,229,766✔

958

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

660,229,766✔

959

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

660,331,842✔

960

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

660,475,956✔

961

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

31,174,043✔

962

         NOTIFY_TYPE_SCAN_PARAM &&

2,983,327✔

963

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

2,983,327✔

964

    STableScanOperatorParam* pNotify =

2,983,327✔

967

    STableScanInfo* pTableScanInfo = pOperator->info;

2,983,327✔

968

    SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

2,983,327✔

969

    SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

2,983,327✔

970

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

2,983,327✔

972

    QUERY_CHECK_CODE(code, lino, _end);

2,983,327✔

974

    if (!pGetParam->reUse) {

2,983,327✔

975

      freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

91,286✔

981

      pGetParam->reUse = false;

2,892,041✔

986

    pOperator->pOperatorGetParam = NULL;

2,983,327✔

989

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

660,470,590✔

990

_end:

82,194,006✔

991

  if (code != TSDB_CODE_SUCCESS) {

660,469,245✔

995

  return code;

660,469,232✔

998

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

621,829,804✔

999

  int32_t         code = TSDB_CODE_SUCCESS;

621,829,804✔

1000

  int32_t         lino = 0;

621,829,804✔

1001

  STableScanInfo* pTableScanInfo = pOperator->info;

621,829,804✔

1002

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

621,950,608✔

1003

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

621,924,359✔

1004

  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;

621,823,642✔

1005

  bool            hasNext = false;

621,910,956✔

1006

  int64_t         st = taosGetTimestampUs();

621,878,818✔

1008

  QRY_PARAM_CHECK(ppRes);

621,878,818✔

1009

  pBlock->info.dataLoad = false;

621,900,815✔

1011

  while (true) {

21,662,630✔

1012

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

643,599,498✔

1013

    if (code != TSDB_CODE_SUCCESS) {

643,382,186✔

1015

      QUERY_CHECK_CODE(code, lino, _end);

3,082✔

1018

    if (!hasNext) {

643,378,435✔

1019

      break;

160,484,569✔

1022

    if (isTaskKilled(pTaskInfo)) {

482,893,866✔

1023

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

12,903✔

1024

      code = pTaskInfo->code;

12,903✔

1025

      QUERY_CHECK_CODE(code, lino, _end);

12,903✔

1028

    if (pOperator->status == OP_EXEC_DONE) {

483,130,388✔

1029

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

147,560✔

1030

      break;

147,560✔

1034

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

482,837,004✔

1035

    if (!processThisBlock) {

482,886,695✔

1039

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

482,886,695✔

1040

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

482,968,719✔

1043

    uint32_t status = 0;

483,016,617✔

1044

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

483,028,396✔

1045

    QUERY_CHECK_CODE(code, lino, _end);

482,888,110✔

1047

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

482,777,350✔

1048

      break;

2,724,720✔

1052

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

480,052,630✔

1053

      continue;

21,662,630✔

1056

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

458,439,870✔

1057

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

458,519,536✔

1059

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

458,488,249✔

1060

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

458,484,713✔

1062

    (*ppRes) = pBlock;

458,405,555✔

1063

    return code;

458,365,725✔

1066

_end:

163,484,061✔

1067

  if (code != TSDB_CODE_SUCCESS) {

163,492,410✔

1068

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

131,506✔

1069

    pTaskInfo->code = code;

131,506✔

1070

    T_LONG_JMP(pTaskInfo->env, code);

123,663✔

1072

  return code;

163,360,904✔

1075

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

668,849,541✔

1076

  int32_t         code = TSDB_CODE_SUCCESS;

668,849,541✔

1077

  int32_t         lino = 0;

668,849,541✔

1078

  STableScanInfo* pTableScanInfo = pOperator->info;

668,849,541✔

1079

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

668,888,220✔

1080

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

668,780,859✔

1082

  QRY_PARAM_CHECK(pBlock);

668,799,146✔

1085

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

668,851,399✔

1086

    return code;

2,735,109✔

1090

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

813,354,189✔

1091

    SSDataBlock* p = NULL;

569,488,080✔

1092

    code = doTableScanImplNext(pOperator, &p);

569,445,782✔

1093

    QUERY_CHECK_CODE(code, lino, _end);

569,306,724✔

1095

    if (p != NULL) {

569,306,724✔

1096

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

421,997,950✔

1097

      *pBlock = p;

421,924,193✔

1098

      return code;

421,899,740✔

1101

    pTableScanInfo->scanTimes += 1;

147,308,774✔

1102

    taosHashClear(pTableScanInfo->pIgnoreTables);

147,317,357✔

1104

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

147,285,218✔

1105

      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

84,013✔

1106

      pTableScanInfo->base.scanFlag = MAIN_SCAN;

84,013✔

1107

      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

84,013✔

1108

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

84,013✔

1111

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

84,013✔

1112

      QUERY_CHECK_CODE(code, lino, _end);

72,909✔

1116

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

243,950,707✔

1117

  if (pTableScanInfo->scanTimes < total) {

243,938,898✔

1118

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

52,450,353✔

1119

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

65,922✔

1120

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

65,922✔

1121

      QUERY_CHECK_CODE(code, lino, _end);

65,922✔

1122

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

65,922✔

1125

    while (pTableScanInfo->scanTimes < total) {

68,508,570✔

1126

      SSDataBlock* p = NULL;

52,451,504✔

1127

      code = doTableScanImplNext(pOperator, &p);

52,434,253✔

1128

      QUERY_CHECK_CODE(code, lino, _end);

52,396,577✔

1130

      if (p != NULL) {

52,396,577✔

1131

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

36,349,254✔

1132

        *pBlock = p;

36,327,137✔

1133

        return code;

36,317,962✔

1136

      pTableScanInfo->scanTimes += 1;

16,047,323✔

1137

      taosHashClear(pTableScanInfo->pIgnoreTables);

16,053,216✔

1139

      if (pTableScanInfo->scanTimes < total) {

16,049,669✔

1150

  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {

207,524,954✔

1151

    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;

62,884,272✔

1152

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

62,882,966✔

1153

      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {

111,691✔

1154

        pTableScanInfo->countState = TABLE_COUNT_STATE_PROCESSED;

1,980✔

1156

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

1,980✔

1158

        if (NULL == pStart) {

1,980✔

1162

        *pBlock = getBlockForEmptyTable(pOperator, pStart);

1,980✔

1163

        return code;

1,980✔

1166

      int32_t numOfTables = 0;

62,769,673✔

1167

      code = tableListGetSize(pTableListInfo, &numOfTables);

62,769,805✔

1168

      QUERY_CHECK_CODE(code, lino, _end);

62,772,581✔

1170

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

62,772,581✔

1172

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

60,526,713✔

1173

        if (pIte != NULL) {

60,524,529✔

1174

          size_t        keySize = 0;

44,332,642✔

1175

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

44,333,145✔

1176

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

44,333,145✔

1177

          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);

44,331,702✔

1178

          markGroupProcessed(pTableScanInfo, *pGroupId);

44,332,642✔

1179

          *pBlock = getBlockForEmptyTable(pOperator, &info);

44,331,133✔

1181

          return code;

44,332,139✔

1185

    pTableScanInfo->countState = TABLE_COUNT_STATE_END;

18,546,462✔

1188

_end:

144,673,471✔

1189

  if (code != TSDB_CODE_SUCCESS) {

163,214,946✔

1194

  return code;

163,214,946✔

1197

static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {

958,306✔

1198

  STableScanInfo*          pInfo = pOperator->info;

958,306✔

1199

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

958,306✔

1200

  int32_t                  code = 0;

958,306✔

1201

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

958,306✔

1202

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

958,306✔

1203

  int32_t                  num = taosArrayGetSize(pParam->pUidList);

958,306✔

1204

  if (num <= 0) {

958,306✔

1209

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

958,306✔

1213

  if (pParam->tableSeq) {

958,306✔

1214

    pListInfo->oneTableForEachGroup = true;

697,062✔

1215

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

697,062✔

1221

    pListInfo->oneTableForEachGroup = false;

261,244✔

1222

    pListInfo->numOfOuputGroups = 1;

261,244✔

1224

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

958,306✔

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};

958,306✔

1232

  int32_t       tableIdx = 0;

958,306✔

1233

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

8,943,932✔

1234

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

7,985,626✔

1235

    if (!pUid) {

7,984,614✔

1240

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

7,984,614✔

1241

      if (TSDB_CODE_DUP_KEY == terrno) {

6,683,333✔

1242

        continue;

6,683,333✔

1247

    info.uid = *pUid;

1,302,293✔

1248

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

1,302,293✔

1249

    if (p == NULL) {

1,302,293✔

1253

    tableIdx++;

1,302,293✔

1254

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

1,302,293✔

1257

  return code;

958,306✔

1260

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

163,975,334✔

1262

  const char* idStr = GET_TASKID(pTaskInfo);

163,975,334✔

1263

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

164,006,009✔

1264

  if (code) {

163,940,370✔

1269

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

163,940,370✔

1274

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

327,888,596✔

1275

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

163,884,127✔

1276

  if (code) {

163,873,019✔

1277

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

2,170✔

1280

  return code;

163,828,863✔

1283

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

294,284,386✔

1284

  SColIdPair* node1 = (SColIdPair*)elem1;

294,284,386✔

1285

  SColIdPair* node2 = (SColIdPair*)elem2;

294,284,386✔

1287

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

294,284,386✔

1288

    return 0;

107,780✔

1291

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

294,188,915✔

1294

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

395,955,629✔

1295

  SColIdSlotIdPair* node1 = (SColIdSlotIdPair*)elem1;

395,955,629✔

1296

  SColIdSlotIdPair* node2 = (SColIdSlotIdPair*)elem2;

395,955,629✔

1298

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

395,955,629✔

1299

    return 0;

164,636✔

1302

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

395,786,948✔

1305

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

704,894,069✔

1306

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

704,894,069✔

1307

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

704,894,069✔

1309

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

704,894,069✔

1313

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

704,894,069✔

1316

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

92,919,262✔

1317

  if (pList == NULL) {

92,919,262✔

1321

  int32_t left = 0;

92,919,262✔

1322

  int32_t right = taosArrayGetSize(pList) - 1;

92,919,262✔

1323

  while (left <= right) {

288,130,091✔

1324

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

288,130,091✔

1325

    SColMatchItem* pItem = taosArrayGet(pList, mid);

288,130,091✔

1326

    if (pItem->colId == colId) {

288,130,091✔

1328

      while (mid > 0) {

92,919,262✔

1329

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

69,685,713✔

1330

        if (pPrev->colId != colId) {

69,685,713✔

1331

          break;

69,685,713✔

1335

      return taosArrayGet(pList, mid);

92,919,262✔

1338

    if (pItem->colId < colId) {

195,210,829✔

1339

      left = mid + 1;

91,699,306✔

1341

      right = mid - 1;

103,511,523✔

1348

static bool isNewScanParam(STableScanOperatorParam* pParam) {

39,739,755✔

1349

  return pParam->isNewParam;

39,739,755✔

1352

static int32_t createVTableScanInfoFromBatchParam(SOperatorInfo* pOperator) {

17,150,533✔

1353

  int32_t                  code = 0;

17,150,533✔

1354

  int32_t                  lino = 0;

17,150,533✔

1355

  STableScanInfo*          pInfo = pOperator->info;

17,150,533✔

1356

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

17,155,833✔

1357

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

17,139,426✔

1358

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

17,146,053✔

1359

  SMetaReader              orgTable = {0};

17,159,676✔

1360

  SMetaReader              superTable = {0};

17,167,431✔

1361

  SSchemaWrapper*          schema = NULL;

17,155,643✔

1362

  SArray*                  pColArray = NULL;

17,155,643✔

1363

  SArray*                  pBlockColArray = NULL;

17,155,643✔

1364

  int32_t                  num = 0;

17,155,643✔

1365

  STableKeyInfo*           pList = NULL;

17,156,713✔

1369

  cleanupQueryTableDataCond(&pInfo->base.cond);

17,156,333✔

1371

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

17,161,003✔

1372

  if (pGetParam &&

17,165,543✔

1373

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

3,991,296✔

1374

    STableScanOperatorParam* pParam =

3,995,516✔

1376

    pInfo->pBatchColMap = pParam->pBatchTbInfo;

4,003,721✔

1377

    pParam->pBatchTbInfo = NULL;

3,996,651✔

1378

    pInfo->cachedTimeWindow = pParam->window;

3,989,531✔

1379

    pInfo->currentBatchIdx = 0;

3,998,361✔

1380

    pInfo->lastBatchIdx = -1;

3,984,736✔

1381

    pInfo->cachedTagList = pParam->pTagList;

3,989,651✔

1382

    pParam->pTagList = NULL;

4,001,196✔

1383

    pInfo->cachedGroupId = pParam->groupid;

3,991,416✔

1385

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

4,000,316✔

1387

    taosHashClear(pInfo->pIgnoreTables);

4,000,316✔

1389

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

4,000,186✔

1390

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

3,804,369✔

1391

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

47,807,923✔

1392

        SqlFunctionCtx ctx = pSup->pCtx[i];

44,020,394✔

1393

        ctx.skipDynDataCheck = false;

44,003,429✔

1398

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

17,167,253✔

1399

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

17,169,511✔

1401

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

17,169,511✔

1402

    isNewTable = false;

10,439,248✔

1404

    isNewTable = true;

6,722,320✔

1405

    pInfo->lastBatchIdx = pInfo->currentBatchIdx;

6,722,320✔

1408

  if (isNewTable) {

17,167,748✔

1409

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

6,728,500✔

1410

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

6,733,175✔

1411

    pAPI->metaReaderFn.readerReleaseLock(&orgTable);

6,702,828✔

1412

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

6,710,705✔

1413

    QUERY_CHECK_CODE(code, lino, _return);

6,716,210✔

1414

    switch (orgTable.me.type) {

6,716,210✔

1415

      case TSDB_CHILD_TABLE:

279,400✔

1416

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

279,400✔

1417

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

279,400✔

1418

        pAPI->metaReaderFn.readerReleaseLock(&superTable);

279,400✔

1419

        QUERY_CHECK_CODE(code, lino, _return);

279,400✔

1420

        schema = &superTable.me.stbEntry.schemaRow;

279,400✔

1421

        break;

279,400✔

1422

      case TSDB_NORMAL_TABLE:

6,395,358✔

1423

        schema = &orgTable.me.ntbEntry.schemaRow;

6,395,358✔

1424

        break;

6,395,358✔

1425

      default:

41,452✔

1426

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

41,452✔

1431

    pListInfo->oneTableForEachGroup = true;

6,674,758✔

1432

    taosHashClear(pListInfo->map);

6,715,585✔

1433

    taosArrayClear(pListInfo->pTableList);

6,735,760✔

1435

    uint64_t      pUid = orgTable.me.uid;

6,735,445✔

1436

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

6,735,445✔

1437

    int32_t       tableIdx = 0;

6,735,130✔

1438

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

6,735,760✔

1439

    QUERY_CHECK_CODE(code, lino, _return);

6,735,760✔

1440

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

13,467,865✔

1441

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

6,732,105✔

1443

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

6,732,105✔

1444

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

6,728,685✔

1445

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

6,728,685✔

1446

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

6,727,043✔

1449

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

6,727,043✔

1450

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

6,733,870✔

1452

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

25,716,489✔

1453

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

18,981,609✔

1454

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

151,234,772✔

1455

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

150,945,520✔

1456

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

18,982,489✔

1457

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

18,990,054✔

1458

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

18,990,054✔

1460

          break;

18,982,737✔

1465

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

32,451,864✔

1466

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

25,712,310✔

1467

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

154,092,167✔

1468

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

153,942,964✔

1469

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

153,961,632✔

1470

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

25,711,814✔

1471

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

25,720,894✔

1472

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

25,720,894✔

1474

          break;

25,705,177✔

1479

    taosArraySort(pColArray, compareColIdPair);

6,733,937✔

1480

    taosArraySort(pBlockColArray, compareColIdSlotIdPair);

6,729,705✔

1481

    if (pInfo->pBlockColMap) {

6,731,220✔

1482

      taosArrayDestroy(pInfo->pBlockColMap);

4,800,847✔

1483

      pInfo->pBlockColMap = NULL;

4,797,187✔

1485

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

6,727,120✔

1486

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

6,728,760✔

1489

    taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

6,721,238✔

1490

    taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

6,724,591✔

1492

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

6,733,995✔

1493

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

6,014,151✔

1494

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

74,634,640✔

1495

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

68,638,399✔

1496

        bool           needScan = false;

68,630,879✔

1497

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

138,188,729✔

1498

          SFunctParam pParam = ctx->param[j];

69,498,260✔

1499

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

333,700,189✔

1500

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

280,394,097✔

1501

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

281,110,392✔

1502

              needScan |= true;

17,315,682✔

1503

              break;

17,315,682✔

1507

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

68,628,814✔

1508

        if (!needScan) {

68,660,331✔

1509

          ctx->skipDynDataCheck = true;

51,355,684✔

1511

          ctx->skipDynDataCheck = false;

17,304,647✔

1518

    pColArray = pInfo->lastColArray;

10,439,248✔

1519

    pBlockColArray = pInfo->lastBlockColArray;

10,439,248✔

1522

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

17,172,298✔

1523

  QUERY_CHECK_CODE(code, lino, _return);

17,172,231✔

1525

  if (pInfo->pResBlock) {

17,172,231✔

1526

    blockDataDestroy(pInfo->pResBlock);

13,164,477✔

1527

    pInfo->pResBlock = NULL;

13,164,477✔

1530

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

17,172,231✔

1538

  if (isNewTable) {

17,163,021✔

1539

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

6,724,088✔

1540

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

6,011,181✔

1541

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

5,988,262✔

1543

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

718,392✔

1544

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

709,752✔

1547

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

10,438,933✔

1548

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

1,369,005✔

1550

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

9,070,243✔

1554

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

17,156,781✔

1555

  pInfo->currentGroupId = 0;

17,149,533✔

1556

  pInfo->ignoreTag = true;

17,145,106✔

1558

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

17,102,823✔

1559

  QUERY_CHECK_CODE(code, lino, _return);

17,161,071✔

1561

  if (isNewTable) {

17,161,071✔

1562

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

6,722,453✔

1563

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

4,797,692✔

1565

    pInfo->base.dataReader = NULL;

6,702,067✔

1567

    taosRLockLatch(&pTaskInfo->lock);

6,706,982✔

1568

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

6,735,445✔

1569

    taosRUnLockLatch(&pTaskInfo->lock);

6,687,833✔

1570

    QUERY_CHECK_CODE(code, lino, _return);

6,715,884✔

1572

    pInfo->newReader = true;

6,715,884✔

1573

    if (pInfo->lastColArray) {

6,716,771✔

1574

      taosArrayDestroy(pInfo->lastColArray);

2,726,049✔

1575

      pInfo->lastColArray = NULL;

2,724,724✔

1577

    if (pInfo->lastBlockColArray) {

6,708,763✔

1578

      taosArrayDestroy(pInfo->lastBlockColArray);

2,725,229✔

1579

      pInfo->lastBlockColArray = NULL;

2,724,599✔

1581

    pInfo->lastColArray = pColArray;

6,680,148✔

1582

    pInfo->lastBlockColArray = pBlockColArray;

6,691,705✔

1584

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

10,438,618✔

1585

    QUERY_CHECK_CODE(code, lino, _return);

10,438,363✔

1587

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

10,438,363✔

1588

    pInfo->newReader = false;

10,438,933✔

1589

    pInfo->scanTimes = 0;

10,438,933✔

1592

  pOperator->status = OP_OPENED;

17,145,524✔

1594

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

17,148,609✔

1598

  pInfo->currentGroupId = -1;

17,152,772✔

1599

_return:

17,071,871✔

1600

  if (code) {

17,109,124✔

1603

  pAPI->metaReaderFn.clearReader(&superTable);

17,109,124✔

1604

  pAPI->metaReaderFn.clearReader(&orgTable);

17,133,071✔

1605

  return code;

17,144,914✔

1608

static int32_t createVTableScanInfoFromParam(SOperatorInfo* pOperator) {

23,233,549✔

1609

  int32_t                  code = 0;

23,233,549✔

1610

  int32_t                  lino = 0;

23,233,549✔

1611

  STableScanInfo*          pInfo = pOperator->info;

23,233,549✔

1612

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

23,233,549✔

1613

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

23,233,549✔

1614

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

23,233,549✔

1615

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

23,233,549✔

1616

  SMetaReader              orgTable = {0};

23,233,549✔

1617

  SMetaReader              superTable = {0};

23,233,549✔

1618

  SSchemaWrapper*          schema = NULL;

23,233,549✔

1619

  SArray*                  pColArray = NULL;

23,233,549✔

1620

  SArray*                  pBlockColArray = NULL;

23,233,549✔

1621

  int32_t                  num = 0;

23,233,549✔

1622

  STableKeyInfo*           pList = NULL;

23,233,549✔

1624

  SArray*                  pMatchList = NULL;

23,233,549✔

1626

  cleanupQueryTableDataCond(&pInfo->base.cond);

23,233,549✔

1628

  pOrgTbInfo = pParam->pOrgTbInfo;

23,233,549✔

1630

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

23,233,549✔

1632

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

23,233,549✔

1633

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

23,233,549✔

1634

  pAPI->metaReaderFn.readerReleaseLock(&orgTable);

23,233,549✔

1635

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

23,233,549✔

1636

  QUERY_CHECK_CODE(code, lino, _return);

23,233,549✔

1637

  switch (orgTable.me.type) {

23,233,549✔

1638

    case TSDB_CHILD_TABLE:

6,343,728✔

1639

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

6,343,728✔

1640

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

6,343,728✔

1641

      pAPI->metaReaderFn.readerReleaseLock(&superTable);

6,343,728✔

1642

      QUERY_CHECK_CODE(code, lino, _return);

6,343,728✔

1643

      schema = &superTable.me.stbEntry.schemaRow;

6,343,728✔

1644

      break;

6,343,728✔

1645

    case TSDB_NORMAL_TABLE:

16,889,821✔

1646

      schema = &orgTable.me.ntbEntry.schemaRow;

16,889,821✔

1647

      break;

16,889,821✔

1654

  pListInfo->oneTableForEachGroup = true;

23,233,549✔

1655

  taosHashClear(pListInfo->map);

23,233,549✔

1656

  taosArrayClear(pListInfo->pTableList);

23,233,549✔

1658

  uint64_t      pUid = orgTable.me.uid;

23,233,549✔

1659

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

23,233,549✔

1660

  int32_t       tableIdx = 0;

23,233,549✔

1661

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

23,233,549✔

1662

  QUERY_CHECK_CODE(code, lino, _return);

23,233,549✔

1663

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

46,467,098✔

1664

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

23,233,549✔

1666

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

23,233,549✔

1667

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

23,233,549✔

1668

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

23,233,549✔

1669

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

23,233,549✔

1670

  if (pInfo->pBlockColMap) {

23,233,549✔

1671

    taosArrayDestroy(pInfo->pBlockColMap);

20,901,285✔

1672

    pInfo->pBlockColMap = NULL;

20,901,285✔

1674

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

23,233,549✔

1675

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

23,233,549✔

1678

  SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID,

23,233,549✔

1682

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

23,233,549✔

1684

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

92,919,262✔

1685

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

69,685,713✔

1686

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

587,957,996✔

1687

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

587,957,996✔

1688

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

69,685,713✔

1689

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

69,685,713✔

1690

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

69,685,713✔

1691

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

69,685,713✔

1692

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

69,685,713✔

1693

        break;

69,685,713✔

1698

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

23,233,549✔

1699

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

23,233,549✔

1700

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

23,233,549✔

1701

    taosArraySort(pMatchList, compareColMatchItemByColId);

23,233,549✔

1705

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

116,152,186✔

1706

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

92,919,262✔

1707

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

92,919,262✔

1708

    if (pItem) {

92,919,262✔

1709

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

92,919,262✔

1710

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

625✔

1712

        code = TSDB_CODE_VTABLE_COLUMN_TYPE_MISMATCH;

625✔

1713

        goto _return;

625✔

1715

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

92,918,637✔

1716

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

92,918,637✔

1717

      if (i > 0) {

92,918,637✔

1718

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

139,370,176✔

1723

  taosArraySort(pColArray, compareColIdPair);

23,232,924✔

1724

  taosArraySort(pBlockColArray, compareColIdSlotIdPair);

23,232,924✔

1725

  taosArraySort(pInfo->pBlockColMap, compareColIdSlotIdPair);

23,232,924✔

1727

  taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

23,232,924✔

1728

  taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

23,232,924✔

1730

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

23,232,924✔

1731

  QUERY_CHECK_CODE(code, lino, _return);

23,232,924✔

1733

  if (pInfo->pResBlock) {

23,232,924✔

1734

    blockDataDestroy(pInfo->pResBlock);

16,509,079✔

1735

    pInfo->pResBlock = NULL;

16,509,079✔

1738

  if (isNewScanParam(pParam)) {

23,232,924✔

1739

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

6,731,126✔

1740

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

6,731,126✔

1741

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

6,731,126✔

1747

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

16,501,798✔

1749

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

23,232,924✔

1750

  pInfo->currentGroupId = 0;

23,232,924✔

1751

  pInfo->ignoreTag = true;

23,232,924✔

1753

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

23,232,924✔

1754

  QUERY_CHECK_CODE(code, lino, _return);

23,232,924✔

1756

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

23,232,924✔

1757

  if (reader) {

23,232,924✔

1758

    if (isNewScanParam(pParam)) {

16,506,831✔

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;

16,501,798✔

1775

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

16,501,798✔

1776

      QUERY_CHECK_CODE(code, lino, _return);

16,501,798✔

1778

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

16,501,798✔

1779

      pInfo->newReader = false;

16,501,798✔

1780

      pInfo->scanTimes = 0;

16,501,798✔

1783

    pInfo->base.dataReader = NULL;

6,726,093✔

1785

    taosRLockLatch(&pTaskInfo->lock);

6,726,093✔

1786

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

6,726,093✔

1787

    taosRUnLockLatch(&pTaskInfo->lock);

6,726,093✔

1788

    QUERY_CHECK_CODE(code, lino, _return);

6,726,093✔

1790

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

6,726,093✔

1791

    QUERY_CHECK_CODE(code, lino, _return);

6,726,093✔

1792

    pInfo->newReader = true;

6,726,093✔

1794

  pOperator->status = OP_OPENED;

23,232,924✔

1796

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

23,232,924✔

1800

  pInfo->currentGroupId = -1;

23,232,924✔

1801

_return:

23,233,549✔

1802

  if (code) {

23,233,549✔

1803

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

625✔

1805

  taosArrayDestroy(pMatchList);

23,233,549✔

1806

  taosArrayDestroy(pColArray);

23,233,549✔

1807

  taosArrayDestroy(pBlockColArray);

23,233,549✔

1808

  pAPI->metaReaderFn.clearReader(&superTable);

23,233,549✔

1809

  pAPI->metaReaderFn.clearReader(&orgTable);

23,233,549✔

1810

  return code;

23,233,549✔

1813

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

160,478,785✔

1814

  int32_t         code = TSDB_CODE_SUCCESS;

160,478,785✔

1815

  int32_t         lino = 0;

160,478,785✔

1816

  STableScanInfo* pInfo = pOperator->info;

160,478,785✔

1817

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

160,477,293✔

1818

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

160,451,391✔

1819

  int32_t         numOfTables = 0;

160,457,897✔

1821

  QRY_PARAM_CHECK(pResult);

160,433,001✔

1823

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

160,461,796✔

1824

  QUERY_CHECK_CODE(code, lino, _end);

160,491,634✔

1826

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

160,491,634✔

1827

    setOperatorCompleted(pOperator);

146,672,756✔

1828

    if (pOperator->dynamicTask) {

146,676,506✔

1829

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

956,092✔

1830

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

956,084✔

1832

    return code;

146,678,409✔

1836

  pOperator->status = OP_OPENED;

13,733,604✔

1837

  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);

13,785,654✔

1839

  int32_t        num = 0;

13,772,440✔

1840

  STableKeyInfo* pList = NULL;

13,774,015✔

1841

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

13,771,055✔

1842

  QUERY_CHECK_CODE(code, lino, _end);

13,795,165✔

1844

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

13,795,165✔

1845

  QUERY_CHECK_CODE(code, lino, _end);

13,788,934✔

1847

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

13,788,934✔

1848

  QUERY_CHECK_CODE(code, lino, _end);

13,781,251✔

1849

  pInfo->scanTimes = 0;

13,781,251✔

1851

  code = doGroupedTableScan(pOperator, pResult);

13,784,656✔

1852

  QUERY_CHECK_CODE(code, lino, _end);

13,815,087✔

1854

  if (*pResult != NULL) {

13,815,087✔

1855

    if (pOperator->dynamicTask) {

11,169,663✔

1856

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

10,921,363✔

1858

    return code;

11,169,978✔

1861

_end:

2,645,929✔

1862

  if (code != TSDB_CODE_SUCCESS) {

2,645,547✔

1866

  return code;

2,648,394✔

1869

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

616,934,415✔

1870

  int32_t         code = TSDB_CODE_SUCCESS;

616,934,415✔

1871

  int32_t         lino = 0;

616,934,415✔

1872

  STableScanInfo* pInfo = pOperator->info;

616,934,415✔

1873

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

616,939,477✔

1874

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

616,936,574✔

1875

  int32_t         num = 0;

616,939,611✔

1876

  STableKeyInfo*  pList = NULL;

616,930,621✔

1877

  SSDataBlock*    pResult = NULL;

616,870,320✔

1878

  const char*     idStr = GET_TASKID(pTaskInfo);

616,851,944✔

1880

  QRY_PARAM_CHECK(pResBlock);

616,887,796✔

1882

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

616,901,714✔

1883

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

150,486,450✔

1888

    taosRLockLatch(&pTaskInfo->lock);

150,509,644✔

1889

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

150,578,156✔

1890

    taosRUnLockLatch(&pTaskInfo->lock);

150,450,802✔

1891

    QUERY_CHECK_CODE(code, lino, _end);

150,526,046✔

1893

    if (pInfo->filesetDelimited) {

150,523,876✔

1894

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

261,592✔

1897

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

150,466,756✔

1902

  code = doGroupedTableScan(pOperator, &pResult);

616,901,069✔

1903

  QUERY_CHECK_CODE(code, lino, _end);

616,580,629✔

1905

  if (pResult != NULL) {

616,580,629✔

1906

    if (pOperator->dynamicTask) {

469,831,590✔

1907

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

2,009,687✔

1910

    *pResBlock = pResult;

469,834,524✔

1911

    return code;

469,625,291✔

1915

    code = startNextGroupScan(pOperator, &pResult);

146,826,734✔

1916

    QUERY_CHECK_CODE(code, lino, _end);

147,025,287✔

1918

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

147,025,287✔

1919

      *pResBlock = pResult;

146,958,410✔

1920

      return code;

146,929,210✔

1924

_end:

2,170✔

1925

  if (code != TSDB_CODE_SUCCESS) {

2,170✔

1926

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

2,170✔

1927

    pTaskInfo->code = code;

2,170✔

1930

  return code;

2,170✔

1933

static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {

23,232,924✔

1934

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

23,232,924✔

1937

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

958,306✔

1938

  int32_t                  code = TSDB_CODE_SUCCESS;

958,306✔

1939

  int32_t                  lino = 0;

958,306✔

1940

  STableScanInfo*          pInfo = pOperator->info;

958,306✔

1941

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

958,306✔

1942

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

958,306✔

1943

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

958,306✔

1945

  code = createTableListInfoFromParam(pOperator);

958,306✔

1946

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

958,306✔

1947

  pOperator->pOperatorGetParam = NULL;

958,306✔

1948

  QUERY_CHECK_CODE(code, lino, _end);

958,306✔

1950

  if (pOperator->status == OP_EXEC_DONE) {

958,306✔

1951

    pInfo->currentGroupId = -1;

22,776✔

1952

    pOperator->status = OP_OPENED;

22,776✔

1953

    SSDataBlock* result = NULL;

22,776✔

1956

      code = startNextGroupScan(pOperator, &result);

22,776✔

1957

      QUERY_CHECK_CODE(code, lino, _end);

22,776✔

1959

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

22,776✔

1960

        (*ppRes) = result;

22,776✔

1961

        return code;

22,776✔

1965

_end:

935,530✔

1966

  if (code) {

935,530✔

1969

  return code;

935,530✔

1972

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

16,510,790✔

1973

  int32_t      code = TSDB_CODE_SUCCESS;

16,510,790✔

1974

  int32_t      lino = 0;

16,510,790✔

1975

  bool         found = false;

16,510,790✔

1977

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

16,510,790✔

1978

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

16,510,790✔

1980

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

82,173,518✔

1981

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

65,662,728✔

1982

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

65,662,728✔

1983

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

188,094,985✔

1984

      SColIdSlotIdPair* pColPair = taosArrayGet(pColMap, j);

171,587,155✔

1985

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

171,587,155✔

1986

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

171,587,155✔

1987

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

49,154,898✔

1988

        found = true;

49,154,898✔

1989

        break;

49,154,898✔

1992

    if (!found) {

65,662,728✔

1993

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

16,507,830✔

1994

        pSrcCol->info.slotId = -1;

16,507,830✔

2003

  return code;

16,510,790✔

2009

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

23,233,549✔

2010

  int32_t                  code = TSDB_CODE_SUCCESS;

23,233,549✔

2011

  int32_t                  lino = 0;

23,233,549✔

2012

  STableScanInfo*          pInfo = pOperator->info;

23,233,549✔

2013

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

23,233,549✔

2014

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

23,233,549✔

2015

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

23,233,549✔

2017

  QRY_PARAM_CHECK(ppRes);

23,233,549✔

2018

  qTrace("%s call", __FUNCTION__);

23,233,549✔

2020

  code = createVTableScanInfoFromParam(pOperator);

23,233,549✔

2021

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

23,233,549✔

2022

  pOperator->pOperatorGetParam = NULL;

23,233,549✔

2024

  QUERY_CHECK_CODE(code, lino, _end);

23,233,549✔

2026

  SSDataBlock* result = NULL;

23,232,924✔

2027

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

23,232,924✔

2028

    (*ppRes) = result;

18,116✔

2029

    return code;

18,116✔

2032

  if (pInfo->newReader) {

23,214,808✔

2033

    code = startNextGroupScan(pOperator, &result);

6,731,126✔

2034

    QUERY_CHECK_CODE(code, lino, _end);

6,731,126✔

2036

    code = doGroupedTableScan(pOperator, &result);

16,483,682✔

2037

    QUERY_CHECK_CODE(code, lino, _end);

16,483,682✔

2040

  if (result) {

23,214,808✔

2041

    SSDataBlock* res = NULL;

16,510,790✔

2042

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

16,510,790✔

2043

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

16,510,790✔

2044

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

16,510,790✔

2046

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

16,510,790✔

2047

    QUERY_CHECK_CODE(code, lino, _end);

16,510,790✔

2053

    (*ppRes) = result;

16,510,790✔

2055

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

6,704,018✔

2056

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

6,704,018✔

2058

    blockDataDestroy(pInfo->pResBlock);

6,704,018✔

2059

    pInfo->pResBlock = NULL;

6,704,018✔

2061

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

6,704,018✔

2062

    if (reader) {

6,704,018✔

2063

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

6,704,018✔

2064

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

6,704,018✔

2065

        pInfo->base.dataReader = NULL;

6,704,018✔

2067

      if (pAPI->tsdReader.tsdReaderClose) {

6,704,018✔

2068

        pAPI->tsdReader.tsdReaderClose(*reader);

6,704,018✔

2072

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

6,704,018✔

2073

    QUERY_CHECK_CODE(code, lino, _end);

6,704,018✔

2075

_end:

23,215,433✔

2076

  if (code) {

23,215,433✔

2077

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

625✔

2079

  return code;

23,215,433✔

2082

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

2,986,346✔

2083

  int32_t                  code = TSDB_CODE_SUCCESS;

2,986,346✔

2084

  int32_t                  lino = 0;

2,986,346✔

2085

  STableScanInfo*          pInfo = pOperator->info;

2,986,346✔

2086

  int32_t                  index = 0;

2,986,346✔

2087

  char*                    tagVal = NULL;

2,986,346✔

2089

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

20,917,972✔

2090

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

17,931,626✔

2091

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

17,931,626✔

2093

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

17,931,626✔

2094

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

17,931,626✔

2096

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

2,147,483,647✔

2097

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

2,147,483,647✔

2098

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

2,147,483,647✔

2099

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

2,147,483,647✔

2101

        varDataSetLen(tagVal, pTagVal->nData);

2,147,483,647✔

2102

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

2,147,483,647✔

2103

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

2,147,483,647✔

2104

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2105

        taosMemoryFreeClear(tagVal);

2,147,483,647✔

2107

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

2,147,483,647✔

2108

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2111

    index++;

17,931,626✔

2113

  return code;

2,986,346✔

2120

static void clearVstbBatchDynamicTableScanInfo(STableScanInfo* pInfo) {

4,008,451✔

2121

  if (pInfo->pBatchColMap) {

4,008,451✔

2122

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

10,740,746✔

2123

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

6,732,610✔

2124

      taosArrayDestroy(pOrgTbInfo->colMap);

6,731,980✔

2126

    taosArrayDestroy(pInfo->pBatchColMap);

4,006,246✔

2127

    pInfo->pBatchColMap = NULL;

4,007,191✔

2129

  if (pInfo->cachedTagList) {

4,006,876✔

2130

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

12,547,847✔

2131

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

10,690,486✔

2132

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

10,691,431✔

2133

        taosMemoryFreeClear(pTagVal->pData);

4,690,050✔

2136

    taosArrayDestroy(pInfo->cachedTagList);

1,855,786✔

2137

    pInfo->cachedTagList = NULL;

1,856,101✔

2139

  if (pInfo->lastColArray) {

4,007,506✔

2140

    taosArrayDestroy(pInfo->lastColArray);

4,008,766✔

2141

    pInfo->lastColArray = NULL;

4,005,931✔

2143

  if (pInfo->lastBlockColArray) {

4,004,986✔

2144

    taosArrayDestroy(pInfo->lastBlockColArray);

4,006,246✔

2145

    pInfo->lastBlockColArray = NULL;

4,007,506✔

2148

  blockDataDestroy(pInfo->pResBlock);

4,006,561✔

2149

  pInfo->pResBlock = NULL;

4,008,136✔

2150

4,008,136✔

2152

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

14,430,664✔

2153

  int32_t                  code = TSDB_CODE_SUCCESS;

14,430,664✔

2154

  int32_t                  lino = 0;

14,430,664✔

2155

  STableScanInfo*          pInfo = pOperator->info;

14,430,664✔

2156

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

14,431,799✔

2157

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

14,437,354✔

2158

  SSDataBlock*             result = NULL;

14,427,699✔

2160

  QRY_PARAM_CHECK(ppRes);

14,429,534✔

2161

  qTrace("%s call", __FUNCTION__);

14,436,284✔

2163

  while (!result) {

27,601,699✔

2164

    code = createVTableScanInfoFromBatchParam(pOperator);

17,163,081✔

2165

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

17,141,530✔

2166

    pOperator->pOperatorGetParam = NULL;

17,127,542✔

2167

    QUERY_CHECK_CODE(code, lino, _end);

17,137,293✔

2169

    if (pInfo->newReader) {

17,137,293✔

2170

      code = startNextGroupScan(pOperator, &result);

6,696,209✔

2171

      QUERY_CHECK_CODE(code, lino, _end);

6,731,841✔

2173

      code = doGroupedTableScan(pOperator, &result);

10,451,494✔

2174

      QUERY_CHECK_CODE(code, lino, _end);

10,438,933✔

2176

    if (!result) {

17,170,774✔

2177

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

6,732,355✔

2179

      pInfo->currentBatchIdx++;

6,732,355✔

2180

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

6,729,504✔

2182

        break;

4,006,876✔

2187

  if (result) {

14,445,494✔

2188

    SSDataBlock* res = NULL;

10,438,933✔

2189

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

10,438,933✔

2190

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

10,436,978✔

2191

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

10,438,113✔

2193

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

10,437,608✔

2194

    QUERY_CHECK_CODE(code, lino, _end);

10,439,248✔

2196

    if (pInfo->cachedTagList) {

10,439,248✔

2197

      code = setTagValFromTagList(pOperator, res);

2,986,346✔

2198

      QUERY_CHECK_CODE(code, lino, _end);

2,986,346✔

2201

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

10,439,248✔

2202

    pInfo->pResBlock = res;

10,439,248✔

2203

    blockDataDestroy(result);

10,439,248✔

2204

    (*ppRes) = res;

10,439,248✔

2206

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

10,439,248✔

2208

    clearVstbBatchDynamicTableScanInfo(pInfo);

4,006,561✔

2209

    setOperatorCompleted(pOperator);

4,007,821✔

2212

_end:

14,448,644✔

2213

  return code;

14,448,644✔

2216

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

38,632,869✔

2217

  int32_t                  code = TSDB_CODE_SUCCESS;

38,632,869✔

2218

  int32_t                  lino = 0;

38,632,869✔

2219

  STableScanInfo*          pInfo = pOperator->info;

38,632,869✔

2220

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

38,634,254✔

2221

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

38,634,509✔

2223

  QRY_PARAM_CHECK(ppRes);

38,636,084✔

2224

  qTrace("%s call", __FUNCTION__);

38,629,334✔

2226

  pOperator->dynamicTask = true;

38,629,334✔

2228

  switch (type) {

38,629,524✔

2229

    case DYN_TYPE_VSTB_WIN_SCAN:

958,306✔

2231

      code = doStbJoinDynamicTableScanNext(pOperator, ppRes);

958,306✔

2232

      QUERY_CHECK_CODE(code, lino, _end);

958,306✔

2233

      break;

958,306✔

2235

    case DYN_TYPE_VSTB_SINGLE_SCAN: {

23,233,549✔

2236

      code = doVstbSingleDynamicTableScanNext(pOperator, ppRes);

23,233,549✔

2237

      QUERY_CHECK_CODE(code, lino, _end);

23,233,549✔

2238

      break;

23,232,924✔

2240

    case DYN_TYPE_VSTB_BATCH_SCAN:

14,437,669✔

2241

      code = doVstbBatchDynamicTableScanNext(pOperator, ppRes);

14,437,669✔

2242

      QUERY_CHECK_CODE(code, lino, _end);

14,447,699✔

2243

      break;

14,447,699✔

2250

_end:

38,639,554✔

2251

  return code;

38,639,554✔

2254

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

660,278,389✔

2255

  int32_t         code = TSDB_CODE_SUCCESS;

660,278,389✔

2256

  int32_t         lino = 0;

660,278,389✔

2257

  STableScanInfo* pInfo = pOperator->info;

660,278,389✔

2258

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

660,382,476✔

2259

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

660,346,286✔

2260

  QRY_PARAM_CHECK(ppRes);

660,417,907✔

2261

  qTrace("%s call", __FUNCTION__);

660,417,883✔

2263

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

660,418,623✔

2264

  QUERY_CHECK_CODE(code, lino, _end);

660,458,951✔

2266

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

660,458,951✔

2267

  if (pGetParam || pInfo->pBatchColMap) {

660,463,013✔

2268

    ETableScanDynType type = pGetParam ?

38,630,784✔

2269

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

38,632,994✔

2272

    code = doDynamicTableScanNext(pOperator, type, ppRes);

38,630,784✔

2273

    QUERY_CHECK_CODE(code, lino, _end);

38,639,554✔

2275

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

38,638,929✔

2276

      return code;

37,704,659✔

2281

  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {

622,747,377✔

2282

    int32_t       numOfTables = 0;

5,842,333✔

2283

    STableKeyInfo tInfo = {0};

5,842,333✔

2284

    pInfo->countState = TABLE_COUNT_STATE_END;

5,842,333✔

2286

    while (1) {

5,442,095✔

2287

      SSDataBlock* result = NULL;

11,284,428✔

2288

      code = doGroupedTableScan(pOperator, &result);

11,282,579✔

2289

      QUERY_CHECK_CODE(code, lino, _end);

11,280,126✔

2291

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

11,280,126✔

2292

        (*ppRes) = result;

5,795,409✔

2293

        return code;

5,797,900✔

2297

      pInfo->currentTable++;

5,486,554✔

2299

      taosRLockLatch(&pTaskInfo->lock);

5,483,116✔

2300

      numOfTables = 0;

5,488,691✔

2301

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

5,488,691✔

2302

      if (code != TSDB_CODE_SUCCESS) {

5,486,220✔

2307

      if (pInfo->currentTable >= numOfTables) {

5,486,220✔

2308

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

44,433✔

2309

        taosRUnLockLatch(&pTaskInfo->lock);

44,433✔

2310

        (*ppRes) = NULL;

44,433✔

2311

        return code;

44,433✔

2314

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

5,433,716✔

2315

      if (!tmp) {

5,438,700✔

2321

      tInfo = *tmp;

5,438,700✔

2322

      taosRUnLockLatch(&pTaskInfo->lock);

5,436,546✔

2324

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

5,443,055✔

2325

      QUERY_CHECK_CODE(code, lino, _end);

5,438,669✔

2326

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

5,438,669✔

2329

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

5,443,950✔

2330

      QUERY_CHECK_CODE(code, lino, _end);

5,439,627✔

2331

      pInfo->scanTimes = 0;

5,439,627✔

2334

    code = groupSeqTableScan(pOperator, ppRes);

616,862,988✔

2335

    QUERY_CHECK_CODE(code, lino, _end);

616,586,428✔

2338

_end:

616,584,258✔

2339

  if (code != TSDB_CODE_SUCCESS) {

616,615,970✔

2340

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

2,795✔

2341

    pTaskInfo->code = code;

2,795✔

2342

    T_LONG_JMP(pTaskInfo->env, code);

2,795✔

2345

  return code;

616,613,175✔

2348

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

1,824,919✔

2349

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

1,824,919✔

2350

  if (!pRecorder) {

1,824,419✔

2353

  STableScanInfo* pTableScanInfo = pOptr->info;

1,824,419✔

2354

  *pRecorder = pTableScanInfo->base.readRecorder;

1,824,919✔

2355

  *pOptrExplain = pRecorder;

1,824,424✔

2356

  *len = sizeof(SFileBlockLoadRecorder);

1,823,924✔

2357

  return 0;

1,823,924✔

2360

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

185,927,805✔

2361

  cleanupQueryTableDataCond(&pBase->cond);

185,927,805✔

2362

  cleanupQueryTableDataCond(&pBase->orgCond);

185,910,605✔

2364

  if (pAPI->tsdReaderClose) {

185,934,859✔

2365

    pAPI->tsdReaderClose(pBase->dataReader);

185,929,161✔

2367

  pBase->dataReader = NULL;

185,883,386✔

2369

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

185,883,483✔

2370

    taosArrayDestroy(pBase->matchInfo.pList);

185,891,356✔

2373

  tableListDestroy(pBase->pTableListInfo);

185,836,996✔

2374

  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);

185,913,053✔

2375

  cleanupExprSupp(&pBase->pseudoSup);

185,894,407✔

2376

185,924,305✔

2378

static void cleanReaderForVTable(STableScanInfo* pInfo){

5,096,683✔

2379

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

5,096,683✔

2380

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

5,097,311✔

2381

    while (pIter != NULL) {

5,119,453✔

2382

      void **reader = pIter;

22,075✔

2383

      if (*reader) {

22,075✔

2384

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

22,075✔

2385

          pInfo->base.dataReader = NULL;

19,827✔

2387

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

22,075✔

2389

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

22,075✔

2392

5,097,065✔

2394

static void destroyTableScanOperatorInfo(void* param) {

160,582,376✔

2395

  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;

160,582,376✔

2396

  blockDataDestroy(pTableScanInfo->pResBlock);

160,582,376✔

2397

  blockDataDestroy(pTableScanInfo->pOrgBlock);

160,563,841✔

2398

  taosHashCleanup(pTableScanInfo->pIgnoreTables);

160,566,117✔

2399

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

160,566,193✔

2400

    cleanReaderForVTable(pTableScanInfo);

5,096,683✔

2401

    taosHashCleanup(pTableScanInfo->readerCache);

5,097,378✔

2403

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

160,568,588✔

2404

  taosArrayDestroy(pTableScanInfo->pBlockColMap);

160,566,909✔

2405

  taosMemoryFreeClear(param);

160,576,304✔

2406

160,534,569✔

2408

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

160,470,845✔

2409

  if (pBlock && dataRequireFlag == FUNC_DATA_REQUIRED_NOT_LOAD) {

160,470,845✔

2410

    int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

7,362,309✔

2411

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

18,693,260✔

2412

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

11,338,587✔

2413

      if (pCol) {

11,338,264✔

2414

        pCol->info.noData = true;

11,338,264✔

2418

160,463,209✔

2420

static int32_t resetTableScanOperatorState(SOperatorInfo* pOper) {

282,803✔

2421

  int32_t         code = TSDB_CODE_SUCCESS;

282,803✔

2422

  STableScanInfo*   pInfo = pOper->info;

282,803✔

2423

  pOper->status = OP_NOT_OPENED;

282,803✔

2425

  pInfo->scanTimes = 0;

282,803✔

2426

  pInfo->currentGroupId = -1;

282,803✔

2427

  pInfo->tableEndIndex = -1;

282,803✔

2428

  pInfo->tableStartIndex = 0;

282,803✔

2429

  pInfo->currentTable = 0;

282,803✔

2430

  pInfo->scanMode = 0;

282,803✔

2431

  pInfo->countState = 0;

282,803✔

2432

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

282,803✔

2434

  if (!pInfo->virtualStableScan) {

282,803✔

2440

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

143,835✔

2441

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

143,835✔

2443

    pInfo->base.dataReader = NULL;

143,671✔

2445

  tableListDestroy(pInfo->base.pTableListInfo);

282,639✔

2446

  pInfo->base.pTableListInfo = tableListCreate();

282,803✔

2447

  if (!pInfo->base.pTableListInfo) {

282,803✔

2451

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

282,803✔

2453

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

282,803✔

2454

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

282,803✔

2455

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

284,919✔

2457

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

284,919✔

2458

    if (code) {

142,839✔

2464

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

282,803✔

2465

  cleanupQueryTableDataCond(&pInfo->base.cond);

282,803✔

2466

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

282,803✔

2467

  if (code) {

282,803✔

2471

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

282,803✔

2472

    cleanupQueryTableDataCond(&pInfo->base.orgCond);

138,968✔

2473

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

138,968✔

2474

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

138,968✔

2477

  pOper->resultInfo.totalRows = 0;

282,803✔

2478

  blockDataEmpty(pInfo->pResBlock);

282,803✔

2479

  blockDataEmpty(pInfo->pOrgBlock);

282,803✔

2480

  taosHashClear(pInfo->pIgnoreTables);

282,803✔

2481

  taosArrayDestroy(pInfo->pBlockColMap);

282,803✔

2482

  pInfo->pBlockColMap = NULL;

282,803✔

2483

  return code;

282,803✔

2486

int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

160,576,743✔

2489

  QRY_PARAM_CHECK(pOptrInfo);

160,576,743✔

2491

  int32_t         code = TSDB_CODE_SUCCESS;

160,589,241✔

2492

  int32_t         lino = 0;

160,589,241✔

2493

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

160,589,241✔

2494

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

160,548,077✔

2495

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

160,538,294✔

2496

    code = terrno;

329✔

2500

  pOperator->pPhyNode = pTableScanNode;

160,538,124✔

2501

  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;

160,537,576✔

2502

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

160,538,870✔

2504

  int32_t numOfCols = 0;

160,565,393✔

2506

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

160,568,369✔

2507

  QUERY_CHECK_CODE(code, lino, _error);

160,511,226✔

2509

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

160,511,226✔

2510

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

160,490,049✔

2511

  QUERY_CHECK_CODE(code, lino, _error);

160,423,865✔

2513

  if (pScanNode->pScanPseudoCols != NULL) {

160,423,865✔

2514

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

52,355,962✔

2515

    pSup->pExprInfo = NULL;

52,400,004✔

2516

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

52,408,851✔

2517

    QUERY_CHECK_CODE(code, lino, _error);

52,353,143✔

2519

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

52,353,143✔

2521

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

52,417,269✔

2524

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

160,442,912✔

2525

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

160,292,859✔

2527

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

160,434,912✔

2528

  pInfo->base.readHandle = *readHandle;

160,274,391✔

2529

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

160,489,260✔

2531

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

160,404,871✔

2532

  pInfo->sample.seed = taosGetTimestampSec();

160,484,076✔

2534

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

160,459,342✔

2535

  initResultSizeInfo(&pOperator->resultInfo, 4096);

160,457,182✔

2536

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

160,286,364✔

2537

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

160,477,451✔

2538

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

160,281,102✔

2540

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

160,284,769✔

2541

  QUERY_CHECK_CODE(code, lino, _error);

160,293,837✔

2543

  pInfo->virtualStableScan = pScanNode->virtualStableScan;

160,293,837✔

2544

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

160,321,243✔

2545

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

5,095,280✔

2546

    pInfo->pResBlock = NULL;

5,090,991✔

2547

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

5,096,425✔

2548

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

5,092,238✔

2551

  if (pInfo->virtualStableScan) {

160,351,189✔

2552

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

5,095,150✔

2553

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

5,098,198✔

2556

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

160,495,958✔

2557

                            pTaskInfo->pStreamRuntimeInfo);

160,484,097✔

2558

  QUERY_CHECK_CODE(code, lino, _error);

160,327,915✔

2560

  pInfo->currentGroupId = -1;

160,305,201✔

2562

  pInfo->tableEndIndex = -1;

160,435,008✔

2563

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

160,420,847✔

2564

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

160,341,805✔

2566

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

160,453,039✔

2568

  pOperator->exprSupp.numOfExprs = numOfCols;

160,294,055✔

2570

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

160,386,801✔

2571

  pInfo->ignoreTag = false;

160,459,084✔

2573

  pInfo->lastBatchIdx = -1;

160,367,508✔

2574

  pInfo->currentBatchIdx = 0;

160,306,067✔

2575

  pInfo->base.pTableListInfo = pTableListInfo;

160,495,098✔

2576

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

160,402,741✔

2586

  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

160,402,741✔

2588

  pOperator->fpSet = createOperatorFpSet(prepareTableScan, doTableScanNext,

160,368,884✔

2594

  setOperatorResetStateFn(pOperator, resetTableScanOperatorState);

160,289,609✔

2596

  pOperator->cost.openCost = 0;

160,302,197✔

2597

  *pOptrInfo = pOperator;

160,303,869✔

2598

  return TSDB_CODE_SUCCESS;

160,224,739✔

2600

_error:

22,714✔

2601

  if (pInfo != NULL) {

22,714✔

2602

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

22,714✔

2603

    destroyTableScanOperatorInfo(pInfo);

22,714✔

2606

  if (pOperator != NULL) {

22,714✔

2607

    pOperator->info = NULL;

22,714✔

2608

    destroyOperator(pOperator);

22,714✔

2610

  pTaskInfo->code = code;

22,714✔

2611

  return code;

22,714✔

2648

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

13,343✔

2649

  int32_t code = TSDB_CODE_SUCCESS;

13,343✔

2650

  int32_t lino = 0;

13,343✔

2651

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

13,343✔

2652

    return code;

1,622✔

2654

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

11,721✔

2655

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

11,721✔

2656

  bool hasUnqualified = false;

11,721✔

2658

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

11,721✔

2659

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

11,721✔

2661

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

11,721✔

2663

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

11,721✔

2666

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

11,721✔

2667

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

11,721✔

2668

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

42,080,871✔

2669

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

42,069,150✔

2670

    void*    data = colDataGetData(pColPk, i);

42,069,150✔

2671

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

42,071,406✔

2672

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

2,256✔

2675

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

2,256✔

2676

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

2,256✔

2677

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

2,256✔

2678

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

2,256✔

2679

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

2,256✔

2680

      taosMemoryFree(tmq);

2,256✔

2682

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

42,066,894✔

2685

    if (!p[i]) {

42,069,150✔

2686

      hasUnqualified = true;

15,016,500✔

2690

  if (hasUnqualified) {

11,721✔

2691

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

3,948✔

2692

    QUERY_CHECK_CODE(code, lino, _end);

3,948✔

2695

_end:

11,721✔

2696

  taosMemoryFree(p);

11,721✔

2697

  if (code != TSDB_CODE_SUCCESS) {

11,721✔

2700

  return code;

11,721✔

2703

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

1,260,050,014✔

2704

  int32_t p1 = *(int32_t*)param1;

1,260,050,014✔

2705

  int32_t p2 = *(int32_t*)param2;

1,260,290,603✔

2707

  if (p1 == p2) {

1,260,294,816✔

2710

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

1,260,294,816✔

2714

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

93,446,903✔

2715

  int32_t         code = TSDB_CODE_SUCCESS;

93,446,903✔

2716

  int32_t         lino = 0;

93,446,903✔

2717

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

93,446,903✔

2718

  SOperatorInfo*  pOperator = pInfo->pStreamScanOp;

93,446,903✔

2719

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

93,447,252✔

2720

  const char*     id = GET_TASKID(pTaskInfo);

93,446,903✔

2721

  SArray* pColList = NULL;

93,447,252✔

2723

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

93,447,252✔

2724

  QUERY_CHECK_CODE(code, lino, _end);

93,444,499✔

2726

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

93,444,499✔

2727

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

93,444,842✔

2728

  pBlockInfo->type = STREAM_NORMAL;

93,444,495✔

2729

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

93,444,842✔

2731

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

93,445,191✔

2732

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

93,445,534✔

2734

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

93,447,252✔

2735

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

93,446,574✔

2738

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

490,935,318✔

2739

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

397,483,136✔

2740

    if (!pColMatchInfo->needOutput) {

397,481,423✔

2741

      continue;

628✔

2744

    bool colExists = false;

397,486,740✔

2745

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

1,226,388,701✔

2746

      SColumnInfoData* pResCol = NULL;

1,225,972,029✔

2747

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

1,225,883,985✔

2748

      QUERY_CHECK_CODE(code, lino, _end);

1,225,943,315✔

2750

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

1,225,943,315✔

2751

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

397,482,390✔

2752

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

397,483,372✔

2753

        QUERY_CHECK_CODE(code, lino, _end);

397,384,997✔

2755

        colExists = true;

397,384,997✔

2756

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

397,384,997✔

2757

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

397,491,218✔

2758

        break;

397,491,218✔

2763

    if (!colExists) {

397,488,116✔

2772

  if (pInfo->numOfPseudoExpr > 0) {

93,446,219✔

2773

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

116,229,796✔

2774

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

58,114,806✔

2776

    if (code) {

58,114,117✔

2781

    terrno = 0;

58,114,117✔

2783

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

232,208,042✔

2784

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

174,089,123✔

2785

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

174,094,954✔

2789

  taosArraySort(pColList, colIdComparFn);

93,433,928✔

2791

  int32_t i = 0, j = 0;

93,445,876✔

2792

  while (i < taosArrayGetSize(pColList)) {

664,906,794✔

2793

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

571,450,334✔

2794

    if (slot1 > j) {

571,460,918✔

2799

      i += 1;

571,460,918✔

2800

      j += 1;

571,460,918✔

2804

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

93,444,865✔

2805

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

628✔

2806

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

628✔

2807

    j += 1;

628✔

2810

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

93,445,263✔

2811

  QUERY_CHECK_CODE(code, lino, _end);

93,443,510✔

2813

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

93,443,510✔

2814

  QUERY_CHECK_CODE(code, lino, _end);

93,442,887✔

2816

_end:

93,442,887✔

2817

  taosArrayDestroy(pColList);

93,057,835✔

2818

  if (code != TSDB_CODE_SUCCESS) {

93,439,471✔

2821

  return code;

93,437,402✔

2824

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

6,518,440✔

2825

  int32_t code = TSDB_CODE_SUCCESS;

6,518,440✔

2826

  int32_t lino = 0;

6,518,440✔

2827

  SValue  val = {0};

6,518,440✔

2828

  if (hasPrimaryKey) {

6,519,616✔

2829

    code = doBlockDataPrimaryKeyFilter(pBlock, offset);

13,343✔

2830

    if (code != TSDB_CODE_SUCCESS) {

13,343✔

2834

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

13,343✔

2836

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

13,343✔

2837

      return code;

3,666✔

2839

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

9,677✔

2840

    val.type = pColPk->info.type;

9,677✔

2841

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

9,677✔

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,060✔

2853

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

6,515,950✔

2855

_end:

6,514,186✔

2856

  if (code != TSDB_CODE_SUCCESS) {

6,514,774✔

2859

  return code;

6,513,547✔

2862

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

118,697,460✔

2863

  int32_t        code = TSDB_CODE_SUCCESS;

118,697,460✔

2864

  int32_t        lino = 0;

118,697,460✔

2865

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

118,697,460✔

2866

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

118,702,538✔

2868

  SStreamScanInfo* pInfo = pOperator->info;

118,703,610✔

2869

  const char*      id = GET_TASKID(pTaskInfo);

118,702,921✔

2871

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

118,701,977✔

2873

  if (isTaskKilled(pTaskInfo)) {

118,703,906✔

2878

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

118,704,945✔

2879

    while (1) {

3,666✔

2880

      SSDataBlock* pResult = NULL;

5,842,333✔

2881

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

5,842,333✔

2882

      QUERY_CHECK_CODE(code, lino, _end);

5,842,333✔

2884

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

5,842,333✔

2885

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

5,797,900✔

2886

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

5,797,900✔

2887

        QUERY_CHECK_CODE(code, lino, _end);

5,797,278✔

2888

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

5,797,278✔

2889

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

5,797,900✔

2890

          (*ppRes) = pResult;

5,794,234✔

2891

          return code;

5,794,234✔

2898

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

44,433✔

2899

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

44,433✔

2901

    pTSInfo->base.dataReader = NULL;

44,433✔

2902

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

44,433✔

2903

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

44,433✔

2904

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

44,433✔

2905

      (*ppRes) = NULL;

12,531✔

2906

      return code;

12,531✔

2909

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

31,902✔

2912

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

112,898,180✔

2913

    while (1) {

748,390✔

2915

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

113,646,227✔

2917

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

113,619,724✔

2918

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

113,637,871✔

2921

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

113,641,627✔

2924

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

113,639,544✔

2926

      if (hasResult) {

113,640,782✔

2927

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

93,440,427✔

2929

        blockDataCleanup(pInfo->pRes);

93,443,117✔

2930

        code = setBlockIntoRes(pInfo, pRes);

93,444,845✔

2931

        QUERY_CHECK_CODE(code, lino, _end);

93,438,403✔

2932

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

93,438,403✔

2934

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

93,442,822✔

2935

          (*ppRes) = pInfo->pRes;

92,698,519✔

2936

          return code;

92,696,804✔

2939

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

20,200,355✔

2940

        (*ppRes) = NULL;

20,200,690✔

2941

        return code;

20,200,690✔

2966

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

322,502✔

2967

  int32_t code = TSDB_CODE_SUCCESS;

322,502✔

2968

  int32_t lino = 0;

322,502✔

2969

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

322,502✔

2970

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

323,357✔

2973

  int32_t size = 0;

323,357✔

2974

  code = tableListGetSize(pTableListInfo, &size);

323,357✔

2975

  QUERY_CHECK_CODE(code, lino, _end);

323,357✔

2976

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

9,612,610✔

2977

    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);

9,289,253✔

2978

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

9,288,010✔

2979

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

9,288,010✔

2980

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

9,289,253✔

2983

  (*ppArrayRes) = tableIdList;

323,357✔

2985

_end:

323,357✔

2986

  if (code != TSDB_CODE_SUCCESS) {

323,357✔

2989

  return code;

323,357✔

2992

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

1,495,665✔

2993

  int32_t        code = TSDB_CODE_SUCCESS;

1,495,665✔

2994

  int32_t        lino = 0;

1,495,665✔

2995

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,495,665✔

2996

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,495,665✔

2998

  SStreamRawScanInfo* pInfo = pOperator->info;

1,495,665✔

2999

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

1,495,665✔

3000

  pTaskInfo->streamInfo.btMetaRsp.batchMetaLen = NULL;

1,495,665✔

3002

  qDebug("tmqsnap doRawScan called");

1,495,665✔

3003

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

1,495,665✔

3004

    bool hasNext = false;

1,490,901✔

3005

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

1,490,901✔

3006

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

1,490,067✔

3007

      if (code != TSDB_CODE_SUCCESS) {

1,486,819✔

3013

    if (pInfo->dataReader && hasNext) {

1,487,653✔

3014

      if (isTaskKilled(pTaskInfo)) {

722,909✔

3019

      SSDataBlock* pBlock = NULL;

722,909✔

3020

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

722,615✔

3021

      QUERY_CHECK_CODE(code, lino, _end);

720,263✔

3023

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

720,263✔

3024

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

722,027✔

3025

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

722,321✔

3026

        QUERY_CHECK_CODE(code, lino, _end);

720,263✔

3027

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

720,263✔

3028

        (*ppRes) = pBlock;

724,379✔

3029

        return code;

724,085✔

3033

    SMetaTableInfo mtInfo = {0};

764,744✔

3034

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

766,522✔

3035

    QUERY_CHECK_CODE(code, lino, _end);

764,464✔

3036

    if (code != 0) {

764,464✔

3040

    STqOffsetVal offset = {0};

764,464✔

3041

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

763,288✔

3042

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

11,150✔

3043

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

11,150✔

3045

      SValue val = {0};

753,916✔

3046

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

753,916✔

3047

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

754,504✔

3050

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

766,816✔

3051

    QUERY_CHECK_CODE(code, lino, _end);

766,816✔

3052

    (*ppRes) = NULL;

766,816✔

3053

    return code;

766,816✔

3054

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

4,764✔

3055

    SSnapContext* sContext = pInfo->sContext;

4,764✔

3056

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

39,898✔

3057

      void*   data = NULL;

39,898✔

3058

      int32_t dataLen = 0;

39,898✔

3059

      int16_t type = 0;

39,898✔

3060

      int64_t uid = 0;

39,898✔

3061

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

39,898✔

3067

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

39,898✔

3068

        STqOffsetVal offset = {0};

4,764✔

3069

        SValue       val = {0};

4,764✔

3071

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

4,764✔

3072

        QUERY_CHECK_CODE(code, lino, _end);

4,764✔

3073

        break;

4,764✔

3075

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

35,134✔

3076

        SMqMetaRsp tmpMetaRsp = {0};

35,134✔

3077

        tmpMetaRsp.resMsgType = type;

35,134✔

3078

        tmpMetaRsp.metaRspLen = dataLen;

35,134✔

3079

        tmpMetaRsp.metaRsp = data;

35,134✔

3080

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

35,134✔

3081

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

4,234✔

3082

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

4,234✔

3084

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

4,234✔

3085

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

4,234✔

3087

        int32_t  tempRes = TSDB_CODE_SUCCESS;

35,134✔

3088

        uint32_t len = 0;

35,134✔

3089

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

35,134✔

3090

        if (TSDB_CODE_SUCCESS != tempRes) {

35,134✔

3096

        int32_t tLen = sizeof(SMqRspHead) + len;

35,134✔

3097

        void*   tBuf = taosMemoryCalloc(1, tLen);

35,134✔

3098

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

35,134✔

3100

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

35,134✔

3101

        SEncoder encoder = {0};

35,134✔

3102

        tEncoderInit(&encoder, metaBuff, len);

35,134✔

3103

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

35,134✔

3104

        if (tempLen < 0) {

35,134✔

3111

        taosMemoryFreeClear(data);

35,134✔

3112

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

35,134✔

3113

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

35,134✔

3115

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

35,134✔

3116

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

35,134✔

3120

    (*ppRes) = NULL;

4,764✔

3121

    return code;

4,764✔

3135

static void destroyTmqRawScanOperatorInfo(void* param) {

115,605✔

3136

  SStreamRawScanInfo* pRawScan = (SStreamRawScanInfo*)param;

115,605✔

3137

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

115,605✔

3138

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

115,605✔

3139

  tableListDestroy(pRawScan->pTableListInfo);

115,605✔

3140

  taosMemoryFree(pRawScan);

115,605✔

3141

115,605✔

3146

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

114,961✔

3152

  QRY_PARAM_CHECK(pOptrInfo);

114,961✔

3153

  int32_t code = TSDB_CODE_SUCCESS;

115,605✔

3154

  int32_t lino = 0;

115,605✔

3156

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

115,605✔

3157

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

115,290✔

3158

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

115,290✔

3164

  pInfo->pTableListInfo = tableListCreate();

115,290✔

3165

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

115,605✔

3166

  pInfo->vnode = pHandle->vnode;

115,605✔

3167

  pInfo->pAPI = &pTaskInfo->storageAPI;

115,605✔

3169

  pInfo->sContext = pHandle->sContext;

115,605✔

3170

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

115,605✔

3171

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

115,605✔

3174

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

115,605✔

3176

  *pOptrInfo = pOperator;

115,605✔

3177

  return code;

115,605✔

3189

void destroyTmqScanOperatorInfo(void* param) {

323,357✔

3190

  if (param == NULL) {

323,357✔

3194

  SStreamScanInfo* pStreamScan = (SStreamScanInfo*)param;

323,357✔

3195

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

323,357✔

3196

    destroyOperator(pStreamScan->pTableScanOp);

323,357✔

3199

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

323,357✔

3200

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

323,060✔

3202

  if (pStreamScan->pVtableMergeHandles) {

323,654✔

3206

  if (pStreamScan->pVtableMergeBuf) {

323,357✔

3210

  if (pStreamScan->pVtableReadyHandles) {

323,357✔

3214

  if (pStreamScan->pTableListInfo) {

323,357✔

3218

  if (pStreamScan->matchInfo.pList) {

323,357✔

3219

    taosArrayDestroy(pStreamScan->matchInfo.pList);

323,037✔

3221

  if (pStreamScan->pPseudoExpr) {

323,677✔

3222

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

186,998✔

3223

    taosMemoryFree(pStreamScan->pPseudoExpr);

186,998✔

3226

  cleanupExprSupp(&pStreamScan->tbnameCalSup);

323,357✔

3227

  cleanupExprSupp(&pStreamScan->tagCalSup);

323,086✔

3229

  blockDataDestroy(pStreamScan->pRes);

323,357✔

3230

  blockDataDestroy(pStreamScan->pUpdateRes);

323,357✔

3231

  blockDataDestroy(pStreamScan->pDeleteDataRes);

323,357✔

3232

  blockDataDestroy(pStreamScan->pUpdateDataRes);

323,357✔

3233

  blockDataDestroy(pStreamScan->pCreateTbRes);

323,357✔

3234

  taosArrayDestroy(pStreamScan->pBlockLists);

323,357✔

3235

  blockDataDestroy(pStreamScan->pCheckpointRes);

323,357✔

3237

  taosMemoryFree(pStreamScan);

323,357✔

3257

 int32_t createTmqScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,

323,357✔

3260

  QRY_PARAM_CHECK(pOptrInfo);

323,357✔

3262

  int32_t          code = TSDB_CODE_SUCCESS;

323,357✔

3263

  int32_t          lino = 0;

323,357✔

3264

  SArray*          pColIds = NULL;

323,357✔

3265

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

323,357✔

3266

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

323,357✔

3267

  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;

323,357✔

3268

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

323,357✔

3270

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

323,357✔

3275

  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;

323,357✔

3276

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;

323,357✔

3278

  pInfo->pTagCond = pTagCond;

323,357✔

3279

  pInfo->pGroupTags = pTableScanNode->pGroupTags;

323,357✔

3281

  int32_t numOfCols = 0;

323,357✔

3282

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

323,357✔

3283

  if (code != TSDB_CODE_SUCCESS) {

323,357✔

3287

  SDataType pkType = {0};

323,357✔

3288

  pInfo->primaryKeyIndex = -1;

323,357✔

3289

  pInfo->basic.primaryPkIndex = -1;

323,357✔

3290

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

323,357✔

3291

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

323,357✔

3292

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

323,357✔

3294

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

2,028,712✔

3295

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

1,705,355✔

3296

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

1,705,017✔

3298

    int16_t colId = id->colId;

1,705,017✔

3299

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

1,705,693✔

3300

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

1,705,693✔

3302

    if (id->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

1,705,693✔

3303

      pInfo->primaryTsIndex = id->dstSlotId;

323,357✔

3305

    if (id->isPk) {

1,705,693✔

3306

      pInfo->primaryKeyIndex = id->dstSlotId;

1,463✔

3307

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

1,463✔

3308

      pkType = id->dataType;

1,463✔

3312

  pInfo->pPartTbnameSup = NULL;

323,357✔

3313

  if (pTableScanNode->pSubtable != NULL) {

323,357✔

3329

  if (pTableScanNode->pTags != NULL) {

323,357✔

3341

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

323,357✔

3342

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

323,357✔

3344

  if (pHandle->vnode) {

323,357✔

3345

    SOperatorInfo* pTableScanOp = NULL;

323,357✔

3346

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

323,357✔

3347

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

323,046✔

3348

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

563✔

3352

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

322,483✔

3353

    if (pHandle->version > 0) {

323,357✔

3354

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

323,357✔

3357

    STableKeyInfo* pList = NULL;

323,357✔

3358

    int32_t        num = 0;

323,357✔

3359

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

323,357✔

3360

    QUERY_CHECK_CODE(code, lino, _error);

323,047✔

3362

    if (pHandle->initTableReader) {

323,047✔

3363

      pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;

323,357✔

3364

      pTSInfo->base.dataReader = NULL;

323,047✔

3367

    if (pHandle->initTqReader) {

323,047✔

3368

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

323,087✔

3369

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

323,357✔

3371

      pInfo->tqReader = pHandle->tqReader;

270✔

3375

    pInfo->pUpdateInfo = NULL;

323,049✔

3376

    pInfo->pTableScanOp = pTableScanOp;

323,357✔

3377

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

323,357✔

3382

    pInfo->readHandle = *pHandle;

323,357✔

3383

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

323,357✔

3384

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

323,357✔

3385

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

323,357✔

3386

    pInfo->hasPart = false;

323,092✔

3388

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

323,357✔

3389

    QUERY_CHECK_CODE(code, lino, _error);

323,357✔

3392

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

323,357✔

3393

    QUERY_CHECK_CODE(code, lino, _error);

323,049✔

3395

    SArray* tableIdList = NULL;

323,049✔

3396

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

323,049✔

3397

    QUERY_CHECK_CODE(code, lino, _error);

323,357✔

3398

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

323,357✔

3399

    QUERY_CHECK_CODE(code, lino, _error);

323,357✔

3400

    taosArrayDestroy(tableIdList);

323,357✔

3401

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

323,357✔

3408

  pColIds = NULL;

323,357✔

3411

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

323,357✔

3412

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

186,998✔

3413

    QUERY_CHECK_CODE(code, lino, _error);

186,998✔

3416

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

323,357✔

3417

  QUERY_CHECK_CODE(code, lino, _error);

323,357✔

3419

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

323,357✔

3420

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

323,357✔

3421

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

323,357✔

3422

  QUERY_CHECK_CODE(code, lino, _error);

323,087✔

3424

  pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;

323,087✔

3425

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

323,357✔

3426

  pInfo->groupId = 0;

323,087✔

3427

  pInfo->igCheckGroupId = false;

323,087✔

3428

  pInfo->pStreamScanOp = pOperator;

323,087✔

3429

  pInfo->deleteDataIndex = 0;

323,357✔

3430

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

323,357✔

3431

  QUERY_CHECK_CODE(code, lino, _error);

323,357✔

3433

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

323,357✔

3434

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

323,357✔

3435

  QUERY_CHECK_CODE(code, lino, _error);

323,357✔

3437

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

323,357✔

3438

  pInfo->partitionSup.needCalc = false;

323,357✔

3439

  pInfo->igCheckUpdate = pTableScanNode->igCheckUpdate;

323,049✔

3440

  pInfo->igExpired = pTableScanNode->igExpired;

323,357✔

3441

  pInfo->twAggSup.maxTs = INT64_MIN;

323,049✔

3442

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

323,357✔

3443

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

323,357✔

3444

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

323,357✔

3445

  pInfo->pFillSup = NULL;

323,357✔

3446

  pInfo->useGetResultRange = false;

323,049✔

3447

  pInfo->pRangeScanRes = NULL;

323,357✔

3449

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

323,357✔

3450

  QUERY_CHECK_CODE(code, lino, _error);

323,357✔

3452

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

323,357✔

3454

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

323,357✔

3456

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

323,357✔

3459

  *pOptrInfo = pOperator;

323,357✔

3460

  return code;

323,357✔

3483

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

9,147,982✔

3484

  int32_t        code = TSDB_CODE_SUCCESS;

9,147,982✔

3485

  int32_t        lino = 0;

9,147,982✔

3486

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,147,982✔

3487

  STagScanInfo*  pInfo = pOperator->info;

9,149,322✔

3488

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

9,151,342✔

3489

  int32_t        count = pRes->info.rows;

9,149,827✔

3491

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

9,150,837✔

3492

  if (!item) {

9,142,088✔

3499

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

9,142,088✔

3500

  tDecoderClear(&(*mr).coder);

9,152,652✔

3501

  if (code != TSDB_CODE_SUCCESS) {

9,150,203✔

3507

  char str[512];

9,150,107✔

3508

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

41,355,731✔

3509

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

32,207,498✔

3512

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

32,205,858✔

3513

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

3,729,152✔

3514

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

3,727,311✔

3515

      QUERY_CHECK_CODE(code, lino, _end);

3,728,826✔

3516

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

28,480,111✔

3517

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

5,231,158✔

3518

      QUERY_CHECK_CODE(code, lino, _end);

5,231,348✔

3519

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

23,249,143✔

3520

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

15,104✔

3521

      QUERY_CHECK_CODE(code, lino, _end);

15,104✔

3523

      STagVal val = {0};

23,234,734✔

3524

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

23,236,374✔

3525

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

23,235,679✔

3527

      char* data = NULL;

23,234,859✔

3528

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

23,234,859✔

3529

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

23,232,852✔

3531

        data = (char*)p;

1,377✔

3534

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

23,233,219✔

3535

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

23,231,704✔

3536

      QUERY_CHECK_CODE(code, lino, _end);

23,234,229✔

3538

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

23,234,229✔

3540

        taosMemoryFree(data);

8,220,712✔

3545

_end:

9,156,517✔

3546

  if (code != TSDB_CODE_SUCCESS) {

9,152,037✔

3550

    pRes->info.rows++;

9,152,037✔

3553

  return code;

9,151,642✔

3556

static void tagScanFreeUidTag(void* p) {

7,080,264✔

3557

  STUidTagInfo* pInfo = p;

7,080,264✔

3558

  if (pInfo->pTagVal != NULL) {

7,080,264✔

3559

    taosMemoryFree(pInfo->pTagVal);

7,078,673✔

3561

7,076,789✔

3563

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

1,580,184✔

3564

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

1,580,184✔

3565

  if (pColumnData == NULL) {

1,578,150✔

3569

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

1,578,150✔

3570

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

1,579,128✔

3571

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

1,579,687✔

3572

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

1,578,013✔

3574

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

1,579,836✔

3575

  if (code != TSDB_CODE_SUCCESS) {

1,577,504✔

3581

  pParam->columnData = pColumnData;

1,577,504✔

3582

  pParam->colAlloced = true;

1,579,691✔

3583

  return TSDB_CODE_SUCCESS;

1,579,691✔

3586

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

9,674,410✔

3587

  int32_t                code = TSDB_CODE_SUCCESS;

9,674,410✔

3588

  int32_t                lino = 0;

9,674,410✔

3589

  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;

9,674,410✔

3590

  SColumnNode*           pSColumnNode = NULL;

9,674,410✔

3591

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

9,674,414✔

3592

    pSColumnNode = *(SColumnNode**)pNode;

2,721,790✔

3593

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

6,960,115✔

3594

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

498✔

3595

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

498✔

3608

      return DEAL_RES_CONTINUE;

498✔

3611

    return DEAL_RES_CONTINUE;

6,956,346✔

3614

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

2,722,283✔

3615

  if (!data) {

2,719,424✔

3616

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

1,580,561✔

3617

    if (code == TSDB_CODE_DUP_KEY) {

1,583,635✔

3620

    QUERY_CHECK_CODE(code, lino, _end);

1,583,635✔

3621

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

1,583,635✔

3622

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

1,583,768✔

3623

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

1,582,301✔

3624

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

1,582,160✔

3625

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

1,581,957✔

3626

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

1,582,438✔

3628

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

1,138,863✔

3629

    pSColumnNode->slotId = col->slotId;

1,138,863✔

3632

_end:

2,721,512✔

3633

  if (code != TSDB_CODE_SUCCESS) {

2,720,679✔

3634

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

1,823✔

3637

  return DEAL_RES_CONTINUE;

2,718,856✔

3640

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

1,581,021✔

3642

  int32_t code = TSDB_CODE_SUCCESS;

1,581,021✔

3643

  int32_t lino = 0;

1,581,021✔

3644

  int32_t numOfTables = taosArrayGetSize(aUidTags);

1,581,021✔

3645

  SArray* pBlockList = NULL;

1,581,021✔

3647

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

1,581,021✔

3648

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

1,578,709✔

3650

  pBlockList = taosArrayInit(1, POINTER_BYTES);

1,578,709✔

3651

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

1,580,528✔

3653

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

1,581,021✔

3654

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

1,581,021✔

3656

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

1,581,021✔

3658

  SScalarParam output = {0};

1,580,528✔

3659

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

1,581,021✔

3660

  QUERY_CHECK_CODE(code, lino, _end);

1,578,846✔

3662

  gTaskScalarExtra.pStreamInfo = (void*)pStreamRuntimeInfo;

1,578,846✔

3663

  gTaskScalarExtra.pStreamRange = NULL;

1,578,846✔

3664

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

1,578,846✔

3665

  QUERY_CHECK_CODE(code, lino, _end);

1,578,515✔

3667

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

1,578,515✔

3668

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

3,945,884✔

3669

    if (result[i]) {

2,368,503✔

3670

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

2,217,174✔

3671

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

2,217,174✔

3675

  colDataDestroy(output.columnData);

1,579,696✔

3676

  taosMemoryFreeClear(output.columnData);

1,577,384✔

3678

_end:

1,577,729✔

3679

  blockDataDestroy(pResBlock);

1,577,730✔

3680

  taosArrayDestroy(pBlockList);

1,574,712✔

3681

  if (code != TSDB_CODE_SUCCESS) {

1,574,407✔

3684

  return code;

1,574,407✔

3687

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

28,831,180✔

3690

  int32_t code = TSDB_CODE_SUCCESS;

28,831,180✔

3691

  int32_t lino = 0;

28,831,180✔

3692

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

28,831,180✔

3693

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

13,754,307✔

3699

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

13,747,175✔

3700

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

6,872,935✔

3701

      QUERY_CHECK_CODE(code, lino, _end);

6,875,598✔

3702

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

6,877,054✔

3703

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

6,875,470✔

3704

      QUERY_CHECK_CODE(code, lino, _end);

6,876,733✔

3707

    STagVal tagVal = {0};

15,077,084✔

3708

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

15,078,075✔

3709

    if (pUidTagInfo->pTagVal == NULL) {

15,077,852✔

3712

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

15,077,728✔

3714

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

15,080,808✔

3715

        colDataSetNULL(pColInfo, rowIndex);

8,403✔

3716

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

15,072,523✔

3719

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

18,798,079✔

3720

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

3,733,513✔

3723

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

3,726,613✔

3724

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

3,728,318✔

3726

        varDataSetLen(tmp, tagVal.nData);

3,728,318✔

3727

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

3,725,621✔

3728

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

3,729,298✔

3729

        taosMemoryFree(tmp);

3,729,184✔

3730

        QUERY_CHECK_CODE(code, lino, _end);

3,727,761✔

3732

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

11,341,354✔

3733

        QUERY_CHECK_CODE(code, lino, _end);

11,345,441✔

3738

_end:

28,834,315✔

3739

  if (code != TSDB_CODE_SUCCESS) {

28,834,315✔

3742

  return code;

28,827,102✔

3745

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

4,526,821✔

3747

  int32_t       code = TSDB_CODE_SUCCESS;

4,526,821✔

3748

  int32_t       lino = 0;

4,526,821✔

3749

  STagScanInfo* pInfo = pOperator->info;

4,526,821✔

3750

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

4,525,150✔

3751

  if (!ignoreFilterIdx) {

4,526,939✔

3752

    size_t szTables = taosArrayGetSize(aFilterIdxs);

1,578,225✔

3753

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

3,796,868✔

3754

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

2,214,032✔

3755

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);

2,215,498✔

3756

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

2,216,042✔

3757

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

12,601,181✔

3758

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

10,381,050✔

3759

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

10,384,286✔

3760

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

10,384,286✔

3761

        QUERY_CHECK_CODE(code, lino, _end);

10,385,139✔

3765

    size_t szTables = taosArrayGetSize(aUidTags);

2,948,714✔

3766

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

7,661,525✔

3767

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);

4,713,595✔

3768

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

4,713,595✔

3769

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

23,154,268✔

3770

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

18,442,471✔

3771

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

18,442,805✔

3772

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

18,442,805✔

3773

        QUERY_CHECK_CODE(code, lino, _end);

18,440,673✔

3778

_end:

2,947,930✔

3779

  if (code != TSDB_CODE_SUCCESS) {

4,530,766✔

3782

  return code;

4,529,146✔

3785

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

9,017,952✔

3786

  int32_t        code = TSDB_CODE_SUCCESS;

9,017,952✔

3787

  int32_t        lino = 0;

9,017,952✔

3788

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,017,952✔

3789

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

9,015,171✔

3791

  STagScanInfo* pInfo = pOperator->info;

9,014,907✔

3792

  SSDataBlock*  pRes = pInfo->pRes;

9,015,247✔

3794

  QRY_PARAM_CHECK(ppRes);

9,017,738✔

3796

  if (pOperator->status == OP_EXEC_DONE) {

9,017,738✔

3797

    return TSDB_CODE_SUCCESS;

4,420,535✔

3799

  blockDataCleanup(pRes);

4,593,837✔

3801

  if (pInfo->pCtbCursor == NULL) {

4,600,600✔

3802

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

4,600,848✔

3803

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

4,601,169✔

3809

  SArray* aUidTags = pInfo->aUidTags;

4,600,355✔

3810

  SArray* aFilterIdxs = pInfo->aFilterIdxs;

4,600,355✔

3811

  int32_t count = 0;

4,601,811✔

3812

  bool    ctbCursorFinished = false;

4,601,811✔

3813

  while (1) {

98,830✔

3814

    taosArrayClearEx(aUidTags, tagScanFreeUidTag);

4,700,641✔

3815

    taosArrayClear(aFilterIdxs);

4,699,684✔

3817

    int32_t numTables = 0;

4,699,684✔

3818

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

11,777,826✔

3819

      SMCtbCursor* pCur = pInfo->pCtbCursor;

11,778,169✔

3820

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

11,777,610✔

3821

      if (uid == 0) {

11,782,890✔

3822

        ctbCursorFinished = true;

4,701,010✔

3823

        break;

4,701,010✔

3825

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

7,081,880✔

3826

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

7,078,163✔

3827

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

7,078,234✔

3829

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

7,078,234✔

3830

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

7,081,269✔

3831

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

7,081,269✔

3832

      ++numTables;

7,081,269✔

3835

    if (numTables == 0) {

4,701,010✔

3836

      break;

170,901✔

3838

    bool ignoreFilterIdx = true;

4,530,109✔

3839

    if (pInfo->pTagCond != NULL) {

4,530,109✔

3840

      ignoreFilterIdx = false;

1,581,021✔

3841

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

1,581,021✔

3842

      QUERY_CHECK_CODE(code, lino, _end);

1,577,232✔

3844

      ignoreFilterIdx = true;

2,949,088✔

3847

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

4,526,320✔

3848

    QUERY_CHECK_CODE(code, lino, _end);

4,529,146✔

3850

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

4,529,146✔

3852

    if (count != 0) {

4,528,651✔

3853

      break;

4,429,821✔

3857

  if (count > 0) {

4,600,722✔

3858

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

4,429,821✔

3860

  if (count == 0 || ctbCursorFinished) {

4,600,324✔

3861

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

4,600,324✔

3862

    pInfo->pCtbCursor = NULL;

4,596,369✔

3863

    setOperatorCompleted(pOperator);

4,597,708✔

3866

  pRes->info.rows = count;

4,595,991✔

3867

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

4,595,077✔

3868

  if (bLimitReached) {

4,598,199✔

3872

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

4,598,199✔

3873

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

4,598,845✔

3875

_end:

4,598,876✔

3876

  if (code != TSDB_CODE_SUCCESS) {

4,598,876✔

3882

  return code;

4,598,876✔

3885

static int32_t createTagScanTableListInfoFromParam(SOperatorInfo* pOperator) {

3,406,794✔

3886

  STagScanInfo*          pInfo = pOperator->info;

3,406,794✔

3887

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

3,406,794✔

3888

  int32_t                code = 0;

3,406,794✔

3889

  STableListInfo*        pListInfo = pInfo->pTableListInfo;

3,406,794✔

3890

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

3,406,794✔

3891

  tb_uid_t               pUid = pParam->vcUid;

3,406,794✔

3897

  STableKeyInfo info = {.groupId = 0};

3,406,794✔

3899

  int32_t tableIdx = 0;

3,406,794✔

3900

  taosHashClear(pListInfo->map);

3,406,794✔

3901

  taosArrayClear(pListInfo->pTableList);

3,406,794✔

3903

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

3,406,794✔

3910

  info.uid = pUid;

3,406,794✔

3911

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

3,406,794✔

3912

  if (p == NULL) {

3,406,794✔

3916

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

3,406,794✔

3918

  return code;

3,406,794✔

3921

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

6,601,407✔

3922

  int32_t       code = TSDB_CODE_SUCCESS;

6,601,407✔

3923

  int32_t       lino = 0;

6,601,407✔

3924

  STagScanInfo* pInfo = pOperator->info;

6,601,407✔

3925

  if (pOperator->pOperatorGetParam) {

6,602,922✔

3926

    pOperator->resultInfo.totalRows = 0;

3,406,794✔

3927

    pOperator->dynamicTask = true;

3,406,794✔

3928

    pInfo->curPos = 0;

3,406,794✔

3929

    code = createTagScanTableListInfoFromParam(pOperator);

3,406,794✔

3930

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

3,406,794✔

3931

    pOperator->pOperatorGetParam = NULL;

3,406,794✔

3932

    QUERY_CHECK_CODE(code, lino, _end);

3,406,794✔

3934

    if (pOperator->status == OP_EXEC_DONE) {

3,406,794✔

3935

      pOperator->status = OP_OPENED;

1,834,288✔

3939

  if (pOperator->status == OP_EXEC_DONE) {

6,601,092✔

3940

    (*ppRes) = NULL;

1,583,590✔

3941

    return code;

1,583,905✔

3944

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

5,017,817✔

3945

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

5,018,007✔

3947

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

5,018,322✔

3948

  SSDataBlock* pRes = pInfo->pRes;

5,017,312✔

3949

  blockDataCleanup(pRes);

5,014,623✔

3951

  int32_t size = 0;

5,017,502✔

3952

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

5,016,997✔

3953

  QUERY_CHECK_CODE(code, lino, _end);

5,014,977✔

3955

  if (size == 0) {

5,014,977✔

3956

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

18,438✔

3957

    (*ppRes) = NULL;

18,438✔

3958

    return code;

18,438✔

3961

  SMetaReader mr = {0};

4,996,539✔

3962

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

4,999,064✔

3963

  pRes->info.rows = 0;

5,000,894✔

3965

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

14,151,397✔

3966

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

9,149,512✔

3967

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

9,151,012✔

3970

      code = TSDB_CODE_SUCCESS;

9,152,143✔

3972

    QUERY_CHECK_CODE(code, lino, _end);

9,151,012✔

3974

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

9,151,012✔

3975

      setOperatorCompleted(pOperator);

5,000,894✔

3979

  pAPI->metaReaderFn.clearReader(&mr);

5,003,540✔

3980

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

5,000,894✔

3981

  if (bLimitReached) {

5,000,389✔

3986

  if (pOperator->status == OP_EXEC_DONE) {

5,000,389✔

3987

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

5,000,894✔

3990

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

5,000,389✔

3992

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

5,000,894✔

3993

_end:

5,000,894✔

3994

  if (code != TSDB_CODE_SUCCESS) {

5,000,894✔

3999

  return code;

5,000,894✔

4002

static void destroyTagScanOperatorInfo(void* param) {

8,112,569✔

4003

  STagScanInfo* pInfo = (STagScanInfo*)param;

8,112,569✔

4004

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

8,112,569✔

4007

  taosHashCleanup(pInfo->filterCtx.colHash);

8,114,778✔

4008

  taosArrayDestroy(pInfo->filterCtx.cInfoList);

8,104,188✔

4009

  taosArrayDestroy(pInfo->aFilterIdxs);

8,103,044✔

4010

  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);

8,101,819✔

4012

  blockDataDestroy(pInfo->pRes);

8,112,564✔

4013

  taosArrayDestroy(pInfo->matchInfo.pList);

8,108,466✔

4014

  tableListDestroy(pInfo->pTableListInfo);

8,108,020✔

4016

  pInfo->pRes = NULL;

8,114,416✔

4017

  pInfo->pTableListInfo = NULL;

8,114,921✔

4018

  taosMemoryFreeClear(param);

8,112,545✔

4019

8,111,188✔

4021

static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {

5,002✔

4022

  int32_t         code = TSDB_CODE_SUCCESS;

5,002✔

4023

  STagScanInfo*   pInfo = pOper->info;

5,002✔

4025

  pOper->status = OP_NOT_OPENED;

5,002✔

4026

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

5,002✔

4027

  if (pTagScanNode->onlyMetaCtbIdx) {

5,002✔

4028

      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

1,992✔

4029

      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,992✔

4030

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

1,992✔

4031

      pInfo->pCtbCursor = NULL;

1,992✔

4033

      pInfo->curPos = 0;

3,010✔

4036

  return code;

5,002✔

4039

int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,

8,113,374✔

4042

  QRY_PARAM_CHECK(pOptrInfo);

8,113,374✔

4044

  int32_t         code = TSDB_CODE_SUCCESS;

8,115,207✔

4045

  int32_t         lino = 0;

8,115,207✔

4046

  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;

8,115,207✔

4047

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

8,115,207✔

4048

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

8,104,004✔

4049

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

8,099,999✔

4053

  pOperator->pPhyNode = pTagScanNode;

8,099,999✔

4054

  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

8,100,469✔

4056

  int32_t    numOfExprs = 0;

8,105,747✔

4057

  SExprInfo* pExprInfo = NULL;

8,106,236✔

4059

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

8,101,439✔

4060

  QUERY_CHECK_CODE(code, lino, _error);

8,107,368✔

4062

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

8,107,368✔

4063

  QUERY_CHECK_CODE(code, lino, _error);

8,097,286✔

4065

  int32_t num = 0;

8,097,286✔

4066

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

8,099,573✔

4067

  QUERY_CHECK_CODE(code, lino, _error);

8,107,854✔

4069

  pInfo->pTagCond = pTagCond;

8,107,854✔

4070

  pInfo->pTagIndexCond = pTagIndexCond;

8,111,249✔

4071

  pInfo->suid = pPhyNode->suid;

8,107,372✔

4072

  pInfo->pStorageAPI = &pTaskInfo->storageAPI;

8,103,443✔

4074

  pInfo->pTableListInfo = pTableListInfo;

8,088,313✔

4075

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

8,109,399✔

4076

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

8,103,768✔

4078

  pInfo->readHandle = *pReadHandle;

8,085,683✔

4079

  pInfo->curPos = 0;

8,111,787✔

4081

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

8,087,560✔

4082

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

8,098,798✔

4084

  initResultSizeInfo(&pOperator->resultInfo, 4096);

8,091,249✔

4085

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

8,088,745✔

4086

  QUERY_CHECK_CODE(code, lino, _error);

8,114,568✔

4088

  if (pTagScanNode->onlyMetaCtbIdx) {

8,114,568✔

4089

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

4,605,773✔

4090

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

4,608,483✔

4092

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

4,607,543✔

4093

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

4,607,154✔

4095

    pInfo->filterCtx.colHash =

4,606,684✔

4096

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

4,607,496✔

4097

    QUERY_CHECK_NULL(pInfo->filterCtx.colHash, code, lino, _error, terrno);

4,607,154✔

4099

    pInfo->filterCtx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));

4,605,993✔

4100

    QUERY_CHECK_NULL(pInfo->filterCtx.cInfoList, code, lino, _error, terrno);

4,608,483✔

4102

    if (pInfo->pTagCond != NULL) {

4,605,693✔

4103

      nodesRewriteExprPostOrder(&pTagCond, tagScanRewriteTagColumn, (void*)&pInfo->filterCtx);

1,584,472✔

4106

  __optr_fn_t tagScanNextFn = (pTagScanNode->onlyMetaCtbIdx) ? doTagScanFromCtbIdxNext : doTagScanFromMetaEntryNext;

8,113,899✔

4107

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, tagScanNextFn, NULL, destroyTagScanOperatorInfo,

8,114,920✔

4109

  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);

8,107,091✔

4111

  *pOptrInfo = pOperator;

8,106,949✔

4112

  return code;

8,105,846✔

4129

static int32_t subTblRowCompareTsFn(const void* pLeft, const void* pRight, void* param) {

232,131,424✔

4130

  int32_t                 left = *(int32_t*)pLeft;

232,131,424✔

4131

  int32_t                 right = *(int32_t*)pRight;

232,138,112✔

4132

  STmsSubTablesMergeInfo* pInfo = (STmsSubTablesMergeInfo*)param;

232,139,520✔

4134

  int32_t leftIdx = pInfo->aInputs[left].rowIdx;

232,139,520✔

4135

  int32_t rightIdx = pInfo->aInputs[right].rowIdx;

232,139,520✔

4137

  if (leftIdx == -1) {

232,134,944✔

4138

    return 1;

198,780,704✔

4139

  } else if (rightIdx == -1) {

33,354,240✔

4140

    return -1;

330,240✔

4143

  int64_t leftTs = pInfo->aInputs[left].aTs[leftIdx];

33,024,000✔

4144

  int64_t rightTs = pInfo->aInputs[right].aTs[rightIdx];

33,024,000✔

4145

  int32_t ret = leftTs > rightTs ? 1 : ((leftTs < rightTs) ? -1 : 0);

33,024,000✔

4146

  if (pInfo->pTsOrderInfo->order == TSDB_ORDER_DESC) {

33,024,000✔

4149

  return ret;

33,024,000✔

4179

int32_t dumpQueryTableCond(const SQueryTableDataCond* src, SQueryTableDataCond* dst) {

2,937,264✔

4180

  int32_t code = TSDB_CODE_SUCCESS;

2,937,264✔

4181

  int32_t lino = 0;

2,937,264✔

4182

  memcpy((void*)dst, (void*)src, sizeof(SQueryTableDataCond));

2,937,264✔

4183

  dst->colList = taosMemoryCalloc(src->numOfCols, sizeof(SColumnInfo));

2,937,264✔

4184

  QUERY_CHECK_NULL(dst->colList, code, lino, _end, terrno);

2,934,832✔

4185

  for (int i = 0; i < src->numOfCols; i++) {

9,790,168✔

4186

    dst->colList[i] = src->colList[i];

6,873,624✔

4189

_end:

2,928,680✔

4190

  if (code != TSDB_CODE_SUCCESS) {

2,928,680✔

4193

  return code;

2,927,632✔

4196

static int32_t fetchNextSubTableBlockFromReader(SOperatorInfo* pOperator, STmsSubTableInput* pInput,

4,414,808✔

4198

  int32_t code = 0;

4,414,808✔

4200

  STableMergeScanInfo*    pInfo = pOperator->info;

4,414,808✔

4201

  SReadHandle*            pHandle = &pInfo->base.readHandle;

4,414,824✔

4202

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

4,413,064✔

4203

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

4,409,576✔

4204

  const SStorageAPI*      pAPI = &pTaskInfo->storageAPI;

4,415,168✔

4206

  blockDataCleanup(pInput->pReaderBlock);

4,415,176✔

4207

  if (!pInput->bInMemReader) {

4,413,480✔

4208

    code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, pInput->pKeyInfo, 1, pInput->pReaderBlock,

4,422,840✔

4209

                                         (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

2,213,496✔

4210

    if (code != 0) {

2,212,112✔

4215

  pInfo->base.dataReader = pInput->pReader;

4,413,488✔

4217

  while (true) {

6,336✔

4218

    bool hasNext = false;

4,416,952✔

4219

    code = pAPI->tsdReader.tsdNextDataBlock(pInfo->base.dataReader, &hasNext);

4,417,664✔

4220

    if (code != 0) {

4,424,296✔

4226

    if (!hasNext || isTaskKilled(pTaskInfo)) {

4,424,296✔

4227

      if (isTaskKilled(pTaskInfo)) {

2,948,352✔

4233

      *pSubTableHasBlock = false;

2,948,008✔

4234

      break;

2,947,320✔

4237

    if (pInput->tblCond.order == TSDB_ORDER_ASC) {

1,475,944✔

4238

      pInput->tblCond.twindows.skey = pInput->pReaderBlock->info.window.ekey + 1;

1,450,944✔

4240

      pInput->tblCond.twindows.ekey = pInput->pReaderBlock->info.window.skey - 1;

25,344✔

4243

    uint32_t status = 0;

1,475,592✔

4244

    code = loadDataBlock(pOperator, &pInfo->base, pInput->pReaderBlock, &status);

1,476,288✔

4245

    if (code != 0) {

1,475,944✔

4250

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

1,475,944✔

4254

    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pInput->pReaderBlock->info.rows == 0) {

1,475,944✔

4255

      continue;

6,336✔

4258

    *pSubTableHasBlock = true;

1,469,952✔

4259

    break;

1,469,952✔

4262

  if (*pSubTableHasBlock) {

4,417,272✔

4263

    pInput->pReaderBlock->info.id.groupId =

2,939,560✔

4264

        tableListGetTableGroupId(pInfo->base.pTableListInfo, pInput->pReaderBlock->info.id.uid);

2,939,216✔

4265

    pOperator->resultInfo.totalRows += pInput->pReaderBlock->info.rows;

1,469,952✔

4267

  if (!pInput->bInMemReader || !*pSubTableHasBlock) {

4,417,272✔

4268

    pAPI->tsdReader.tsdReaderClose(pInput->pReader);

3,318,888✔

4269

    pInput->pReader = NULL;

3,316,128✔

4272

  pInfo->base.dataReader = NULL;

4,415,544✔

4273

  return TSDB_CODE_SUCCESS;

4,416,240✔

4276

static int32_t setGroupStartEndIndex(STableMergeScanInfo* pInfo) {

1,829,784✔

4277

  pInfo->bGroupProcessed = false;

1,829,784✔

4279

  int32_t numOfTables = 0;

1,831,528✔

4280

  int32_t code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

1,805,288✔

4281

  if (code != TSDB_CODE_SUCCESS) {

1,800,384✔

4286

  int32_t i = pInfo->tableStartIndex + 1;

1,800,384✔

4287

  for (; i < numOfTables; ++i) {

2,906,760✔

4288

    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

1,157,248✔

4289

    if (!tableKeyInfo) {

1,156,544✔

4293

    if (tableKeyInfo->groupId != pInfo->groupId) {

1,156,544✔

4294

      break;

55,360✔

4297

  pInfo->tableEndIndex = i - 1;

1,804,872✔

4298

  return TSDB_CODE_SUCCESS;

1,834,584✔

4301

static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {

1,843,296✔

4302

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,791,296✔

4303

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

2,947,656✔

4304

    if (pInput->rowIdx == -1) {

2,945,936✔

4305

      continue;

1,478,048✔

4308

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,469,600✔

4309

      pInput->rowIdx = 0;

1,469,952✔

4310

      pInput->pageIdx = -1;

1,469,952✔

4313

    pInput->pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,469,952✔

4314

    SColumnInfoData* col = taosArrayGet(pInput->pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

1,469,600✔

4315

    if (!col) {

1,469,952✔

4319

    pInput->aTs = (int64_t*)col->pData;

1,469,952✔

4322

  __merge_compare_fn_t mergeCompareFn = (!pSubTblsInfo->pPkOrderInfo) ? subTblRowCompareTsFn : subTblRowCompareTsPkFn;

1,843,296✔

4323

  return tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, mergeCompareFn);

1,843,296✔

4326

static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {

1,830,472✔

4327

  int32_t code = setGroupStartEndIndex(pInfo);

1,830,472✔

4328

  if (code != TSDB_CODE_SUCCESS) {

1,829,384✔

4332

  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));

1,829,384✔

4333

  if (pSubTblsInfo == NULL) {

1,821,112✔

4336

  pSubTblsInfo->pTsOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);

1,821,112✔

4337

  if (!pSubTblsInfo->pTsOrderInfo) {

1,826,304✔

4341

  if (taosArrayGetSize(pInfo->pSortInfo) == 2) {

1,831,496✔

4348

    pSubTblsInfo->pPkOrderInfo = NULL;

1,812,832✔

4350

  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

1,837,392✔

4351

  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));

1,813,520✔

4352

  if (pSubTblsInfo->aInputs == NULL) {

1,813,592✔

4356

  int32_t bufPageSize = pInfo->bufPageSize;

1,815,960✔

4357

  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;

1,832,928✔

4358

  code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);

1,836,384✔

4359

  if (code != TSDB_CODE_SUCCESS) {

1,819,464✔

4364

  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;

1,819,464✔

4365

  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;

1,834,648✔

4367

  pInfo->pSubTablesMergeInfo = pSubTblsInfo;

1,821,176✔

4368

  return TSDB_CODE_SUCCESS;

1,814,944✔

4371

static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {

1,831,864✔

4372

  int32_t        code = TSDB_CODE_SUCCESS;

1,831,864✔

4373

  int32_t        lino = 0;

1,831,864✔

4374

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,831,864✔

4375

  SReadHandle*   pHandle = &pInfo->base.readHandle;

1,830,488✔

4376

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,832,904✔

4378

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

1,839,504✔

4380

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,748,464✔

4381

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

2,938,680✔

4382

    pInput->type = SUB_TABLE_MEM_BLOCK;

2,917,224✔

4384

    code = dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);

2,936,944✔

4385

    QUERY_CHECK_CODE(code, lino, _end);

2,925,184✔

4387

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pReaderBlock);

2,925,184✔

4388

    QUERY_CHECK_CODE(code, lino, _end);

2,947,312✔

4390

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pPageBlock);

2,947,312✔

4391

    QUERY_CHECK_CODE(code, lino, _end);

2,948,352✔

4393

    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);

2,948,352✔

4394

    pInput->pKeyInfo = keyInfo;

2,946,624✔

4396

    if (isTaskKilled(pTaskInfo)) {

2,943,144✔

4400

    if (i + 1 < pSubTblsInfo->numInMemReaders) {

2,935,880✔

4401

      code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, keyInfo, 1, pInput->pReaderBlock,

2,204,128✔

4402

                                           (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

1,104,000✔

4403

      QUERY_CHECK_CODE(code, lino, _end);

1,102,240✔

4404

      pInput->bInMemReader = true;

1,102,240✔

4406

      pInput->pReader = NULL;

1,841,224✔

4407

      pInput->bInMemReader = false;

1,841,920✔

4409

    bool hasNext = true;

2,940,712✔

4410

    code = fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);

2,939,976✔

4411

    QUERY_CHECK_CODE(code, lino, _end);

2,946,624✔

4412

    if (!hasNext) {

2,946,624✔

4413

      pInput->rowIdx = -1;

1,476,672✔

4414

      ++pSubTblsInfo->numSubTablesCompleted;

1,478,056✔

4415

      continue;

1,477,712✔

4417

      pInput->rowIdx = 0;

1,469,952✔

4418

      pInput->pageIdx = -1;

1,469,952✔

4422

_end:

1,842,952✔

4423

  if (code != TSDB_CODE_SUCCESS) {

1,842,952✔

4426

  return code;

1,843,304✔

4429

static void adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

1,469,952✔

4430

  int32_t              code = TSDB_CODE_SUCCESS;

1,469,952✔

4431

  int32_t              lino = 0;

1,469,952✔

4432

  SExecTaskInfo*       pTaskInfo = pOperatorInfo->pTaskInfo;

1,469,952✔

4433

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

1,469,952✔

4434

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,469,952✔

4435

  bool                 hasNext = true;

1,469,952✔

4436

  code = fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);

1,469,952✔

4437

  QUERY_CHECK_CODE(code, lino, _end);

1,469,952✔

4439

  if (!hasNext) {

1,469,952✔

4440

    pInput->rowIdx = -1;

1,469,952✔

4441

    ++pSubTblsInfo->numSubTablesCompleted;

1,469,952✔

4446

_end:

1,469,952✔

4447

  if (code != TSDB_CODE_SUCCESS) {

1,469,952✔

4452

1,469,952✔

4454

static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

267,613,664✔

4455

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

267,613,664✔

4456

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

267,617,488✔

4458

  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

267,612,280✔

4459

  if (pInput->rowIdx < pInputBlock->info.rows - 1) {

267,634,856✔

4460

    ++pInput->rowIdx;

266,164,568✔

4461

  } else if (pInput->rowIdx == pInputBlock->info.rows - 1) {

1,469,952✔

4462

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,469,952✔

4463

      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);

1,469,952✔

4465

    if (pInput->rowIdx != -1) {

1,469,952✔

4476

  return tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));

267,631,032✔

4479

static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {

267,474,688✔

4480

  int32_t            code = TSDB_CODE_SUCCESS;

267,474,688✔

4481

  int32_t            lino = 0;

267,474,688✔

4482

  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

267,474,688✔

4483

  SSDataBlock*       pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

267,515,632✔

4485

  for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {

1,099,932,160✔

4486

    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);

831,818,368✔

4487

    QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);

831,917,704✔

4489

    SColumnInfoData* pSrcColInfo = taosArrayGet(pInputBlock->pDataBlock, i);

831,917,704✔

4490

    QUERY_CHECK_NULL(pSrcColInfo, code, lino, _end, terrno);

831,659,880✔

4491

    bool isNull = colDataIsNull(pSrcColInfo, pInputBlock->info.rows, pInput->rowIdx, NULL);

831,659,880✔

4493

    if (isNull) {

832,075,280✔

4494

      code = colDataSetVal(pColInfo, pBlock->info.rows, NULL, true);

90,190,840✔

4495

      QUERY_CHECK_CODE(code, lino, _end);

90,191,144✔

4497

      if (pSrcColInfo->pData != NULL) {

741,884,440✔

4498

        char* pData = colDataGetData(pSrcColInfo, pInput->rowIdx);

741,893,744✔

4499

        code = colDataSetVal(pColInfo, pBlock->info.rows, pData, false);

742,477,544✔

4500

        QUERY_CHECK_CODE(code, lino, _end);

742,230,616✔

4504

  pBlock->info.dataLoad = 1;

267,619,208✔

4505

  pBlock->info.scanFlag = pInputBlock->info.scanFlag;

267,619,552✔

4506

  pBlock->info.rows += 1;

267,618,512✔

4508

_end:

267,618,536✔

4509

  if (code != TSDB_CODE_SUCCESS) {

267,618,536✔

4512

  return code;

267,612,672✔

4515

static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,

2,983,008✔

4517

  int32_t                 code = TSDB_CODE_SUCCESS;

2,983,008✔

4518

  int32_t                 lino = 0;

2,983,008✔

4519

  STableMergeScanInfo*    pInfo = pOperator->info;

2,983,008✔

4520

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

2,983,360✔

4521

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

2,983,016✔

4522

  bool                    finished = false;

2,981,976✔

4524

  QRY_PARAM_CHECK(pResBlock);

2,981,976✔

4526

  blockDataCleanup(pBlock);

2,981,624✔

4530

      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {

270,493,336✔

4531

        finished = true;

2,983,360✔

4532

        break;

2,983,360✔

4535

      code = appendChosenRowToDataBlock(pSubTblsInfo, pBlock);

267,489,272✔

4536

      QUERY_CHECK_CODE(code, lino, _end);

267,618,176✔

4538

      code = adjustSubTableForNextRow(pOperator, pSubTblsInfo);

267,618,176✔

4539

      QUERY_CHECK_CODE(code, lino, _end);

267,510,712✔

4541

      if (pBlock->info.rows >= capacity) {

267,510,712✔

4546

    if (isTaskKilled(pTaskInfo)) {

2,983,360✔

4550

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pBlock, pTaskInfo);

2,983,360✔

4551

    if (finished || limitReached || pBlock->info.rows > 0) {

2,983,360✔

4556

  if (pBlock->info.rows > 0) {

2,983,360✔

4557

    *pResBlock = pBlock;

1,139,712✔

4560

_end:

1,843,304✔

4561

  if (code != TSDB_CODE_SUCCESS) {

2,983,016✔

4565

  return code;

2,983,016✔

4568

static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {

1,830,464✔

4569

  int32_t              code = TSDB_CODE_SUCCESS;

1,830,464✔

4570

  int32_t              lino = 0;

1,830,464✔

4571

  STableMergeScanInfo* pInfo = pOperator->info;

1,830,464✔

4573

  code = initSubTablesMergeInfo(pInfo);

1,831,880✔

4574

  QUERY_CHECK_CODE(code, lino, _end);

1,813,912✔

4576

  code = initSubTableInputs(pOperator, pInfo);

1,813,912✔

4577

  QUERY_CHECK_CODE(code, lino, _end);

1,842,608✔

4579

  code = openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);

1,842,608✔

4580

  QUERY_CHECK_CODE(code, lino, _end);

1,843,304✔

4582

_end:

1,843,304✔

4583

  if (code != TSDB_CODE_SUCCESS) {

1,843,304✔

4586

  return code;

1,843,296✔

4589

static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {

29,334,262✔

4590

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

29,334,262✔

4591

  if (pSubTblsInfo != NULL) {

29,337,748✔

4592

    tMergeTreeDestroy(&pSubTblsInfo->pTree);

1,843,648✔

4594

    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,790,960✔

4595

      STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

2,948,352✔

4596

      taosMemoryFree(pInput->tblCond.colList);

2,946,288✔

4597

      blockDataDestroy(pInput->pReaderBlock);

2,947,312✔

4598

      blockDataDestroy(pInput->pPageBlock);

2,946,968✔

4599

      taosArrayDestroy(pInput->aBlockPages);

2,948,008✔

4600

      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);

2,948,008✔

4601

      pInput->pReader = NULL;

2,947,656✔

4604

    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);

1,843,304✔

4605

    taosMemoryFree(pSubTblsInfo->aInputs);

1,841,928✔

4607

    taosMemoryFree(pSubTblsInfo);

1,840,896✔

4608

    pInfo->pSubTablesMergeInfo = NULL;

1,840,888✔

4612

29,334,988✔

4614

int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

2,973,608✔

4615

  QRY_PARAM_CHECK(ppRes);

2,973,608✔

4617

  int32_t lino = 0;

2,975,336✔

4618

  int32_t tableListSize = 0;

2,975,336✔

4619

  int64_t st = taosGetTimestampUs();

2,970,136✔

4621

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

2,970,136✔

4622

  STableMergeScanInfo* pInfo = pOperator->info;

2,972,576✔

4624

  if (pOperator->status == OP_EXEC_DONE) {

2,978,144✔

4628

  int32_t code = pOperator->fpSet._openFn(pOperator);

2,940,032✔

4629

  QUERY_CHECK_CODE(code, lino, _end);

2,969,792✔

4631

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

2,969,792✔

4632

  QUERY_CHECK_CODE(code, lino, _end);

2,968,088✔

4634

  if (!pInfo->hasGroupId) {

2,968,088✔

4635

    pInfo->hasGroupId = true;

1,832,160✔

4637

    if (tableListSize == 0) {

1,810,016✔

4638

      setOperatorCompleted(pOperator);

58,432✔

4639

      (*ppRes) = NULL;

58,432✔

4640

      return code;

58,432✔

4643

    pInfo->tableStartIndex = 0;

1,751,584✔

4644

    STableKeyInfo* pTmpGpId = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,754,672✔

4645

    QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

1,741,240✔

4646

    pInfo->groupId = pTmpGpId->groupId;

1,741,240✔

4647

    code = startSubTablesTableMergeScan(pOperator);

1,776,872✔

4648

    QUERY_CHECK_CODE(code, lino, _end);

1,787,936✔

4651

  SSDataBlock* pBlock = NULL;

2,926,304✔

4652

  while (pInfo->tableStartIndex < tableListSize) {

2,983,016✔

4653

    if (isTaskKilled(pTaskInfo)) {

2,982,328✔

4657

    code = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity, &pBlock);

2,980,600✔

4658

    QUERY_CHECK_CODE(code, lino, _end);

2,983,016✔

4660

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

2,983,016✔

4667

    if (pBlock != NULL) {

2,983,016✔

4668

      pBlock->info.id.groupId = pInfo->groupId;

1,139,712✔

4669

      pOperator->resultInfo.totalRows += pBlock->info.rows;

1,139,712✔

4670

      pInfo->bGroupProcessed = true;

1,139,712✔

4671

      break;

1,139,712✔

4674

      stopSubTablesTableMergeScan(pInfo);

1,843,304✔

4675

      if (pInfo->tableEndIndex >= tableListSize - 1) {

1,842,264✔

4676

        setOperatorCompleted(pOperator);

1,785,184✔

4677

        break;

1,787,936✔

4680

      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

55,360✔

4681

      STableKeyInfo* pTmpGpId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

55,360✔

4682

      QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

55,360✔

4684

      pInfo->groupId = pTmpGpId->groupId;

55,360✔

4685

      code = startSubTablesTableMergeScan(pOperator);

55,360✔

4686

      QUERY_CHECK_CODE(code, lino, _end);

55,360✔

4687

      resetLimitInfoForNextGroup(&pInfo->limitInfo);

55,360✔

4691

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

2,927,296✔

4693

_end:

2,925,240✔

4694

  if (code != TSDB_CODE_SUCCESS) {

2,927,656✔

4699

    (*ppRes) = pBlock;

2,927,656✔

4702

  return code;

2,927,304✔

4705

static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {

3,447,670✔

4706

  int32_t              code = TSDB_CODE_SUCCESS;

3,447,670✔

4707

  int32_t              lino = 0;

3,447,670✔

4708

  SOperatorInfo*       pOperator = (SOperatorInfo*)pTableMergeOpInfo;

3,447,670✔

4709

  STableMergeScanInfo* pInfo = pOperator->info;

3,447,670✔

4710

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,447,670✔

4712

  if (pInfo->mSkipTables == NULL) {

3,447,670✔

4713

    pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,

372,071✔

4715

    QUERY_CHECK_NULL(pInfo->mSkipTables, code, lino, _end, terrno);

372,071✔

4717

  int bSkip = 1;

3,447,670✔

4718

  if (pInfo->mSkipTables != NULL) {

3,447,670✔

4719

    code = taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));

3,447,670✔

4720

    if (code == TSDB_CODE_DUP_KEY) {

3,447,670✔

4723

    QUERY_CHECK_CODE(code, lino, _end);

3,447,670✔

4726

_end:

3,447,670✔

4727

  if (code != TSDB_CODE_SUCCESS) {

3,447,670✔

4732

3,447,670✔

4734

static int32_t doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {

154,114,053✔

4735

  STableMergeScanInfo* pInfo = pOperator->info;

154,114,053✔

4736

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

154,120,557✔

4737

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

154,116,666✔

4738

  SSDataBlock*         pBlock = pInfo->pReaderBlock;

154,114,469✔

4739

  int32_t              code = 0;

154,094,022✔

4740

  bool                 hasNext = false;

154,094,022✔

4741

  STsdbReader*         reader = pInfo->base.dataReader;

154,101,466✔

4743

  code = pAPI->tsdReader.tsdNextDataBlock(reader, &hasNext);

154,090,964✔

4744

  if (code != 0) {

154,095,047✔

4751

  if (!hasNext || isTaskKilled(pTaskInfo)) {

154,095,047✔

4752

    if (isTaskKilled(pTaskInfo)) {

28,409,812✔

4756

    *pFinished = true;

28,403,246✔

4757

    return code;

28,393,520✔

4760

  uint32_t status = 0;

125,697,312✔

4761

  code = loadDataBlock(pOperator, &pInfo->base, pBlock, &status);

125,700,828✔

4763

  if (code != TSDB_CODE_SUCCESS) {

125,687,179✔

4764

    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));

16,640✔

4765

    pTaskInfo->code = code;

16,640✔

4766

    return code;

16,640✔

4769

  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

125,670,539✔

4775

  if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {

125,670,539✔

4776

    *pSkipped = true;

5,152,996✔

4777

    return code;

5,171,299✔

4780

  return code;

120,505,713✔

4783

static int32_t getBlockForTableMergeScan(void* param, SSDataBlock** ppBlock) {

152,017,502✔

4784

  STableMergeScanSortSourceParam* source = param;

152,017,502✔

4786

  SOperatorInfo*       pOperator = source->pOperator;

152,017,502✔

4787

  STableMergeScanInfo* pInfo = pOperator->info;

152,017,539✔

4788

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

152,015,093✔

4789

  SSDataBlock*         pBlock = NULL;

151,999,358✔

4790

  int64_t              st = taosGetTimestampUs();

151,991,745✔

4791

  int32_t              code = TSDB_CODE_SUCCESS;

151,991,745✔

4794

    if (pInfo->rtnNextDurationBlocks) {

160,122,575✔

4795

      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d",

6,029,827✔

4798

      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {

6,033,487✔

4799

        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];

2,964,963✔

4800

        ++pInfo->nextDurationBlocksIdx;

2,964,963✔

4802

        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

6,026,225✔

4803

          blockDataDestroy(pInfo->nextDurationBlocks[i]);

2,963,923✔

4804

          pInfo->nextDurationBlocks[i] = NULL;

2,960,629✔

4807

        pInfo->rtnNextDurationBlocks = false;

3,063,766✔

4808

        pInfo->nextDurationBlocksIdx = 0;

3,064,864✔

4809

        pInfo->numNextDurationBlocks = 0;

3,061,936✔

4810

        continue;

3,063,400✔

4813

      bool bFinished = false;

154,111,944✔

4814

      bool bSkipped = false;

154,116,743✔

4816

      code = doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);

154,131,741✔

4817

      if (code != 0) {

154,071,589✔

4818

        return code;

16,640✔

4821

      pBlock = pInfo->pReaderBlock;

154,054,949✔

4822

      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d",

154,078,209✔

4824

      if (bFinished) {

154,132,465✔

4825

        pInfo->bNewFilesetEvent = false;

28,414,074✔

4826

        break;

28,413,214✔

4829

      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {

125,718,391✔

4830

        if (!bSkipped) {

3,067,611✔

4831

          code = createOneDataBlock(pBlock, true, &pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks]);

2,964,963✔

4832

          if (code) {

2,953,574✔

4837

          ++pInfo->numNextDurationBlocks;

2,953,574✔

4838

          if (pInfo->numNextDurationBlocks > 2) {

2,954,306✔

4846

        if (pInfo->bNewFilesetEvent) {

3,061,389✔

4847

          pInfo->rtnNextDurationBlocks = true;

3,060,371✔

4848

          *ppBlock = NULL;

3,058,907✔

4849

          return code;

3,052,319✔

4852

        if (pInfo->bNextDurationBlockEvent) {

366✔

4857

      if (bSkipped) continue;

122,650,294✔

4860

    pBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pBlock->info.id.uid);

120,520,311✔

4862

    pOperator->resultInfo.totalRows += pBlock->info.rows;

120,547,026✔

4863

    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

120,541,004✔

4864

    *ppBlock = pBlock;

120,542,096✔

4866

    return code;

120,541,686✔

4869

  *ppBlock = NULL;

28,413,088✔

4870

  return code;

28,414,074✔

4873

int32_t generateSortByTsPkInfo(SArray* colMatchInfo, int32_t order, SArray** ppSortArray) {

25,330,156✔

4874

  int32_t code = TSDB_CODE_SUCCESS;

25,330,156✔

4875

  int32_t lino = 0;

25,330,156✔

4876

  SArray* pSortInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));

25,330,156✔

4877

  QUERY_CHECK_NULL(pSortInfo, code, lino, _end, terrno);

25,291,626✔

4878

  SBlockOrderInfo biTs = {0};

25,291,626✔

4879

  SBlockOrderInfo biPk = {0};

25,297,781✔

4881

  int32_t tsTargetSlotId = 0;

25,307,071✔

4882

  int32_t pkTargetSlotId = -1;

25,307,071✔

4883

  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {

97,820,307✔

4884

    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);

72,584,496✔

4885

    QUERY_CHECK_NULL(colInfo, code, lino, _end, terrno);

72,588,143✔

4886

    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

72,588,143✔

4887

      tsTargetSlotId = colInfo->dstSlotId;

25,327,333✔

4888

      biTs.order = order;

25,325,070✔

4889

      biTs.slotId = tsTargetSlotId;

25,325,070✔

4890

      biTs.nullFirst = (order == TSDB_ORDER_ASC);

25,325,070✔

4891

      biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, order);

25,325,070✔

4894

    if (colInfo->isPk) {

72,525,504✔

4895

      pkTargetSlotId = colInfo->dstSlotId;

2,356,389✔

4896

      biPk.order = order;

2,356,389✔

4897

      biPk.slotId = pkTargetSlotId;

2,356,389✔

4898

      biPk.nullFirst = (order == TSDB_ORDER_ASC);

2,356,389✔

4899

      biPk.compFn = getKeyComparFunc(colInfo->dataType.type, order);

2,356,389✔

4903

  void* tmp = taosArrayPush(pSortInfo, &biTs);

25,336,638✔

4904

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

25,336,638✔

4905

  if (pkTargetSlotId != -1) {

25,336,638✔

4906

    tmp = taosArrayPush(pSortInfo, &biPk);

2,357,009✔

4907

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

2,357,009✔

4910

  (*ppSortArray) = pSortInfo;

25,337,878✔

4912

_end:

25,312,428✔

4913

  if (code != TSDB_CODE_SUCCESS) {

25,312,866✔

4916

  return code;

25,312,866✔

4919

void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {

4,389,843✔

4920

  STableMergeScanInfo* pTmsInfo = param;

4,389,843✔

4921

  if (type == TSD_READER_NOTIFY_DURATION_START) {

4,389,843✔

4922

    pTmsInfo->bNewFilesetEvent = true;

4,390,941✔

4926

  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);

4,378,540✔

4928

  return;

4,394,805✔

4931

int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

31,678,938✔

4932

  STableMergeScanInfo* pInfo = pOperator->info;

31,678,938✔

4933

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

31,694,448✔

4935

  int32_t code = TSDB_CODE_SUCCESS;

31,681,176✔

4936

  int32_t lino = 0;

31,681,176✔

4937

  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

31,681,176✔

4939

  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));

31,652,863✔

4940

  pInfo->bNewFilesetEvent = false;

31,683,081✔

4941

  pInfo->bNextDurationBlockEvent = false;

31,709,629✔

4943

  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;

31,702,753✔

4944

  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;

31,706,686✔

4946

  pInfo->pSortHandle = NULL;

31,702,803✔

4947

  code = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,

63,408,461✔

4948

                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0, &pInfo->pSortHandle);

31,689,610✔

4949

  if (code) {

31,656,482✔

4953

  if (pInfo->bSortRowId && numOfTable != 1) {

31,656,482✔

4954

    int32_t memSize = 512 * 1024 * 1024;

1,688,294✔

4955

    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);

1,688,294✔

4956

    QUERY_CHECK_CODE(code, lino, _end);

1,688,764✔

4959

  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);

31,668,891✔

4960

  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pOperator);

31,687,866✔

4961

  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);

31,694,861✔

4963

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);

31,671,162✔

4964

  QUERY_CHECK_CODE(code, lino, _end);

31,645,831✔

4966

  STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));

31,645,831✔

4967

  QUERY_CHECK_NULL(param, code, lino, _end, terrno);

31,643,144✔

4968

  param->pOperator = pOperator;

31,643,144✔

4970

  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));

31,658,190✔

4971

  if (ps == NULL) {

31,647,876✔

4976

  ps->param = param;

31,647,876✔

4977

  ps->onlyRef = false;

31,677,093✔

4978

  code = tsortAddSource(pInfo->pSortHandle, ps);

31,688,510✔

4979

  QUERY_CHECK_CODE(code, lino, _end);

31,685,496✔

4981

  if (numOfTable == 1) {

31,685,496✔

4982

    tsortSetSingleTableMerge(pInfo->pSortHandle);

22,786,645✔

4984

    code = tsortOpen(pInfo->pSortHandle);

8,898,851✔

4985

    QUERY_CHECK_CODE(code, lino, _end);

8,903,424✔

4988

_end:

8,895,104✔

4989

  if (code != TSDB_CODE_SUCCESS) {

31,689,487✔

4990

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,320✔

4992

  return code;

31,698,413✔

4995

void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

31,698,570✔

4996

  STableMergeScanInfo* pInfo = pOperator->info;

31,698,570✔

4997

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

31,698,970✔

4998

  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));

31,702,264✔

5000

  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

31,707,022✔

5001

  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;

31,696,042✔

5002

  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;

31,696,408✔

5003

  pInfo->sortExecInfo.loops += sortExecInfo.loops;

31,692,085✔

5004

  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;

31,690,608✔

5005

  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

31,686,709✔

5007

  tsortDestroySortHandle(pInfo->pSortHandle);

31,691,260✔

5008

  pInfo->pSortHandle = NULL;

31,696,002✔

5009

31,693,448✔

5011

void startGroupTableMergeScan(SOperatorInfo* pOperator) {

28,622,138✔

5012

  int32_t              code = TSDB_CODE_SUCCESS;

28,622,138✔

5013

  int32_t              lino = 0;

28,622,138✔

5014

  STableMergeScanInfo* pInfo = pOperator->info;

28,622,138✔

5015

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,615,741✔

5016

  SReadHandle*         pHandle = &pInfo->base.readHandle;

28,603,155✔

5017

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,582,730✔

5018

  qDebug("%s table merge scan start group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,620,274✔

5021

    int32_t numOfTables = 0;

28,625,093✔

5022

    code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

28,608,955✔

5023

    QUERY_CHECK_CODE(code, lino, _end);

28,616,504✔

5025

    int32_t i = pInfo->tableStartIndex + 1;

28,616,504✔

5026

    for (; i < numOfTables; ++i) {

132,306,174✔

5027

      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

109,874,319✔

5028

      QUERY_CHECK_NULL(tableKeyInfo, code, lino, _end, terrno);

109,872,909✔

5029

      if (tableKeyInfo->groupId != pInfo->groupId) {

109,872,909✔

5030

        break;

6,182,249✔

5033

    pInfo->tableEndIndex = i - 1;

28,614,104✔

5035

  pInfo->bGroupProcessed = false;

28,642,137✔

5036

  int32_t tableStartIdx = pInfo->tableStartIndex;

28,643,524✔

5037

  int32_t tableEndIdx = pInfo->tableEndIndex;

28,638,717✔

5039

  int32_t        numOfTable = tableEndIdx - tableStartIdx + 1;

28,641,829✔

5040

  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);

28,641,829✔

5041

  code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,

57,251,232✔

5042

                                       (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->mSkipTables);

28,624,988✔

5043

  QUERY_CHECK_CODE(code, lino, _end);

28,618,170✔

5044

  if (pInfo->filesetDelimited) {

28,618,170✔

5045

    pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);

28,612,991✔

5047

  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);

28,637,058✔

5049

  code = startDurationForGroupTableMergeScan(pOperator);

28,570,462✔

5050

  QUERY_CHECK_CODE(code, lino, _end);

28,597,637✔

5052

_end:

28,589,317✔

5053

  if (code != TSDB_CODE_SUCCESS) {

28,596,574✔

5054

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,320✔

5055

    pTaskInfo->code = code;

8,320✔

5056

    T_LONG_JMP(pTaskInfo->env, code);

8,320✔

5058

28,588,254✔

5060

void stopGroupTableMergeScan(SOperatorInfo* pOperator) {

28,633,915✔

5061

  STableMergeScanInfo* pInfo = pOperator->info;

28,633,915✔

5062

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,634,164✔

5063

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,634,164✔

5065

  stopDurationForGroupTableMergeScan(pOperator);

28,631,699✔

5067

  if (pInfo->base.dataReader != NULL) {

28,630,472✔

5068

    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);

28,631,578✔

5069

    pInfo->base.dataReader = NULL;

28,599,748✔

5071

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

28,600,704✔

5072

    if (pInfo->nextDurationBlocks[i]) {

1,040✔

5073

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

1,040✔

5074

      pInfo->nextDurationBlocks[i] = NULL;

1,040✔

5076

    pInfo->numNextDurationBlocks = 0;

1,040✔

5077

    pInfo->nextDurationBlocksIdx = 0;

1,040✔

5079

  resetLimitInfoForNextGroup(&pInfo->limitInfo);

28,615,445✔

5080

  taosHashCleanup(pInfo->mSkipTables);

28,592,971✔

5081

  pInfo->mSkipTables = NULL;

28,590,560✔

5082

  qDebug("%s table merge scan stop group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,590,076✔

5083

28,639,977✔

5087

SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,

67,283,690✔

5089

  STableMergeScanInfo* pInfo = pOperator->info;

67,283,690✔

5090

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

67,290,850✔

5091

  STupleHandle*        pTupleHandle = NULL;

67,291,558✔

5093

  blockDataCleanup(pResBlock);

67,302,078✔

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));

3,387,123✔

5101

        T_LONG_JMP(pOperator->pTaskInfo->env, code);

3,387,123✔

5103

      if (pTupleHandle == NULL) {

2,147,483,647✔

5104

        break;

51,510,787✔

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,781,005✔

5118

    if (tsortIsClosed(pHandle)) {

67,291,792✔

5123

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);

67,298,152✔

5124

    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,

67,291,784✔

5126

    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {

67,311,515✔

5130

  return (pResBlock->info.rows > 0) ? pResBlock : NULL;

67,311,515✔

5133

int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

61,267,068✔

5134

  if (pOperator->status == OP_EXEC_DONE) {

61,267,068✔

5135

    (*ppRes) = NULL;

41,694✔

5136

    return TSDB_CODE_SUCCESS;

41,694✔

5139

  int32_t              code = TSDB_CODE_SUCCESS;

61,209,576✔

5140

  int32_t              lino = 0;

61,209,576✔

5141

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

61,209,576✔

5142

  STableMergeScanInfo* pInfo = pOperator->info;

61,226,100✔

5144

  code = pOperator->fpSet._openFn(pOperator);

61,222,580✔

5145

  QUERY_CHECK_CODE(code, lino, _end);

61,200,154✔

5147

  int64_t st = taosGetTimestampUs();

61,216,118✔

5149

  int32_t tableListSize = 0;

61,216,118✔

5150

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

61,210,398✔

5151

  QUERY_CHECK_CODE(code, lino, _end);

61,218,848✔

5153

  if (!pInfo->hasGroupId) {

61,218,848✔

5154

    pInfo->hasGroupId = true;

25,620,365✔

5156

    if (tableListSize == 0) {

25,590,608✔

5157

      setOperatorCompleted(pOperator);

3,181,568✔

5158

      (*ppRes) = NULL;

3,180,366✔

5159

      return code;

3,180,358✔

5161

    pInfo->tableStartIndex = 0;

22,409,040✔

5162

    STableKeyInfo* tmp = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

22,440,606✔

5163

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

22,415,406✔

5164

    pInfo->groupId = tmp->groupId;

22,415,406✔

5165

    startGroupTableMergeScan(pOperator);

22,438,489✔

5168

  SSDataBlock* pBlock = NULL;

58,033,720✔

5169

  while (pInfo->tableStartIndex < tableListSize) {

67,282,605✔

5170

    if (isTaskKilled(pTaskInfo)) {

67,299,892✔

5174

    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,

67,289,610✔

5176

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

67,310,529✔

5177

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,803,797✔

5178

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

1,803,797✔

5179

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

1,803,797✔

5181

    if (pBlock != NULL) {

67,306,727✔

5182

      pBlock->info.id.groupId = pInfo->groupId;

35,607,660✔

5183

      pOperator->resultInfo.totalRows += pBlock->info.rows;

35,607,660✔

5184

      pInfo->bGroupProcessed = true;

35,607,660✔

5185

      break;

35,607,660✔

5187

      if (pInfo->bNewFilesetEvent) {

31,699,067✔

5188

        stopDurationForGroupTableMergeScan(pOperator);

3,065,538✔

5189

        code = startDurationForGroupTableMergeScan(pOperator);

3,063,342✔

5190

        QUERY_CHECK_CODE(code, lino, _end);

3,066,636✔

5193

        stopGroupTableMergeScan(pOperator);

28,636,121✔

5194

        if (pInfo->tableEndIndex >= tableListSize - 1) {

28,630,782✔

5195

          setOperatorCompleted(pOperator);

22,447,543✔

5196

          break;

22,446,834✔

5199

        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

6,182,249✔

5200

        STableKeyInfo* tmp = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

6,182,249✔

5201

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

6,182,249✔

5202

        pInfo->groupId = tmp->groupId;

6,182,249✔

5203

        startGroupTableMergeScan(pOperator);

6,182,249✔

5204

        resetLimitInfoForNextGroup(&pInfo->limitInfo);

6,181,745✔

5209

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

58,053,800✔

5211

_end:

58,052,019✔

5212

  if (code != TSDB_CODE_SUCCESS) {

58,053,141✔

5217

    (*ppRes) = pBlock;

58,053,141✔

5220

  return code;

58,053,795✔

5229

void destroyTableMergeScanOperatorInfo(void* param) {

25,351,435✔

5230

  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;

25,351,435✔

5233

  if (pTableScanInfo->base.readerAPI.tsdReaderClose != NULL) {

25,351,435✔

5234

    pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);

25,352,606✔

5235

    pTableScanInfo->base.dataReader = NULL;

25,342,320✔

5238

  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {

25,346,207✔

5245

  tsortDestroySortHandle(pTableScanInfo->pSortHandle);

25,346,333✔

5246

  pTableScanInfo->pSortHandle = NULL;

25,344,612✔

5247

  taosHashCleanup(pTableScanInfo->mSkipTables);

25,350,273✔

5248

  pTableScanInfo->mSkipTables = NULL;

25,345,099✔

5249

  blockDataDestroy(pTableScanInfo->pSortInputBlock);

25,344,734✔

5250

  pTableScanInfo->pSortInputBlock = NULL;

25,341,611✔

5253

  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);

25,347,220✔

5255

  blockDataDestroy(pTableScanInfo->pResBlock);

25,349,313✔

5256

  pTableScanInfo->pResBlock = NULL;

25,346,173✔

5259

  blockDataDestroy(pTableScanInfo->pReaderBlock);

25,347,751✔

5260

  pTableScanInfo->pReaderBlock = NULL;

25,349,101✔

5261

  taosArrayDestroy(pTableScanInfo->pSortInfo);

25,351,515✔

5263

  stopSubTablesTableMergeScan(pTableScanInfo);

25,345,259✔

5265

  taosMemoryFreeClear(param);

25,340,180✔

5266

25,340,071✔

5268

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {

1,200,223✔

5269

  if (pOptr == NULL) {

1,200,223✔

5274

  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));

1,200,223✔

5275

  if (!execInfo) {

1,197,645✔

5278

  STableMergeScanInfo* pInfo = pOptr->info;

1,197,645✔

5279

  execInfo->blockRecorder = pInfo->base.readRecorder;

1,198,385✔

5280

  execInfo->sortExecInfo = pInfo->sortExecInfo;

1,198,759✔

5282

  *pOptrExplain = execInfo;

1,201,337✔

5283

  *len = sizeof(STableMergeScanExecInfo);

1,198,759✔

5285

  return TSDB_CODE_SUCCESS;

1,199,117✔

5288

static int32_t resetTableMergeScanOperatorState(SOperatorInfo* pOper) {

2,142,309✔

5289

  int32_t         code = TSDB_CODE_SUCCESS;

2,142,309✔

5290

  STableMergeScanInfo*   pInfo = pOper->info;

2,142,309✔

5291

  pOper->status = OP_NOT_OPENED;

2,142,309✔

5293

  pInfo->tableEndIndex = 0;

2,142,309✔

5294

  pInfo->tableStartIndex = 0;

2,142,309✔

5295

  pInfo->hasGroupId = false;

2,142,309✔

5296

  if (pInfo->base.readerAPI.tsdReaderClose) {

2,142,309✔

5297

    pInfo->base.readerAPI.tsdReaderClose(pInfo->base.dataReader);

2,142,309✔

5299

  pInfo->base.dataReader = NULL;

2,142,309✔

5300

  pInfo->base.scanFlag = MAIN_SCAN;

2,142,309✔

5302

  pInfo->base.limitInfo = (SLimitInfo){0};

2,142,309✔

5303

  pInfo->base.limitInfo.limit.limit = -1;

2,142,309✔

5304

  pInfo->base.limitInfo.slimit.limit = -1;

2,142,309✔

5306

  tableListDestroy(pInfo->base.pTableListInfo);

2,142,309✔

5307

  pInfo->base.pTableListInfo = tableListCreate();

2,142,309✔

5308

  if (!pInfo->base.pTableListInfo) {

2,142,309✔

5312

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

2,142,309✔

5314

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

2,142,309✔

5315

  code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

4,284,618✔

5317

                                  pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

4,284,618✔

5318

  if (code) {

2,142,309✔

5323

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

2,142,309✔

5324

  cleanupQueryTableDataCond(&pInfo->base.cond);

2,142,309✔

5325

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, &pInfo->base.readHandle, false);

2,142,309✔

5326

  if (code) {

2,142,309✔

5331

  tsortDestroySortHandle(pInfo->pSortHandle);

2,142,309✔

5332

  pInfo->pSortHandle = NULL;

2,142,309✔

5334

  taosHashCleanup(pInfo->mSkipTables);

2,142,309✔

5335

  pInfo->mSkipTables = NULL;

2,142,309✔

5336

  pOper->resultInfo.totalRows = 0;

2,142,309✔

5338

  pInfo->bGroupProcessed = false;

2,142,309✔

5339

  pInfo->bNewFilesetEvent = false;

2,142,309✔

5340

  pInfo->bNextDurationBlockEvent = false;

2,142,309✔

5341

  pInfo->rtnNextDurationBlocks = false;

2,142,309✔

5342

  pInfo->nextDurationBlocksIdx = 0;

2,142,309✔

5343

  pInfo->bSortRowId = false;

2,142,309✔

5345

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

2,142,309✔

5351

  pInfo->numNextDurationBlocks = 0;

2,142,309✔

5353

  stopSubTablesTableMergeScan(pInfo);

2,142,309✔

5354

  return code;

2,142,309✔

5357

int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

25,309,783✔

5360

  QRY_PARAM_CHECK(pOptrInfo);

25,309,783✔

5362

  int32_t              code = TSDB_CODE_SUCCESS;

25,322,153✔

5363

  int32_t              lino = 0;

25,322,153✔

5364

  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));

25,322,153✔

5365

  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

25,292,422✔

5366

  if (pInfo == NULL || pOperator == NULL) {

25,242,665✔

5371

  pOperator->pPhyNode = pTableScanNode;

25,246,781✔

5372

  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

25,270,259✔

5374

  int32_t numOfCols = 0;

25,277,438✔

5375

  code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,

25,272,760✔

5377

  QUERY_CHECK_CODE(code, lino, _error);

25,289,456✔

5379

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle, false);

25,289,456✔

5380

  QUERY_CHECK_CODE(code, lino, _error);

25,268,087✔

5382

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

25,268,087✔

5383

    SExprSupp* pSup = &pInfo->base.pseudoSup;

10,617,159✔

5384

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->pExprInfo, &pSup->numOfExprs);

10,594,266✔

5385

    QUERY_CHECK_CODE(code, lino, _error);

10,591,341✔

5387

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

10,591,341✔

5389

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

10,576,289✔

5392

  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};

25,221,266✔

5394

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

25,187,939✔

5399

  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;

25,187,939✔

5400

  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

25,280,914✔

5401

  pInfo->base.scanFlag = MAIN_SCAN;

25,175,166✔

5402

  pInfo->base.readHandle = *readHandle;

25,221,474✔

5404

  pInfo->readIdx = -1;

25,292,840✔

5406

  pInfo->base.limitInfo.limit.limit = -1;

25,165,958✔

5407

  pInfo->base.limitInfo.slimit.limit = -1;

25,191,931✔

5408

  pInfo->base.pTableListInfo = pTableListInfo;

25,230,990✔

5410

  pInfo->sample.sampleRatio = pTableScanNode->ratio;

25,302,992✔

5411

  pInfo->sample.seed = taosGetTimestampSec();

25,192,142✔

5413

  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,

25,290,545✔

5414

                            pTaskInfo->pStreamRuntimeInfo);

25,288,649✔

5415

  QUERY_CHECK_CODE(code, lino, _error);

25,252,638✔

5417

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

25,252,638✔

5419

  pInfo->mergeLimit = -1;

25,255,482✔

5420

  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;

25,293,332✔

5421

  if (hasLimit) {

25,206,334✔

5422

    pInfo->mergeLimit = pInfo->limitInfo.limit.offset != -1

4,168,537✔

5423

                            ? pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset

795,354✔

5424

                            : pInfo->limitInfo.limit.limit;

2,875,950✔

5425

    pInfo->mSkipTables = NULL;

2,085,935✔

5428

  initResultSizeInfo(&pOperator->resultInfo, 1024);

25,204,148✔

5429

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

25,277,839✔

5430

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

25,303,039✔

5431

  code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

25,246,303✔

5432

  QUERY_CHECK_CODE(code, lino, _error);

25,339,200✔

5433

  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {

25,339,200✔

5434

    pInfo->bSortRowId = true;

5,232,743✔

5436

    pInfo->bSortRowId = false;

20,107,645✔

5439

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

25,321,401✔

5440

  QUERY_CHECK_CODE(code, lino, _error);

25,291,720✔

5442

  code = generateSortByTsPkInfo(pInfo->base.matchInfo.pList, pInfo->base.cond.order, &pInfo->pSortInfo);

25,291,720✔

5443

  QUERY_CHECK_CODE(code, lino, _error);

25,307,584✔

5445

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pReaderBlock);

25,307,584✔

5446

  QUERY_CHECK_CODE(code, lino, _error);

25,313,415✔

5448

  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;

25,313,415✔

5450

  int32_t  rowSize = pInfo->pResBlock->info.rowSize;

25,330,934✔

5451

  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);

25,312,709✔

5453

  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);

25,247,395✔

5456

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pSortInputBlock);

25,319,335✔

5457

  QUERY_CHECK_CODE(code, lino, _error);

25,328,653✔

5459

  if (!tsExperimental) {

25,328,653✔

5462

    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

25,328,653✔

5465

  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,

25,311,514✔

5467

  pOperator->exprSupp.numOfExprs = numOfCols;

25,284,717✔

5469

  pOperator->fpSet = createOperatorFpSet(

25,269,640✔

5470

      optrDummyOpenFn, pTableScanNode->paraTablesSort ? doTableMergeScanParaSubTablesNext : doTableMergeScanNext, NULL,

25,322,781✔

5473

  setOperatorResetStateFn(pOperator, resetTableMergeScanOperatorState);

25,282,789✔

5474

  pOperator->cost.openCost = 0;

25,316,605✔

5476

  *pOptrInfo = pOperator;

25,319,247✔

5477

  return code;

25,239,361✔

5516

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {

25,254✔

5517

  if (scanCols != NULL) {

25,254✔

5518

    SNode* pNode = NULL;

14,583✔

5519

    FOREACH(pNode, scanCols) {

34,050✔

5520

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

19,467✔

5523

      STargetNode* targetNode = (STargetNode*)pNode;

19,467✔

5524

      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {

19,467✔

5527

      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);

19,467✔

5528

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

19,467✔

5529

        supp->dbNameSlotId = targetNode->slotId;

9,600✔

5530

      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

9,867✔

5531

        supp->stbNameSlotId = targetNode->slotId;

9,867✔

5535

  return TSDB_CODE_SUCCESS;

25,254✔

5538

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {

25,254✔

5539

  if (pseudoCols != NULL) {

25,254✔

5540

    SNode* pNode = NULL;

25,254✔

5541

    FOREACH(pNode, pseudoCols) {

50,508✔

5542

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

25,254✔

5545

      STargetNode* targetNode = (STargetNode*)pNode;

25,254✔

5546

      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {

25,254✔

5549

      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);

25,254✔

5550

      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {

25,254✔

5551

        supp->tbCountSlotId = targetNode->slotId;

25,254✔

5555

  return TSDB_CODE_SUCCESS;

25,254✔

5558

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {

25,254✔

5559

  if (groupTags != NULL) {

25,254✔

5560

    SNode* pNode = NULL;

14,583✔

5561

    FOREACH(pNode, groupTags) {

34,050✔

5562

      if (nodeType(pNode) != QUERY_NODE_COLUMN) {

19,467✔

5565

      SColumnNode* colNode = (SColumnNode*)pNode;

19,467✔

5566

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

19,467✔

5567

        supp->groupByDbName = true;

9,600✔

5569

      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

19,467✔

5570

        supp->groupByStbName = true;

9,867✔

5574

    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);

10,671✔

5575

    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);

10,671✔

5577

  return TSDB_CODE_SUCCESS;

25,254✔

5580

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,

25,254✔

5582

  int32_t code = 0;

25,254✔

5583

  code = tblCountScanGetInputs(groupTags, tableName, supp);

25,254✔

5584

  if (code != TSDB_CODE_SUCCESS) {

25,254✔

5589

  supp->dbNameSlotId = -1;

25,254✔

5590

  supp->stbNameSlotId = -1;

25,254✔

5591

  supp->tbCountSlotId = -1;

25,254✔

5593

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);

25,254✔

5594

  if (code != TSDB_CODE_SUCCESS) {

25,254✔

5599

  code = tblCountScanGetCountSlotId(pseudoCols, supp);

25,254✔

5600

  if (code != TSDB_CODE_SUCCESS) {

25,254✔

5604

  return code;

25,254✔

5607

int32_t createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,

25,083✔

5609

  QRY_PARAM_CHECK(pOptrInfo);

25,083✔

5611

  int32_t                      code = TSDB_CODE_SUCCESS;

25,254✔

5612

  int32_t                      lino = 0;

25,254✔

5613

  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;

25,254✔

5614

  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));

25,254✔

5615

  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

25,254✔

5616

  if (!pInfo || !pOperator) {

25,254✔

5621

  pInfo->readHandle = *readHandle;

25,254✔

5623

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

25,254✔

5624

  initResultSizeInfo(&pOperator->resultInfo, 1);

25,254✔

5625

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

25,254✔

5626

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

25,254✔

5628

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

25,254✔

5629

  QUERY_CHECK_CODE(code, lino, _error);

25,254✔

5631

  code = getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,

25,254✔

5634

  QUERY_CHECK_CODE(code, lino, _error);

25,254✔

5636

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,

25,254✔

5638

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScanNext, NULL, destoryTableCountScanOperator,

25,254✔

5640

  *pOptrInfo = pOperator;

25,254✔

5641

  return code;

25,254✔

5655

int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,

35,251✔

5657

  int32_t code = TSDB_CODE_SUCCESS;

35,251✔

5658

  int32_t lino = 0;

35,251✔

5659

  if (pSupp->dbNameSlotId != -1) {

35,251✔

5660

    QUERY_CHECK_CONDITION((strlen(dbName) > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

14,494✔

5661

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);

14,494✔

5662

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

14,494✔

5664

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

14,494✔

5665

    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

14,494✔

5667

    varDataSetLen(varDbName, strlen(dbName));

14,494✔

5668

    code = colDataSetVal(colInfoData, 0, varDbName, false);

14,494✔

5669

    QUERY_CHECK_CODE(code, lino, _end);

14,494✔

5672

  if (pSupp->stbNameSlotId != -1) {

35,251✔

5673

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);

18,906✔

5674

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

18,906✔

5675

    if (strlen(stbName) != 0) {

18,906✔

5676

      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

13,592✔

5677

      tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);

13,592✔

5678

      varDataSetLen(varStbName, strlen(stbName));

13,592✔

5679

      code = colDataSetVal(colInfoData, 0, varStbName, false);

13,592✔

5680

      QUERY_CHECK_CODE(code, lino, _end);

13,592✔

5686

  if (pSupp->tbCountSlotId != -1) {

35,251✔

5687

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);

35,251✔

5688

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

35,251✔

5689

    code = colDataSetVal(colInfoData, 0, (char*)&count, false);

35,251✔

5690

    QUERY_CHECK_CODE(code, lino, _end);

35,251✔

5692

  pRes->info.rows = 1;

35,251✔

5694

_end:

35,251✔

5695

  if (code != TSDB_CODE_SUCCESS) {

35,251✔

5698

  return code;

35,251✔

5701

static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {

10,692✔

5702

  STableCountScanSupp* pSupp = &pInfo->supp;

10,692✔

5703

  SSDataBlock*         pRes = pInfo->pRes;

10,692✔

5705

  size_t infodbTableNum;

10,692✔

5706

  getInfosDbMeta(NULL, &infodbTableNum);

10,692✔

5707

  infodbTableNum -= 1;

10,692✔

5708

  size_t perfdbTableNum;

10,692✔

5709

  getPerfDbMeta(NULL, &perfdbTableNum);

10,692✔

5711

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

10,692✔

5712

    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);

9,183✔

5713

    return (pRes->info.rows > 0) ? pRes : NULL;

9,183✔

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,183✔

5756

  int32_t        code = TSDB_CODE_SUCCESS;

9,183✔

5757

  int32_t        lino = 0;

9,183✔

5758

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,183✔

5759

  if (pInfo->currGrpIdx == 0) {

9,183✔

5760

    uint64_t groupId = 0;

3,061✔

5761

    if (pSupp->groupByDbName) {

3,061✔

5762

      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));

2,086✔

5764

      groupId = calcGroupId("", 0);

975✔

5767

    pRes->info.id.groupId = groupId;

3,061✔

5768

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

3,061✔

5769

    QUERY_CHECK_CODE(code, lino, _end);

3,061✔

5770

  } else if (pInfo->currGrpIdx == 1) {

6,122✔

5771

    uint64_t groupId = 0;

3,061✔

5772

    if (pSupp->groupByDbName) {

3,061✔

5773

      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));

2,086✔

5775

      groupId = calcGroupId("", 0);

975✔

5778

    pRes->info.id.groupId = groupId;

3,061✔

5779

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

3,061✔

5780

    QUERY_CHECK_CODE(code, lino, _end);

3,061✔

5782

    setOperatorCompleted(pOperator);

3,061✔

5784

  pInfo->currGrpIdx++;

9,183✔

5786

_end:

9,183✔

5787

  if (code != TSDB_CODE_SUCCESS) {

9,183✔

5792

9,183✔

5794

static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

60,505✔

5795

  int32_t                      code = TSDB_CODE_SUCCESS;

60,505✔

5796

  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;

60,505✔

5797

  STableCountScanOperatorInfo* pInfo = pOperator->info;

60,505✔

5798

  STableCountScanSupp*         pSupp = &pInfo->supp;

60,505✔

5799

  SSDataBlock*                 pRes = pInfo->pRes;

60,505✔

5801

  blockDataCleanup(pRes);

60,505✔

5802

  QRY_PARAM_CHECK(ppRes);

60,505✔

5804

  if (pOperator->status == OP_EXEC_DONE) {

60,505✔

5805

    return code;

14,429✔

5808

  if (pInfo->readHandle.mnd != NULL) {

46,076✔

5809

    (*ppRes) = buildSysDbTableCount(pOperator, pInfo);

10,692✔

5810

    return code;

10,692✔

5813

  code = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);

35,384✔

5814

  if (code != TSDB_CODE_SUCCESS) {

35,384✔

5819

  if (pRes->info.rows > 0) {

35,384✔

5820

    *ppRes = pRes;

27,620✔

5823

  return code;

35,384✔

5826

static int32_t buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

35,384✔

5828

  int32_t        code = TSDB_CODE_SUCCESS;

35,384✔

5829

  int32_t        lino = 0;

35,384✔

5830

  const char*    db = NULL;

35,384✔

5831

  int32_t        vgId = 0;

35,384✔

5832

  char           dbName[TSDB_DB_NAME_LEN] = {0};

35,384✔

5833

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

35,384✔

5834

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

35,384✔

5837

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);

35,384✔

5838

  SName sn = {0};

35,384✔

5840

  code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

35,384✔

5841

  QUERY_CHECK_CODE(code, lino, _end);

35,384✔

5843

  code = tNameGetDbName(&sn, dbName);

35,384✔

5844

  QUERY_CHECK_CODE(code, lino, _end);

35,384✔

5846

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

35,384✔

5847

    code = buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);

26,222✔

5849

    code = buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);

9,162✔

5852

_end:

35,384✔

5853

  if (code != TSDB_CODE_SUCCESS) {

35,384✔

5858

  return code;

35,384✔

5861

static int32_t buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

26,222✔

5863

  int32_t        code = TSDB_CODE_SUCCESS;

26,222✔

5864

  int32_t        lino = 0;

26,222✔

5865

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

26,222✔

5866

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

26,222✔

5868

  if (pSupp->groupByStbName) {

26,222✔

5869

    if (pInfo->stbUidList == NULL) {

22,464✔

5870

      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));

7,764✔

5871

      QUERY_CHECK_NULL(pInfo->stbUidList, code, lino, _end, terrno);

7,764✔

5872

      code = pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList);

7,764✔

5873

      QUERY_CHECK_CODE(code, lino, _end);

7,764✔

5876

    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {

22,464✔

5877

      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);

13,592✔

5878

      code = buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);

13,592✔

5879

      QUERY_CHECK_CODE(code, lino, _end);

13,592✔

5881

      pInfo->currGrpIdx++;

13,592✔

5882

    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {

8,872✔

5883

      code = buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);

7,764✔

5884

      QUERY_CHECK_CODE(code, lino, _end);

7,764✔

5886

      pInfo->currGrpIdx++;

7,764✔

5888

      setOperatorCompleted(pOperator);

1,108✔

5891

    uint64_t groupId = calcGroupId(dbName, strlen(dbName));

3,758✔

5892

    pRes->info.id.groupId = groupId;

3,758✔

5894

    int64_t dbTableCount = 0;

3,758✔

5895

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);

3,758✔

5896

    code = fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);

3,758✔

5897

    QUERY_CHECK_CODE(code, lino, _end);

3,758✔

5898

    setOperatorCompleted(pOperator);

3,758✔

5901

_end:

26,222✔

5902

  if (code != TSDB_CODE_SUCCESS) {

26,222✔

5906

  return code;

26,222✔

5909

static int32_t buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,162✔

5911

  int32_t        code = TSDB_CODE_SUCCESS;

9,162✔

5912

  int32_t        lino = 0;

9,162✔

5913

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,162✔

5914

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

9,162✔

5916

  if (strlen(pSupp->dbNameFilter) != 0 && strcmp(dbName, pSupp->dbNameFilter) != 0) {

9,162✔

5920

  } else if (strlen(pSupp->stbNameFilter) != 0) {

9,162✔

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,144✔

5939

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);

6,144✔

5940

    code = fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);

6,144✔

5941

    QUERY_CHECK_CODE(code, lino, _end);

6,144✔

5944

_end:

9,162✔

5945

  if (code != TSDB_CODE_SUCCESS) {

9,162✔

5950

  setOperatorCompleted(pOperator);

9,162✔

5951

  return code;

9,162✔

5954

static int32_t buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

7,764✔

5956

  int32_t code = TSDB_CODE_SUCCESS;

7,764✔

5957

  int32_t lino = 0;

7,764✔

5958

  char    fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

7,764✔

5959

  if (pSupp->groupByDbName) {

7,764✔

5960

    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");

3,756✔

5963

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

7,764✔

5964

  pRes->info.id.groupId = groupId;

7,764✔

5966

  int64_t numOfTables = 0;

7,764✔

5967

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);

7,764✔

5969

  if (numOfTables != 0) {

7,764✔

5970

    code = fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);

1,108✔

5973

_end:

6,656✔

5974

  if (code != TSDB_CODE_SUCCESS) {

7,764✔

5977

  return code;

7,764✔

5980

static int32_t buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

13,592✔

5982

  int32_t code = TSDB_CODE_SUCCESS;

13,592✔

5983

  int32_t lino = 0;

13,592✔

5984

  char    stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

13,592✔

5985

  code = pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);

13,592✔

5986

  QUERY_CHECK_CODE(code, lino, _end);

13,592✔

5988

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

13,592✔

5989

  if (pSupp->groupByDbName) {

13,592✔

5990

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));

6,260✔

5992

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));

7,332✔

5995

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

13,592✔

5996

  pRes->info.id.groupId = groupId;

13,592✔

5998

  int64_t ctbNum = 0;

13,592✔

5999

  code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL, NULL);

13,592✔

6000

  QUERY_CHECK_CODE(code, lino, _end);

13,592✔

6001

  code = fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);

13,592✔

6002

  QUERY_CHECK_CODE(code, lino, _end);

13,592✔

6004

_end:

13,592✔

6005

  if (code != TSDB_CODE_SUCCESS) {

13,592✔

6008

  return code;

13,592✔

6011

static void destoryTableCountScanOperator(void* param) {

25,254✔

6012

  STableCountScanOperatorInfo* pTableCountScanInfo = param;

25,254✔

6013

  blockDataDestroy(pTableCountScanInfo->pRes);

25,254✔

6015

  taosArrayDestroy(pTableCountScanInfo->stbUidList);

25,254✔

6016

  taosMemoryFreeClear(param);

25,254✔

6017

25,254✔

taosdata / TDengine / #4948

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4948

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