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

79

bool processBlockWithProbability(const SSampleExecInfo* pInfo) {

682,915,257✔

88

  return true;

682,915,257✔

92

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

67,828✔

93

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

67,828✔

96

67,828✔

98

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

689,350,609✔

99

  int32_t     code = TSDB_CODE_SUCCESS;

689,350,609✔

100

  STimeWindow w = {0};

689,350,609✔

103

  if (pInterval->interval == 0) {

689,498,791✔

104

    *overlap = false;

649,869,891✔

105

    return code;

649,858,896✔

108

  if (order == TSDB_ORDER_ASC) {

39,626,161✔

109

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

37,503,254✔

110

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

37,494,230✔

115

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

37,494,946✔

116

      *overlap = true;

32,590,691✔

117

      return code;

32,590,631✔

121

      getNextTimeWindow(pInterval, &w, order);

4,910,243✔

122

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

4,911,677✔

123

        break;

4,633,343✔

126

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

277,408✔

130

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

277,408✔

131

        *overlap = true;

277,408✔

132

        return code;

277,408✔

136

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

2,122,907✔

137

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

2,124,992✔

142

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

2,122,916✔

143

      *overlap = true;

1,666,867✔

144

      return code;

1,666,867✔

148

      getNextTimeWindow(pInterval, &w, order);

458,356✔

149

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

457,102✔

150

        break;

454,444✔

153

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

2,658✔

157

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

2,658✔

158

        *overlap = true;

2,658✔

159

        return code;

2,658✔

164

  *overlap = false;

5,087,787✔

165

  return code;

5,089,041✔

169

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

177,232,490✔

170

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

177,232,490✔

174

  int64_t buf[2] = {0};

177,234,413✔

175

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

177,238,852✔

177

  STableScanInfo* pTableScanInfo = pOperator->info;

177,231,556✔

179

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

177,228,360✔

182

  if (p1 == NULL) {

177,238,800✔

183

    return NULL;

51,179,810✔

186

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

126,058,990✔

187

  if (NULL == *pPage) {

126,047,247✔

191

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

126,053,934✔

194

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

13,389,385✔

195

  if (NULL == pTableScanInfo->pIgnoreTables) {

13,389,385✔

196

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

1,745,603✔

197

    pTableScanInfo->pIgnoreTables =

1,744,297✔

198

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

1,744,903✔

199

    if (NULL == pTableScanInfo->pIgnoreTables) {

1,744,903✔

204

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

13,390,226✔

206

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

13,388,751✔

210

  return TSDB_CODE_SUCCESS;

13,388,751✔

213

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

770,998,565✔

214

  STableScanInfo* pTableScanInfo = pOperator->info;

770,998,565✔

215

  int32_t         code = TSDB_CODE_SUCCESS;

771,222,570✔

217

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

771,222,570✔

218

    return TSDB_CODE_SUCCESS;

593,995,097✔

221

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

177,228,477✔

223

  SFilePage*  pPage = NULL;

177,245,441✔

224

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

177,240,545✔

226

  if (pRow == NULL) {

177,218,949✔

227

    return TSDB_CODE_SUCCESS;

51,181,362✔

230

  bool notLoadBlock = true;

126,037,587✔

231

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

2,147,483,647✔

232

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

2,147,483,647✔

234

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

2,147,483,647✔

236

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

2,147,483,647✔

237

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

2,147,483,647✔

238

      notLoadBlock = false;

112,664,550✔

239

      break;

112,664,550✔

244

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

125,742,845✔

246

  if (notLoadBlock) {

126,054,649✔

247

    *status = FUNC_DATA_REQUIRED_NOT_LOAD;

13,391,510✔

248

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

13,391,109✔

251

  return code;

126,042,060✔

254

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

10,032,351✔

256

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

10,032,351✔

261

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

10,032,351✔

264

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

206,113,935✔

266

  SStorageAPI* pAPI = &pTaskInfo->storageAPI;

206,113,935✔

267

  bool         allColumnsHaveAgg = true;

206,166,568✔

268

  bool         hasNullSMA = false;

206,172,069✔

269

  if (pLoad != NULL) {

206,167,370✔

270

    *pLoad = false;

206,166,030✔

273

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

206,168,290✔

275

  if (code != TSDB_CODE_SUCCESS) {

206,139,683✔

279

  if (!allColumnsHaveAgg || hasNullSMA) {

206,139,683✔

280

    *pLoad = false;

173,632,347✔

282

    *pLoad = true;

32,507,336✔

285

  return code;

206,146,685✔

288

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

754,871,393✔

290

  int32_t    code = 0;

754,871,393✔

291

  SExprSupp* pSup = &pTableScanInfo->pseudoSup;

754,871,393✔

292

  if (pSup->numOfExprs > 0) {

755,108,883✔

293

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

298,211,280✔

296

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

298,237,452✔

301

  return code;

755,010,782✔

304

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

1,234,635,264✔

305

  SLimit*     pLimit = &pLimitInfo->limit;

1,234,635,264✔

306

  const char* id = GET_TASKID(pTaskInfo);

1,234,731,857✔

308

  if (pLimitInfo->remainOffset > 0) {

1,234,791,602✔

309

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

5,226,445✔

310

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

2,708,394✔

311

      blockDataEmpty(pBlock);

2,708,394✔

312

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

2,708,394✔

313

      return false;

2,708,394✔

315

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

2,518,051✔

316

      if (code != TSDB_CODE_SUCCESS) {

2,518,051✔

321

      pLimitInfo->remainOffset = 0;

2,518,051✔

325

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

1,232,081,033✔

327

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

26,974,337✔

328

    blockDataKeepFirstNRows(pBlock, keep);

26,973,962✔

330

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

26,972,894✔

331

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

26,973,179✔

332

    return true;

26,974,593✔

335

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

1,204,985,541✔

336

  return false;

1,205,069,434✔

339

static bool skipSetTagColumnData(SOperatorInfo* pOperator) {

802,411,335✔

340

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

1,470,561,800✔

341

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

668,055,581✔

344

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

819,418,595✔

346

  int32_t        code = TSDB_CODE_SUCCESS;

819,418,595✔

347

  int32_t        lino = 0;

819,418,595✔

348

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

819,418,595✔

349

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

819,550,516✔

350

  bool           loadSMA = false;

819,495,079✔

352

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

819,495,079✔

354

  pCost->totalBlocks += 1;

819,596,317✔

355

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

819,456,873✔

356

  *status = pTableScanInfo->dataBlockLoadFlag;

819,560,583✔

358

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

819,471,346✔

359

    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

129,898,553✔

361

    bool overlap = false;

689,520,118✔

363

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

689,601,843✔

364

    if (ret != TSDB_CODE_SUCCESS) {

689,297,663✔

367

    if (overlap) {

689,297,663✔

368

      (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

34,534,085✔

372

  SDataBlockInfo* pBlockInfo = &pBlock->info;

819,256,623✔

373

  taosMemoryFreeClear(pBlock->pBlockAgg);

819,434,916✔

375

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {

819,555,329✔

382

  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

819,199,825✔

383

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

22,367,977✔

386

    if (!skipSetTagColumnData(pOperator)) {

22,371,662✔

387

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

22,370,498✔

389

    pCost->skipBlocks += 1;

22,369,958✔

390

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

22,370,498✔

391

    return code;

22,369,966✔

392

  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {

796,823,008✔

393

    pCost->loadBlockStatis += 1;

76,286,007✔

394

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

76,250,256✔

395

    bool success = true;

76,250,256✔

396

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

76,269,765✔

397

    if (code) {

76,259,347✔

403

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

76,235,205✔

404

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

22,455,893✔

406

      if (!skipSetTagColumnData(pOperator)) {

22,455,893✔

407

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

5,449,331✔

409

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

22,476,779✔

410

      return code;

22,471,397✔

412

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

53,779,312✔

413

      *status = FUNC_DATA_REQUIRED_DATA_LOAD;

53,807,179✔

417

  if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {

774,327,921✔

424

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

774,404,600✔

425

    bool success = true;

129,907,189✔

426

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

129,913,251✔

427

    if (code) {

129,880,976✔

430

      QUERY_CHECK_CODE(code, lino, _end);

14,334✔

433

    if (success) {

129,895,310✔

434

      size_t size = taosArrayGetSize(pBlock->pDataBlock);

10,032,351✔

435

      bool   keep = false;

10,032,351✔

436

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

10,032,351✔

437

      if (code) {

10,032,351✔

443

      if (!keep) {

10,031,811✔

444

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

3,398,511✔

446

        pCost->filterOutBlocks += 1;

3,398,511✔

447

        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

3,398,511✔

448

        taosMemoryFreeClear(pBlock->pBlockAgg);

3,398,511✔

450

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

3,398,511✔

451

        return TSDB_CODE_SUCCESS;

3,398,511✔

457

  taosMemoryFreeClear(pBlock->pBlockAgg);

771,163,368✔

460

  code = doDynamicPruneDataBlock(pOperator, pBlockInfo, status);

770,995,639✔

461

  if (code) {

771,062,289✔

466

  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

771,062,289✔

467

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

13,393,060✔

469

    pCost->skipBlocks += 1;

13,395,516✔

470

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

13,393,187✔

472

    STableScanInfo* p1 = pOperator->info;

13,393,187✔

473

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

13,392,837✔

474

      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;

4,242,324✔

476

      *status = FUNC_DATA_REQUIRED_FILTEROUT;

9,151,213✔

478

    return TSDB_CODE_SUCCESS;

13,393,187✔

481

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

757,615,202✔

482

  pCost->loadBlocks += 1;

757,717,449✔

484

  SSDataBlock* p = NULL;

757,706,563✔

485

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

757,823,795✔

486

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

757,837,766✔

490

  if (!skipSetTagColumnData(pOperator)) {

757,837,766✔

491

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

679,869,559✔

492

    QUERY_CHECK_CODE(code, lino, _end);

679,799,121✔

496

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

757,697,271✔

498

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

757,798,680✔

499

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

126,489,931✔

500

    QUERY_CHECK_CODE(code, lino, _end);

126,460,754✔

502

    int64_t st = taosGetTimestampUs();

125,541,473✔

503

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

125,542,423✔

504

    pTableScanInfo->readRecorder.filterTime += el;

125,542,423✔

506

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

125,533,939✔

507

      pCost->filterOutBlocks += 1;

34,458,433✔

508

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

34,453,785✔

511

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

91,059,124✔

515

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

756,803,858✔

516

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

756,850,237✔

517

    setOperatorCompleted(pOperator);

3,375,777✔

520

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

756,850,237✔

522

_end:

757,849,249✔

523

  if (code != TSDB_CODE_SUCCESS) {

757,825,991✔

524

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

935,388✔

526

  return code;

757,704,847✔

529

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

67,828✔

530

  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

67,828✔

532

  switchCtxOrder(pCtx, numOfOutput);

67,828✔

533

  pTableScanInfo->cond.order = TSDB_ORDER_DESC;

67,828✔

536

67,828✔

599

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

298,632,988✔

601

  int32_t          code = TSDB_CODE_SUCCESS;

298,632,988✔

602

  int32_t          lino = 0;

298,632,988✔

603

  bool             freeReader = false;

298,632,988✔

604

  LRUHandle*       h = NULL;

298,632,988✔

605

  STableCachedVal  val = {0};

298,632,988✔

606

  SMetaReader      mr = {0};

298,632,988✔

607

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

298,621,875✔

608

  int32_t          insertRet = TAOS_LRU_STATUS_OK;

298,708,745✔

609

  STableCachedVal* pVal = NULL;

298,708,745✔

612

  if (numOfExpr <= 0) {

298,739,906✔

613

    return TSDB_CODE_SUCCESS;

394,107✔

619

  int32_t backupRows = pBlock->info.rows;

298,345,799✔

620

  pBlock->info.rows = rows;

298,337,348✔

626

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

298,376,649✔

627

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

298,379,940✔

628

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

298,408,241✔

629

    if (code != TSDB_CODE_SUCCESS) {

298,180,416✔

644

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

298,180,416✔

646

    val.pName = mr.me.name;

298,370,124✔

647

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

298,370,124✔

649

    freeReader = true;

298,370,124✔

691

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

982,513,765✔

692

    const SExprInfo* pExpr1 = &pExpr[j];

684,105,002✔

693

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

684,108,306✔

695

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

684,237,257✔

696

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

684,322,363✔

697

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

684,322,363✔

699

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

684,325,151✔

702

    if (fmIsScanPseudoColumnFunc(functionId)) {

684,387,467✔

703

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

101,255,267✔

704

      if (fType == FUNCTION_TYPE_TBNAME) {

101,266,005✔

705

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

101,266,005✔

706

        QUERY_CHECK_CODE(code, lino, _end);

101,258,840✔

715

      STagVal tagVal = {0};

582,860,137✔

716

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

582,895,924✔

717

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

583,106,713✔

719

      char* data = NULL;

583,073,486✔

720

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

583,073,486✔

721

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

577,336,727✔

723

        data = (char*)p;

5,771,520✔

726

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

583,021,661✔

727

      if (isNullVal) {

583,030,293✔

728

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

5,275,947✔

729

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

577,754,346✔

730

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

577,287,831✔

731

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

577,307,828✔

732

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

210,587,655✔

733

          if (tmp != NULL) {

210,620,344✔

734

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

210,620,344✔

735

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

210,635,676✔

737

            taosMemoryFree(tmp);

210,674,212✔

739

          taosMemoryFree(data);

210,674,568✔

741

        QUERY_CHECK_CODE(code, lino, _end);

577,211,411✔

743

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

3,089,179✔

744

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

2,591,788✔

745

          QUERY_CHECK_CODE(code, lino, _end);

2,592,180✔

752

  pBlock->info.rows = backupRows;

298,408,763✔

754

_end:

298,383,725✔

756

  if (NULL != pVal) {

298,381,979✔

765

  if (freeReader) {

298,381,979✔

766

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

298,321,697✔

768

  if (code != TSDB_CODE_SUCCESS) {

298,370,041✔

771

  return code;

298,315,279✔

774

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

101,258,468✔

776

  int32_t                     code = TSDB_CODE_SUCCESS;

101,258,468✔

777

  int32_t                     lino = 0;

101,258,468✔

778

  struct SScalarFuncExecFuncs fpSet = {0};

101,258,468✔

779

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

101,263,880✔

780

  QUERY_CHECK_CODE(code, lino, _end);

101,268,950✔

782

  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;

101,268,950✔

783

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

101,268,950✔

784

  STR_TO_VARSTR(buf, name)

101,275,125✔

786

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

101,248,894✔

788

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

101,257,730✔

789

  QUERY_CHECK_CODE(code, lino, _end);

101,270,749✔

791

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

101,270,749✔

792

  QUERY_CHECK_CODE(code, lino, _end);

101,253,714✔

794

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

101,253,714✔

795

  SScalarParam param = {.columnData = pColInfoData};

101,260,714✔

797

  if (fpSet.process != NULL) {

101,251,161✔

798

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

101,251,161✔

799

    QUERY_CHECK_CODE(code, lino, _end);

101,270,129✔

804

  colDataDestroy(&infoData);

101,270,129✔

806

_end:

101,250,558✔

807

  if (code != TSDB_CODE_SUCCESS) {

101,261,477✔

810

  return code;

101,261,477✔

879

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

250,337,107✔

880

  int32_t code = TSDB_CODE_SUCCESS;

250,337,107✔

881

  int32_t lino = 0;

250,337,107✔

882

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

250,337,107✔

883

  QUERY_CHECK_CODE(code, lino, _end);

250,321,594✔

885

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

250,321,594✔

886

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

250,384,443✔

887

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

250,241,208✔

888

  taosMemoryFreeClear(pInfo->pResBlock->pBlockAgg);

250,343,322✔

890

  if (!pInfo->needCountEmptyTable) {

250,305,508✔

891

    pInfo->countState = TABLE_COUNT_STATE_END;

235,654,948✔

893

    pInfo->countState = TABLE_COUNT_STATE_SCAN;

14,670,270✔

896

_end:

250,304,112✔

897

  if (code != TSDB_CODE_SUCCESS) {

250,304,112✔

900

  return code;

250,312,688✔

903

void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {

681,821,038✔

904

  if (pInfo->countState == TABLE_COUNT_STATE_END) {

681,821,038✔

905

    return;

599,657,368✔

907

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

82,319,187✔

908

    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;

223,834✔

910

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

82,087,249✔

911

    if (code != TSDB_CODE_SUCCESS) {

82,090,880✔

912

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

16,505,988✔

917

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

47,351,753✔

919

  blockDataEmpty(pBlock);

47,351,753✔

920

  pBlock->info.rows = 1;

47,352,269✔

921

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

47,352,269✔

922

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

47,351,237✔

923

  pBlock->info.blankFill = true;

47,352,269✔

926

  int32_t col_num = blockDataGetNumOfCols(pBlock);

47,352,269✔

927

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

195,799,446✔

928

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

148,449,241✔

933

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

47,350,205✔

935

  return pBlock;

47,349,173✔

938

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

45,497,448✔

939

  STableScanInfo* pTableScanInfo = pOperator->info;

45,497,448✔

941

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

45,498,480✔

943

  pOperator->resultInfo.totalRows++;

45,493,320✔

944

  return pBlock;

45,497,964✔

951

static int32_t prepareTableScan(SOperatorInfo* pOperator) {

905,958,930✔

952

  int32_t code = TSDB_CODE_SUCCESS;

905,958,930✔

953

  int32_t lino = 0;

905,958,930✔

955

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

905,958,930✔

956

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

906,026,067✔

957

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

906,095,362✔

958

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

35,554,499✔

959

         NOTIFY_TYPE_SCAN_PARAM &&

3,085,939✔

960

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

3,085,939✔

961

    STableScanOperatorParam* pNotify =

3,085,939✔

964

    STableScanInfo* pTableScanInfo = pOperator->info;

3,085,443✔

965

    SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

3,085,939✔

966

    SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

3,085,939✔

967

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

3,085,939✔

969

    QUERY_CHECK_CODE(code, lino, _end);

3,085,939✔

971

    if (!pGetParam->reUse) {

3,085,939✔

972

      freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

98,690✔

978

      pGetParam->reUse = false;

2,987,249✔

983

    pOperator->pOperatorGetParam = NULL;

3,085,939✔

986

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

906,114,586✔

987

_end:

264,588,637✔

988

  if (code != TSDB_CODE_SUCCESS) {

906,107,254✔

992

  return code;

906,108,756✔

995

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

868,122,473✔

996

  int32_t         code = TSDB_CODE_SUCCESS;

868,122,473✔

997

  int32_t         lino = 0;

868,122,473✔

998

  STableScanInfo* pTableScanInfo = pOperator->info;

868,122,473✔

999

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

868,211,522✔

1000

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

868,146,066✔

1001

  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;

868,045,283✔

1002

  bool            hasNext = false;

868,146,011✔

1003

  int64_t         st = taosGetTimestampUs();

868,159,936✔

1005

  QRY_PARAM_CHECK(ppRes);

868,159,936✔

1006

  pBlock->info.dataLoad = false;

868,201,157✔

1008

  while (true) {

41,415,614✔

1009

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

909,605,937✔

1010

    if (code != TSDB_CODE_SUCCESS) {

909,445,546✔

1011

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

13✔

1012

      QUERY_CHECK_CODE(code, lino, _end);

41✔

1015

    if (!hasNext) {

909,329,662✔

1016

      break;

226,365,076✔

1019

    if (isTaskKilled(pTaskInfo)) {

682,964,586✔

1020

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

31,525✔

1021

      code = pTaskInfo->code;

31,525✔

1022

      QUERY_CHECK_CODE(code, lino, _end);

31,525✔

1025

    if (pOperator->status == OP_EXEC_DONE) {

683,226,930✔

1026

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

91,224✔

1027

      break;

91,224✔

1031

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

683,056,871✔

1032

    if (!processThisBlock) {

682,975,631✔

1036

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

682,975,631✔

1037

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

683,105,537✔

1040

    uint32_t status = 0;

683,141,882✔

1041

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

683,193,963✔

1042

    QUERY_CHECK_CODE(code, lino, _end);

682,987,538✔

1044

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

682,069,430✔

1045

      break;

4,241,624✔

1049

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

677,827,806✔

1050

      continue;

41,415,090✔

1053

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

636,490,728✔

1054

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

636,535,725✔

1056

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

636,549,640✔

1057

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

636,535,551✔

1059

    (*ppRes) = pBlock;

636,503,887✔

1060

    return code;

636,398,220✔

1063

_end:

231,660,461✔

1064

  if (code != TSDB_CODE_SUCCESS) {

231,671,955✔

1065

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

963,824✔

1066

    pTaskInfo->code = code;

963,824✔

1067

    T_LONG_JMP(pTaskInfo->env, code);

949,633✔

1069

  return code;

230,708,131✔

1072

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

916,515,543✔

1073

  int32_t         code = TSDB_CODE_SUCCESS;

916,515,543✔

1074

  int32_t         lino = 0;

916,515,543✔

1075

  STableScanInfo* pTableScanInfo = pOperator->info;

916,515,543✔

1076

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

916,587,289✔

1077

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

916,551,401✔

1079

  QRY_PARAM_CHECK(pBlock);

916,531,834✔

1082

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

916,530,841✔

1083

    return code;

2,999,204✔

1087

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

1,132,813,215✔

1088

    SSDataBlock* p = NULL;

817,877,945✔

1089

    code = doTableScanImplNext(pOperator, &p);

817,840,719✔

1090

    QUERY_CHECK_CODE(code, lino, _end);

816,718,252✔

1092

    if (p != NULL) {

816,718,252✔

1093

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

597,393,316✔

1094

      *pBlock = p;

597,444,954✔

1095

      return code;

597,435,547✔

1098

    pTableScanInfo->scanTimes += 1;

219,324,936✔

1099

    taosHashClear(pTableScanInfo->pIgnoreTables);

219,335,534✔

1101

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

219,306,788✔

1102

      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

84,862✔

1103

      pTableScanInfo->base.scanFlag = MAIN_SCAN;

84,862✔

1104

      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

84,862✔

1105

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

84,862✔

1108

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

84,862✔

1109

      QUERY_CHECK_CODE(code, lino, _end);

69,322✔

1113

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

315,009,432✔

1114

  if (pTableScanInfo->scanTimes < total) {

315,010,919✔

1115

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

50,323,236✔

1116

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

67,828✔

1117

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

67,828✔

1118

      QUERY_CHECK_CODE(code, lino, _end);

67,828✔

1119

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

67,828✔

1122

    while (pTableScanInfo->scanTimes < total) {

61,698,039✔

1123

      SSDataBlock* p = NULL;

50,320,695✔

1124

      code = doTableScanImplNext(pOperator, &p);

50,308,989✔

1125

      QUERY_CHECK_CODE(code, lino, _end);

50,302,452✔

1127

      if (p != NULL) {

50,302,452✔

1128

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

38,929,547✔

1129

        *pBlock = p;

38,937,657✔

1130

        return code;

38,935,576✔

1133

      pTableScanInfo->scanTimes += 1;

11,372,905✔

1134

      taosHashClear(pTableScanInfo->pIgnoreTables);

11,372,205✔

1136

      if (pTableScanInfo->scanTimes < total) {

11,371,910✔

1147

  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {

276,035,023✔

1148

    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;

59,425,823✔

1149

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

59,424,894✔

1150

      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {

109,198✔

1163

      int32_t numOfTables = 0;

59,315,180✔

1164

      code = tableListGetSize(pTableListInfo, &numOfTables);

59,315,180✔

1165

      QUERY_CHECK_CODE(code, lino, _end);

59,316,007✔

1167

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

59,316,007✔

1169

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

59,041,427✔

1170

        if (pIte != NULL) {

59,041,529✔

1171

          size_t        keySize = 0;

45,497,964✔

1172

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

45,499,512✔

1173

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

45,497,448✔

1174

          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);

45,498,480✔

1175

          markGroupProcessed(pTableScanInfo, *pGroupId);

45,498,996✔

1176

          *pBlock = getBlockForEmptyTable(pOperator, &info);

45,497,448✔

1178

          return code;

45,496,932✔

1182

    pTableScanInfo->countState = TABLE_COUNT_STATE_END;

13,926,311✔

1185

_end:

216,635,293✔

1186

  if (code != TSDB_CODE_SUCCESS) {

230,549,128✔

1191

  return code;

230,549,128✔

1194

static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {

982,043✔

1195

  STableScanInfo*          pInfo = pOperator->info;

982,043✔

1196

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

982,043✔

1197

  int32_t                  code = 0;

982,043✔

1198

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

982,043✔

1199

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

982,043✔

1200

  int32_t                  num = taosArrayGetSize(pParam->pUidList);

982,426✔

1201

  if (num <= 0) {

982,043✔

1206

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

982,043✔

1210

  if (pParam->tableSeq) {

982,426✔

1211

    pListInfo->oneTableForEachGroup = true;

721,518✔

1212

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

721,518✔

1218

    pListInfo->oneTableForEachGroup = false;

260,908✔

1219

    pListInfo->numOfOuputGroups = 1;

260,908✔

1221

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

982,426✔

1224

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

716,898✔

1225

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

716,898✔

1228

  STableKeyInfo info = {.groupId = 0};

982,426✔

1229

  int32_t       tableIdx = 0;

982,426✔

1230

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

9,222,802✔

1231

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

8,240,376✔

1232

    if (!pUid) {

8,228,870✔

1237

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

8,228,870✔

1238

      if (TSDB_CODE_DUP_KEY == terrno) {

6,907,415✔

1239

        continue;

6,900,616✔

1244

    info.uid = *pUid;

1,339,760✔

1245

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

1,339,760✔

1246

    if (p == NULL) {

1,340,143✔

1250

    tableIdx++;

1,340,143✔

1251

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

1,340,143✔

1254

  return code;

982,426✔

1257

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

232,388,619✔

1259

  const char* idStr = GET_TASKID(pTaskInfo);

232,388,619✔

1260

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

232,420,249✔

1261

  if (code) {

232,362,544✔

1266

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

232,362,544✔

1271

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

464,729,462✔

1272

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

232,318,714✔

1273

  if (code) {

232,317,842✔

1274

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

2,300✔

1277

  return code;

232,291,946✔

1280

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

340,009,525✔

1281

  SColIdPair* node1 = (SColIdPair*)elem1;

340,009,525✔

1282

  SColIdPair* node2 = (SColIdPair*)elem2;

340,009,525✔

1284

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

340,009,525✔

1285

    return 0;

112,184✔

1288

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

339,920,196✔

1291

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

447,977,219✔

1292

  SColIdSlotIdPair* node1 = (SColIdSlotIdPair*)elem1;

447,977,219✔

1293

  SColIdSlotIdPair* node2 = (SColIdSlotIdPair*)elem2;

447,977,219✔

1295

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

447,977,219✔

1296

    return 0;

171,368✔

1299

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

447,799,810✔

1313

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

100,191,635✔

1314

  if (pList == NULL) {

100,191,635✔

1318

  int32_t left = 0;

100,191,635✔

1319

  int32_t right = taosArrayGetSize(pList) - 1;

100,191,635✔

1320

  while (left <= right) {

308,874,102✔

1321

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

308,874,102✔

1322

    SColMatchItem* pItem = taosArrayGet(pList, mid);

308,874,102✔

1323

    if (pItem->colId == colId) {

308,874,102✔

1325

      while (mid > 0) {

100,191,635✔

1326

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

74,766,770✔

1327

        if (pPrev->colId != colId) {

74,766,770✔

1328

          break;

74,766,770✔

1332

      return taosArrayGet(pList, mid);

100,191,635✔

1335

    if (pItem->colId < colId) {

208,682,467✔

1336

      left = mid + 1;

98,350,323✔

1338

      right = mid - 1;

110,332,144✔

1345

static bool isNewScanParam(STableScanOperatorParam* pParam) {

43,524,482✔

1346

  return pParam->isNewParam;

43,524,482✔

1349

static int32_t createVTableScanInfoFromBatchParam(SOperatorInfo* pOperator) {

87,053,634✔

1350

  int32_t                  code = 0;

87,053,634✔

1351

  int32_t                  lino = 0;

87,053,634✔

1352

  STableScanInfo*          pInfo = pOperator->info;

87,053,634✔

1353

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

87,089,738✔

1354

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

87,061,399✔

1355

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

87,077,234✔

1356

  SMetaReader              orgTable = {0};

87,067,629✔

1357

  SMetaReader              superTable = {0};

87,120,970✔

1358

  SSchemaWrapper*          schema = NULL;

87,091,229✔

1359

  SArray*                  pColArray = NULL;

87,091,229✔

1360

  SArray*                  pBlockColArray = NULL;

87,091,229✔

1361

  int32_t                  num = 0;

87,091,229✔

1362

  STableKeyInfo*           pList = NULL;

87,093,931✔

1366

  cleanupQueryTableDataCond(&pInfo->base.cond);

87,090,776✔

1368

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

87,100,923✔

1369

  if (pGetParam &&

87,101,545✔

1370

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

6,054,725✔

1371

    STableScanOperatorParam* pParam =

6,050,667✔

1373

    pInfo->pBatchColMap = pParam->pBatchTbInfo;

6,062,582✔

1374

    pParam->pBatchTbInfo = NULL;

6,051,111✔

1375

    pInfo->cachedTimeWindow = pParam->window;

6,028,038✔

1376

    pInfo->currentBatchIdx = 0;

6,056,244✔

1377

    pInfo->lastBatchIdx = -1;

6,037,909✔

1378

    pInfo->cachedTagList = pParam->pTagList;

6,024,068✔

1379

    pParam->pTagList = NULL;

6,047,134✔

1380

    pInfo->cachedGroupId = pParam->groupid;

6,039,725✔

1382

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

6,057,737✔

1384

    taosHashClear(pInfo->pIgnoreTables);

6,057,737✔

1386

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

6,049,245✔

1387

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

5,560,545✔

1388

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

69,513,901✔

1389

        SqlFunctionCtx ctx = pSup->pCtx[i];

63,962,114✔

1390

        ctx.skipDynDataCheck = false;

63,952,971✔

1395

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

87,086,334✔

1396

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

87,134,696✔

1398

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

87,134,696✔

1399

    isNewTable = false;

76,857,611✔

1401

    isNewTable = true;

10,273,291✔

1402

    pInfo->lastBatchIdx = pInfo->currentBatchIdx;

10,273,291✔

1405

  if (isNewTable) {

87,099,681✔

1406

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

10,268,689✔

1407

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

10,279,378✔

1408

    pAPI->metaReaderFn.readerReleaseLock(&orgTable);

10,257,683✔

1409

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

10,272,272✔

1410

    QUERY_CHECK_CODE(code, lino, _return);

10,252,420✔

1411

    switch (orgTable.me.type) {

10,252,420✔

1412

      case TSDB_CHILD_TABLE:

583,546✔

1413

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

583,546✔

1414

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

583,546✔

1415

        pAPI->metaReaderFn.readerReleaseLock(&superTable);

583,546✔

1416

        QUERY_CHECK_CODE(code, lino, _return);

583,546✔

1417

        schema = &superTable.me.stbEntry.schemaRow;

583,546✔

1418

        break;

583,546✔

1419

      case TSDB_NORMAL_TABLE:

9,633,305✔

1420

        schema = &orgTable.me.ntbEntry.schemaRow;

9,633,305✔

1421

        break;

9,633,305✔

1422

      default:

35,569✔

1423

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

35,569✔

1428

    pListInfo->oneTableForEachGroup = true;

10,216,851✔

1429

    taosHashClear(pListInfo->map);

10,255,806✔

1430

    taosArrayClear(pListInfo->pTableList);

10,285,254✔

1432

    uint64_t      pUid = orgTable.me.uid;

10,284,856✔

1433

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

10,284,856✔

1434

    int32_t       tableIdx = 0;

10,284,506✔

1435

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

10,284,506✔

1436

    QUERY_CHECK_CODE(code, lino, _return);

10,286,304✔

1437

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

20,568,463✔

1438

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

10,282,159✔

1440

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

10,282,159✔

1441

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

10,273,176✔

1442

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

10,273,176✔

1443

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

10,270,954✔

1446

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

10,270,954✔

1447

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

10,284,231✔

1449

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

38,185,614✔

1450

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

27,898,741✔

1451

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

218,876,647✔

1452

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

218,490,758✔

1453

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

27,915,763✔

1454

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

27,911,979✔

1455

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

27,911,979✔

1457

          break;

27,897,422✔

1462

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

48,464,436✔

1463

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

38,179,219✔

1464

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

222,894,071✔

1465

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

222,658,599✔

1466

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

222,720,895✔

1467

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

38,181,652✔

1468

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

38,193,714✔

1469

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

38,193,714✔

1471

          break;

38,164,871✔

1476

    taosArraySort(pColArray, compareColIdPair);

10,279,634✔

1477

    taosArraySort(pBlockColArray, compareColIdSlotIdPair);

10,274,962✔

1478

    if (pInfo->pBlockColMap) {

10,279,646✔

1479

      taosArrayDestroy(pInfo->pBlockColMap);

7,305,711✔

1480

      pInfo->pBlockColMap = NULL;

7,301,959✔

1482

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

10,277,482✔

1483

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

10,278,203✔

1486

    taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

10,260,619✔

1487

    taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

10,264,466✔

1489

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

10,276,657✔

1490

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

8,668,989✔

1491

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

107,049,055✔

1492

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

98,392,827✔

1493

        bool           needScan = false;

98,392,966✔

1494

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

198,201,572✔

1495

          SFunctParam pParam = ctx->param[j];

99,776,313✔

1496

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

479,365,963✔

1497

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

403,008,171✔

1498

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

403,926,863✔

1499

              needScan |= true;

24,830,440✔

1500

              break;

24,830,440✔

1504

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

98,396,269✔

1505

        if (!needScan) {

98,441,696✔

1506

          ctx->skipDynDataCheck = true;

73,619,967✔

1508

          ctx->skipDynDataCheck = false;

24,821,729✔

1515

    pColArray = pInfo->lastColArray;

76,830,992✔

1516

    pBlockColArray = pInfo->lastBlockColArray;

76,834,070✔

1519

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

87,128,978✔

1520

  QUERY_CHECK_CODE(code, lino, _return);

87,137,370✔

1522

  if (pInfo->pResBlock) {

87,137,370✔

1523

    blockDataDestroy(pInfo->pResBlock);

81,070,631✔

1524

    pInfo->pResBlock = NULL;

81,031,574✔

1527

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

87,123,731✔

1535

  if (isNewTable) {

87,058,817✔

1536

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

10,265,579✔

1537

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

8,820,737✔

1538

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

8,798,584✔

1540

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

1,453,286✔

1541

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

1,445,644✔

1544

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

76,793,238✔

1545

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

1,956,835✔

1547

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

74,845,080✔

1551

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

87,081,596✔

1552

  pInfo->currentGroupId = 0;

87,085,944✔

1553

  pInfo->ignoreTag = true;

87,093,523✔

1555

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

87,067,512✔

1556

  QUERY_CHECK_CODE(code, lino, _return);

87,108,198✔

1558

  if (isNewTable) {

87,108,198✔

1559

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

10,267,666✔

1560

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

7,309,329✔

1562

    pInfo->base.dataReader = NULL;

10,238,434✔

1564

    taosRLockLatch(&pTaskInfo->lock);

10,247,064✔

1565

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

10,284,314✔

1566

    taosRUnLockLatch(&pTaskInfo->lock);

10,228,729✔

1567

    QUERY_CHECK_CODE(code, lino, _return);

10,241,468✔

1569

    pInfo->newReader = true;

10,241,468✔

1570

    if (pInfo->lastColArray) {

10,245,865✔

1571

      taosArrayDestroy(pInfo->lastColArray);

4,214,743✔

1572

      pInfo->lastColArray = NULL;

4,213,614✔

1574

    if (pInfo->lastBlockColArray) {

10,231,434✔

1575

      taosArrayDestroy(pInfo->lastBlockColArray);

4,212,741✔

1576

      pInfo->lastBlockColArray = NULL;

4,215,093✔

1578

    pInfo->lastColArray = pColArray;

10,212,718✔

1579

    pInfo->lastBlockColArray = pBlockColArray;

10,232,225✔

1581

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

76,840,532✔

1582

    QUERY_CHECK_CODE(code, lino, _return);

76,847,245✔

1584

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

76,847,245✔

1585

    pInfo->newReader = false;

76,857,355✔

1586

    pInfo->scanTimes = 0;

76,857,355✔

1589

  pOperator->status = OP_OPENED;

87,107,902✔

1591

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

87,102,953✔

1595

  pInfo->currentGroupId = -1;

87,112,376✔

1596

_return:

87,104,080✔

1597

  if (code) {

87,071,245✔

1600

  pAPI->metaReaderFn.clearReader(&superTable);

87,071,245✔

1601

  pAPI->metaReaderFn.clearReader(&orgTable);

87,087,025✔

1602

  return code;

87,115,459✔

1605

static int32_t createVTableScanInfoFromParam(SOperatorInfo* pOperator) {

25,424,865✔

1606

  int32_t                  code = 0;

25,424,865✔

1607

  int32_t                  lino = 0;

25,424,865✔

1608

  STableScanInfo*          pInfo = pOperator->info;

25,424,865✔

1609

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

25,424,865✔

1610

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

25,424,865✔

1611

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

25,424,865✔

1612

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

25,424,865✔

1613

  SMetaReader              orgTable = {0};

25,424,865✔

1614

  SMetaReader              superTable = {0};

25,424,865✔

1615

  SSchemaWrapper*          schema = NULL;

25,424,865✔

1616

  SArray*                  pColArray = NULL;

25,424,865✔

1617

  SArray*                  pBlockColArray = NULL;

25,424,865✔

1618

  int32_t                  num = 0;

25,424,865✔

1619

  STableKeyInfo*           pList = NULL;

25,424,865✔

1621

  SArray*                  pMatchList = NULL;

25,424,865✔

1623

  cleanupQueryTableDataCond(&pInfo->base.cond);

25,424,865✔

1625

  pOrgTbInfo = pParam->pOrgTbInfo;

25,424,865✔

1627

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

25,424,865✔

1629

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

25,424,865✔

1630

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

25,424,865✔

1631

  pAPI->metaReaderFn.readerReleaseLock(&orgTable);

25,424,865✔

1632

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

25,424,865✔

1633

  QUERY_CHECK_CODE(code, lino, _return);

25,424,865✔

1634

  switch (orgTable.me.type) {

25,424,865✔

1635

    case TSDB_CHILD_TABLE:

7,069,854✔

1636

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

7,069,854✔

1637

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

7,069,854✔

1638

      pAPI->metaReaderFn.readerReleaseLock(&superTable);

7,069,854✔

1639

      QUERY_CHECK_CODE(code, lino, _return);

7,069,854✔

1640

      schema = &superTable.me.stbEntry.schemaRow;

7,069,854✔

1641

      break;

7,069,854✔

1642

    case TSDB_NORMAL_TABLE:

18,355,011✔

1643

      schema = &orgTable.me.ntbEntry.schemaRow;

18,355,011✔

1644

      break;

18,355,011✔

1651

  pListInfo->oneTableForEachGroup = true;

25,424,865✔

1652

  taosHashClear(pListInfo->map);

25,424,865✔

1653

  taosArrayClear(pListInfo->pTableList);

25,424,865✔

1655

  uint64_t      pUid = orgTable.me.uid;

25,424,865✔

1656

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

25,424,865✔

1657

  int32_t       tableIdx = 0;

25,424,865✔

1658

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

25,424,865✔

1659

  QUERY_CHECK_CODE(code, lino, _return);

25,424,865✔

1660

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

50,849,730✔

1661

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

25,424,865✔

1663

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

25,424,865✔

1664

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

25,424,865✔

1665

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

25,424,865✔

1666

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

25,424,865✔

1667

  if (pInfo->pBlockColMap) {

25,424,865✔

1668

    taosArrayDestroy(pInfo->pBlockColMap);

22,776,271✔

1669

    pInfo->pBlockColMap = NULL;

22,776,271✔

1671

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

25,424,865✔

1672

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

25,424,865✔

1675

  SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID,

25,424,865✔

1679

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

25,424,865✔

1681

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

100,191,635✔

1682

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

74,766,770✔

1683

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

629,599,645✔

1684

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

629,599,645✔

1685

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

74,766,770✔

1686

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

74,766,770✔

1687

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

74,766,770✔

1688

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

74,766,770✔

1689

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

74,766,770✔

1690

        break;

74,766,770✔

1695

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

25,424,865✔

1696

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

25,424,865✔

1701

      pMatchList = pInfo->base.matchInfo.pList;

25,424,865✔

1706

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

125,615,849✔

1707

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

100,191,635✔

1708

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

100,191,635✔

1709

    if (pItem) {

100,191,635✔

1710

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

100,191,635✔

1711

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

651✔

1713

        code = TSDB_CODE_VTABLE_COLUMN_TYPE_MISMATCH;

651✔

1714

        goto _return;

651✔

1716

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

100,190,984✔

1717

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

100,190,984✔

1718

      if (i > 0) {

100,190,984✔

1719

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

149,532,238✔

1724

  taosArraySort(pColArray, compareColIdPair);

25,424,214✔

1725

  taosArraySort(pBlockColArray, compareColIdSlotIdPair);

25,424,214✔

1726

  taosArraySort(pInfo->pBlockColMap, compareColIdSlotIdPair);

25,424,214✔

1728

  taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

25,424,214✔

1729

  taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

25,424,214✔

1731

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

25,424,214✔

1732

  QUERY_CHECK_CODE(code, lino, _return);

25,424,214✔

1734

  if (pInfo->pResBlock) {

25,424,214✔

1735

    blockDataDestroy(pInfo->pResBlock);

18,102,596✔

1736

    pInfo->pResBlock = NULL;

18,102,596✔

1739

  if (isNewScanParam(pParam)) {

25,424,214✔

1740

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

7,330,642✔

1741

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

7,330,642✔

1742

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

7,330,642✔

1748

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

18,093,572✔

1750

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

25,424,214✔

1751

  pInfo->currentGroupId = 0;

25,424,214✔

1752

  pInfo->ignoreTag = true;

25,424,214✔

1754

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

25,424,214✔

1755

  QUERY_CHECK_CODE(code, lino, _return);

25,424,214✔

1757

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

25,424,214✔

1758

  if (reader) {

25,424,214✔

1759

    if (isNewScanParam(pParam)) {

18,100,268✔

1760

      pAPI->tsdReader.tsdReaderClose(*reader);

6,696✔

1761

      pInfo->base.dataReader = NULL;

6,696✔

1763

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

6,696✔

1764

      QUERY_CHECK_CODE(code, lino, _return);

6,696✔

1766

      taosRLockLatch(&pTaskInfo->lock);

6,696✔

1767

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

6,696✔

1768

      taosRUnLockLatch(&pTaskInfo->lock);

6,696✔

1769

      QUERY_CHECK_CODE(code, lino, _return);

6,696✔

1771

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

6,696✔

1772

      QUERY_CHECK_CODE(code, lino, _return);

6,696✔

1773

      pInfo->newReader = true;

6,696✔

1775

      pInfo->base.dataReader = *reader;

18,093,572✔

1776

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

18,093,572✔

1777

      QUERY_CHECK_CODE(code, lino, _return);

18,093,572✔

1779

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

18,093,572✔

1780

      pInfo->newReader = false;

18,093,572✔

1781

      pInfo->scanTimes = 0;

18,093,572✔

1784

    pInfo->base.dataReader = NULL;

7,323,946✔

1786

    taosRLockLatch(&pTaskInfo->lock);

7,323,946✔

1787

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

7,323,946✔

1788

    taosRUnLockLatch(&pTaskInfo->lock);

7,323,946✔

1789

    QUERY_CHECK_CODE(code, lino, _return);

7,323,946✔

1791

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

7,323,946✔

1792

    QUERY_CHECK_CODE(code, lino, _return);

7,323,946✔

1793

    pInfo->newReader = true;

7,323,946✔

1795

  pOperator->status = OP_OPENED;

25,424,214✔

1797

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

25,424,214✔

1801

  pInfo->currentGroupId = -1;

25,424,214✔

1802

_return:

25,424,865✔

1803

  if (code) {

25,424,865✔

1804

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

651✔

1806

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

25,424,865✔

1809

  taosArrayDestroy(pColArray);

25,424,865✔

1810

  taosArrayDestroy(pBlockColArray);

25,424,865✔

1811

  pAPI->metaReaderFn.clearReader(&superTable);

25,424,865✔

1812

  pAPI->metaReaderFn.clearReader(&orgTable);

25,424,865✔

1813

  return code;

25,424,865✔

1816

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

227,608,210✔

1817

  int32_t         code = TSDB_CODE_SUCCESS;

227,608,210✔

1818

  int32_t         lino = 0;

227,608,210✔

1819

  STableScanInfo* pInfo = pOperator->info;

227,608,210✔

1820

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

227,597,506✔

1821

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

227,588,113✔

1822

  int32_t         numOfTables = 0;

227,561,449✔

1824

  QRY_PARAM_CHECK(pResult);

227,560,319✔

1826

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

227,580,272✔

1827

  QUERY_CHECK_CODE(code, lino, _end);

227,626,366✔

1829

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

227,626,366✔

1830

    setOperatorCompleted(pOperator);

209,648,218✔

1831

    if (pOperator->dynamicTask) {

209,651,149✔

1832

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

980,890✔

1833

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

981,656✔

1835

    return code;

209,663,519✔

1839

  pOperator->status = OP_OPENED;

17,863,677✔

1840

  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);

17,933,868✔

1842

  int32_t        num = 0;

17,927,018✔

1843

  STableKeyInfo* pList = NULL;

17,927,297✔

1844

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

17,929,197✔

1845

  QUERY_CHECK_CODE(code, lino, _end);

17,950,435✔

1847

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

17,950,435✔

1848

  QUERY_CHECK_CODE(code, lino, _end);

17,942,348✔

1850

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

17,942,348✔

1851

  QUERY_CHECK_CODE(code, lino, _end);

17,952,045✔

1852

  pInfo->scanTimes = 0;

17,952,045✔

1854

  code = doGroupedTableScan(pOperator, pResult);

17,962,424✔

1855

  QUERY_CHECK_CODE(code, lino, _end);

17,977,557✔

1857

  if (*pResult != NULL) {

17,977,557✔

1858

    if (pOperator->dynamicTask) {

14,700,707✔

1859

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

14,443,739✔

1861

    return code;

14,700,380✔

1864

_end:

3,277,967✔

1865

  if (code != TSDB_CODE_SUCCESS) {

3,279,303✔

1869

  return code;

3,277,557✔

1872

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

791,441,445✔

1873

  int32_t         code = TSDB_CODE_SUCCESS;

791,441,445✔

1874

  int32_t         lino = 0;

791,441,445✔

1875

  STableScanInfo* pInfo = pOperator->info;

791,441,445✔

1876

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

791,477,519✔

1877

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

791,464,725✔

1878

  int32_t         num = 0;

791,479,433✔

1879

  STableKeyInfo*  pList = NULL;

791,474,675✔

1880

  SSDataBlock*    pResult = NULL;

791,461,322✔

1881

  const char*     idStr = GET_TASKID(pTaskInfo);

791,452,418✔

1883

  QRY_PARAM_CHECK(pResBlock);

791,409,652✔

1885

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

791,450,399✔

1886

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

214,773,318✔

1891

    taosRLockLatch(&pTaskInfo->lock);

214,776,260✔

1892

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

214,852,427✔

1893

    taosRUnLockLatch(&pTaskInfo->lock);

214,779,003✔

1894

    QUERY_CHECK_CODE(code, lino, _end);

214,815,112✔

1896

    if (pInfo->filesetDelimited) {

214,812,812✔

1897

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

261,659✔

1900

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

214,745,853✔

1905

  code = doGroupedTableScan(pOperator, &pResult);

791,477,581✔

1906

  QUERY_CHECK_CODE(code, lino, _end);

790,261,232✔

1908

  if (pResult != NULL) {

790,261,232✔

1909

    if (pOperator->dynamicTask) {

580,522,383✔

1910

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

2,074,118✔

1913

    *pResBlock = pResult;

580,527,568✔

1914

    return code;

580,340,067✔

1918

    code = startNextGroupScan(pOperator, &pResult);

209,820,482✔

1919

    QUERY_CHECK_CODE(code, lino, _end);

210,012,220✔

1921

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

210,012,220✔

1922

      *pResBlock = pResult;

209,936,239✔

1923

      return code;

209,918,275✔

1927

_end:

2,300✔

1928

  if (code != TSDB_CODE_SUCCESS) {

2,300✔

1929

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

2,300✔

1930

    pTaskInfo->code = code;

2,300✔

1933

  return code;

2,300✔

1936

static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {

25,424,214✔

1937

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

25,424,214✔

1940

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

982,426✔

1941

  int32_t                  code = TSDB_CODE_SUCCESS;

982,426✔

1942

  int32_t                  lino = 0;

982,426✔

1943

  STableScanInfo*          pInfo = pOperator->info;

982,426✔

1944

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

982,426✔

1945

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

982,426✔

1946

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

982,043✔

1948

  code = createTableListInfoFromParam(pOperator);

982,043✔

1949

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

982,426✔

1950

  pOperator->pOperatorGetParam = NULL;

982,043✔

1951

  QUERY_CHECK_CODE(code, lino, _end);

982,043✔

1953

  if (pOperator->status == OP_EXEC_DONE) {

982,043✔

1954

    pInfo->currentGroupId = -1;

23,604✔

1955

    pOperator->status = OP_OPENED;

23,604✔

1956

    SSDataBlock* result = NULL;

23,604✔

1959

      code = startNextGroupScan(pOperator, &result);

23,604✔

1960

      QUERY_CHECK_CODE(code, lino, _end);

23,604✔

1962

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

23,604✔

1963

        (*ppRes) = result;

23,604✔

1964

        return code;

23,604✔

1968

_end:

958,439✔

1969

  if (code) {

958,439✔

1972

  return code;

958,439✔

1975

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

18,102,884✔

1976

  int32_t      code = TSDB_CODE_SUCCESS;

18,102,884✔

1977

  int32_t      lino = 0;

18,102,884✔

1978

  bool         found = false;

18,102,884✔

1980

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

18,102,884✔

1981

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

18,102,884✔

1983

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

89,014,399✔

1984

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

70,911,515✔

1985

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

70,911,515✔

1986

    found = false;

70,911,515✔

1987

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

201,608,322✔

1988

      SColIdSlotIdPair* pColPair = taosArrayGet(pColMap, j);

183,508,518✔

1989

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

183,508,518✔

1990

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

183,508,518✔

1991

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

52,811,711✔

1992

        found = true;

52,811,711✔

1993

        break;

52,811,711✔

1996

    if (!found) {

70,911,515✔

1997

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

18,099,804✔

1998

        pSrcCol->info.slotId = -1;

18,099,804✔

2007

  return code;

18,102,884✔

2013

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

25,424,865✔

2014

  int32_t                  code = TSDB_CODE_SUCCESS;

25,424,865✔

2015

  int32_t                  lino = 0;

25,424,865✔

2016

  STableScanInfo*          pInfo = pOperator->info;

25,424,865✔

2017

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

25,424,865✔

2018

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

25,424,865✔

2019

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

25,424,865✔

2021

  QRY_PARAM_CHECK(ppRes);

25,424,865✔

2022

  qTrace("%s call", __FUNCTION__);

25,424,865✔

2024

  code = createVTableScanInfoFromParam(pOperator);

25,424,865✔

2025

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

25,424,865✔

2026

  pOperator->pOperatorGetParam = NULL;

25,424,865✔

2028

  QUERY_CHECK_CODE(code, lino, _end);

25,424,865✔

2030

  SSDataBlock* result = NULL;

25,424,214✔

2031

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

25,424,214✔

2032

    (*ppRes) = result;

20,742✔

2033

    return code;

20,742✔

2036

  if (pInfo->newReader) {

25,403,472✔

2037

    code = startNextGroupScan(pOperator, &result);

7,330,642✔

2038

    QUERY_CHECK_CODE(code, lino, _end);

7,330,642✔

2040

    code = doGroupedTableScan(pOperator, &result);

18,072,830✔

2041

    QUERY_CHECK_CODE(code, lino, _end);

18,072,830✔

2044

  if (result) {

25,403,472✔

2045

    SSDataBlock* res = NULL;

18,102,884✔

2046

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

18,102,884✔

2047

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

18,102,884✔

2048

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

18,102,884✔

2050

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

18,102,884✔

2051

    QUERY_CHECK_CODE(code, lino, _end);

18,102,884✔

2057

    (*ppRes) = result;

18,102,884✔

2059

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

7,300,588✔

2060

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

7,300,588✔

2062

    blockDataDestroy(pInfo->pResBlock);

7,300,588✔

2063

    pInfo->pResBlock = NULL;

7,300,588✔

2065

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

7,300,588✔

2066

    if (reader) {

7,300,588✔

2067

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

7,300,588✔

2068

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

7,300,588✔

2069

        pInfo->base.dataReader = NULL;

7,300,588✔

2071

      if (pAPI->tsdReader.tsdReaderClose) {

7,300,588✔

2072

        pAPI->tsdReader.tsdReaderClose(*reader);

7,300,588✔

2076

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

7,300,588✔

2077

    QUERY_CHECK_CODE(code, lino, _end);

7,300,588✔

2079

_end:

25,404,123✔

2080

  if (code) {

25,404,123✔

2081

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

651✔

2083

  return code;

25,404,123✔

2086

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

16,947,320✔

2087

  int32_t                  code = TSDB_CODE_SUCCESS;

16,947,320✔

2088

  int32_t                  lino = 0;

16,947,320✔

2089

  STableScanInfo*          pInfo = pOperator->info;

16,947,320✔

2090

  int32_t                  index = 0;

16,947,914✔

2091

  char*                    tagVal = NULL;

16,947,914✔

2093

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

120,916,192✔

2094

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

103,962,908✔

2095

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

103,965,296✔

2097

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

103,964,120✔

2098

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

103,964,120✔

2100

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

2,147,483,647✔

2101

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

2,147,483,647✔

2102

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

2,147,483,647✔

2103

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

2,147,483,647✔

2105

        varDataSetLen(tagVal, pTagVal->nData);

2,147,483,647✔

2106

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

2,147,483,647✔

2107

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

2,147,483,647✔

2108

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2109

        taosMemoryFreeClear(tagVal);

2,147,483,647✔

2111

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

2,147,483,647✔

2112

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2115

    index++;

103,968,278✔

2117

  return code;

16,947,564✔

2124

static void clearVstbBatchDynamicTableScanInfo(STableScanInfo* pInfo) {

6,067,573✔

2125

  if (pInfo->pBatchColMap) {

6,067,573✔

2126

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

16,352,583✔

2127

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

10,283,472✔

2128

      taosArrayDestroy(pOrgTbInfo->colMap);

10,283,016✔

2130

    taosArrayDestroy(pInfo->pBatchColMap);

6,069,217✔

2131

    pInfo->pBatchColMap = NULL;

6,069,461✔

2133

  if (pInfo->cachedTagList) {

6,070,755✔

2134

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

18,299,910✔

2135

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

15,566,650✔

2136

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

15,568,400✔

2137

        taosMemoryFreeClear(pTagVal->pData);

6,837,018✔

2140

    taosArrayDestroy(pInfo->cachedTagList);

2,735,010✔

2141

    pInfo->cachedTagList = NULL;

2,734,310✔

2143

  if (pInfo->lastColArray) {

6,067,223✔

2144

    taosArrayDestroy(pInfo->lastColArray);

6,068,623✔

2145

    pInfo->lastColArray = NULL;

6,068,167✔

2147

  if (pInfo->lastBlockColArray) {

6,065,579✔

2148

    taosArrayDestroy(pInfo->lastBlockColArray);

6,067,817✔

2149

    pInfo->lastBlockColArray = NULL;

6,067,923✔

2152

  blockDataDestroy(pInfo->pResBlock);

6,065,685✔

2153

  pInfo->pResBlock = NULL;

6,067,923✔

2154

6,068,273✔

2156

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

82,845,795✔

2157

  int32_t                  code = TSDB_CODE_SUCCESS;

82,845,795✔

2158

  int32_t                  lino = 0;

82,845,795✔

2159

  STableScanInfo*          pInfo = pOperator->info;

82,845,795✔

2160

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

82,879,534✔

2161

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

82,878,938✔

2162

  SSDataBlock*             result = NULL;

82,871,671✔

2164

  QRY_PARAM_CHECK(ppRes);

82,884,714✔

2165

  qTrace("%s call", __FUNCTION__);

82,873,656✔

2167

  while (!result) {

163,845,427✔

2168

    code = createVTableScanInfoFromBatchParam(pOperator);

87,075,931✔

2169

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

87,073,108✔

2170

    pOperator->pOperatorGetParam = NULL;

87,073,357✔

2171

    QUERY_CHECK_CODE(code, lino, _end);

87,089,916✔

2173

    if (pInfo->newReader) {

87,089,916✔

2174

      code = startNextGroupScan(pOperator, &result);

10,244,485✔

2175

      QUERY_CHECK_CODE(code, lino, _end);

10,282,760✔

2177

      code = doGroupedTableScan(pOperator, &result);

76,847,834✔

2178

      QUERY_CHECK_CODE(code, lino, _end);

76,716,878✔

2180

    if (!result) {

86,999,638✔

2181

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

10,281,993✔

2183

      pInfo->currentBatchIdx++;

10,281,993✔

2184

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

10,278,598✔

2186

        break;

6,069,217✔

2191

  if (result) {

82,838,713✔

2192

    SSDataBlock* res = NULL;

76,770,546✔

2193

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

76,805,282✔

2194

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

76,826,264✔

2195

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

76,825,806✔

2197

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

76,829,697✔

2198

    QUERY_CHECK_CODE(code, lino, _end);

76,850,186✔

2200

    if (pInfo->cachedTagList) {

76,850,186✔

2201

      code = setTagValFromTagList(pOperator, res);

16,947,320✔

2202

      QUERY_CHECK_CODE(code, lino, _end);

16,947,914✔

2205

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

76,854,660✔

2206

    pInfo->pResBlock = res;

76,855,829✔

2207

    blockDataDestroy(result);

76,858,555✔

2208

    (*ppRes) = res;

76,851,980✔

2210

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

76,852,574✔

2212

    clearVstbBatchDynamicTableScanInfo(pInfo);

6,068,167✔

2213

    setOperatorCompleted(pOperator);

6,067,573✔

2216

_end:

82,910,754✔

2217

  return code;

82,921,820✔

2220

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

109,274,716✔

2221

  int32_t                  code = TSDB_CODE_SUCCESS;

109,274,716✔

2222

  int32_t                  lino = 0;

109,274,716✔

2223

  STableScanInfo*          pInfo = pOperator->info;

109,274,716✔

2224

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

109,303,549✔

2225

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

109,288,706✔

2227

  QRY_PARAM_CHECK(ppRes);

109,319,148✔

2228

  qTrace("%s call", __FUNCTION__);

109,312,176✔

2230

  pOperator->dynamicTask = true;

109,312,176✔

2232

  switch (type) {

109,281,536✔

2233

    case DYN_TYPE_VSTB_WIN_SCAN:

982,426✔

2235

      code = doStbJoinDynamicTableScanNext(pOperator, ppRes);

982,426✔

2236

      QUERY_CHECK_CODE(code, lino, _end);

982,043✔

2237

      break;

982,043✔

2239

    case DYN_TYPE_VSTB_SINGLE_SCAN: {

25,424,865✔

2240

      code = doVstbSingleDynamicTableScanNext(pOperator, ppRes);

25,424,865✔

2241

      QUERY_CHECK_CODE(code, lino, _end);

25,424,865✔

2242

      break;

25,424,214✔

2244

    case DYN_TYPE_VSTB_BATCH_SCAN:

82,874,245✔

2245

      code = doVstbBatchDynamicTableScanNext(pOperator, ppRes);

82,874,245✔

2246

      QUERY_CHECK_CODE(code, lino, _end);

82,901,422✔

2247

      break;

82,901,422✔

2254

_end:

109,308,330✔

2255

  return code;

109,308,330✔

2258

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

905,993,243✔

2259

  int32_t         code = TSDB_CODE_SUCCESS;

905,993,243✔

2260

  int32_t         lino = 0;

905,993,243✔

2261

  STableScanInfo* pInfo = pOperator->info;

905,993,243✔

2262

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

906,097,495✔

2263

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

906,055,003✔

2264

  QRY_PARAM_CHECK(ppRes);

906,122,850✔

2265

  qTrace("%s call", __FUNCTION__);

906,116,440✔

2267

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

906,117,999✔

2268

  QUERY_CHECK_CODE(code, lino, _end);

906,135,588✔

2270

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

906,135,588✔

2271

  if (pGetParam || pInfo->pBatchColMap) {

906,141,331✔

2272

    ETableScanDynType type = pGetParam ?

109,280,587✔

2273

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

109,281,822✔

2276

    code = doDynamicTableScanNext(pOperator, type, ppRes);

109,280,587✔

2277

    QUERY_CHECK_CODE(code, lino, _end);

109,303,704✔

2279

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

109,303,053✔

2280

      return code;

108,364,862✔

2285

  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {

797,802,643✔

2286

    int32_t       numOfTables = 0;

6,333,027✔

2287

    STableKeyInfo tInfo = {0};

6,333,027✔

2288

    pInfo->countState = TABLE_COUNT_STATE_END;

6,333,347✔

2290

    while (1) {

5,916,048✔

2291

      SSDataBlock* result = NULL;

12,249,717✔

2292

      code = doGroupedTableScan(pOperator, &result);

12,246,813✔

2293

      QUERY_CHECK_CODE(code, lino, _end);

12,239,773✔

2295

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

12,239,773✔

2296

        (*ppRes) = result;

6,276,176✔

2297

        return code;

6,289,990✔

2301

      pInfo->currentTable++;

5,959,109✔

2303

      taosRLockLatch(&pTaskInfo->lock);

5,949,772✔

2304

      numOfTables = 0;

5,961,674✔

2305

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

5,961,674✔

2306

      if (code != TSDB_CODE_SUCCESS) {

5,959,407✔

2311

      if (pInfo->currentTable >= numOfTables) {

5,959,407✔

2312

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

43,679✔

2313

        taosRUnLockLatch(&pTaskInfo->lock);

43,679✔

2314

        (*ppRes) = NULL;

43,679✔

2315

        return code;

43,679✔

2318

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

5,903,510✔

2319

      if (!tmp) {

5,915,407✔

2325

      tInfo = *tmp;

5,915,407✔

2326

      taosRUnLockLatch(&pTaskInfo->lock);

5,906,068✔

2328

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

5,918,005✔

2329

      QUERY_CHECK_CODE(code, lino, _end);

5,916,098✔

2330

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

5,916,098✔

2333

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

5,922,814✔

2334

      QUERY_CHECK_CODE(code, lino, _end);

5,907,023✔

2335

      pInfo->scanTimes = 0;

5,907,023✔

2338

    code = groupSeqTableScan(pOperator, ppRes);

791,389,488✔

2339

    QUERY_CHECK_CODE(code, lino, _end);

790,277,780✔

2342

_end:

790,275,480✔

2343

  if (code != TSDB_CODE_SUCCESS) {

790,319,886✔

2344

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

2,951✔

2345

    pTaskInfo->code = code;

2,951✔

2346

    T_LONG_JMP(pTaskInfo->env, code);

2,951✔

2349

  return code;

790,316,935✔

2352

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

1,929,138✔

2353

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

1,929,138✔

2354

  if (!pRecorder) {

1,929,138✔

2357

  STableScanInfo* pTableScanInfo = pOptr->info;

1,929,138✔

2358

  *pRecorder = pTableScanInfo->base.readRecorder;

1,929,138✔

2359

  *pOptrExplain = pRecorder;

1,929,138✔

2360

  *len = sizeof(SFileBlockLoadRecorder);

1,929,138✔

2361

  return 0;

1,929,138✔

2364

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

251,139,635✔

2365

  cleanupQueryTableDataCond(&pBase->cond);

251,139,635✔

2366

  cleanupQueryTableDataCond(&pBase->orgCond);

251,140,511✔

2368

  if (pAPI->tsdReaderClose) {

251,148,143✔

2369

    pAPI->tsdReaderClose(pBase->dataReader);

251,140,167✔

2371

  pBase->dataReader = NULL;

251,089,837✔

2373

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

251,094,739✔

2374

    taosArrayDestroy(pBase->matchInfo.pList);

251,106,327✔

2377

  tableListDestroy(pBase->pTableListInfo);

251,087,366✔

2378

  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);

251,137,918✔

2379

  cleanupExprSupp(&pBase->pseudoSup);

251,110,030✔

2380

251,144,907✔

2382

static void cleanReaderForVTable(STableScanInfo* pInfo){

6,726,185✔

2383

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

6,726,185✔

2384

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

6,726,991✔

2385

    while (pIter != NULL) {

6,750,125✔

2386

      void **reader = pIter;

23,358✔

2387

      if (*reader) {

23,358✔

2388

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

23,358✔

2389

          pInfo->base.dataReader = NULL;

21,030✔

2391

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

23,358✔

2393

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

23,358✔

2396

6,725,532✔

2398

static void destroyTableScanOperatorInfo(void* param) {

226,629,596✔

2399

  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;

226,629,596✔

2400

  blockDataDestroy(pTableScanInfo->pResBlock);

226,629,596✔

2401

  blockDataDestroy(pTableScanInfo->pOrgBlock);

226,617,273✔

2402

  taosHashCleanup(pTableScanInfo->pIgnoreTables);

226,621,053✔

2403

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

226,610,940✔

2404

    cleanReaderForVTable(pTableScanInfo);

6,724,652✔

2405

    taosHashCleanup(pTableScanInfo->readerCache);

6,727,117✔

2407

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

226,618,503✔

2408

  taosArrayDestroy(pTableScanInfo->pBlockColMap);

226,614,030✔

2409

  taosMemoryFreeClear(param);

226,620,992✔

2410

226,587,198✔

2412

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

226,537,319✔

2413

  if (pBlock && dataRequireFlag == FUNC_DATA_REQUIRED_NOT_LOAD) {

226,537,319✔

2414

    int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

7,753,640✔

2415

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

19,602,735✔

2416

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

11,857,235✔

2417

      if (pCol) {

11,853,475✔

2418

        pCol->info.noData = true;

11,853,991✔

2422

226,529,179✔

2424

static int32_t resetTableScanOperatorState(SOperatorInfo* pOper) {

361,566✔

2425

  int32_t         code = TSDB_CODE_SUCCESS;

361,566✔

2426

  STableScanInfo*   pInfo = pOper->info;

361,566✔

2427

  pOper->status = OP_NOT_OPENED;

361,566✔

2429

  pInfo->scanTimes = 0;

361,566✔

2430

  pInfo->currentGroupId = -1;

361,566✔

2431

  pInfo->tableEndIndex = -1;

361,566✔

2432

  pInfo->tableStartIndex = 0;

361,566✔

2433

  pInfo->currentTable = 0;

361,566✔

2434

  pInfo->scanMode = 0;

361,566✔

2435

  pInfo->countState = 0;

361,566✔

2436

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

361,566✔

2438

  if (!pInfo->virtualStableScan) {

361,566✔

2444

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

195,240✔

2445

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

195,240✔

2447

    pInfo->base.dataReader = NULL;

195,240✔

2449

  tableListDestroy(pInfo->base.pTableListInfo);

361,566✔

2450

  pInfo->base.pTableListInfo = tableListCreate();

361,566✔

2451

  if (!pInfo->base.pTableListInfo) {

361,566✔

2455

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

361,566✔

2457

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

361,566✔

2458

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

361,566✔

2459

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

387,674✔

2461

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

387,674✔

2462

    if (code) {

194,188✔

2468

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

361,566✔

2469

  cleanupQueryTableDataCond(&pInfo->base.cond);

361,566✔

2470

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

361,566✔

2471

  if (code) {

361,566✔

2475

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

361,566✔

2476

    cleanupQueryTableDataCond(&pInfo->base.orgCond);

166,326✔

2477

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

166,326✔

2478

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

166,326✔

2481

  pOper->resultInfo.totalRows = 0;

361,566✔

2482

  blockDataEmpty(pInfo->pResBlock);

361,566✔

2483

  blockDataEmpty(pInfo->pOrgBlock);

361,566✔

2484

  taosHashClear(pInfo->pIgnoreTables);

361,566✔

2485

  taosArrayDestroy(pInfo->pBlockColMap);

361,566✔

2486

  pInfo->pBlockColMap = NULL;

361,566✔

2487

  return code;

361,566✔

2490

int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

226,622,566✔

2493

  QRY_PARAM_CHECK(pOptrInfo);

226,622,566✔

2495

  int32_t         code = TSDB_CODE_SUCCESS;

226,630,733✔

2496

  int32_t         lino = 0;

226,630,733✔

2497

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

226,630,733✔

2498

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

226,583,538✔

2499

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

226,592,179✔

2500

    code = terrno;

143✔

2504

  pOperator->pPhyNode = pTableScanNode;

226,593,366✔

2505

  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;

226,594,248✔

2506

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

226,596,974✔

2508

  int32_t numOfCols = 0;

226,615,242✔

2510

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

226,616,682✔

2511

  QUERY_CHECK_CODE(code, lino, _error);

226,552,920✔

2513

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

226,552,920✔

2514

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

226,547,870✔

2515

  QUERY_CHECK_CODE(code, lino, _error);

226,497,719✔

2517

  if (pScanNode->pScanPseudoCols != NULL) {

226,497,719✔

2518

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

49,354,521✔

2519

    pSup->pExprInfo = NULL;

49,363,044✔

2520

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

49,367,351✔

2521

    QUERY_CHECK_CODE(code, lino, _error);

49,334,335✔

2523

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

49,334,335✔

2525

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

49,369,804✔

2528

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

226,508,939✔

2529

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

226,412,778✔

2531

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

226,490,463✔

2532

  pInfo->base.readHandle = *readHandle;

226,383,095✔

2533

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

226,463,573✔

2535

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

226,495,134✔

2536

  pInfo->sample.seed = taosGetTimestampSec();

226,547,325✔

2538

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

226,518,502✔

2539

  initResultSizeInfo(&pOperator->resultInfo, 4096);

226,550,900✔

2540

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

226,387,940✔

2541

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

226,535,829✔

2542

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

226,413,980✔

2544

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

226,331,080✔

2545

  QUERY_CHECK_CODE(code, lino, _error);

226,398,354✔

2547

  pInfo->virtualStableScan = pScanNode->virtualStableScan;

226,398,354✔

2548

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

226,422,776✔

2549

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

6,724,643✔

2550

    pInfo->pResBlock = NULL;

6,719,784✔

2551

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

6,725,351✔

2552

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

6,718,394✔

2555

  if (pInfo->virtualStableScan) {

226,412,731✔

2556

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

6,723,281✔

2557

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

6,728,517✔

2560

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

226,545,093✔

2561

                            pTaskInfo->pStreamRuntimeInfo);

226,538,029✔

2562

  QUERY_CHECK_CODE(code, lino, _error);

226,425,064✔

2564

  pInfo->currentGroupId = -1;

226,401,494✔

2566

  pInfo->tableEndIndex = -1;

226,492,370✔

2567

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

226,489,272✔

2568

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

226,379,280✔

2570

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

226,551,680✔

2572

  pOperator->exprSupp.numOfExprs = numOfCols;

226,282,069✔

2574

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

226,455,087✔

2575

  pInfo->ignoreTag = false;

226,536,257✔

2577

  pInfo->lastBatchIdx = -1;

226,444,332✔

2578

  pInfo->currentBatchIdx = 0;

226,373,430✔

2579

  pInfo->base.pTableListInfo = pTableListInfo;

226,505,524✔

2580

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

226,401,938✔

2590

  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

226,401,938✔

2592

  pOperator->fpSet = createOperatorFpSet(prepareTableScan, doTableScanNext,

226,460,195✔

2598

  setOperatorResetStateFn(pOperator, resetTableScanOperatorState);

226,301,894✔

2600

  pOperator->cost.openCost = 0;

226,239,407✔

2601

  *pOptrInfo = pOperator;

226,374,341✔

2602

  return TSDB_CODE_SUCCESS;

226,252,818✔

2604

_error:

23,570✔

2605

  if (pInfo != NULL) {

23,570✔

2606

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

23,570✔

2607

    destroyTableScanOperatorInfo(pInfo);

23,570✔

2610

  if (pOperator != NULL) {

23,570✔

2611

    pOperator->info = NULL;

23,570✔

2612

    destroyOperator(pOperator);

23,570✔

2614

  pTaskInfo->code = code;

23,570✔

2615

  return code;

23,570✔

2652

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

14,015✔

2653

  int32_t code = TSDB_CODE_SUCCESS;

14,015✔

2654

  int32_t lino = 0;

14,015✔

2655

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

14,015✔

2656

    return code;

1,701✔

2658

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

12,314✔

2659

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

12,314✔

2660

  bool hasUnqualified = false;

12,314✔

2662

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

12,314✔

2663

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

12,314✔

2665

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

12,314✔

2667

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

12,314✔

2670

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

12,314✔

2671

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

12,314✔

2672

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

44,219,152✔

2673

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

44,206,838✔

2674

    void*    data = colDataGetData(pColPk, i);

44,206,838✔

2675

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

44,209,214✔

2676

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

2,376✔

2679

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

2,376✔

2680

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

2,376✔

2681

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

2,376✔

2682

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

2,376✔

2683

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

2,376✔

2684

      taosMemoryFree(tmq);

2,376✔

2686

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

44,204,462✔

2689

    if (!p[i]) {

44,206,838✔

2690

      hasUnqualified = true;

15,762,002✔

2694

  if (hasUnqualified) {

12,314✔

2695

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

4,145✔

2696

    QUERY_CHECK_CODE(code, lino, _end);

4,145✔

2699

_end:

12,314✔

2700

  taosMemoryFree(p);

12,314✔

2701

  if (code != TSDB_CODE_SUCCESS) {

12,314✔

2704

  return code;

12,314✔

2718

static int32_t processBlock(STmqQueryScanInfo* pInfo) {

25,852,106✔

2719

  int32_t         code = TSDB_CODE_SUCCESS;

25,852,106✔

2720

  int32_t         lino = 0;

25,852,106✔

2721

  SOperatorInfo*  pOperator = pInfo->pTmqScanOp;

25,852,106✔

2723

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

25,852,382✔

2724

  QUERY_CHECK_CODE(code, lino, _end);

25,852,382✔

2726

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

25,852,382✔

2727

  QUERY_CHECK_CODE(code, lino, _end);

25,852,382✔

2729

_end:

25,852,382✔

2730

  if (code != TSDB_CODE_SUCCESS) {

25,852,382✔

2733

  return code;

25,852,382✔

2736

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

7,048,176✔

2737

  int32_t code = TSDB_CODE_SUCCESS;

7,048,176✔

2738

  int32_t lino = 0;

7,048,176✔

2739

  SValue  val = {0};

7,048,176✔

2740

  if (hasPrimaryKey) {

7,048,496✔

2741

    code = doBlockDataPrimaryKeyFilter(pBlock, offset);

14,015✔

2742

    if (code != TSDB_CODE_SUCCESS) {

14,015✔

2746

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

14,015✔

2748

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

14,015✔

2749

      return code;

3,848✔

2751

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

10,167✔

2752

    val.type = pColPk->info.type;

10,167✔

2753

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

10,167✔

2754

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

648✔

2757

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

648✔

2758

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

648✔

2759

      val.nData = varDataLen(tmp);

648✔

2760

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

648✔

2762

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

9,519✔

2765

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

7,044,648✔

2767

_end:

7,043,386✔

2768

  if (code != TSDB_CODE_SUCCESS) {

7,043,694✔

2771

  return code;

7,043,386✔

2774

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

26,041,155✔

2775

  int32_t        code = TSDB_CODE_SUCCESS;

26,041,155✔

2776

  int32_t        lino = 0;

26,041,155✔

2777

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

26,041,155✔

2778

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

26,042,508✔

2780

  STmqQueryScanInfo* pInfo = pOperator->info;

26,041,836✔

2781

  const char*      id = GET_TASKID(pTaskInfo);

26,042,186✔

2783

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

26,041,839✔

2785

  if (isTaskKilled(pTaskInfo)) {

26,043,897✔

2790

  blockDataCleanup(pInfo->pRes);

26,043,208✔

2791

  int64_t st = taosGetTimestampMs();

26,042,858✔

2793

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

26,042,858✔

2794

    while (1) {

6,011,184✔

2795

      SSDataBlock* pResult = NULL;

6,331,735✔

2796

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

6,331,416✔

2797

      QUERY_CHECK_CODE(code, lino, _end);

6,333,669✔

2799

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

6,333,669✔

2800

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

6,289,990✔

2801

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

6,289,666✔

2802

        QUERY_CHECK_CODE(code, lino, _end);

6,289,342✔

2803

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

6,289,342✔

2804

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

6,289,990✔

2805

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

6,286,142✔

2806

          QUERY_CHECK_CODE(code, lino, _end);

6,285,171✔

2812

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

6,289,019✔

2813

        (*ppRes) = pInfo->pRes;

269,798✔

2814

        return code;

269,798✔

2816

      int64_t elapsed = taosGetTimestampMs() - st;

6,017,287✔

2817

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

6,017,287✔

2818

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

8,038✔

2821

        (*ppRes) = pInfo->pRes;

7,074✔

2822

        return code;

7,074✔

2826

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

43,679✔

2827

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

43,679✔

2829

    pTSInfo->base.dataReader = NULL;

43,679✔

2830

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

43,679✔

2831

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

43,679✔

2832

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

43,679✔

2833

    if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) {

43,679✔

2834

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

11,294✔

2835

        (*ppRes) = NULL;

4,836✔

2837

        (*ppRes) = pInfo->pRes;

6,458✔

2839

      return 0;

11,294✔

2841

    QUERY_CHECK_CODE(code, lino, _end);

32,385✔

2843

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

32,385✔

2846

  while (1) {

97,340✔

2847

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

77,555,407✔

2851

                                                              pTaskInfo->tmqInfo.sourceExcluded,

25,852,032✔

2854

                                                              pTaskInfo->tmqInfo.enableReplay);

25,852,032✔

2855

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

25,852,119✔

2858

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

25,852,119✔

2861

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

25,852,382✔

2864

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

25,851,765✔

2866

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

25,851,821✔

2868

    int32_t ret = processBlock(pInfo);

25,852,403✔

2869

    if (ret != 0) {

25,852,382✔

2873

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

25,852,382✔

2876

    if (code != 0) {

25,852,032✔

2877

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

25,247,938✔

2878

        (*ppRes) = NULL;

11,901,229✔

2880

        (*ppRes) = pInfo->pRes;

13,347,059✔

2882

      return 0;

25,248,288✔

2885

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

604,094✔

2886

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

102,506✔

2887

      (*ppRes) = pInfo->pRes;

506,754✔

2888

      return 0;

506,754✔

2902

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

342,435✔

2903

  int32_t code = TSDB_CODE_SUCCESS;

342,435✔

2904

  int32_t lino = 0;

342,435✔

2905

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

342,435✔

2906

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

343,437✔

2909

  int32_t size = 0;

343,437✔

2910

  code = tableListGetSize(pTableListInfo, &size);

343,437✔

2911

  QUERY_CHECK_CODE(code, lino, _end);

343,118✔

2912

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

9,970,941✔

2913

    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);

9,627,499✔

2914

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

9,632,022✔

2915

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

9,632,022✔

2916

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

9,627,823✔

2919

  (*ppArrayRes) = tableIdList;

343,442✔

2921

_end:

343,437✔

2922

  if (code != TSDB_CODE_SUCCESS) {

343,437✔

2925

  return code;

343,437✔

2928

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

1,571,008✔

2929

  int32_t        code = TSDB_CODE_SUCCESS;

1,571,008✔

2930

  int32_t        lino = 0;

1,571,008✔

2931

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,571,008✔

2932

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,571,008✔

2934

  STmqRawScanInfo* pInfo = pOperator->info;

1,571,008✔

2935

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

1,571,008✔

2936

  pTaskInfo->tmqInfo.btMetaRsp.batchMetaLen = NULL;

1,571,008✔

2938

  qDebug("tmqsnap doRawScan called");

1,571,008✔

2939

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

1,571,008✔

2940

    bool hasNext = false;

1,564,597✔

2941

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

1,564,597✔

2942

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

1,563,360✔

2943

      if (code != TSDB_CODE_SUCCESS) {

1,561,820✔

2949

    if (pInfo->dataReader && hasNext) {

1,563,057✔

2950

      if (isTaskKilled(pTaskInfo)) {

760,062✔

2955

      SSDataBlock* pBlock = NULL;

760,062✔

2956

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

759,754✔

2957

      QUERY_CHECK_CODE(code, lino, _end);

759,138✔

2959

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

759,138✔

2960

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

759,754✔

2961

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

759,754✔

2962

        QUERY_CHECK_CODE(code, lino, _end);

758,830✔

2963

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

758,830✔

2964

        (*ppRes) = pBlock;

760,678✔

2965

        return code;

760,370✔

2969

    SMetaTableInfo mtInfo = {0};

802,995✔

2970

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

803,611✔

2971

    QUERY_CHECK_CODE(code, lino, _end);

802,687✔

2972

    if (code != 0) {

802,687✔

2976

    STqOffsetVal offset = {0};

802,687✔

2977

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

801,763✔

2978

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

12,076✔

2979

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

12,076✔

2981

      SValue val = {0};

790,932✔

2982

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

790,932✔

2983

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

791,869✔

2986

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

804,227✔

2987

    QUERY_CHECK_CODE(code, lino, _end);

804,227✔

2988

    (*ppRes) = NULL;

804,227✔

2989

    return code;

804,227✔

2990

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

6,411✔

2991

    SSnapContext* sContext = pInfo->sContext;

6,411✔

2992

    int64_t st = taosGetTimestampMs();

6,411✔

2993

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

49,540✔

2994

      void*   data = NULL;

48,838✔

2995

      int32_t dataLen = 0;

48,838✔

2996

      int16_t type = 0;

48,838✔

2997

      int64_t uid = 0;

48,838✔

2998

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

48,838✔

3004

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

48,838✔

3005

        STqOffsetVal offset = {0};

5,709✔

3006

        SValue       val = {0};

5,709✔

3008

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

5,709✔

3009

        QUERY_CHECK_CODE(code, lino, _end);

5,709✔

3010

        break;

5,709✔

3012

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

43,129✔

3013

        SMqMetaRsp tmpMetaRsp = {0};

43,129✔

3014

        tmpMetaRsp.resMsgType = type;

43,129✔

3015

        tmpMetaRsp.metaRspLen = dataLen;

43,129✔

3016

        tmpMetaRsp.metaRsp = data;

43,129✔

3017

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

43,129✔

3018

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

5,157✔

3019

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

5,157✔

3021

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

5,157✔

3022

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

5,157✔

3024

        int32_t  tempRes = TSDB_CODE_SUCCESS;

43,129✔

3025

        uint32_t len = 0;

43,129✔

3026

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

43,129✔

3027

        if (TSDB_CODE_SUCCESS != tempRes) {

43,129✔

3033

        int32_t tLen = sizeof(SMqRspHead) + len;

43,129✔

3034

        void*   tBuf = taosMemoryCalloc(1, tLen);

43,129✔

3035

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

43,129✔

3037

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

43,129✔

3038

        SEncoder encoder = {0};

43,129✔

3039

        tEncoderInit(&encoder, metaBuff, len);

43,129✔

3040

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

43,129✔

3041

        if (tempLen < 0) {

43,129✔

3048

        taosMemoryFreeClear(data);

43,129✔

3049

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

43,129✔

3050

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

43,129✔

3052

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

43,129✔

3053

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

43,129✔

3055

        int64_t elapsed = taosGetTimestampMs() - st;

43,129✔

3056

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

43,129✔

3062

    (*ppRes) = NULL;

6,411✔

3063

    return code;

6,411✔

3077

static void destroyTmqRawScanOperatorInfo(void* param) {

125,295✔

3078

  STmqRawScanInfo* pRawScan = (STmqRawScanInfo*)param;

125,295✔

3079

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

125,295✔

3080

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

125,295✔

3081

  tableListDestroy(pRawScan->pTableListInfo);

125,295✔

3082

  taosMemoryFree(pRawScan);

125,295✔

3083

125,295✔

3088

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

125,295✔

3094

  QRY_PARAM_CHECK(pOptrInfo);

125,295✔

3095

  int32_t code = TSDB_CODE_SUCCESS;

125,295✔

3096

  int32_t lino = 0;

125,295✔

3098

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

125,295✔

3099

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

125,295✔

3100

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

125,295✔

3106

  pInfo->pTableListInfo = tableListCreate();

125,295✔

3107

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

125,295✔

3108

  pInfo->vnode = pHandle->vnode;

125,295✔

3109

  pInfo->pAPI = &pTaskInfo->storageAPI;

125,295✔

3111

  pInfo->sContext = pHandle->sContext;

125,295✔

3112

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

125,295✔

3113

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

125,295✔

3116

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

125,295✔

3118

  *pOptrInfo = pOperator;

125,295✔

3119

  return code;

125,295✔

3131

void destroyTmqScanOperatorInfo(void* param) {

343,437✔

3132

  if (param == NULL) {

343,437✔

3136

  STmqQueryScanInfo* pTmqScan = (STmqQueryScanInfo*)param;

343,437✔

3137

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

343,437✔

3138

    destroyOperator(pTmqScan->pTableScanOp);

343,437✔

3141

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

343,771✔

3142

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

343,437✔

3145

  if (pTmqScan->pTableListInfo) {

343,437✔

3149

  if (pTmqScan->pPseudoExpr) {

343,437✔

3150

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

200,874✔

3151

    taosMemoryFree(pTmqScan->pPseudoExpr);

200,874✔

3154

  taosHashCleanup(pTmqScan->pCol2SlotId);

343,437✔

3155

  pTmqScan->pCol2SlotId = NULL;

343,437✔

3157

  cleanupExprSupp(&pTmqScan->tagCalSup);

343,437✔

3159

  blockDataDestroy(pTmqScan->pRes);

342,800✔

3161

  taosMemoryFree(pTmqScan);

343,153✔

3181

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

343,437✔

3182

  size_t  numOfCols = LIST_LENGTH(pNodeList);

343,437✔

3183

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

2,190,009✔

3184

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

1,846,748✔

3185

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

1,846,572✔

3186

      *pTsSlotId = pNode->slotId;

343,261✔

3189

343,261✔

3191

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

343,261✔

3192

  size_t  numOfCols = LIST_LENGTH(pNodeList);

343,261✔

3193

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

2,190,544✔

3194

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

1,846,389✔

3195

    if (pNode == NULL) {

1,845,600✔

3198

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

1,845,600✔

3199

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

1,845,600✔

3200

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

1,845,424✔

3201

      if (code != TSDB_CODE_SUCCESS) {

1,846,389✔

3206

  return TSDB_CODE_SUCCESS;

344,155✔

3209

int32_t createTmqScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,

343,078✔

3212

  QRY_PARAM_CHECK(pOptrInfo);

343,078✔

3214

  int32_t          code = TSDB_CODE_SUCCESS;

343,437✔

3215

  int32_t          lino = 0;

343,437✔

3216

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

343,437✔

3217

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

343,437✔

3218

  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;

343,437✔

3219

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

343,437✔

3221

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

343,437✔

3226

  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;

343,437✔

3227

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;

343,437✔

3229

  pInfo->pTagCond = pTagCond;

343,261✔

3230

  pInfo->pGroupTags = pTableScanNode->pGroupTags;

343,437✔

3232

  pInfo->primaryTsIndex = -1;

343,437✔

3233

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

343,261✔

3234

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

343,261✔

3235

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

343,437✔

3236

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

343,261✔

3237

  QUERY_CHECK_CODE(code, lino, _error);

343,437✔

3239

  SDataBlockDescNode* pNode = (SDataBlockDescNode*)pDescNode;

343,437✔

3240

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

343,437✔

3246

  if (pTableScanNode->pTags != NULL) {

343,078✔

3258

  SOperatorInfo* pTableScanOp = NULL;

343,078✔

3259

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

343,437✔

3260

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

342,788✔

3261

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

642✔

3265

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

342,146✔

3266

  if (pHandle->version > 0) {

342,467✔

3267

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

343,113✔

3270

  STableKeyInfo* pList = NULL;

341,226✔

3271

  int32_t        num = 0;

342,826✔

3272

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

341,841✔

3273

  QUERY_CHECK_CODE(code, lino, _error);

342,826✔

3275

  if (pHandle->initTableReader) {

342,826✔

3276

    pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;

343,150✔

3277

    pTSInfo->base.dataReader = NULL;

343,150✔

3280

  if (pHandle->initTqReader) {

341,227✔

3281

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

342,430✔

3282

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

343,437✔

3284

    pInfo->tqReader = pHandle->tqReader;

54✔

3288

  pInfo->pTableScanOp = pTableScanOp;

343,437✔

3290

  pInfo->readHandle = *pHandle;

343,078✔

3291

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

343,078✔

3293

  SArray* tableIdList = NULL;

343,078✔

3294

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

343,078✔

3295

  QUERY_CHECK_CODE(code, lino, _error);

343,078✔

3296

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

343,078✔

3297

  QUERY_CHECK_CODE(code, lino, _error);

343,437✔

3298

  taosArrayDestroy(tableIdList);

343,437✔

3299

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

343,437✔

3302

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

343,437✔

3303

  QUERY_CHECK_CODE(code, lino, _error);

343,437✔

3305

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

343,437✔

3306

  QUERY_CHECK_CODE(code, lino, _error);

343,114✔

3308

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

343,114✔

3309

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

343,161✔

3311

  pInfo->pTmqScanOp = pOperator;

343,161✔

3312

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

342,838✔

3314

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

343,437✔

3316

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

343,161✔

3318

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

343,437✔

3321

  *pOptrInfo = pOperator;

342,838✔

3322

  return code;

343,161✔

3341

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

11,347,294✔

3342

  int32_t        code = TSDB_CODE_SUCCESS;

11,347,294✔

3343

  int32_t        lino = 0;

11,347,294✔

3344

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

11,347,294✔

3345

  STagScanInfo*  pInfo = pOperator->info;

11,350,089✔

3346

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

11,350,089✔

3347

  int32_t        count = pRes->info.rows;

11,350,442✔

3349

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

11,349,046✔

3350

  if (!item) {

11,344,190✔

3357

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

11,344,190✔

3358

  tDecoderClear(&(*mr).coder);

11,351,479✔

3359

  if (code != TSDB_CODE_SUCCESS) {

11,350,365✔

3365

  char str[512];

11,350,261✔

3366

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

51,922,131✔

3367

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

40,574,310✔

3370

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

40,576,647✔

3371

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

4,773,916✔

3372

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

4,772,514✔

3373

      QUERY_CHECK_CODE(code, lino, _end);

4,773,913✔

3374

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

35,803,257✔

3375

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

7,232,299✔

3376

      QUERY_CHECK_CODE(code, lino, _end);

7,230,207✔

3377

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

28,570,785✔

3378

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

15,837✔

3379

      QUERY_CHECK_CODE(code, lino, _end);

15,837✔

3381

      STagVal val = {0};

28,556,517✔

3382

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

28,554,948✔

3383

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

28,557,563✔

3385

      char* data = NULL;

28,556,690✔

3386

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

28,556,690✔

3387

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

28,555,277✔

3389

        data = (char*)p;

1,413✔

3392

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

28,554,944✔

3393

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

28,554,071✔

3394

      QUERY_CHECK_CODE(code, lino, _end);

28,551,841✔

3396

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

28,551,841✔

3398

        taosMemoryFree(data);

10,327,050✔

3403

_end:

11,353,058✔

3404

  if (code != TSDB_CODE_SUCCESS) {

11,350,789✔

3408

    pRes->info.rows++;

11,350,789✔

3411

  return code;

11,350,615✔

3414

static void tagScanFreeUidTag(void* p) {

7,318,965✔

3415

  STUidTagInfo* pInfo = p;

7,318,965✔

3416

  if (pInfo->pTagVal != NULL) {

7,318,965✔

3417

    taosMemoryFree(pInfo->pTagVal);

7,318,740✔

3419

7,318,873✔

3421

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

1,616,510✔

3422

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

1,616,510✔

3423

  if (pColumnData == NULL) {

1,616,499✔

3427

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

1,616,499✔

3428

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

1,616,499✔

3429

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

1,616,639✔

3430

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

1,615,155✔

3432

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

1,616,499✔

3433

  if (code != TSDB_CODE_SUCCESS) {

1,616,132✔

3439

  pParam->columnData = pColumnData;

1,616,132✔

3440

  pParam->colAlloced = true;

1,616,499✔

3441

  return TSDB_CODE_SUCCESS;

1,616,499✔

3444

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

9,954,539✔

3445

  int32_t                code = TSDB_CODE_SUCCESS;

9,954,539✔

3446

  int32_t                lino = 0;

9,954,539✔

3447

  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;

9,954,539✔

3448

  SColumnNode*           pSColumnNode = NULL;

9,954,539✔

3449

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

9,956,276✔

3450

    pSColumnNode = *(SColumnNode**)pNode;

2,826,082✔

3451

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

7,134,225✔

3452

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

526✔

3453

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

526✔

3466

      return DEAL_RES_CONTINUE;

526✔

3469

    return DEAL_RES_CONTINUE;

7,131,476✔

3472

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

2,826,051✔

3473

  if (!data) {

2,824,845✔

3474

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

1,617,037✔

3475

    if (code == TSDB_CODE_DUP_KEY) {

1,620,613✔

3478

    QUERY_CHECK_CODE(code, lino, _end);

1,620,613✔

3479

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

1,620,613✔

3480

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

1,619,618✔

3481

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

1,616,887✔

3482

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

1,617,528✔

3483

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

1,617,529✔

3484

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

1,619,257✔

3486

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

1,207,808✔

3487

    pSColumnNode->slotId = col->slotId;

1,207,808✔

3490

_end:

2,825,559✔

3491

  if (code != TSDB_CODE_SUCCESS) {

2,826,554✔

3492

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

2,232✔

3495

  return DEAL_RES_CONTINUE;

2,824,322✔

3498

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

1,617,001✔

3500

  int32_t code = TSDB_CODE_SUCCESS;

1,617,001✔

3501

  int32_t lino = 0;

1,617,001✔

3502

  int32_t numOfTables = taosArrayGetSize(aUidTags);

1,617,001✔

3503

  SArray* pBlockList = NULL;

1,616,639✔

3505

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

1,616,639✔

3506

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

1,616,277✔

3508

  pBlockList = taosArrayInit(1, POINTER_BYTES);

1,616,277✔

3509

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

1,616,277✔

3511

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

1,617,001✔

3512

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

1,617,001✔

3514

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

1,617,001✔

3516

  SScalarParam output = {0};

1,617,001✔

3517

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

1,617,001✔

3518

  QUERY_CHECK_CODE(code, lino, _end);

1,617,001✔

3520

  gTaskScalarExtra.pStreamInfo = (void*)pStreamRuntimeInfo;

1,617,001✔

3521

  gTaskScalarExtra.pStreamRange = NULL;

1,617,001✔

3522

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

1,617,001✔

3523

  QUERY_CHECK_CODE(code, lino, _end);

1,613,187✔

3525

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

1,613,187✔

3526

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

4,037,304✔

3527

    if (result[i]) {

2,422,888✔

3528

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

2,244,273✔

3529

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

2,244,273✔

3533

  colDataDestroy(output.columnData);

1,616,510✔

3534

  taosMemoryFreeClear(output.columnData);

1,613,908✔

3536

_end:

1,612,533✔

3537

  blockDataDestroy(pResBlock);

1,612,029✔

3538

  taosArrayDestroy(pBlockList);

1,612,685✔

3539

  if (code != TSDB_CODE_SUCCESS) {

1,608,600✔

3542

  return code;

1,608,600✔

3545

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

29,640,818✔

3548

  int32_t code = TSDB_CODE_SUCCESS;

29,640,818✔

3549

  int32_t lino = 0;

29,640,818✔

3550

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

29,640,818✔

3551

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

14,165,394✔

3557

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

14,156,971✔

3558

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

7,080,218✔

3559

      QUERY_CHECK_CODE(code, lino, _end);

7,082,838✔

3560

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

7,083,342✔

3561

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

7,080,599✔

3562

      QUERY_CHECK_CODE(code, lino, _end);

7,082,840✔

3565

    STagVal tagVal = {0};

15,475,322✔

3566

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

15,477,040✔

3567

    if (pUidTagInfo->pTagVal == NULL) {

15,475,738✔

3570

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

15,471,520✔

3572

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

15,478,422✔

3573

        colDataSetNULL(pColInfo, rowIndex);

8,195✔

3574

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

15,469,736✔

3577

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

19,273,420✔

3578

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

3,824,694✔

3581

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

3,807,324✔

3582

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

3,806,822✔

3584

        varDataSetLen(tmp, tagVal.nData);

3,806,822✔

3585

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

3,802,855✔

3586

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

3,808,311✔

3587

        taosMemoryFree(tmp);

3,806,683✔

3588

        QUERY_CHECK_CODE(code, lino, _end);

3,805,852✔

3590

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

11,656,214✔

3591

        QUERY_CHECK_CODE(code, lino, _end);

11,660,191✔

3596

_end:

29,642,106✔

3597

  if (code != TSDB_CODE_SUCCESS) {

29,642,106✔

3600

  return code;

29,630,760✔

3603

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

4,680,851✔

3605

  int32_t       code = TSDB_CODE_SUCCESS;

4,680,851✔

3606

  int32_t       lino = 0;

4,680,851✔

3607

  STagScanInfo* pInfo = pOperator->info;

4,680,851✔

3608

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

4,680,140✔

3609

  if (!ignoreFilterIdx) {

4,678,261✔

3610

    size_t szTables = taosArrayGetSize(aFilterIdxs);

1,612,179✔

3611

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

3,856,229✔

3612

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

2,242,543✔

3613

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);

2,242,194✔

3614

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

2,243,055✔

3615

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

12,780,888✔

3616

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

10,535,985✔

3617

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

10,538,990✔

3618

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

10,538,990✔

3619

        QUERY_CHECK_CODE(code, lino, _end);

10,537,833✔

3623

    size_t szTables = taosArrayGetSize(aUidTags);

3,066,082✔

3624

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

7,965,722✔

3625

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);

4,899,273✔

3626

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

4,898,767✔

3627

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

23,990,993✔

3628

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

19,092,687✔

3629

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

19,095,415✔

3630

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

19,095,415✔

3631

        QUERY_CHECK_CODE(code, lino, _end);

19,092,226✔

3636

_end:

3,066,449✔

3637

  if (code != TSDB_CODE_SUCCESS) {

4,680,135✔

3640

  return code;

4,682,592✔

3643

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

9,312,520✔

3644

  int32_t        code = TSDB_CODE_SUCCESS;

9,312,520✔

3645

  int32_t        lino = 0;

9,312,520✔

3646

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,312,520✔

3647

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

9,309,060✔

3649

  STagScanInfo* pInfo = pOperator->info;

9,310,534✔

3650

  SSDataBlock*  pRes = pInfo->pRes;

9,311,384✔

3652

  QRY_PARAM_CHECK(ppRes);

9,309,180✔

3654

  if (pOperator->status == OP_EXEC_DONE) {

9,311,283✔

3655

    return TSDB_CODE_SUCCESS;

4,554,249✔

3657

  blockDataCleanup(pRes);

4,749,534✔

3659

  if (pInfo->pCtbCursor == NULL) {

4,757,628✔

3660

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

4,756,775✔

3661

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

4,758,130✔

3667

  SArray* aUidTags = pInfo->aUidTags;

4,756,677✔

3668

  SArray* aFilterIdxs = pInfo->aFilterIdxs;

4,757,639✔

3669

  int32_t count = 0;

4,757,995✔

3670

  bool    ctbCursorFinished = false;

4,757,995✔

3671

  while (1) {

116,364✔

3672

    taosArrayClearEx(aUidTags, tagScanFreeUidTag);

4,874,359✔

3673

    taosArrayClear(aFilterIdxs);

4,872,612✔

3675

    int32_t numTables = 0;

4,871,365✔

3676

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

12,193,133✔

3677

      SMCtbCursor* pCur = pInfo->pCtbCursor;

12,195,375✔

3678

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

12,196,233✔

3679

      if (uid == 0) {

12,192,927✔

3680

        ctbCursorFinished = true;

4,874,499✔

3681

        break;

4,874,499✔

3683

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

7,318,428✔

3684

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

7,316,075✔

3685

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

7,315,991✔

3687

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

7,315,991✔

3688

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

7,323,105✔

3689

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

7,323,105✔

3690

      ++numTables;

7,323,105✔

3693

    if (numTables == 0) {

4,874,499✔

3694

      break;

191,778✔

3696

    bool ignoreFilterIdx = true;

4,682,721✔

3697

    if (pInfo->pTagCond != NULL) {

4,682,721✔

3698

      ignoreFilterIdx = false;

1,616,639✔

3699

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

1,616,639✔

3700

      QUERY_CHECK_CODE(code, lino, _end);

1,613,920✔

3702

      ignoreFilterIdx = true;

3,066,082✔

3705

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

4,680,002✔

3706

    QUERY_CHECK_CODE(code, lino, _end);

4,682,592✔

3708

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

4,682,592✔

3710

    if (count != 0) {

4,681,694✔

3711

      break;

4,565,330✔

3715

  if (count > 0) {

4,757,108✔

3716

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

4,565,821✔

3718

  if (count == 0 || ctbCursorFinished) {

4,758,105✔

3719

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

4,758,105✔

3720

    pInfo->pCtbCursor = NULL;

4,753,069✔

3721

    setOperatorCompleted(pOperator);

4,755,157✔

3724

  pRes->info.rows = count;

4,754,179✔

3725

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

4,754,665✔

3726

  if (bLimitReached) {

4,754,780✔

3730

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

4,754,780✔

3731

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

4,754,776✔

3733

_end:

4,754,776✔

3734

  if (code != TSDB_CODE_SUCCESS) {

4,754,776✔

3740

  return code;

4,754,776✔

3743

static int32_t createTagScanTableListInfoFromParam(SOperatorInfo* pOperator) {

3,594,428✔

3744

  STagScanInfo*          pInfo = pOperator->info;

3,594,428✔

3745

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

3,594,428✔

3746

  int32_t                code = 0;

3,594,428✔

3747

  STableListInfo*        pListInfo = pInfo->pTableListInfo;

3,594,428✔

3748

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

3,594,428✔

3749

  tb_uid_t               pUid = pParam->vcUid;

3,594,428✔

3755

  STableKeyInfo info = {.groupId = 0};

3,594,428✔

3757

  int32_t tableIdx = 0;

3,594,428✔

3758

  taosHashClear(pListInfo->map);

3,594,428✔

3759

  taosArrayClear(pListInfo->pTableList);

3,594,428✔

3761

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

3,594,428✔

3768

  info.uid = pUid;

3,594,428✔

3769

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

3,594,428✔

3770

  if (p == NULL) {

3,594,428✔

3774

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

3,594,428✔

3776

  return code;

3,594,428✔

3779

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

7,711,369✔

3780

  int32_t       code = TSDB_CODE_SUCCESS;

7,711,369✔

3781

  int32_t       lino = 0;

7,711,369✔

3782

  STagScanInfo* pInfo = pOperator->info;

7,711,369✔

3783

  if (pOperator->pOperatorGetParam) {

7,711,719✔

3784

    pOperator->resultInfo.totalRows = 0;

3,594,428✔

3785

    pOperator->dynamicTask = true;

3,594,428✔

3786

    pInfo->curPos = 0;

3,594,428✔

3787

    code = createTagScanTableListInfoFromParam(pOperator);

3,594,428✔

3788

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

3,594,428✔

3789

    pOperator->pOperatorGetParam = NULL;

3,594,428✔

3790

    QUERY_CHECK_CODE(code, lino, _end);

3,594,428✔

3792

    if (pOperator->status == OP_EXEC_DONE) {

3,594,428✔

3793

      pOperator->status = OP_OPENED;

1,939,991✔

3797

  if (pOperator->status == OP_EXEC_DONE) {

7,710,326✔

3798

    (*ppRes) = NULL;

2,044,320✔

3799

    return code;

2,044,320✔

3802

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

5,665,310✔

3803

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

5,665,306✔

3805

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

5,667,402✔

3806

  SSDataBlock* pRes = pInfo->pRes;

5,666,529✔

3807

  blockDataCleanup(pRes);

5,665,412✔

3809

  int32_t size = 0;

5,667,049✔

3810

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

5,667,049✔

3811

  QUERY_CHECK_CODE(code, lino, _end);

5,663,663✔

3813

  if (size == 0) {

5,663,663✔

3814

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

19,267✔

3815

    (*ppRes) = NULL;

19,267✔

3816

    return code;

19,267✔

3819

  SMetaReader mr = {0};

5,644,396✔

3820

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

5,647,259✔

3821

  pRes->info.rows = 0;

5,649,005✔

3823

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

16,997,498✔

3824

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

11,347,647✔

3825

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

11,349,880✔

3828

      code = TSDB_CODE_SUCCESS;

11,350,935✔

3830

    QUERY_CHECK_CODE(code, lino, _end);

11,349,880✔

3832

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

11,349,880✔

3833

      setOperatorCompleted(pOperator);

5,649,008✔

3837

  pAPI->metaReaderFn.clearReader(&mr);

5,649,294✔

3838

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

5,649,008✔

3839

  if (bLimitReached) {

5,649,008✔

3844

  if (pOperator->status == OP_EXEC_DONE) {

5,649,008✔

3845

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

5,649,008✔

3848

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

5,649,008✔

3850

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

5,649,008✔

3851

_end:

5,649,008✔

3852

  if (code != TSDB_CODE_SUCCESS) {

5,649,008✔

3857

  return code;

5,649,008✔

3860

static void destroyTagScanOperatorInfo(void* param) {

8,822,017✔

3861

  STagScanInfo* pInfo = (STagScanInfo*)param;

8,822,017✔

3862

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

8,822,017✔

3865

  taosHashCleanup(pInfo->filterCtx.colHash);

8,822,912✔

3866

  taosArrayDestroy(pInfo->filterCtx.cInfoList);

8,814,938✔

3867

  taosArrayDestroy(pInfo->aFilterIdxs);

8,809,718✔

3868

  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);

8,812,612✔

3870

  blockDataDestroy(pInfo->pRes);

8,817,683✔

3871

  taosArrayDestroy(pInfo->matchInfo.pList);

8,816,897✔

3872

  tableListDestroy(pInfo->pTableListInfo);

8,813,752✔

3874

  pInfo->pRes = NULL;

8,821,116✔

3875

  pInfo->pTableListInfo = NULL;

8,822,581✔

3876

  taosMemoryFreeClear(param);

8,821,475✔

3877

8,819,310✔

3879

static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {

4,840✔

3880

  int32_t         code = TSDB_CODE_SUCCESS;

4,840✔

3881

  STagScanInfo*   pInfo = pOper->info;

4,840✔

3883

  pOper->status = OP_NOT_OPENED;

4,840✔

3884

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

4,840✔

3885

  if (pTagScanNode->onlyMetaCtbIdx) {

4,840✔

3886

      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

2,104✔

3887

      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,104✔

3888

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

2,104✔

3889

      pInfo->pCtbCursor = NULL;

2,104✔

3891

      pInfo->curPos = 0;

2,736✔

3894

  return code;

4,840✔

3897

int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,

8,824,118✔

3900

  QRY_PARAM_CHECK(pOptrInfo);

8,824,118✔

3902

  int32_t         code = TSDB_CODE_SUCCESS;

8,825,014✔

3903

  int32_t         lino = 0;

8,825,014✔

3904

  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;

8,825,014✔

3905

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

8,825,014✔

3906

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

8,814,183✔

3907

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

8,816,820✔

3911

  pOperator->pPhyNode = pTagScanNode;

8,817,184✔

3912

  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

8,817,187✔

3914

  int32_t    numOfExprs = 0;

8,820,933✔

3915

  SExprInfo* pExprInfo = NULL;

8,822,694✔

3917

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

8,821,340✔

3918

  QUERY_CHECK_CODE(code, lino, _error);

8,822,432✔

3920

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

8,822,432✔

3921

  QUERY_CHECK_CODE(code, lino, _error);

8,796,369✔

3923

  int32_t num = 0;

8,796,369✔

3924

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

8,815,360✔

3925

  QUERY_CHECK_CODE(code, lino, _error);

8,818,402✔

3927

  pInfo->pTagCond = pTagCond;

8,818,402✔

3928

  pInfo->pTagIndexCond = pTagIndexCond;

8,819,777✔

3929

  pInfo->suid = pPhyNode->suid;

8,815,752✔

3930

  pInfo->pStorageAPI = &pTaskInfo->storageAPI;

8,819,000✔

3932

  pInfo->pTableListInfo = pTableListInfo;

8,809,370✔

3933

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

8,823,527✔

3934

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

8,821,841✔

3936

  pInfo->readHandle = *pReadHandle;

8,814,275✔

3937

  pInfo->curPos = 0;

8,825,778✔

3939

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

8,801,149✔

3940

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

8,817,989✔

3942

  initResultSizeInfo(&pOperator->resultInfo, 4096);

8,802,844✔

3943

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

8,809,683✔

3944

  QUERY_CHECK_CODE(code, lino, _error);

8,825,357✔

3946

  if (pTagScanNode->onlyMetaCtbIdx) {

8,825,357✔

3947

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

4,763,668✔

3948

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

4,764,150✔

3950

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

4,764,139✔

3951

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

4,764,531✔

3953

    pInfo->filterCtx.colHash =

4,764,666✔

3954

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

4,764,164✔

3955

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

4,764,666✔

3957

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

4,763,300✔

3958

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

4,764,169✔

3960

    if (pInfo->pTagCond != NULL) {

4,761,806✔

3961

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

1,619,610✔

3964

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

8,826,443✔

3965

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

8,826,427✔

3967

  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);

8,820,090✔

3969

  *pOptrInfo = pOperator;

8,821,736✔

3970

  return code;

8,819,336✔

3987

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

2,147,483,647✔

3988

  int32_t                 left = *(int32_t*)pLeft;

2,147,483,647✔

3989

  int32_t                 right = *(int32_t*)pRight;

2,147,483,647✔

3990

  STmsSubTablesMergeInfo* pInfo = (STmsSubTablesMergeInfo*)param;

2,147,483,647✔

3992

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

2,147,483,647✔

3993

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

2,147,483,647✔

3995

  if (leftIdx == -1) {

2,147,483,647✔

3996

    return 1;

1,138,561,843✔

3997

  } else if (rightIdx == -1) {

1,430,416,707✔

3998

    return -1;

438,930✔

4001

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

1,429,977,777✔

4002

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

1,430,180,113✔

4003

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

1,430,035,974✔

4004

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

1,430,035,974✔

4007

  return ret;

1,430,430,812✔

4037

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

3,340,997✔

4038

  int32_t code = TSDB_CODE_SUCCESS;

3,340,997✔

4039

  int32_t lino = 0;

3,340,997✔

4040

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

3,340,997✔

4041

  dst->colList = NULL;

3,340,997✔

4042

  dst->pSlotList = NULL;

3,336,623✔

4044

  if (src->numOfCols > 0) {

3,337,367✔

4045

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

3,338,443✔

4046

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

3,335,180✔

4047

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

3,323,933✔

4049

    if (src->pSlotList != NULL) {

3,326,854✔

4050

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

3,340,263✔

4051

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

3,332,636✔

4052

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

3,338,448✔

4057

  if (code != TSDB_CODE_SUCCESS) {

3,326,824✔

4062

  return code;

3,340,263✔

4065

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

5,289,116✔

4067

  int32_t code = 0;

5,289,116✔

4069

  STableMergeScanInfo*    pInfo = pOperator->info;

5,289,116✔

4070

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

5,291,665✔

4071

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

5,289,116✔

4072

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

5,291,675✔

4073

  const SStorageAPI*      pAPI = &pTaskInfo->storageAPI;

5,289,126✔

4075

  blockDataCleanup(pInput->pReaderBlock);

5,286,572✔

4076

  if (!pInput->bInMemReader) {

5,289,488✔

4077

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

5,169,225✔

4078

                                         (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

2,583,710✔

4079

    if (code != 0) {

2,582,262✔

4084

  pInfo->base.dataReader = pInput->pReader;

5,288,045✔

4086

  while (true) {

7,340✔

4087

    bool hasNext = false;

5,294,611✔

4088

    code = pAPI->tsdReader.tsdNextDataBlock(pInfo->base.dataReader, &hasNext);

5,295,345✔

4089

    if (code != 0) {

5,301,197✔

4095

    if (!hasNext || isTaskKilled(pTaskInfo)) {

5,301,197✔

4096

      if (isTaskKilled(pTaskInfo)) {

3,288,905✔

4102

      *pSubTableHasBlock = false;

3,288,905✔

4103

      break;

3,288,543✔

4106

    if (pInput->tblCond.order == TSDB_ORDER_ASC) {

2,012,292✔

4107

      pInput->tblCond.twindows.skey = pInput->pReaderBlock->info.window.ekey + 1;

1,986,602✔

4109

      pInput->tblCond.twindows.ekey = pInput->pReaderBlock->info.window.skey - 1;

25,690✔

4112

    uint32_t status = 0;

2,012,292✔

4113

    code = loadDataBlock(pOperator, &pInfo->base, pInput->pReaderBlock, &status);

2,012,292✔

4114

    if (code != 0) {

2,012,292✔

4119

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

2,012,292✔

4123

    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pInput->pReaderBlock->info.rows == 0) {

2,012,292✔

4124

      continue;

7,340✔

4127

    *pSubTableHasBlock = true;

2,004,952✔

4128

    break;

2,004,952✔

4131

  if (*pSubTableHasBlock) {

5,293,128✔

4132

    pInput->pReaderBlock->info.id.groupId =

4,009,904✔

4133

        tableListGetTableGroupId(pInfo->base.pTableListInfo, pInput->pReaderBlock->info.id.uid);

4,009,904✔

4134

    pOperator->resultInfo.totalRows += pInput->pReaderBlock->info.rows;

2,004,952✔

4136

  if (!pInput->bInMemReader || !*pSubTableHasBlock) {

5,292,042✔

4137

    pAPI->tsdReader.tsdReaderClose(pInput->pReader);

3,854,185✔

4138

    pInput->pReader = NULL;

3,853,099✔

4141

  pInfo->base.dataReader = NULL;

5,291,323✔

4142

  return TSDB_CODE_SUCCESS;

5,291,685✔

4145

static int32_t setGroupStartEndIndex(STableMergeScanInfo* pInfo) {

2,003,543✔

4146

  pInfo->bGroupProcessed = false;

2,003,543✔

4148

  int32_t numOfTables = 0;

2,009,350✔

4149

  int32_t code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

1,986,112✔

4150

  if (code != TSDB_CODE_SUCCESS) {

1,981,064✔

4155

  int32_t i = pInfo->tableStartIndex + 1;

1,981,064✔

4156

  for (; i < numOfTables; ++i) {

3,256,602✔

4157

    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

1,324,201✔

4158

    if (!tableKeyInfo) {

1,321,265✔

4162

    if (tableKeyInfo->groupId != pInfo->groupId) {

1,321,265✔

4163

      break;

57,753✔

4166

  pInfo->tableEndIndex = i - 1;

1,990,154✔

4167

  return TSDB_CODE_SUCCESS;

2,014,433✔

4170

static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {

2,019,888✔

4171

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,302,991✔

4172

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,288,900✔

4173

    if (pInput->rowIdx == -1) {

3,287,090✔

4174

      continue;

1,576,882✔

4177

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,708,760✔

4178

      pInput->rowIdx = 0;

1,708,760✔

4179

      pInput->pageIdx = -1;

1,708,760✔

4182

    pInput->pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,708,393✔

4183

    SColumnInfoData* col = taosArrayGet(pInput->pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

1,708,031✔

4184

    if (!col) {

1,708,393✔

4188

    pInput->aTs = (int64_t*)col->pData;

1,708,393✔

4191

  __merge_compare_fn_t mergeCompareFn = (!pSubTblsInfo->pPkOrderInfo) ? subTblRowCompareTsFn : subTblRowCompareTsPkFn;

2,014,453✔

4192

  return tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, mergeCompareFn);

2,019,164✔

4195

static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {

2,007,183✔

4196

  int32_t code = setGroupStartEndIndex(pInfo);

2,007,183✔

4197

  if (code != TSDB_CODE_SUCCESS) {

2,006,399✔

4201

  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));

2,006,399✔

4202

  if (pSubTblsInfo == NULL) {

1,994,081✔

4205

  pSubTblsInfo->pTsOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);

1,994,081✔

4206

  if (!pSubTblsInfo->pTsOrderInfo) {

2,008,269✔

4210

  if (taosArrayGetSize(pInfo->pSortInfo) == 2) {

1,990,461✔

4217

    pSubTblsInfo->pPkOrderInfo = NULL;

1,989,013✔

4219

  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

2,008,254✔

4220

  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));

2,001,391✔

4221

  if (pSubTblsInfo->aInputs == NULL) {

1,984,674✔

4225

  int32_t bufPageSize = pInfo->bufPageSize;

2,000,647✔

4226

  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;

2,001,738✔

4227

  code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);

1,995,569✔

4228

  if (code != TSDB_CODE_SUCCESS) {

1,992,296✔

4233

  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;

1,992,296✔

4234

  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;

2,006,087✔

4236

  pInfo->pSubTablesMergeInfo = pSubTblsInfo;

2,012,628✔

4237

  return TSDB_CODE_SUCCESS;

1,987,932✔

4250

static int32_t setSubTableCondSuidByUid(STableMergeScanInfo* pInfo, SStorageAPI* pAPI, uint64_t uid,

3,279,468✔

4252

  int32_t     code = TSDB_CODE_SUCCESS;

3,279,468✔

4253

  int32_t     lino = 0;

3,279,468✔

4254

  SMetaReader tableReader = {0};

3,279,468✔

4256

  QUERY_CHECK_NULL(pInfo, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,282,017✔

4257

  QUERY_CHECK_NULL(pAPI, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,282,017✔

4258

  QUERY_CHECK_NULL(pCond, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,282,017✔

4260

  pAPI->metaReaderFn.initReader(&tableReader, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

3,282,017✔

4261

  code = pAPI->metaReaderFn.getTableEntryByUid(&tableReader, uid);

3,285,280✔

4262

  pAPI->metaReaderFn.readerReleaseLock(&tableReader);

3,278,377✔

4263

  QUERY_CHECK_CODE(code, lino, _end);

3,278,382✔

4265

  if (tableReader.me.type == TSDB_CHILD_TABLE) {

3,278,382✔

4266

    pCond->suid = tableReader.me.ctbEntry.suid;

3,174,252✔

4267

  } else if (tableReader.me.type == TSDB_NORMAL_TABLE) {

104,130✔

4268

    pCond->suid = 0;

104,130✔

4274

_end:

3,285,637✔

4275

  if (code != TSDB_CODE_SUCCESS) {

3,276,562✔

4278

  pAPI->metaReaderFn.clearReader(&tableReader);

3,276,562✔

4279

  return code;

3,269,634✔

4282

static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {

2,008,641✔

4283

  int32_t        code = TSDB_CODE_SUCCESS;

2,008,641✔

4284

  int32_t        lino = 0;

2,008,641✔

4285

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

2,008,641✔

4286

  SReadHandle*   pHandle = &pInfo->base.readHandle;

2,002,477✔

4287

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,006,469✔

4289

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

2,004,292✔

4291

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,268,496✔

4292

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,276,924✔

4293

    pInput->type = SUB_TABLE_MEM_BLOCK;

3,266,039✔

4295

    code = dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);

3,268,226✔

4296

    QUERY_CHECK_CODE(code, lino, _end);

3,276,547✔

4298

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pReaderBlock);

3,276,547✔

4299

    QUERY_CHECK_CODE(code, lino, _end);

3,287,090✔

4301

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pPageBlock);

3,287,090✔

4302

    QUERY_CHECK_CODE(code, lino, _end);

3,288,181✔

4304

    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);

3,288,181✔

4305

    pInput->pKeyInfo = keyInfo;

3,285,994✔

4306

    QUERY_CHECK_NULL(pInput->pKeyInfo, code, lino, _end, terrno);

3,281,645✔

4308

    code = setSubTableCondSuidByUid(pInfo, pAPI, pInput->pKeyInfo->uid, &pInput->tblCond);

3,279,468✔

4309

    QUERY_CHECK_CODE(code, lino, _end);

3,267,819✔

4311

    if (isTaskKilled(pTaskInfo)) {

3,267,819✔

4315

    if (i + 1 < pSubTblsInfo->numInMemReaders) {

3,272,907✔

4316

      code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, keyInfo, 1, pInput->pReaderBlock,

2,533,997✔

4317

                                           (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

1,266,448✔

4318

      QUERY_CHECK_CODE(code, lino, _end);

1,264,980✔

4319

      pInput->bInMemReader = true;

1,264,980✔

4321

      pInput->pReader = NULL;

2,017,711✔

4322

      pInput->bInMemReader = false;

2,018,802✔

4324

    bool hasNext = true;

3,282,691✔

4325

    code = fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);

3,283,435✔

4326

    QUERY_CHECK_CODE(code, lino, _end);

3,287,457✔

4327

    if (!hasNext) {

3,287,457✔

4328

      pInput->rowIdx = -1;

1,578,697✔

4329

      ++pSubTblsInfo->numSubTablesCompleted;

1,578,335✔

4330

      continue;

1,579,783✔

4332

      pInput->rowIdx = 0;

1,708,760✔

4333

      pInput->pageIdx = -1;

1,708,760✔

4337

_end:

2,018,440✔

4338

  if (code != TSDB_CODE_SUCCESS) {

2,018,440✔

4341

  return code;

2,019,164✔

4344

static void adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

2,004,952✔

4345

  int32_t              code = TSDB_CODE_SUCCESS;

2,004,952✔

4346

  int32_t              lino = 0;

2,004,952✔

4347

  SExecTaskInfo*       pTaskInfo = pOperatorInfo->pTaskInfo;

2,004,952✔

4348

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

2,004,952✔

4349

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

2,004,952✔

4350

  bool                 hasNext = true;

2,004,952✔

4351

  code = fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);

2,004,952✔

4352

  QUERY_CHECK_CODE(code, lino, _end);

2,004,952✔

4354

  if (!hasNext) {

2,004,952✔

4355

    pInput->rowIdx = -1;

1,708,760✔

4356

    ++pSubTblsInfo->numSubTablesCompleted;

1,708,760✔

4358

    pInput->rowIdx = 0;

296,192✔

4361

_end:

2,004,952✔

4362

  if (code != TSDB_CODE_SUCCESS) {

2,004,952✔

4367

2,004,952✔

4369

static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

1,763,563,903✔

4370

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

1,763,563,903✔

4371

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,763,584,334✔

4373

  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,763,542,567✔

4374

  if (pInput->rowIdx < pInputBlock->info.rows - 1) {

1,763,901,836✔

4375

    ++pInput->rowIdx;

1,761,840,360✔

4376

  } else if (pInput->rowIdx == pInputBlock->info.rows - 1) {

2,004,952✔

4377

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

2,004,952✔

4378

      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);

2,004,952✔

4380

    if (pInput->rowIdx != -1) {

2,004,952✔

4381

      SColumnInfoData* col = taosArrayGet(pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

296,192✔

4382

      if (!col) {

296,192✔

4386

      pInput->pInputBlock = pInputBlock;

296,192✔

4387

      pInput->aTs = (int64_t*)col->pData;

296,192✔

4391

  return tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));

1,763,850,937✔

4394

static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {

1,761,953,638✔

4395

  int32_t            code = TSDB_CODE_SUCCESS;

1,761,953,638✔

4396

  int32_t            lino = 0;

1,761,953,638✔

4397

  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,761,953,638✔

4398

  SSDataBlock*       pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,762,362,544✔

4400

  for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {

2,147,483,647✔

4401

    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);

2,147,483,647✔

4402

    QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);

2,147,483,647✔

4404

    SColumnInfoData* pSrcColInfo = taosArrayGet(pInputBlock->pDataBlock, i);

2,147,483,647✔

4405

    QUERY_CHECK_NULL(pSrcColInfo, code, lino, _end, terrno);

2,147,483,647✔

4406

    bool isNull = colDataIsNull(pSrcColInfo, pInputBlock->info.rows, pInput->rowIdx, NULL);

2,147,483,647✔

4408

    if (isNull) {

2,147,483,647✔

4409

      code = colDataSetVal(pColInfo, pBlock->info.rows, NULL, true);

120,287,223✔

4410

      QUERY_CHECK_CODE(code, lino, _end);

120,288,636✔

4412

      if (pSrcColInfo->pData != NULL) {

2,147,483,647✔

4413

        char* pData = colDataGetData(pSrcColInfo, pInput->rowIdx);

2,147,483,647✔

4414

        code = colDataSetVal(pColInfo, pBlock->info.rows, pData, false);

2,147,483,647✔

4415

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

4419

  pBlock->info.dataLoad = 1;

1,763,673,109✔

4420

  pBlock->info.scanFlag = pInputBlock->info.scanFlag;

1,763,690,829✔

4421

  pBlock->info.rows += 1;

1,763,694,720✔

4423

_end:

1,763,665,537✔

4424

  if (code != TSDB_CODE_SUCCESS) {

1,763,665,537✔

4427

  return code;

1,763,619,015✔

4430

static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,

4,709,002✔

4432

  int32_t                 code = TSDB_CODE_SUCCESS;

4,709,002✔

4433

  int32_t                 lino = 0;

4,709,002✔

4434

  STableMergeScanInfo*    pInfo = pOperator->info;

4,709,002✔

4435

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

4,708,997✔

4436

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

4,708,640✔

4437

  bool                    finished = false;

4,708,278✔

4439

  QRY_PARAM_CHECK(pResBlock);

4,708,278✔

4441

  blockDataCleanup(pBlock);

4,707,916✔

4445

      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {

1,765,992,061✔

4446

        finished = true;

3,290,882✔

4447

        break;

3,290,882✔

4450

      code = appendChosenRowToDataBlock(pSubTblsInfo, pBlock);

1,762,344,240✔

4451

      QUERY_CHECK_CODE(code, lino, _end);

1,763,601,889✔

4453

      code = adjustSubTableForNextRow(pOperator, pSubTblsInfo);

1,763,601,889✔

4454

      QUERY_CHECK_CODE(code, lino, _end);

1,762,810,845✔

4456

      if (pBlock->info.rows >= capacity) {

1,762,810,845✔

4457

        break;

1,418,482✔

4461

    if (isTaskKilled(pTaskInfo)) {

4,709,364✔

4465

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pBlock, pTaskInfo);

4,709,002✔

4466

    if (finished || limitReached || pBlock->info.rows > 0) {

4,707,916✔

4471

  if (pBlock->info.rows > 0) {

4,707,916✔

4472

    *pResBlock = pBlock;

2,689,476✔

4475

_end:

2,019,526✔

4476

  if (code != TSDB_CODE_SUCCESS) {

4,709,002✔

4480

  return code;

4,709,002✔

4483

static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {

2,003,196✔

4484

  int32_t              code = TSDB_CODE_SUCCESS;

2,003,196✔

4485

  int32_t              lino = 0;

2,003,196✔

4486

  STableMergeScanInfo* pInfo = pOperator->info;

2,003,196✔

4488

  code = initSubTablesMergeInfo(pInfo);

2,011,195✔

4489

  QUERY_CHECK_CODE(code, lino, _end);

1,986,122✔

4491

  code = initSubTableInputs(pOperator, pInfo);

1,986,122✔

4492

  QUERY_CHECK_CODE(code, lino, _end);

2,019,526✔

4494

  code = openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);

2,019,526✔

4495

  QUERY_CHECK_CODE(code, lino, _end);

2,019,164✔

4497

_end:

2,019,164✔

4498

  if (code != TSDB_CODE_SUCCESS) {

2,019,164✔

4501

  return code;

2,018,802✔

4504

static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {

28,842,857✔

4505

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

28,842,857✔

4506

  if (pSubTblsInfo != NULL) {

28,850,820✔

4507

    tMergeTreeDestroy(&pSubTblsInfo->pTree);

2,016,625✔

4509

    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,305,530✔

4510

      STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,287,090✔

4511

      cleanupQueryTableDataCond(&pInput->tblCond);

3,285,280✔

4512

      blockDataDestroy(pInput->pReaderBlock);

3,287,814✔

4513

      blockDataDestroy(pInput->pPageBlock);

3,288,181✔

4514

      taosArrayDestroy(pInput->aBlockPages);

3,286,723✔

4515

      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);

3,286,723✔

4516

      pInput->pReader = NULL;

3,288,176✔

4519

    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);

2,019,164✔

4520

    taosMemoryFree(pSubTblsInfo->aInputs);

2,017,354✔

4522

    taosMemoryFree(pSubTblsInfo);

2,015,906✔

4523

    pInfo->pSubTablesMergeInfo = NULL;

2,016,630✔

4527

28,850,825✔

4529

void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

33,478,937✔

4530

  STableMergeScanInfo* pInfo = pOperator->info;

33,478,937✔

4531

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

33,482,411✔

4532

  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));

33,478,189✔

4534

  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

33,483,560✔

4535

  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;

33,475,614✔

4536

  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;

33,479,703✔

4537

  pInfo->sortExecInfo.loops += sortExecInfo.loops;

33,473,396✔

4538

  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;

33,467,496✔

4539

  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

33,469,548✔

4541

  tsortDestroySortHandle(pInfo->pSortHandle);

33,465,311✔

4542

  pInfo->pSortHandle = NULL;

33,471,573✔

4543

33,472,740✔

4552

static int32_t createVTableMergeScanInfoFromBatchParam(SOperatorInfo* pOperator) {

64,792✔

4553

  int32_t                  code = TSDB_CODE_SUCCESS;

64,792✔

4554

  int32_t                  lino = 0;

64,792✔

4555

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

64,792✔

4556

  STableMergeScanInfo*     pInfo = pOperator->info;

64,792✔

4557

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

64,792✔

4558

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

64,792✔

4559

  SMetaReader              orgTable = {0};

64,792✔

4560

  SOperatorParam*          pGetParam = pOperator->pOperatorGetParam;

64,792✔

4561

  STableScanOperatorParam* pParam = (STableScanOperatorParam*)pGetParam->value;

64,792✔

4563

  if (pParam->paramType != DYN_TYPE_SCAN_PARAM) {

64,792✔

4567

  if (pInfo->pSubTablesMergeInfo) {

64,792✔

4570

  if (pInfo->pSortHandle) {

64,792✔

4573

  if (pInfo->base.dataReader) {

64,792✔

4578

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

64,792✔

4584

  pInfo->numNextDurationBlocks = 0;

64,792✔

4585

  pInfo->nextDurationBlocksIdx = 0;

64,792✔

4586

  pInfo->rtnNextDurationBlocks = false;

64,792✔

4587

  pInfo->bNewFilesetEvent = false;

64,792✔

4588

  pInfo->bNextDurationBlockEvent = false;

64,792✔

4589

  taosHashCleanup(pInfo->mSkipTables);

64,792✔

4590

  pInfo->mSkipTables = NULL;

64,792✔

4592

  if (pInfo->base.orgCond.colList && pInfo->base.orgCond.numOfCols > 0) {

64,792✔

4593

    cleanupQueryTableDataCond(&pInfo->base.cond);

64,792✔

4594

    code = dumpQueryTableCond(&pInfo->base.orgCond, &pInfo->base.cond);

64,792✔

4595

    QUERY_CHECK_CODE(code, lino, _return);

64,792✔

4598

  if (pParam->window.skey == INT64_MAX && pParam->window.ekey == INT64_MIN) {

64,792✔

4599

    if (pInfo->base.orgCond.colList && pInfo->base.orgCond.numOfCols > 0) {

64,792✔

4600

      pInfo->base.cond.twindows.skey = pInfo->base.orgCond.twindows.skey;

64,792✔

4601

      pInfo->base.cond.twindows.ekey = pInfo->base.orgCond.twindows.ekey;

64,792✔

4608

  pInfo->tableStartIndex = 0;

64,792✔

4609

  pInfo->tableEndIndex = 0;

64,792✔

4610

  pInfo->hasGroupId = false;

64,792✔

4611

  pInfo->groupId = pParam->groupid;

64,792✔

4612

  pInfo->bGroupProcessed = false;

64,792✔

4613

  pOperator->status = OP_NOT_OPENED;

64,792✔

4615

  pListInfo->oneTableForEachGroup = true;

64,792✔

4616

  pListInfo->numOfOuputGroups = 1;

64,792✔

4617

  taosHashClear(pListInfo->map);

64,792✔

4618

  taosArrayClear(pListInfo->pTableList);

64,792✔

4620

  int32_t tableIdx = 0;

64,792✔

4621

  if (pParam->pBatchTbInfo) {

64,792✔

4622

    int32_t numOfBatchTables = taosArrayGetSize(pParam->pBatchTbInfo);

64,792✔

4623

    for (int32_t i = 0; i < numOfBatchTables; ++i) {

227,929✔

4624

      SOrgTbInfo* pOrgTbInfo = taosArrayGet(pParam->pBatchTbInfo, i);

163,137✔

4625

      QUERY_CHECK_NULL(pOrgTbInfo, code, lino, _return, terrno);

163,137✔

4627

      const char* pTableName = strstr(pOrgTbInfo->tbName, ".");

163,137✔

4628

      pTableName = pTableName ? (pTableName + 1) : pOrgTbInfo->tbName;

163,137✔

4629

      pAPI->metaReaderFn.initReader(&orgTable, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

163,137✔

4630

      code = pAPI->metaReaderFn.getTableEntryByName(&orgTable, pTableName);

163,137✔

4631

      pAPI->metaReaderFn.readerReleaseLock(&orgTable);

163,137✔

4632

      QUERY_CHECK_CODE(code, lino, _return);

163,137✔

4634

      uint64_t uid = orgTable.me.uid;

163,137✔

4635

      code = taosHashPut(pListInfo->map, &uid, sizeof(uid), &tableIdx, sizeof(tableIdx));

163,137✔

4636

      if (code == TSDB_CODE_DUP_KEY) {

163,137✔

4640

      QUERY_CHECK_CODE(code, lino, _return);

163,137✔

4642

      STableKeyInfo keyInfo = {.groupId = pParam->groupid, .uid = uid};

163,137✔

4643

      QUERY_CHECK_NULL(taosArrayPush(pListInfo->pTableList, &keyInfo), code, lino, _return, terrno);

326,274✔

4644

      ++tableIdx;

163,137✔

4646

      qDebug("%s table merge dynamic batch add table name:%s uid:%" PRIu64 " group:%" PRIu64, GET_TASKID(pTaskInfo),

163,137✔

4648

      pAPI->metaReaderFn.clearReader(&orgTable);

163,137✔

4652

  qDebug("%s table merge dynamic batch param consumed, group:%" PRIu64 ", tables:%d", GET_TASKID(pTaskInfo),

64,792✔

4655

_return:

64,792✔

4656

  if (code != TSDB_CODE_SUCCESS) {

64,792✔

4659

  pAPI->metaReaderFn.clearReader(&orgTable);

64,792✔

4660

  return code;

64,792✔

4663

int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

4,676,355✔

4664

  int32_t              code = TSDB_CODE_SUCCESS;

4,676,355✔

4665

  int32_t              lino = 0;

4,676,355✔

4666

  int32_t              tableListSize = 0;

4,676,355✔

4667

  int64_t              st = taosGetTimestampUs();

4,671,267✔

4668

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

4,671,267✔

4669

  STableMergeScanInfo* pInfo = pOperator->info;

4,666,184✔

4671

  QRY_PARAM_CHECK(ppRes);

4,676,003✔

4673

  if (pOperator->status == OP_EXEC_DONE && !pOperator->pOperatorGetParam) {

4,679,256✔

4678

  if (pOperator->pOperatorGetParam) {

4,671,644✔

4679

    code = createVTableMergeScanInfoFromBatchParam(pOperator);

64,792✔

4680

    QUERY_CHECK_CODE(code, lino, _end);

64,792✔

4682

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

64,792✔

4683

    pOperator->pOperatorGetParam = NULL;

64,792✔

4684

    pOperator->dynamicTask = true;

64,792✔

4687

  code = pOperator->fpSet._openFn(pOperator);

4,679,990✔

4688

  QUERY_CHECK_CODE(code, lino, _end);

4,673,449✔

4690

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

4,673,449✔

4691

  QUERY_CHECK_CODE(code, lino, _end);

4,672,358✔

4693

  if (!pInfo->hasGroupId) {

4,672,358✔

4694

    pInfo->hasGroupId = true;

1,984,697✔

4696

    if (tableListSize == 0) {

1,969,795✔

4697

      setOperatorCompleted(pOperator);

38,902✔

4698

      (*ppRes) = NULL;

38,902✔

4699

      return code;

38,902✔

4702

    pInfo->tableStartIndex = 0;

1,930,893✔

4703

    STableKeyInfo* pTmpGpId = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,923,291✔

4704

    QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

1,920,752✔

4705

    pInfo->groupId = pTmpGpId->groupId;

1,920,752✔

4706

    code = startSubTablesTableMergeScan(pOperator);

1,950,164✔

4707

    QUERY_CHECK_CODE(code, lino, _end);

1,960,320✔

4710

  SSDataBlock* pBlock = NULL;

4,648,705✔

4711

  while (pInfo->tableStartIndex < tableListSize) {

4,708,278✔

4712

    if (isTaskKilled(pTaskInfo)) {

4,707,192✔

4716

    code = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity, &pBlock);

4,706,106✔

4717

    QUERY_CHECK_CODE(code, lino, _end);

4,709,364✔

4719

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

4,709,364✔

4726

    if (pBlock != NULL) {

4,709,002✔

4727

      pBlock->info.id.groupId = pInfo->groupId;

2,689,476✔

4728

      pOperator->resultInfo.totalRows += pBlock->info.rows;

2,689,476✔

4729

      pInfo->bGroupProcessed = true;

2,689,476✔

4730

      break;

2,689,476✔

4733

      stopSubTablesTableMergeScan(pInfo);

2,019,526✔

4734

      if (pInfo->tableEndIndex >= tableListSize - 1) {

2,017,711✔

4735

        setOperatorCompleted(pOperator);

1,958,510✔

4736

        break;

1,960,320✔

4739

      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

58,120✔

4740

      STableKeyInfo* pTmpGpId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

58,120✔

4741

      QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

58,120✔

4743

      pInfo->groupId = pTmpGpId->groupId;

58,120✔

4744

      code = startSubTablesTableMergeScan(pOperator);

58,120✔

4745

      QUERY_CHECK_CODE(code, lino, _end);

58,120✔

4746

      resetLimitInfoForNextGroup(&pInfo->limitInfo);

58,120✔

4750

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

4,649,796✔

4752

_end:

4,649,796✔

4753

  if (code != TSDB_CODE_SUCCESS) {

4,649,072✔

4758

    (*ppRes) = pBlock;

4,649,072✔

4761

  return code;

4,648,710✔

4764

static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {

3,600,328✔

4765

  int32_t              code = TSDB_CODE_SUCCESS;

3,600,328✔

4766

  int32_t              lino = 0;

3,600,328✔

4767

  SOperatorInfo*       pOperator = (SOperatorInfo*)pTableMergeOpInfo;

3,600,328✔

4768

  STableMergeScanInfo* pInfo = pOperator->info;

3,600,328✔

4769

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,600,328✔

4771

  if (pInfo->mSkipTables == NULL) {

3,600,328✔

4772

    pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,

326,821✔

4774

    QUERY_CHECK_NULL(pInfo->mSkipTables, code, lino, _end, terrno);

326,821✔

4776

  int bSkip = 1;

3,600,328✔

4777

  if (pInfo->mSkipTables != NULL) {

3,600,328✔

4778

    code = taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));

3,600,328✔

4779

    if (code == TSDB_CODE_DUP_KEY) {

3,600,328✔

4782

    QUERY_CHECK_CODE(code, lino, _end);

3,600,328✔

4785

_end:

3,600,328✔

4786

  if (code != TSDB_CODE_SUCCESS) {

3,600,328✔

4791

3,600,328✔

4793

static int32_t doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {

162,515,782✔

4794

  STableMergeScanInfo* pInfo = pOperator->info;

162,515,782✔

4795

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

162,550,205✔

4796

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

162,528,978✔

4797

  SSDataBlock*         pBlock = pInfo->pReaderBlock;

162,548,935✔

4798

  int32_t              code = 0;

162,539,672✔

4799

  bool                 hasNext = false;

162,539,672✔

4800

  STsdbReader*         reader = pInfo->base.dataReader;

162,535,363✔

4802

  code = pAPI->tsdReader.tsdNextDataBlock(reader, &hasNext);

162,524,505✔

4803

  if (code != 0) {

162,487,881✔

4810

  if (!hasNext || isTaskKilled(pTaskInfo)) {

162,487,881✔

4811

    if (isTaskKilled(pTaskInfo)) {

28,137,138✔

4812

      qInfo("table merge scan fetch next data block found task killed. %s", GET_TASKID(pTaskInfo));

540✔

4813

      pAPI->tsdReader.tsdReaderReleaseDataBlock(reader);

540✔

4815

    *pFinished = true;

28,155,978✔

4816

    return code;

28,141,419✔

4819

  uint32_t status = 0;

134,376,082✔

4820

  code = loadDataBlock(pOperator, &pInfo->base, pBlock, &status);

134,378,046✔

4822

  if (code != TSDB_CODE_SUCCESS) {

134,364,395✔

4823

    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));

17,280✔

4824

    pTaskInfo->code = code;

17,280✔

4825

    return code;

17,280✔

4828

  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

134,347,115✔

4834

  if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {

134,347,115✔

4835

    *pSkipped = true;

6,418,743✔

4836

    return code;

6,464,659✔

4839

  return code;

127,901,082✔

4842

static int32_t getBlockForTableMergeScan(void* param, SSDataBlock** ppBlock) {

161,162,108✔

4843

  STableMergeScanSortSourceParam* source = param;

161,162,108✔

4845

  SOperatorInfo*       pOperator = source->pOperator;

161,162,108✔

4846

  STableMergeScanInfo* pInfo = pOperator->info;

161,169,047✔

4847

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

161,170,006✔

4848

  SSDataBlock*         pBlock = NULL;

161,162,984✔

4849

  int64_t              st = taosGetTimestampUs();

161,163,197✔

4850

  int32_t              code = TSDB_CODE_SUCCESS;

161,163,197✔

4853

    if (pInfo->rtnNextDurationBlocks) {

172,085,274✔

4854

      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d",

9,540,244✔

4857

      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {

9,547,695✔

4858

        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];

4,463,477✔

4859

        ++pInfo->nextDurationBlocksIdx;

4,463,094✔

4861

        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

9,544,256✔

4862

          blockDataDestroy(pInfo->nextDurationBlocks[i]);

4,462,780✔

4863

          pInfo->nextDurationBlocks[i] = NULL;

4,459,716✔

4866

        pInfo->rtnNextDurationBlocks = false;

5,083,391✔

4867

        pInfo->nextDurationBlocksIdx = 0;

5,083,008✔

4868

        pInfo->numNextDurationBlocks = 0;

5,079,944✔

4869

        continue;

5,081,859✔

4872

      bool bFinished = false;

162,547,021✔

4873

      bool bSkipped = false;

162,546,288✔

4875

      code = doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);

162,560,653✔

4876

      if (code != 0) {

162,500,479✔

4877

        return code;

17,280✔

4880

      pBlock = pInfo->pReaderBlock;

162,483,199✔

4881

      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d",

162,502,810✔

4883

      if (bFinished) {

162,555,996✔

4884

        pInfo->bNewFilesetEvent = false;

28,163,525✔

4885

        break;

28,163,555✔

4888

      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {

134,392,471✔

4889

        if (!bSkipped) {

5,089,123✔

4890

          code = createOneDataBlock(pBlock, true, &pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks]);

4,463,860✔

4891

          if (code) {

4,455,817✔

4896

          ++pInfo->numNextDurationBlocks;

4,455,817✔

4897

          if (pInfo->numNextDurationBlocks > 2) {

4,454,050✔

4905

        if (pInfo->bNewFilesetEvent) {

5,084,910✔

4906

          pInfo->rtnNextDurationBlocks = true;

5,074,043✔

4907

          *ppBlock = NULL;

5,079,405✔

4908

          return code;

5,069,830✔

4911

        if (pInfo->bNextDurationBlockEvent) {

2,681✔

4916

      if (bSkipped) continue;

129,310,363✔

4919

    pBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pBlock->info.id.uid);

127,934,835✔

4921

    pOperator->resultInfo.totalRows += pBlock->info.rows;

127,932,356✔

4922

    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

127,928,139✔

4923

    *ppBlock = pBlock;

127,925,334✔

4925

    return code;

127,923,908✔

4928

  *ppBlock = NULL;

28,163,555✔

4929

  return code;

28,164,057✔

4932

int32_t generateSortByTsPkInfo(SArray* colMatchInfo, int32_t order, SArray** ppSortArray) {

24,498,740✔

4933

  int32_t code = TSDB_CODE_SUCCESS;

24,498,740✔

4934

  int32_t lino = 0;

24,498,740✔

4935

  SArray* pSortInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));

24,498,740✔

4936

  QUERY_CHECK_NULL(pSortInfo, code, lino, _end, terrno);

24,474,307✔

4937

  SBlockOrderInfo biTs = {0};

24,474,307✔

4938

  SBlockOrderInfo biPk = {0};

24,471,188✔

4940

  int32_t tsTargetSlotId = 0;

24,483,408✔

4941

  int32_t pkTargetSlotId = -1;

24,483,408✔

4942

  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {

86,794,937✔

4943

    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);

62,370,114✔

4944

    QUERY_CHECK_NULL(colInfo, code, lino, _end, terrno);

62,377,733✔

4945

    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

62,377,733✔

4946

      tsTargetSlotId = colInfo->dstSlotId;

24,498,442✔

4947

      biTs.order = order;

24,512,712✔

4948

      biTs.slotId = tsTargetSlotId;

24,512,712✔

4949

      biTs.nullFirst = (order == TSDB_ORDER_ASC);

24,512,712✔

4950

      biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, order);

24,512,712✔

4953

    if (colInfo->isPk) {

62,373,524✔

4954

      pkTargetSlotId = colInfo->dstSlotId;

1,074,997✔

4955

      biPk.order = order;

1,074,997✔

4956

      biPk.slotId = pkTargetSlotId;

1,074,997✔

4957

      biPk.nullFirst = (order == TSDB_ORDER_ASC);

1,074,997✔

4958

      biPk.compFn = getKeyComparFunc(colInfo->dataType.type, order);

1,074,997✔

4962

  void* tmp = taosArrayPush(pSortInfo, &biTs);

24,518,274✔

4963

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

24,518,274✔

4964

  if (pkTargetSlotId != -1) {

24,518,274✔

4965

    tmp = taosArrayPush(pSortInfo, &biPk);

1,075,645✔

4966

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

1,075,645✔

4969

  (*ppSortArray) = pSortInfo;

24,520,218✔

4971

_end:

24,490,663✔

4972

  if (code != TSDB_CODE_SUCCESS) {

24,489,601✔

4975

  return code;

24,489,601✔

4978

void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {

14,410,538✔

4979

  STableMergeScanInfo* pTmsInfo = param;

14,410,538✔

4980

  if (type == TSD_READER_NOTIFY_DURATION_START) {

14,410,538✔

4981

    pTmsInfo->bNewFilesetEvent = true;

14,421,253✔

4985

  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);

14,395,519✔

4987

  return;

14,437,824✔

4990

int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

33,446,873✔

4991

  STableMergeScanInfo* pInfo = pOperator->info;

33,446,873✔

4992

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

33,466,802✔

4994

  int32_t code = TSDB_CODE_SUCCESS;

33,473,790✔

4995

  int32_t lino = 0;

33,473,790✔

4996

  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

33,473,790✔

4998

  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));

33,448,773✔

4999

  pInfo->bNewFilesetEvent = false;

33,475,511✔

5000

  pInfo->bNextDurationBlockEvent = false;

33,492,761✔

5002

  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;

33,488,805✔

5003

  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;

33,486,151✔

5005

  pInfo->pSortHandle = NULL;

33,484,206✔

5006

  code = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,

66,984,488✔

5007

                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0, &pInfo->pSortHandle);

33,488,639✔

5008

  if (code) {

33,445,517✔

5012

  if (pInfo->bSortRowId && numOfTable != 1) {

33,445,517✔

5013

    int32_t memSize = 512 * 1024 * 1024;

3,164,718✔

5014

    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);

3,164,718✔

5015

    QUERY_CHECK_CODE(code, lino, _end);

3,166,570✔

5018

  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);

33,452,311✔

5019

  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pOperator);

33,451,444✔

5020

  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);

33,438,981✔

5022

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);

33,431,689✔

5023

  QUERY_CHECK_CODE(code, lino, _end);

33,455,759✔

5025

  STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));

33,455,759✔

5026

  QUERY_CHECK_NULL(param, code, lino, _end, terrno);

33,441,235✔

5027

  param->pOperator = pOperator;

33,441,235✔

5029

  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));

33,449,095✔

5030

  if (ps == NULL) {

33,423,459✔

5035

  ps->param = param;

33,423,459✔

5036

  ps->onlyRef = false;

33,452,500✔

5037

  code = tsortAddSource(pInfo->pSortHandle, ps);

33,459,808✔

5038

  QUERY_CHECK_CODE(code, lino, _end);

33,477,018✔

5040

  if (numOfTable == 1) {

33,477,018✔

5041

    tsortSetSingleTableMerge(pInfo->pSortHandle);

23,493,593✔

5043

    code = tsortOpen(pInfo->pSortHandle);

9,983,425✔

5044

    QUERY_CHECK_CODE(code, lino, _end);

9,987,599✔

5047

_end:

9,978,959✔

5048

  if (code != TSDB_CODE_SUCCESS) {

33,458,803✔

5049

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,640✔

5051

  return code;

33,470,501✔

5054

void startGroupTableMergeScan(SOperatorInfo* pOperator) {

28,394,928✔

5055

  int32_t              code = TSDB_CODE_SUCCESS;

28,394,928✔

5056

  int32_t              lino = 0;

28,394,928✔

5057

  STableMergeScanInfo* pInfo = pOperator->info;

28,394,928✔

5058

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,399,659✔

5059

  SReadHandle*         pHandle = &pInfo->base.readHandle;

28,373,770✔

5060

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,376,978✔

5061

  qDebug("%s table merge scan start group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,395,389✔

5064

    int32_t numOfTables = 0;

28,401,125✔

5065

    code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

28,380,852✔

5066

    QUERY_CHECK_CODE(code, lino, _end);

28,393,838✔

5068

    int32_t i = pInfo->tableStartIndex + 1;

28,393,838✔

5069

    for (; i < numOfTables; ++i) {

137,256,882✔

5070

      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

115,241,140✔

5071

      QUERY_CHECK_NULL(tableKeyInfo, code, lino, _end, terrno);

115,238,336✔

5072

      if (tableKeyInfo->groupId != pInfo->groupId) {

115,238,336✔

5073

        break;

6,377,688✔

5076

    pInfo->tableEndIndex = i - 1;

28,393,430✔

5078

  pInfo->bGroupProcessed = false;

28,406,868✔

5079

  int32_t tableStartIdx = pInfo->tableStartIndex;

28,401,254✔

5080

  int32_t tableEndIdx = pInfo->tableEndIndex;

28,401,835✔

5082

  int32_t        numOfTable = tableEndIdx - tableStartIdx + 1;

28,407,850✔

5083

  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);

28,407,850✔

5084

  code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,

56,795,577✔

5085

                                       (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->mSkipTables);

28,401,244✔

5086

  QUERY_CHECK_CODE(code, lino, _end);

28,373,267✔

5087

  if (pInfo->filesetDelimited) {

28,373,267✔

5088

    pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);

28,381,247✔

5090

  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);

28,392,660✔

5092

  code = startDurationForGroupTableMergeScan(pOperator);

28,359,409✔

5093

  QUERY_CHECK_CODE(code, lino, _end);

28,371,207✔

5095

_end:

28,362,567✔

5096

  if (code != TSDB_CODE_SUCCESS) {

28,370,057✔

5097

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,640✔

5098

    pTaskInfo->code = code;

8,640✔

5099

    T_LONG_JMP(pTaskInfo->env, code);

8,640✔

5101

28,361,417✔

5103

void stopGroupTableMergeScan(SOperatorInfo* pOperator) {

28,394,876✔

5104

  STableMergeScanInfo* pInfo = pOperator->info;

28,394,876✔

5105

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,394,876✔

5106

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,394,374✔

5108

  stopDurationForGroupTableMergeScan(pOperator);

28,393,872✔

5110

  if (pInfo->base.dataReader != NULL) {

28,386,011✔

5111

    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);

28,391,125✔

5112

    pInfo->base.dataReader = NULL;

28,346,315✔

5114

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

28,349,099✔

5115

    if (pInfo->nextDurationBlocks[i]) {

1,080✔

5116

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

1,080✔

5117

      pInfo->nextDurationBlocks[i] = NULL;

1,080✔

5119

    pInfo->numNextDurationBlocks = 0;

1,080✔

5120

    pInfo->nextDurationBlocksIdx = 0;

1,080✔

5122

  resetLimitInfoForNextGroup(&pInfo->limitInfo);

28,369,745✔

5123

  taosHashCleanup(pInfo->mSkipTables);

28,330,875✔

5124

  pInfo->mSkipTables = NULL;

28,341,699✔

5125

  qDebug("%s table merge scan stop group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,345,929✔

5126

28,386,430✔

5130

SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,

70,700,712✔

5132

  STableMergeScanInfo* pInfo = pOperator->info;

70,700,712✔

5133

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

70,714,151✔

5134

  STupleHandle*        pTupleHandle = NULL;

70,707,088✔

5136

  blockDataCleanup(pResBlock);

70,719,952✔

5139

    while (1) {

2,147,483,647✔

5140

      pTupleHandle = NULL;

2,147,483,647✔

5141

      int32_t code = tsortNextTuple(pHandle, &pTupleHandle);

2,147,483,647✔

5142

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5143

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

6,720✔

5144

        T_LONG_JMP(pOperator->pTaskInfo->env, code);

6,720✔

5146

      if (pTupleHandle == NULL) {

2,147,483,647✔

5147

        break;

54,035,664✔

5150

      code = tsortAppendTupleToBlock(pInfo->pSortHandle, pResBlock, pTupleHandle);

2,147,483,647✔

5151

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5156

      if (pResBlock->info.rows >= capacity) {

2,147,483,647✔

5157

        break;

16,671,812✔

5161

    if (tsortIsClosed(pHandle)) {

70,707,476✔

5166

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);

70,711,676✔

5167

    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,

70,713,550✔

5169

    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {

70,723,647✔

5173

  return (pResBlock->info.rows > 0) ? pResBlock : NULL;

70,723,647✔

5176

int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

62,108,025✔

5177

  if (pOperator->status == OP_EXEC_DONE) {

62,108,025✔

5178

    (*ppRes) = NULL;

45,850✔

5179

    return TSDB_CODE_SUCCESS;

45,850✔

5182

  int32_t              code = TSDB_CODE_SUCCESS;

62,044,309✔

5183

  int32_t              lino = 0;

62,044,309✔

5184

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

62,044,309✔

5185

  STableMergeScanInfo* pInfo = pOperator->info;

62,065,236✔

5187

  code = pOperator->fpSet._openFn(pOperator);

62,066,198✔

5188

  QUERY_CHECK_CODE(code, lino, _end);

62,050,329✔

5190

  int64_t st = taosGetTimestampUs();

62,055,851✔

5192

  int32_t tableListSize = 0;

62,055,851✔

5193

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

62,056,175✔

5194

  QUERY_CHECK_CODE(code, lino, _end);

62,061,990✔

5196

  if (!pInfo->hasGroupId) {

62,061,990✔

5197

    pInfo->hasGroupId = true;

24,814,997✔

5199

    if (tableListSize == 0) {

24,807,049✔

5200

      setOperatorCompleted(pOperator);

2,804,314✔

5201

      (*ppRes) = NULL;

2,803,931✔

5202

      return code;

2,803,147✔

5204

    pInfo->tableStartIndex = 0;

22,002,735✔

5205

    STableKeyInfo* tmp = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

22,016,798✔

5206

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

21,987,625✔

5207

    pInfo->groupId = tmp->groupId;

21,987,625✔

5208

    startGroupTableMergeScan(pOperator);

22,007,647✔

5211

  SSDataBlock* pBlock = NULL;

59,231,519✔

5212

  while (pInfo->tableStartIndex < tableListSize) {

70,692,192✔

5213

    if (isTaskKilled(pTaskInfo)) {

70,706,151✔

5214

      goto _end;

540✔

5217

    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,

70,698,649✔

5219

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

70,726,431✔

5220

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,852,757✔

5221

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

1,852,757✔

5222

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

1,852,757✔

5224

    if (pBlock != NULL) {

70,727,859✔

5225

      pBlock->info.id.groupId = pInfo->groupId;

37,245,831✔

5226

      pOperator->resultInfo.totalRows += pBlock->info.rows;

37,246,347✔

5227

      pInfo->bGroupProcessed = true;

37,245,283✔

5228

      break;

37,245,423✔

5230

      if (pInfo->bNewFilesetEvent) {

33,482,028✔

5231

        stopDurationForGroupTableMergeScan(pOperator);

5,086,386✔

5232

        code = startDurationForGroupTableMergeScan(pOperator);

5,081,407✔

5233

        QUERY_CHECK_CODE(code, lino, _end);

5,082,469✔

5236

        stopGroupTableMergeScan(pOperator);

28,394,876✔

5237

        if (pInfo->tableEndIndex >= tableListSize - 1) {

28,392,860✔

5238

          setOperatorCompleted(pOperator);

22,009,552✔

5239

          break;

22,013,104✔

5242

        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

6,378,204✔

5243

        STableKeyInfo* tmp = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

6,378,204✔

5244

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

6,378,204✔

5245

        pInfo->groupId = tmp->groupId;

6,378,204✔

5246

        startGroupTableMergeScan(pOperator);

6,378,204✔

5247

        resetLimitInfoForNextGroup(&pInfo->limitInfo);

6,377,688✔

5252

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

59,253,470✔

5254

_end:

59,257,421✔

5255

  if (code != TSDB_CODE_SUCCESS) {

59,257,389✔

5260

    (*ppRes) = pBlock;

59,257,389✔

5263

  return code;

59,255,259✔

5272

void destroyTableMergeScanOperatorInfo(void* param) {

24,524,536✔

5273

  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;

24,524,536✔

5276

  if (pTableScanInfo->base.readerAPI.tsdReaderClose != NULL) {

24,524,536✔

5277

    pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);

24,528,240✔

5278

    pTableScanInfo->base.dataReader = NULL;

24,512,336✔

5281

  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {

24,517,574✔

5288

  tsortDestroySortHandle(pTableScanInfo->pSortHandle);

24,525,499✔

5289

  pTableScanInfo->pSortHandle = NULL;

24,520,790✔

5290

  taosHashCleanup(pTableScanInfo->mSkipTables);

24,526,796✔

5291

  pTableScanInfo->mSkipTables = NULL;

24,516,264✔

5292

  blockDataDestroy(pTableScanInfo->pSortInputBlock);

24,516,716✔

5293

  pTableScanInfo->pSortInputBlock = NULL;

24,523,305✔

5296

  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);

24,523,697✔

5298

  blockDataDestroy(pTableScanInfo->pResBlock);

24,519,916✔

5299

  pTableScanInfo->pResBlock = NULL;

24,525,178✔

5302

  blockDataDestroy(pTableScanInfo->pReaderBlock);

24,526,061✔

5303

  pTableScanInfo->pReaderBlock = NULL;

24,522,407✔

5304

  taosArrayDestroy(pTableScanInfo->pSortInfo);

24,528,544✔

5306

  stopSubTablesTableMergeScan(pTableScanInfo);

24,519,610✔

5308

  taosMemoryFreeClear(param);

24,515,759✔

5309

24,517,670✔

5311

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {

1,254,618✔

5312

  if (pOptr == NULL) {

1,254,618✔

5317

  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));

1,254,618✔

5318

  if (!execInfo) {

1,253,059✔

5321

  STableMergeScanInfo* pInfo = pOptr->info;

1,253,059✔

5322

  execInfo->blockRecorder = pInfo->base.readRecorder;

1,253,059✔

5323

  execInfo->sortExecInfo = pInfo->sortExecInfo;

1,252,293✔

5325

  *pOptrExplain = execInfo;

1,255,785✔

5326

  *len = sizeof(STableMergeScanExecInfo);

1,253,059✔

5328

  return TSDB_CODE_SUCCESS;

1,255,812✔

5331

static int32_t resetTableMergeScanOperatorState(SOperatorInfo* pOper) {

2,304,014✔

5332

  int32_t         code = TSDB_CODE_SUCCESS;

2,304,014✔

5333

  STableMergeScanInfo*   pInfo = pOper->info;

2,304,014✔

5334

  pOper->status = OP_NOT_OPENED;

2,304,014✔

5336

  pInfo->tableEndIndex = 0;

2,304,014✔

5337

  pInfo->tableStartIndex = 0;

2,304,014✔

5338

  pInfo->hasGroupId = false;

2,304,014✔

5339

  if (pInfo->base.readerAPI.tsdReaderClose) {

2,304,014✔

5340

    pInfo->base.readerAPI.tsdReaderClose(pInfo->base.dataReader);

2,304,014✔

5342

  pInfo->base.dataReader = NULL;

2,304,014✔

5343

  pInfo->base.scanFlag = MAIN_SCAN;

2,304,014✔

5345

  pInfo->base.limitInfo = (SLimitInfo){0};

2,304,014✔

5346

  pInfo->base.limitInfo.limit.limit = -1;

2,304,014✔

5347

  pInfo->base.limitInfo.slimit.limit = -1;

2,304,014✔

5349

  tableListDestroy(pInfo->base.pTableListInfo);

2,304,014✔

5350

  pInfo->base.pTableListInfo = tableListCreate();

2,304,014✔

5351

  if (!pInfo->base.pTableListInfo) {

2,304,014✔

5355

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

2,304,014✔

5357

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

2,304,014✔

5358

  code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

4,608,028✔

5360

                                  pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

4,608,028✔

5361

  if (code) {

2,304,014✔

5366

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

2,304,014✔

5367

  cleanupQueryTableDataCond(&pInfo->base.cond);

2,304,014✔

5368

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, &pInfo->base.readHandle, false);

2,303,751✔

5369

  if (code) {

2,303,765✔

5374

  tsortDestroySortHandle(pInfo->pSortHandle);

2,303,765✔

5375

  pInfo->pSortHandle = NULL;

2,303,751✔

5377

  taosHashCleanup(pInfo->mSkipTables);

2,304,014✔

5378

  pInfo->mSkipTables = NULL;

2,304,014✔

5379

  pOper->resultInfo.totalRows = 0;

2,304,014✔

5381

  pInfo->bGroupProcessed = false;

2,303,751✔

5382

  pInfo->bNewFilesetEvent = false;

2,304,014✔

5383

  pInfo->bNextDurationBlockEvent = false;

2,304,014✔

5384

  pInfo->rtnNextDurationBlocks = false;

2,304,014✔

5385

  pInfo->nextDurationBlocksIdx = 0;

2,304,014✔

5386

  pInfo->bSortRowId = false;

2,303,751✔

5388

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

2,303,751✔

5394

  pInfo->numNextDurationBlocks = 0;

2,304,014✔

5396

  stopSubTablesTableMergeScan(pInfo);

2,303,751✔

5397

  return code;

2,304,014✔

5400

int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

24,481,280✔

5403

  QRY_PARAM_CHECK(pOptrInfo);

24,481,280✔

5405

  int32_t              code = TSDB_CODE_SUCCESS;

24,503,000✔

5406

  int32_t              lino = 0;

24,503,000✔

5407

  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));

24,503,000✔

5408

  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

24,436,764✔

5409

  if (pInfo == NULL || pOperator == NULL) {

24,439,874✔

5414

  pOperator->pPhyNode = pTableScanNode;

24,444,962✔

5415

  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

24,463,184✔

5417

  int32_t numOfCols = 0;

24,462,826✔

5418

  code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,

24,472,118✔

5420

  QUERY_CHECK_CODE(code, lino, _error);

24,465,023✔

5422

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle, false);

24,465,023✔

5423

  QUERY_CHECK_CODE(code, lino, _error);

24,451,578✔

5425

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

24,451,578✔

5426

    SExprSupp* pSup = &pInfo->base.pseudoSup;

10,533,443✔

5427

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->pExprInfo, &pSup->numOfExprs);

10,527,858✔

5428

    QUERY_CHECK_CODE(code, lino, _error);

10,509,463✔

5430

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

10,509,463✔

5432

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

10,503,843✔

5435

  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};

24,451,654✔

5437

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

24,448,438✔

5442

  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;

24,448,438✔

5443

  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

24,482,474✔

5444

  pInfo->base.scanFlag = MAIN_SCAN;

24,358,642✔

5445

  pInfo->base.readHandle = *readHandle;

24,398,599✔

5447

  pInfo->readIdx = -1;

24,479,098✔

5449

  pInfo->base.limitInfo.limit.limit = -1;

24,376,391✔

5450

  pInfo->base.limitInfo.slimit.limit = -1;

24,446,866✔

5451

  pInfo->base.pTableListInfo = pTableListInfo;

24,409,362✔

5453

  pInfo->sample.sampleRatio = pTableScanNode->ratio;

24,479,108✔

5454

  pInfo->sample.seed = taosGetTimestampSec();

24,441,010✔

5456

  int32_t sortOrder = pInfo->base.cond.order;

24,462,599✔

5457

  if (pTableScanNode->scan.node.dynamicOp) {

24,480,381✔

5458

    memcpy(&pInfo->base.orgCond, &pInfo->base.cond, sizeof(SQueryTableDataCond));

64,792✔

5459

    sortOrder = pInfo->base.orgCond.order;

64,792✔

5460

    memset(&pInfo->base.cond, 0, sizeof(SQueryTableDataCond));

64,792✔

5463

  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,

24,487,698✔

5464

                            pTaskInfo->pStreamRuntimeInfo);

24,480,633✔

5465

  QUERY_CHECK_CODE(code, lino, _error);

24,425,730✔

5467

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

24,425,730✔

5469

  pInfo->mergeLimit = -1;

24,435,180✔

5470

  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;

24,472,416✔

5471

  if (hasLimit) {

24,413,980✔

5472

    pInfo->mergeLimit = pInfo->limitInfo.limit.offset != -1

4,096,234✔

5473

                            ? pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset

811,415✔

5474

                            : pInfo->limitInfo.limit.limit;

2,858,384✔

5475

    pInfo->mSkipTables = NULL;

2,047,897✔

5478

  initResultSizeInfo(&pOperator->resultInfo, 1024);

24,413,098✔

5479

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

24,439,689✔

5480

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

24,480,628✔

5481

  code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

24,430,715✔

5482

  QUERY_CHECK_CODE(code, lino, _error);

24,513,863✔

5483

  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {

24,513,863✔

5484

    pInfo->bSortRowId = true;

4,888,851✔

5486

    pInfo->bSortRowId = false;

19,611,051✔

5489

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

24,499,876✔

5490

  QUERY_CHECK_CODE(code, lino, _error);

24,483,414✔

5492

  code = generateSortByTsPkInfo(pInfo->base.matchInfo.pList, sortOrder, &pInfo->pSortInfo);

24,483,414✔

5493

  QUERY_CHECK_CODE(code, lino, _error);

24,475,650✔

5495

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pReaderBlock);

24,475,650✔

5496

  QUERY_CHECK_CODE(code, lino, _error);

24,492,485✔

5498

  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;

24,492,485✔

5500

  int32_t  rowSize = pInfo->pResBlock->info.rowSize;

24,512,019✔

5501

  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);

24,473,683✔

5503

  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);

24,472,535✔

5506

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pSortInputBlock);

24,509,945✔

5507

  QUERY_CHECK_CODE(code, lino, _error);

24,508,157✔

5509

  if (!tsExperimental) {

24,508,157✔

5512

    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

24,508,157✔

5515

  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,

24,493,935✔

5517

  pOperator->exprSupp.numOfExprs = numOfCols;

24,460,669✔

5519

  pOperator->fpSet = createOperatorFpSet(

24,485,047✔

5520

      optrDummyOpenFn, (pTableScanNode->paraTablesSort || pTableScanNode->scan.node.dynamicOp) ? doTableMergeScanParaSubTablesNext : doTableMergeScanNext, NULL,

24,484,471✔

5523

  setOperatorResetStateFn(pOperator, resetTableMergeScanOperatorState);

24,486,360✔

5524

  pOperator->cost.openCost = 0;

24,454,110✔

5526

  *pOptrInfo = pOperator;

24,460,685✔

5527

  return code;

24,430,824✔

5566

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {

28,354✔

5567

  if (scanCols != NULL) {

28,354✔

5568

    SNode* pNode = NULL;

14,918✔

5569

    FOREACH(pNode, scanCols) {

34,664✔

5570

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

19,922✔

5573

      STargetNode* targetNode = (STargetNode*)pNode;

19,922✔

5574

      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {

19,922✔

5577

      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);

19,922✔

5578

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

19,922✔

5579

        supp->dbNameSlotId = targetNode->slotId;

9,832✔

5580

      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

10,090✔

5581

        supp->stbNameSlotId = targetNode->slotId;

9,914✔

5585

  return TSDB_CODE_SUCCESS;

28,478✔

5588

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {

28,654✔

5589

  if (pseudoCols != NULL) {

28,654✔

5590

    SNode* pNode = NULL;

28,654✔

5591

    FOREACH(pNode, pseudoCols) {

57,308✔

5592

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

28,654✔

5595

      STargetNode* targetNode = (STargetNode*)pNode;

28,654✔

5596

      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {

28,654✔

5599

      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);

28,654✔

5600

      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {

28,654✔

5601

        supp->tbCountSlotId = targetNode->slotId;

28,478✔

5605

  return TSDB_CODE_SUCCESS;

28,654✔

5608

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {

28,654✔

5609

  if (groupTags != NULL) {

28,654✔

5610

    SNode* pNode = NULL;

14,918✔

5611

    FOREACH(pNode, groupTags) {

34,840✔

5612

      if (nodeType(pNode) != QUERY_NODE_COLUMN) {

19,922✔

5615

      SColumnNode* colNode = (SColumnNode*)pNode;

19,922✔

5616

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

19,922✔

5617

        supp->groupByDbName = true;

9,832✔

5619

      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

19,922✔

5620

        supp->groupByStbName = true;

10,090✔

5624

    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);

13,736✔

5625

    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);

13,736✔

5627

  return TSDB_CODE_SUCCESS;

28,654✔

5630

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,

28,654✔

5632

  int32_t code = 0;

28,654✔

5633

  code = tblCountScanGetInputs(groupTags, tableName, supp);

28,654✔

5634

  if (code != TSDB_CODE_SUCCESS) {

28,654✔

5639

  supp->dbNameSlotId = -1;

28,654✔

5640

  supp->stbNameSlotId = -1;

28,654✔

5641

  supp->tbCountSlotId = -1;

28,654✔

5643

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);

28,654✔

5644

  if (code != TSDB_CODE_SUCCESS) {

28,478✔

5649

  code = tblCountScanGetCountSlotId(pseudoCols, supp);

28,478✔

5650

  if (code != TSDB_CODE_SUCCESS) {

28,654✔

5651

    qError("%s get table count scan supp. get count error", GET_TASKID(taskInfo));

176✔

5654

  return code;

28,478✔

5657

int32_t createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,

28,354✔

5659

  QRY_PARAM_CHECK(pOptrInfo);

28,354✔

5661

  int32_t                      code = TSDB_CODE_SUCCESS;

28,654✔

5662

  int32_t                      lino = 0;

28,654✔

5663

  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;

28,654✔

5664

  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));

28,654✔

5665

  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

28,654✔

5666

  if (!pInfo || !pOperator) {

28,654✔

5671

  pInfo->readHandle = *readHandle;

28,654✔

5673

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

28,654✔

5674

  initResultSizeInfo(&pOperator->resultInfo, 1);

28,654✔

5675

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

28,654✔

5676

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

28,654✔

5678

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

28,654✔

5679

  QUERY_CHECK_CODE(code, lino, _error);

28,654✔

5681

  code = getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,

28,654✔

5684

  QUERY_CHECK_CODE(code, lino, _error);

28,478✔

5686

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,

28,478✔

5688

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScanNext, NULL, destoryTableCountScanOperator,

28,478✔

5690

  *pOptrInfo = pOperator;

28,478✔

5691

  return code;

28,478✔

5705

int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,

38,848✔

5707

  int32_t code = TSDB_CODE_SUCCESS;

38,848✔

5708

  int32_t lino = 0;

38,848✔

5709

  if (pSupp->dbNameSlotId != -1) {

38,848✔

5710

    QUERY_CHECK_CONDITION((strlen(dbName) > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

14,822✔

5711

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);

14,822✔

5712

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

14,822✔

5714

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

14,822✔

5715

    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

14,822✔

5717

    varDataSetLen(varDbName, strlen(dbName));

14,822✔

5718

    code = colDataSetVal(colInfoData, 0, varDbName, false);

14,822✔

5719

    QUERY_CHECK_CODE(code, lino, _end);

14,822✔

5722

  if (pSupp->stbNameSlotId != -1) {

38,672✔

5723

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);

19,300✔

5724

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

19,300✔

5725

    if (strlen(stbName) != 0) {

19,300✔

5726

      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

13,856✔

5727

      tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);

13,856✔

5728

      varDataSetLen(varStbName, strlen(stbName));

13,856✔

5729

      code = colDataSetVal(colInfoData, 0, varStbName, false);

13,856✔

5730

      QUERY_CHECK_CODE(code, lino, _end);

13,856✔

5736

  if (pSupp->tbCountSlotId != -1) {

38,848✔

5737

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);

38,672✔

5738

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

38,813✔

5739

    code = colDataSetVal(colInfoData, 0, (char*)&count, false);

38,813✔

5740

    QUERY_CHECK_CODE(code, lino, _end);

38,372✔

5742

  pRes->info.rows = 1;

38,372✔

5744

_end:

38,548✔

5745

  if (code != TSDB_CODE_SUCCESS) {

38,548✔

5748

  return code;

38,698✔

5751

static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {

10,984✔

5752

  STableCountScanSupp* pSupp = &pInfo->supp;

10,984✔

5753

  SSDataBlock*         pRes = pInfo->pRes;

10,984✔

5755

  size_t infodbTableNum;

10,984✔

5756

  getInfosDbMeta(NULL, &infodbTableNum);

10,984✔

5757

  infodbTableNum -= 1;

10,984✔

5758

  size_t perfdbTableNum;

10,984✔

5759

  getPerfDbMeta(NULL, &perfdbTableNum);

10,984✔

5761

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

10,984✔

5762

    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);

9,426✔

5763

    return (pRes->info.rows > 0) ? pRes : NULL;

9,426✔

5765

    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);

1,558✔

5766

    return (pRes->info.rows > 0) ? pRes : NULL;

1,558✔

5770

static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,

1,558✔

5772

  int32_t        code = TSDB_CODE_SUCCESS;

1,558✔

5773

  int32_t        lino = 0;

1,558✔

5774

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,558✔

5775

  if (strlen(pSupp->stbNameFilter) != 0) {

1,558✔

5777

    code = fillTableCountScanDataBlock(pSupp, "", "", 0, pRes);

528✔

5778

    QUERY_CHECK_CODE(code, lino, _end);

528✔

5779

  } else if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {

1,030✔

5780

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

176✔

5781

    QUERY_CHECK_CODE(code, lino, _end);

176✔

5782

  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

854✔

5785

  } else if (strlen(pSupp->dbNameFilter) == 0) {

854✔

5786

    code = fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);

678✔

5787

    QUERY_CHECK_CODE(code, lino, _end);

678✔

5790

    code = fillTableCountScanDataBlock(pSupp, pSupp->dbNameFilter, pSupp->stbNameFilter, 0, pRes);

176✔

5791

    QUERY_CHECK_CODE(code, lino, _end);

176✔

5794

_end:

176✔

5795

  if (code != TSDB_CODE_SUCCESS) {

1,558✔

5800

  setOperatorCompleted(pOperator);

1,558✔

5801

1,558✔

5803

static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,426✔

5806

  int32_t        code = TSDB_CODE_SUCCESS;

9,426✔

5807

  int32_t        lino = 0;

9,426✔

5808

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,426✔

5809

  if (pInfo->currGrpIdx == 0) {

9,426✔

5810

    uint64_t groupId = 0;

3,142✔

5811

    if (pSupp->groupByDbName) {

3,142✔

5812

      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));

2,144✔

5814

      groupId = calcGroupId("", 0);

998✔

5817

    pRes->info.id.groupId = groupId;

3,142✔

5818

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

3,142✔

5819

    QUERY_CHECK_CODE(code, lino, _end);

3,142✔

5820

  } else if (pInfo->currGrpIdx == 1) {

6,284✔

5821

    uint64_t groupId = 0;

3,142✔

5822

    if (pSupp->groupByDbName) {

3,142✔

5823

      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));

2,144✔

5825

      groupId = calcGroupId("", 0);

998✔

5828

    pRes->info.id.groupId = groupId;

3,142✔

5829

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

3,142✔

5830

    QUERY_CHECK_CODE(code, lino, _end);

3,142✔

5832

    setOperatorCompleted(pOperator);

3,142✔

5834

  pInfo->currGrpIdx++;

9,426✔

5836

_end:

9,426✔

5837

  if (code != TSDB_CODE_SUCCESS) {

9,426✔

5842

9,426✔

5844

static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

67,468✔

5845

  int32_t                      code = TSDB_CODE_SUCCESS;

67,468✔

5846

  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;

67,468✔

5847

  STableCountScanOperatorInfo* pInfo = pOperator->info;

67,468✔

5848

  STableCountScanSupp*         pSupp = &pInfo->supp;

67,468✔

5849

  SSDataBlock*                 pRes = pInfo->pRes;

67,468✔

5851

  blockDataCleanup(pRes);

67,502✔

5852

  QRY_PARAM_CHECK(ppRes);

67,502✔

5854

  if (pOperator->status == OP_EXEC_DONE) {

67,502✔

5855

    return code;

17,580✔

5858

  if (pInfo->readHandle.mnd != NULL) {

49,922✔

5859

    (*ppRes) = buildSysDbTableCount(pOperator, pInfo);

10,984✔

5860

    return code;

10,984✔

5863

  code = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);

38,938✔

5864

  if (code != TSDB_CODE_SUCCESS) {

38,759✔

5869

  if (pRes->info.rows > 0) {

38,759✔

5870

    *ppRes = pRes;

31,006✔

5873

  return code;

38,903✔

5876

static int32_t buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

38,788✔

5878

  int32_t        code = TSDB_CODE_SUCCESS;

38,788✔

5879

  int32_t        lino = 0;

38,788✔

5880

  const char*    db = NULL;

38,788✔

5881

  int32_t        vgId = 0;

38,938✔

5882

  char           dbName[TSDB_DB_NAME_LEN] = {0};

38,938✔

5883

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

38,938✔

5884

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

38,938✔

5887

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);

38,938✔

5888

  SName sn = {0};

38,938✔

5890

  code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

38,938✔

5891

  QUERY_CHECK_CODE(code, lino, _end);

38,938✔

5893

  code = tNameGetDbName(&sn, dbName);

38,938✔

5894

  QUERY_CHECK_CODE(code, lino, _end);

38,938✔

5896

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

38,938✔

5897

    code = buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);

26,794✔

5899

    code = buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);

12,144✔

5902

_end:

38,909✔

5903

  if (code != TSDB_CODE_SUCCESS) {

38,909✔

5908

  return code;

38,909✔

5911

static int32_t buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

26,760✔

5913

  int32_t        code = TSDB_CODE_SUCCESS;

26,760✔

5914

  int32_t        lino = 0;

26,760✔

5915

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

26,760✔

5916

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

26,760✔

5918

  if (pSupp->groupByStbName) {

26,760✔

5919

    if (pInfo->stbUidList == NULL) {

22,916✔

5920

      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));

7,932✔

5921

      QUERY_CHECK_NULL(pInfo->stbUidList, code, lino, _end, terrno);

7,932✔

5922

      code = pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList);

7,932✔

5923

      QUERY_CHECK_CODE(code, lino, _end);

7,932✔

5926

    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {

22,916✔

5927

      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);

13,856✔

5928

      code = buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);

13,856✔

5929

      QUERY_CHECK_CODE(code, lino, _end);

13,856✔

5931

      pInfo->currGrpIdx++;

13,856✔

5932

    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {

9,060✔

5933

      code = buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);

7,932✔

5934

      QUERY_CHECK_CODE(code, lino, _end);

7,932✔

5936

      pInfo->currGrpIdx++;

7,932✔

5938

      setOperatorCompleted(pOperator);

1,128✔

5941

    uint64_t groupId = calcGroupId(dbName, strlen(dbName));

3,844✔

5942

    pRes->info.id.groupId = groupId;

3,844✔

5944

    int64_t dbTableCount = 0;

3,844✔

5945

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);

3,844✔

5946

    code = fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);

3,844✔

5947

    QUERY_CHECK_CODE(code, lino, _end);

3,844✔

5948

    setOperatorCompleted(pOperator);

3,844✔

5951

_end:

26,760✔

5952

  if (code != TSDB_CODE_SUCCESS) {

26,760✔

5956

  return code;

26,760✔

5959

static int32_t buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

12,178✔

5961

  int32_t        code = TSDB_CODE_SUCCESS;

12,178✔

5962

  int32_t        lino = 0;

12,178✔

5963

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

12,178✔

5964

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

12,178✔

5966

  if (strlen(pSupp->dbNameFilter) != 0 && strcmp(dbName, pSupp->dbNameFilter) != 0) {

12,178✔

5970

  } else if (strlen(pSupp->stbNameFilter) != 0) {

12,110✔

5971

    uint64_t uid = 0;

6,016✔

5972

    code = pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid);

6,016✔

5973

    if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

6,016✔

5975

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, 0, pRes);

1,408✔

5976

      QUERY_CHECK_CODE(code, lino, _end);

1,408✔

5978

      QUERY_CHECK_CODE(code, lino, _end);

4,608✔

5980

      int64_t numOfChildTables = 0;

4,608✔

5981

      code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL, NULL);

4,608✔

5982

      QUERY_CHECK_CODE(code, lino, _end);

4,608✔

5984

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes);

4,608✔

5985

      QUERY_CHECK_CODE(code, lino, _end);

4,608✔

5988

    int64_t tbNumVnode = 0;

6,128✔

5989

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);

6,162✔

5990

    code = fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);

6,162✔

5991

    QUERY_CHECK_CODE(code, lino, _end);

6,162✔

5994

_end:

12,178✔

5995

  if (code != TSDB_CODE_SUCCESS) {

12,178✔

6000

  setOperatorCompleted(pOperator);

12,178✔

6001

  return code;

12,149✔

6004

static int32_t buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

7,932✔

6006

  int32_t code = TSDB_CODE_SUCCESS;

7,932✔

6007

  int32_t lino = 0;

7,932✔

6008

  char    fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

7,932✔

6009

  if (pSupp->groupByDbName) {

7,932✔

6010

    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");

3,844✔

6013

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

7,932✔

6014

  pRes->info.id.groupId = groupId;

7,932✔

6016

  int64_t numOfTables = 0;

7,932✔

6017

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);

7,932✔

6019

  if (numOfTables != 0) {

7,932✔

6020

    code = fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);

1,128✔

6023

_end:

6,804✔

6024

  if (code != TSDB_CODE_SUCCESS) {

7,932✔

6027

  return code;

7,932✔

6030

static int32_t buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

13,856✔

6032

  int32_t code = TSDB_CODE_SUCCESS;

13,856✔

6033

  int32_t lino = 0;

13,856✔

6034

  char    stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

13,856✔

6035

  code = pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);

13,856✔

6036

  QUERY_CHECK_CODE(code, lino, _end);

13,856✔

6038

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

13,856✔

6039

  if (pSupp->groupByDbName) {

13,856✔

6040

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));

6,384✔

6042

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));

7,472✔

6045

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

13,856✔

6046

  pRes->info.id.groupId = groupId;

13,856✔

6048

  int64_t ctbNum = 0;

13,856✔

6049

  code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL, NULL);

13,856✔

6050

  QUERY_CHECK_CODE(code, lino, _end);

13,856✔

6051

  code = fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);

13,856✔

6052

  QUERY_CHECK_CODE(code, lino, _end);

13,856✔

6054

_end:

13,856✔

6055

  if (code != TSDB_CODE_SUCCESS) {

13,856✔

6058

  return code;

13,856✔

6061

static void destoryTableCountScanOperator(void* param) {

28,504✔

6062

  STableCountScanOperatorInfo* pTableCountScanInfo = param;

28,504✔

6063

  blockDataDestroy(pTableCountScanInfo->pRes);

28,504✔

6065

  taosArrayDestroy(pTableCountScanInfo->stbUidList);

28,654✔

6066

  taosMemoryFreeClear(param);

28,654✔

6067

28,654✔

taosdata / TDengine / #4984

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4984

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