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

79

bool processBlockWithProbability(const SSampleExecInfo* pInfo) {

682,974,565✔

88

  return true;

682,974,565✔

92

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

69,145✔

93

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

69,145✔

96

69,145✔

98

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

692,297,143✔

99

  int32_t     code = TSDB_CODE_SUCCESS;

692,297,143✔

100

  STimeWindow w = {0};

692,297,143✔

103

  if (pInterval->interval == 0) {

692,369,284✔

104

    *overlap = false;

653,052,402✔

105

    return code;

653,058,098✔

108

  if (order == TSDB_ORDER_ASC) {

39,319,005✔

109

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

38,066,823✔

110

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

38,066,898✔

115

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

38,060,194✔

116

      *overlap = true;

33,221,523✔

117

      return code;

33,223,607✔

121

      getNextTimeWindow(pInterval, &w, order);

4,843,684✔

122

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

4,846,710✔

123

        break;

4,566,940✔

126

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

279,209✔

130

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

279,209✔

131

        *overlap = true;

279,209✔

132

        return code;

279,209✔

136

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

1,252,182✔

137

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

1,253,069✔

142

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

1,251,785✔

143

      *overlap = true;

1,079,914✔

144

      return code;

1,079,120✔

148

      getNextTimeWindow(pInterval, &w, order);

173,155✔

149

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

173,949✔

150

        break;

171,948✔

153

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

2,640✔

157

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

2,640✔

158

        *overlap = true;

2,640✔

159

        return code;

2,640✔

164

  *overlap = false;

4,738,888✔

165

  return code;

4,738,337✔

169

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

189,250,677✔

170

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

189,250,677✔

174

  int64_t buf[2] = {0};

189,264,989✔

175

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

189,248,138✔

177

  STableScanInfo* pTableScanInfo = pOperator->info;

189,249,341✔

179

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

189,270,444✔

182

  if (p1 == NULL) {

189,287,057✔

183

    return NULL;

49,423,293✔

186

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

139,863,764✔

187

  if (NULL == *pPage) {

139,840,756✔

191

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

139,840,910✔

194

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

11,838,974✔

195

  if (NULL == pTableScanInfo->pIgnoreTables) {

11,838,974✔

196

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

870,693✔

197

    pTableScanInfo->pIgnoreTables =

870,693✔

198

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

870,693✔

199

    if (NULL == pTableScanInfo->pIgnoreTables) {

870,693✔

204

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

11,838,974✔

206

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

11,839,452✔

210

  return TSDB_CODE_SUCCESS;

11,839,452✔

213

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

778,427,347✔

214

  STableScanInfo* pTableScanInfo = pOperator->info;

778,427,347✔

215

  int32_t         code = TSDB_CODE_SUCCESS;

778,594,334✔

217

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

778,594,334✔

218

    return TSDB_CODE_SUCCESS;

589,315,829✔

221

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

189,254,887✔

223

  SFilePage*  pPage = NULL;

189,274,361✔

224

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

189,231,945✔

226

  if (pRow == NULL) {

189,273,139✔

227

    return TSDB_CODE_SUCCESS;

49,419,823✔

230

  bool notLoadBlock = true;

139,853,316✔

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;

128,017,705✔

239

      break;

128,017,705✔

244

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

140,048,775✔

246

  if (notLoadBlock) {

139,855,607✔

247

    *status = FUNC_DATA_REQUIRED_NOT_LOAD;

11,839,581✔

248

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

11,838,974✔

251

  return code;

139,822,629✔

254

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

10,196,488✔

256

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

10,196,488✔

261

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

10,198,264✔

264

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

192,626,863✔

266

  SStorageAPI* pAPI = &pTaskInfo->storageAPI;

192,626,863✔

267

  bool         allColumnsHaveAgg = true;

192,652,900✔

268

  bool         hasNullSMA = false;

192,649,060✔

269

  if (pLoad != NULL) {

192,643,465✔

270

    *pLoad = false;

192,679,981✔

273

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

192,636,276✔

275

  if (code != TSDB_CODE_SUCCESS) {

192,620,527✔

279

  if (!allColumnsHaveAgg || hasNullSMA) {

192,620,527✔

280

    *pLoad = false;

159,046,596✔

282

    *pLoad = true;

33,573,931✔

285

  return code;

192,620,982✔

288

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

767,551,097✔

290

  int32_t    code = 0;

767,551,097✔

291

  SExprSupp* pSup = &pTableScanInfo->pseudoSup;

767,551,097✔

292

  if (pSup->numOfExprs > 0) {

767,713,969✔

293

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

329,421,778✔

296

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

329,352,430✔

301

  return code;

767,622,457✔

304

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

1,237,746,465✔

305

  SLimit*     pLimit = &pLimitInfo->limit;

1,237,746,465✔

306

  const char* id = GET_TASKID(pTaskInfo);

1,237,804,660✔

308

  if (pLimitInfo->remainOffset > 0) {

1,237,862,092✔

309

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

5,890,979✔

310

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

3,105,260✔

311

      blockDataEmpty(pBlock);

3,105,260✔

312

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

3,105,260✔

313

      return false;

3,105,260✔

315

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

2,785,719✔

316

      if (code != TSDB_CODE_SUCCESS) {

2,785,719✔

321

      pLimitInfo->remainOffset = 0;

2,785,719✔

325

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

1,234,743,802✔

327

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

27,247,378✔

328

    blockDataKeepFirstNRows(pBlock, keep);

27,247,938✔

330

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

27,247,413✔

331

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

27,247,844✔

332

    return true;

27,248,344✔

335

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

1,207,378,526✔

336

  return false;

1,207,490,199✔

339

static bool skipSetTagColumnData(SOperatorInfo* pOperator) {

811,544,446✔

340

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

1,481,138,339✔

341

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

669,536,095✔

344

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

827,033,523✔

346

  int32_t        code = TSDB_CODE_SUCCESS;

827,033,523✔

347

  int32_t        lino = 0;

827,033,523✔

348

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

827,033,523✔

349

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

827,142,507✔

350

  bool           loadSMA = false;

827,122,194✔

352

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

827,122,194✔

354

  pCost->totalBlocks += 1;

827,089,156✔

355

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

827,126,428✔

356

  *status = pTableScanInfo->dataBlockLoadFlag;

827,026,876✔

358

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

827,062,735✔

359

    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

134,668,583✔

361

    bool overlap = false;

692,369,345✔

363

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

692,375,658✔

364

    if (ret != TSDB_CODE_SUCCESS) {

692,213,492✔

367

    if (overlap) {

692,213,492✔

368

      (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

34,580,745✔

372

  SDataBlockInfo* pBlockInfo = &pBlock->info;

827,086,677✔

373

  taosMemoryFreeClear(pBlock->pBlockAgg);

827,059,790✔

375

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {

827,061,240✔

382

  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

826,861,537✔

383

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

21,570,001✔

386

    if (!skipSetTagColumnData(pOperator)) {

21,571,831✔

387

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

21,576,107✔

389

    pCost->skipBlocks += 1;

21,576,107✔

390

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

21,576,107✔

391

    return code;

21,576,107✔

392

  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {

805,282,566✔

393

    pCost->loadBlockStatis += 1;

57,995,055✔

394

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

57,984,282✔

395

    bool success = true;

57,984,282✔

396

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

57,983,739✔

397

    if (code) {

57,980,635✔

403

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

57,973,189✔

404

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

23,371,995✔

406

      if (!skipSetTagColumnData(pOperator)) {

23,371,995✔

407

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

6,039,740✔

409

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

23,371,498✔

410

      return code;

23,367,229✔

412

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

34,601,194✔

413

      *status = FUNC_DATA_REQUIRED_DATA_LOAD;

34,612,797✔

417

  if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {

781,858,013✔

424

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

781,918,475✔

425

    bool success = true;

134,675,410✔

426

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

134,680,389✔

427

    if (code) {

134,630,156✔

430

      QUERY_CHECK_CODE(code, lino, _end);

25,849✔

433

    if (success) {

134,656,005✔

434

      size_t size = taosArrayGetSize(pBlock->pDataBlock);

10,195,156✔

435

      bool   keep = false;

10,198,856✔

436

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

10,199,300✔

437

      if (code) {

10,196,044✔

443

      if (!keep) {

10,195,748✔

444

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

3,482,504✔

446

        pCost->filterOutBlocks += 1;

3,482,504✔

447

        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

3,482,504✔

448

        taosMemoryFreeClear(pBlock->pBlockAgg);

3,482,504✔

450

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

3,482,504✔

451

        return TSDB_CODE_SUCCESS;

3,482,504✔

457

  taosMemoryFreeClear(pBlock->pBlockAgg);

778,561,523✔

460

  code = doDynamicPruneDataBlock(pOperator, pBlockInfo, status);

778,458,890✔

461

  if (code) {

778,494,180✔

466

  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

778,494,180✔

467

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

11,840,059✔

469

    pCost->skipBlocks += 1;

11,840,906✔

470

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

11,840,474✔

472

    STableScanInfo* p1 = pOperator->info;

11,839,867✔

473

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

11,840,474✔

474

      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;

2,805,696✔

476

      *status = FUNC_DATA_REQUIRED_FILTEROUT;

9,034,778✔

478

    return TSDB_CODE_SUCCESS;

11,840,474✔

481

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

766,520,779✔

482

  pCost->loadBlocks += 1;

766,678,655✔

484

  SSDataBlock* p = NULL;

766,640,467✔

485

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

766,615,039✔

486

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

766,719,259✔

490

  if (!skipSetTagColumnData(pOperator)) {

766,719,259✔

491

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

691,966,147✔

492

    QUERY_CHECK_CODE(code, lino, _end);

691,930,693✔

496

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

766,661,949✔

498

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

766,863,524✔

499

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

131,161,042✔

500

    QUERY_CHECK_CODE(code, lino, _end);

131,139,626✔

502

    int64_t st = taosGetTimestampUs();

130,195,557✔

503

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

130,189,554✔

504

    pTableScanInfo->readRecorder.filterTime += el;

130,189,554✔

506

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

130,195,631✔

507

      pCost->filterOutBlocks += 1;

36,082,764✔

508

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

36,083,049✔

511

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

94,089,483✔

515

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

765,871,843✔

516

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

765,865,901✔

517

    setOperatorCompleted(pOperator);

3,491,600✔

520

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

765,865,901✔

522

_end:

766,864,622✔

523

  if (code != TSDB_CODE_SUCCESS) {

766,790,714✔

524

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

956,496✔

526

  return code;

766,698,926✔

529

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

69,145✔

530

  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

69,145✔

532

  switchCtxOrder(pCtx, numOfOutput);

69,145✔

533

  pTableScanInfo->cond.order = TSDB_ORDER_DESC;

69,145✔

536

69,145✔

599

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

329,937,441✔

601

  int32_t          code = TSDB_CODE_SUCCESS;

329,937,441✔

602

  int32_t          lino = 0;

329,937,441✔

603

  bool             freeReader = false;

329,937,441✔

604

  LRUHandle*       h = NULL;

329,937,441✔

605

  STableCachedVal  val = {0};

329,937,441✔

606

  SMetaReader      mr = {0};

329,937,441✔

607

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

329,933,106✔

608

  int32_t          insertRet = TAOS_LRU_STATUS_OK;

329,971,538✔

609

  STableCachedVal* pVal = NULL;

329,971,538✔

612

  if (numOfExpr <= 0) {

329,992,609✔

613

    return TSDB_CODE_SUCCESS;

437,901✔

619

  int32_t backupRows = pBlock->info.rows;

329,554,708✔

620

  pBlock->info.rows = rows;

329,618,998✔

626

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

329,610,720✔

627

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

329,609,120✔

628

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

329,638,445✔

629

    if (code != TSDB_CODE_SUCCESS) {

329,533,423✔

644

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

329,533,423✔

646

    val.pName = mr.me.name;

329,557,467✔

647

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

329,557,467✔

649

    freeReader = true;

329,557,467✔

691

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

1,286,148,333✔

692

    const SExprInfo* pExpr1 = &pExpr[j];

956,478,448✔

693

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

956,482,322✔

695

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

956,663,930✔

696

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

956,584,955✔

697

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

956,584,955✔

699

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

956,677,963✔

702

    if (fmIsScanPseudoColumnFunc(functionId)) {

956,709,864✔

703

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

132,321,872✔

704

      if (fType == FUNCTION_TYPE_TBNAME) {

132,330,944✔

705

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

132,330,944✔

706

        QUERY_CHECK_CODE(code, lino, _end);

132,323,498✔

715

      STagVal tagVal = {0};

824,220,051✔

716

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

824,234,207✔

717

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

824,355,239✔

719

      char* data = NULL;

824,413,609✔

720

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

824,413,609✔

721

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

818,494,162✔

723

        data = (char*)p;

5,912,862✔

726

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

824,295,583✔

727

      if (isNullVal) {

824,229,892✔

728

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

5,415,384✔

729

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

818,814,508✔

730

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

818,444,406✔

731

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

818,492,559✔

732

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

285,135,720✔

733

          if (tmp != NULL) {

285,168,808✔

734

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

285,168,808✔

735

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

285,170,111✔

737

            taosMemoryFree(tmp);

285,215,863✔

739

          taosMemoryFree(data);

285,214,763✔

741

        QUERY_CHECK_CODE(code, lino, _end);

818,409,462✔

743

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

3,103,904✔

744

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

2,605,642✔

745

          QUERY_CHECK_CODE(code, lino, _end);

2,606,034✔

752

  pBlock->info.rows = backupRows;

329,669,885✔

754

_end:

329,679,580✔

756

  if (NULL != pVal) {

329,613,660✔

765

  if (freeReader) {

329,613,660✔

766

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

329,563,401✔

768

  if (code != TSDB_CODE_SUCCESS) {

329,606,885✔

771

  return code;

329,580,540✔

774

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

132,313,319✔

776

  int32_t                     code = TSDB_CODE_SUCCESS;

132,313,319✔

777

  int32_t                     lino = 0;

132,313,319✔

778

  struct SScalarFuncExecFuncs fpSet = {0};

132,313,319✔

779

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

132,316,148✔

780

  QUERY_CHECK_CODE(code, lino, _end);

132,324,491✔

782

  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;

132,324,491✔

783

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

132,324,491✔

784

  STR_TO_VARSTR(buf, name)

132,305,542✔

786

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

132,323,984✔

788

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

132,308,924✔

789

  QUERY_CHECK_CODE(code, lino, _end);

132,315,762✔

791

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

132,315,762✔

792

  QUERY_CHECK_CODE(code, lino, _end);

132,316,224✔

794

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

132,316,224✔

795

  SScalarParam param = {.columnData = pColInfoData};

132,321,123✔

797

  if (fpSet.process != NULL) {

132,299,883✔

798

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

132,299,883✔

799

    QUERY_CHECK_CODE(code, lino, _end);

132,327,607✔

804

  colDataDestroy(&infoData);

132,327,607✔

806

_end:

132,317,975✔

807

  if (code != TSDB_CODE_SUCCESS) {

132,318,789✔

810

  return code;

132,318,789✔

879

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

256,691,337✔

880

  int32_t code = TSDB_CODE_SUCCESS;

256,691,337✔

881

  int32_t lino = 0;

256,691,337✔

882

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

256,691,337✔

883

  QUERY_CHECK_CODE(code, lino, _end);

256,665,690✔

885

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

256,665,690✔

886

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

256,619,929✔

887

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

256,645,367✔

888

  taosMemoryFreeClear(pInfo->pResBlock->pBlockAgg);

256,597,227✔

890

  if (!pInfo->needCountEmptyTable) {

256,722,896✔

891

    pInfo->countState = TABLE_COUNT_STATE_END;

237,021,948✔

893

    pInfo->countState = TABLE_COUNT_STATE_SCAN;

19,655,578✔

896

_end:

256,644,994✔

897

  if (code != TSDB_CODE_SUCCESS) {

256,644,994✔

900

  return code;

256,633,535✔

903

void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {

682,290,275✔

904

  if (pInfo->countState == TABLE_COUNT_STATE_END) {

682,290,275✔

905

    return;

572,140,218✔

907

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

110,237,094✔

908

    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;

226,098✔

910

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

110,016,353✔

911

    if (code != TSDB_CODE_SUCCESS) {

110,019,547✔

912

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

43,927,440✔

917

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

48,080,970✔

919

  blockDataEmpty(pBlock);

48,080,970✔

920

  pBlock->info.rows = 1;

48,080,970✔

921

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

48,081,494✔

922

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

48,081,494✔

923

  pBlock->info.blankFill = true;

48,080,446✔

926

  int32_t col_num = blockDataGetNumOfCols(pBlock);

48,080,970✔

927

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

198,795,623✔

928

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

150,714,129✔

933

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

48,081,494✔

935

  return pBlock;

48,081,494✔

938

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

46,201,509✔

939

  STableScanInfo* pTableScanInfo = pOperator->info;

46,201,509✔

941

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

46,203,081✔

943

  pOperator->resultInfo.totalRows++;

46,200,563✔

944

  return pBlock;

46,204,653✔

951

static int32_t prepareTableScan(SOperatorInfo* pOperator) {

914,291,351✔

952

  int32_t code = TSDB_CODE_SUCCESS;

914,291,351✔

953

  int32_t lino = 0;

914,291,351✔

955

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

914,291,351✔

956

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

914,352,909✔

957

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

914,463,687✔

958

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

34,174,571✔

959

         NOTIFY_TYPE_SCAN_PARAM &&

3,141,151✔

960

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

3,141,151✔

961

    STableScanOperatorParam* pNotify =

3,141,151✔

964

    STableScanInfo* pTableScanInfo = pOperator->info;

3,141,151✔

965

    SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

3,141,151✔

966

    SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

3,141,151✔

967

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

3,141,151✔

969

    QUERY_CHECK_CODE(code, lino, _end);

3,141,151✔

971

    if (!pGetParam->reUse) {

3,141,151✔

972

      freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

99,894✔

978

      pGetParam->reUse = false;

3,041,257✔

983

    pOperator->pOperatorGetParam = NULL;

3,141,151✔

986

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

914,447,008✔

987

_end:

263,534,180✔

988

  if (code != TSDB_CODE_SUCCESS) {

914,447,539✔

992

  return code;

914,446,173✔

995

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

875,775,695✔

996

  int32_t         code = TSDB_CODE_SUCCESS;

875,775,695✔

997

  int32_t         lino = 0;

875,775,695✔

998

  STableScanInfo* pTableScanInfo = pOperator->info;

875,775,695✔

999

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

875,811,159✔

1000

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

875,779,646✔

1001

  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;

875,846,967✔

1002

  bool            hasNext = false;

875,785,366✔

1003

  int64_t         st = taosGetTimestampUs();

875,768,117✔

1005

  QRY_PARAM_CHECK(ppRes);

875,768,117✔

1006

  pBlock->info.dataLoad = false;

875,778,152✔

1008

  while (true) {

43,143,694✔

1009

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

918,856,583✔

1010

    if (code != TSDB_CODE_SUCCESS) {

918,784,145✔

1012

      QUERY_CHECK_CODE(code, lino, _end);

117✔

1015

    if (!hasNext) {

918,745,383✔

1016

      break;

235,565,908✔

1019

    if (isTaskKilled(pTaskInfo)) {

683,179,475✔

1020

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

20,640✔

1021

      code = pTaskInfo->code;

20,640✔

1022

      QUERY_CHECK_CODE(code, lino, _end);

20,640✔

1025

    if (pOperator->status == OP_EXEC_DONE) {

683,342,275✔

1026

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

216,384✔

1027

      break;

216,384✔

1031

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

683,056,617✔

1032

    if (!processThisBlock) {

683,067,009✔

1036

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

683,067,009✔

1037

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

683,128,750✔

1040

    uint32_t status = 0;

683,168,458✔

1041

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

683,179,368✔

1042

    QUERY_CHECK_CODE(code, lino, _end);

683,025,325✔

1044

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

682,085,662✔

1045

      break;

2,805,696✔

1049

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

679,279,966✔

1050

      continue;

43,142,634✔

1053

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

636,183,699✔

1054

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

636,202,456✔

1056

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

636,218,784✔

1057

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

636,200,778✔

1059

    (*ppRes) = pBlock;

636,249,548✔

1060

    return code;

636,141,340✔

1063

_end:

239,573,147✔

1064

  if (code != TSDB_CODE_SUCCESS) {

239,574,292✔

1065

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

974,967✔

1066

    pTaskInfo->code = code;

974,967✔

1067

    T_LONG_JMP(pTaskInfo->env, code);

960,303✔

1069

  return code;

238,599,325✔

1072

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

924,854,051✔

1073

  int32_t         code = TSDB_CODE_SUCCESS;

924,854,051✔

1074

  int32_t         lino = 0;

924,854,051✔

1075

  STableScanInfo* pTableScanInfo = pOperator->info;

924,854,051✔

1076

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

924,904,418✔

1077

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

924,875,083✔

1079

  QRY_PARAM_CHECK(pBlock);

924,895,980✔

1082

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

924,831,178✔

1083

    return code;

3,021,626✔

1087

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

1,144,829,557✔

1088

    SSDataBlock* p = NULL;

786,096,022✔

1089

    code = doTableScanImplNext(pOperator, &p);

786,062,315✔

1090

    QUERY_CHECK_CODE(code, lino, _end);

784,965,169✔

1092

    if (p != NULL) {

784,965,169✔

1093

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

561,989,460✔

1094

      *pBlock = p;

562,038,315✔

1095

      return code;

562,028,078✔

1098

    pTableScanInfo->scanTimes += 1;

222,975,709✔

1099

    taosHashClear(pTableScanInfo->pIgnoreTables);

222,980,184✔

1101

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

222,988,569✔

1102

      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

86,838✔

1103

      pTableScanInfo->base.scanFlag = MAIN_SCAN;

86,838✔

1104

      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

86,838✔

1105

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

86,838✔

1108

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

86,838✔

1109

      QUERY_CHECK_CODE(code, lino, _end);

85,531✔

1113

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

358,747,398✔

1114

  if (pTableScanInfo->scanTimes < total) {

358,758,464✔

1115

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

89,709,576✔

1116

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

69,145✔

1117

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

69,145✔

1118

      QUERY_CHECK_CODE(code, lino, _end);

69,145✔

1119

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

69,145✔

1122

    while (pTableScanInfo->scanTimes < total) {

105,330,077✔

1123

      SSDataBlock* p = NULL;

89,717,787✔

1124

      code = doTableScanImplNext(pOperator, &p);

89,704,160✔

1125

      QUERY_CHECK_CODE(code, lino, _end);

89,688,695✔

1127

      if (p != NULL) {

89,688,695✔

1128

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

74,074,958✔

1129

        *pBlock = p;

74,085,997✔

1130

        return code;

74,084,308✔

1133

      pTableScanInfo->scanTimes += 1;

15,613,737✔

1134

      taosHashClear(pTableScanInfo->pIgnoreTables);

15,616,182✔

1136

      if (pTableScanInfo->scanTimes < total) {

15,614,194✔

1142

        QUERY_CHECK_CODE(code, lino, _end);

3,328✔

1147

  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {

284,652,608✔

1148

    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;

65,246,529✔

1149

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

65,244,348✔

1150

      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {

110,279✔

1163

      int32_t numOfTables = 0;

65,131,973✔

1164

      code = tableListGetSize(pTableListInfo, &numOfTables);

65,132,936✔

1165

      QUERY_CHECK_CODE(code, lino, _end);

65,133,768✔

1167

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

65,133,768✔

1169

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

63,271,101✔

1170

        if (pIte != NULL) {

63,270,331✔

1171

          size_t        keySize = 0;

46,203,228✔

1172

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

46,203,228✔

1173

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

46,203,605✔

1174

          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);

46,202,557✔

1175

          markGroupProcessed(pTableScanInfo, *pGroupId);

46,202,557✔

1176

          *pBlock = getBlockForEmptyTable(pOperator, &info);

46,201,509✔

1178

          return code;

46,204,180✔

1182

    pTableScanInfo->countState = TABLE_COUNT_STATE_END;

19,041,267✔

1185

_end:

219,407,748✔

1186

  if (code != TSDB_CODE_SUCCESS) {

238,448,001✔

1191

  return code;

238,448,001✔

1194

static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {

979,472✔

1195

  STableScanInfo*          pInfo = pOperator->info;

979,472✔

1196

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

979,472✔

1197

  int32_t                  code = 0;

979,472✔

1198

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

979,472✔

1199

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

979,472✔

1200

  int32_t                  num = taosArrayGetSize(pParam->pUidList);

979,472✔

1201

  if (num <= 0) {

979,472✔

1206

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

979,472✔

1210

  if (pParam->tableSeq) {

979,472✔

1211

    pListInfo->oneTableForEachGroup = true;

726,681✔

1212

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

726,681✔

1218

    pListInfo->oneTableForEachGroup = false;

252,791✔

1219

    pListInfo->numOfOuputGroups = 1;

252,791✔

1221

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

979,472✔

1224

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

722,049✔

1225

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

722,049✔

1228

  STableKeyInfo info = {.groupId = 0};

979,472✔

1229

  int32_t       tableIdx = 0;

979,472✔

1230

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

9,240,210✔

1231

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

8,260,738✔

1232

    if (!pUid) {

8,260,738✔

1237

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

8,260,738✔

1238

      if (TSDB_CODE_DUP_KEY == terrno) {

6,922,754✔

1239

        continue;

6,922,754✔

1244

    info.uid = *pUid;

1,337,984✔

1245

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

1,337,984✔

1246

    if (p == NULL) {

1,337,592✔

1250

    tableIdx++;

1,337,592✔

1251

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

1,337,592✔

1254

  return code;

979,472✔

1257

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

240,431,080✔

1259

  const char* idStr = GET_TASKID(pTaskInfo);

240,431,080✔

1260

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

240,455,956✔

1261

  if (code) {

240,285,610✔

1266

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

240,285,610✔

1271

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

480,753,734✔

1272

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

240,416,292✔

1273

  if (code) {

240,391,151✔

1274

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

2,295✔

1277

  return code;

240,408,745✔

1280

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

313,170,377✔

1281

  SColIdPair* node1 = (SColIdPair*)elem1;

313,170,377✔

1282

  SColIdPair* node2 = (SColIdPair*)elem2;

313,170,377✔

1284

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

313,170,377✔

1285

    return 0;

112,184✔

1288

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

313,107,833✔

1291

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

416,940,051✔

1292

  SColIdSlotIdPair* node1 = (SColIdSlotIdPair*)elem1;

416,940,051✔

1293

  SColIdSlotIdPair* node2 = (SColIdSlotIdPair*)elem2;

416,940,051✔

1295

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

416,940,051✔

1296

    return 0;

171,368✔

1299

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

416,810,319✔

1313

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

98,859,995✔

1314

  if (pList == NULL) {

98,859,995✔

1318

  int32_t left = 0;

98,859,995✔

1319

  int32_t right = taosArrayGetSize(pList) - 1;

98,859,995✔

1320

  while (left <= right) {

302,172,221✔

1321

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

302,172,221✔

1322

    SColMatchItem* pItem = taosArrayGet(pList, mid);

302,172,221✔

1323

    if (pItem->colId == colId) {

302,172,221✔

1325

      while (mid > 0) {

98,859,995✔

1326

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

73,103,106✔

1327

        if (pPrev->colId != colId) {

73,103,106✔

1328

          break;

73,103,106✔

1332

      return taosArrayGet(pList, mid);

98,859,995✔

1335

    if (pItem->colId < colId) {

203,312,226✔

1336

      left = mid + 1;

96,168,507✔

1338

      right = mid - 1;

107,143,719✔

1345

static bool isNewScanParam(STableScanOperatorParam* pParam) {

44,165,062✔

1346

  return pParam->isNewParam;

44,165,062✔

1349

static int32_t createVTableScanInfoFromBatchParam(SOperatorInfo* pOperator) {

82,245,732✔

1350

  int32_t                  code = 0;

82,245,732✔

1351

  int32_t                  lino = 0;

82,245,732✔

1352

  STableScanInfo*          pInfo = pOperator->info;

82,245,732✔

1353

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

82,271,923✔

1354

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

82,261,725✔

1355

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

82,275,804✔

1356

  SMetaReader              orgTable = {0};

82,272,761✔

1357

  SMetaReader              superTable = {0};

82,280,561✔

1358

  SSchemaWrapper*          schema = NULL;

82,276,666✔

1359

  SArray*                  pColArray = NULL;

82,276,666✔

1360

  SArray*                  pBlockColArray = NULL;

82,276,666✔

1361

  int32_t                  num = 0;

82,276,666✔

1362

  STableKeyInfo*           pList = NULL;

82,278,778✔

1366

  cleanupQueryTableDataCond(&pInfo->base.cond);

82,255,676✔

1368

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

82,263,759✔

1369

  if (pGetParam &&

82,270,678✔

1370

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

4,295,501✔

1371

    STableScanOperatorParam* pParam =

4,299,775✔

1373

    pInfo->pBatchColMap = pParam->pBatchTbInfo;

4,291,580✔

1374

    pParam->pBatchTbInfo = NULL;

4,296,529✔

1375

    pInfo->cachedTimeWindow = pParam->window;

4,283,300✔

1376

    pInfo->currentBatchIdx = 0;

4,281,321✔

1377

    pInfo->lastBatchIdx = -1;

4,296,529✔

1378

    pInfo->cachedTagList = pParam->pTagList;

4,276,846✔

1379

    pParam->pTagList = NULL;

4,286,343✔

1380

    pInfo->cachedGroupId = pParam->groupid;

4,293,215✔

1382

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

4,283,443✔

1384

    taosHashClear(pInfo->pIgnoreTables);

4,283,443✔

1386

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

4,293,377✔

1387

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

3,569,991✔

1388

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

43,440,940✔

1389

        SqlFunctionCtx ctx = pSup->pCtx[i];

39,881,884✔

1390

        ctx.skipDynDataCheck = false;

39,870,419✔

1395

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

82,251,003✔

1396

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

82,283,732✔

1398

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

82,283,732✔

1399

    isNewTable = false;

73,732,472✔

1401

    isNewTable = true;

8,553,387✔

1402

    pInfo->lastBatchIdx = pInfo->currentBatchIdx;

8,553,387✔

1405

  if (isNewTable) {

82,273,473✔

1406

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

8,549,976✔

1407

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

8,556,786✔

1408

    pAPI->metaReaderFn.readerReleaseLock(&orgTable);

8,546,589✔

1409

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

8,550,231✔

1410

    QUERY_CHECK_CODE(code, lino, _return);

8,545,166✔

1411

    switch (orgTable.me.type) {

8,545,166✔

1412

      case TSDB_CHILD_TABLE:

838,252✔

1413

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

838,252✔

1414

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

838,252✔

1415

        pAPI->metaReaderFn.readerReleaseLock(&superTable);

838,252✔

1416

        QUERY_CHECK_CODE(code, lino, _return);

838,252✔

1417

        schema = &superTable.me.stbEntry.schemaRow;

838,252✔

1418

        break;

838,252✔

1419

      case TSDB_NORMAL_TABLE:

7,709,481✔

1420

        schema = &orgTable.me.ntbEntry.schemaRow;

7,709,481✔

1421

        break;

7,709,481✔

1422

      default:

83✔

1423

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

83✔

1428

    pListInfo->oneTableForEachGroup = true;

8,547,733✔

1429

    taosHashClear(pListInfo->map);

8,547,733✔

1430

    taosArrayClear(pListInfo->pTableList);

8,558,508✔

1432

    uint64_t      pUid = orgTable.me.uid;

8,557,911✔

1433

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

8,557,911✔

1434

    int32_t       tableIdx = 0;

8,557,325✔

1435

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

8,557,922✔

1436

    QUERY_CHECK_CODE(code, lino, _return);

8,558,508✔

1437

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

17,116,426✔

1438

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

8,557,918✔

1440

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

8,557,918✔

1441

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

8,554,611✔

1442

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

8,554,611✔

1443

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

8,554,014✔

1446

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

8,554,014✔

1447

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

8,555,212✔

1449

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

30,184,334✔

1450

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

21,630,231✔

1451

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

164,716,915✔

1452

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

164,629,369✔

1453

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

21,633,551✔

1454

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

21,635,742✔

1455

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

21,635,742✔

1457

          break;

21,629,729✔

1462

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

38,747,003✔

1463

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

30,188,007✔

1464

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

166,763,410✔

1465

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

166,666,749✔

1466

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

166,691,684✔

1467

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

30,191,989✔

1468

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

30,195,359✔

1469

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

30,195,359✔

1471

          break;

30,185,231✔

1476

    taosArraySort(pColArray, compareColIdPair);

8,556,795✔

1477

    taosArraySort(pBlockColArray, compareColIdSlotIdPair);

8,554,033✔

1478

    if (pInfo->pBlockColMap) {

8,553,085✔

1479

      taosArrayDestroy(pInfo->pBlockColMap);

6,417,829✔

1480

      pInfo->pBlockColMap = NULL;

6,417,885✔

1482

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

8,550,776✔

1483

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

8,556,199✔

1486

    taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

8,542,228✔

1487

    taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

8,545,567✔

1489

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

8,553,538✔

1490

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

6,218,692✔

1491

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

76,181,541✔

1492

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

69,971,617✔

1493

        bool           needScan = false;

69,971,201✔

1494

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

141,022,850✔

1495

          SFunctParam pParam = ctx->param[j];

71,055,567✔

1496

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

342,032,391✔

1497

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

288,101,745✔

1498

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

288,504,494✔

1499

              needScan |= true;

17,660,169✔

1500

              break;

17,660,169✔

1504

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

69,973,335✔

1505

        if (!needScan) {

69,984,249✔

1506

          ctx->skipDynDataCheck = true;

52,325,613✔

1508

          ctx->skipDynDataCheck = false;

17,658,636✔

1515

    pColArray = pInfo->lastColArray;

73,723,497✔

1516

    pBlockColArray = pInfo->lastBlockColArray;

73,724,691✔

1519

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

82,282,362✔

1520

  QUERY_CHECK_CODE(code, lino, _return);

82,289,464✔

1522

  if (pInfo->pResBlock) {

82,289,464✔

1523

    blockDataDestroy(pInfo->pResBlock);

77,990,675✔

1524

    pInfo->pResBlock = NULL;

77,977,481✔

1527

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

82,273,356✔

1535

  if (isNewTable) {

82,237,161✔

1536

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

8,548,370✔

1537

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

6,426,343✔

1538

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

6,432,142✔

1540

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

2,124,323✔

1541

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

2,111,332✔

1544

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

73,688,791✔

1545

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

1,340,952✔

1547

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

72,363,948✔

1551

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

82,270,289✔

1552

  pInfo->currentGroupId = 0;

82,274,726✔

1553

  pInfo->ignoreTag = true;

82,257,529✔

1555

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

82,265,183✔

1556

  QUERY_CHECK_CODE(code, lino, _return);

82,271,880✔

1558

  if (isNewTable) {

82,271,880✔

1559

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

8,552,025✔

1560

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

6,420,026✔

1562

    pInfo->base.dataReader = NULL;

8,536,909✔

1564

    taosRLockLatch(&pTaskInfo->lock);

8,540,434✔

1565

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

8,558,508✔

1566

    taosRUnLockLatch(&pTaskInfo->lock);

8,527,583✔

1567

    QUERY_CHECK_CODE(code, lino, _return);

8,545,721✔

1569

    pInfo->newReader = true;

8,545,721✔

1570

    if (pInfo->lastColArray) {

8,545,748✔

1571

      taosArrayDestroy(pInfo->lastColArray);

4,256,419✔

1572

      pInfo->lastColArray = NULL;

4,255,816✔

1574

    if (pInfo->lastBlockColArray) {

8,537,915✔

1575

      taosArrayDestroy(pInfo->lastBlockColArray);

4,255,230✔

1576

      pInfo->lastBlockColArray = NULL;

4,255,893✔

1578

    pInfo->lastColArray = pColArray;

8,546,093✔

1579

    pInfo->lastBlockColArray = pBlockColArray;

8,549,596✔

1581

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

73,719,855✔

1582

    QUERY_CHECK_CODE(code, lino, _return);

73,726,492✔

1584

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

73,726,492✔

1585

    pInfo->newReader = false;

73,731,278✔

1586

    pInfo->scanTimes = 0;

73,731,278✔

1589

  pOperator->status = OP_OPENED;

82,254,027✔

1591

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

82,268,976✔

1595

  pInfo->currentGroupId = -1;

82,265,858✔

1596

_return:

82,277,667✔

1597

  if (code) {

82,249,724✔

1600

  pAPI->metaReaderFn.clearReader(&superTable);

82,249,724✔

1601

  pAPI->metaReaderFn.clearReader(&orgTable);

82,260,113✔

1602

  return code;

82,260,245✔

1605

static int32_t createVTableScanInfoFromParam(SOperatorInfo* pOperator) {

25,756,889✔

1606

  int32_t                  code = 0;

25,756,889✔

1607

  int32_t                  lino = 0;

25,756,889✔

1608

  STableScanInfo*          pInfo = pOperator->info;

25,756,889✔

1609

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

25,756,889✔

1610

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

25,756,889✔

1611

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

25,756,889✔

1612

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

25,756,889✔

1613

  SMetaReader              orgTable = {0};

25,756,889✔

1614

  SMetaReader              superTable = {0};

25,756,889✔

1615

  SSchemaWrapper*          schema = NULL;

25,756,889✔

1616

  SArray*                  pColArray = NULL;

25,756,889✔

1617

  SArray*                  pBlockColArray = NULL;

25,756,889✔

1618

  int32_t                  num = 0;

25,756,889✔

1619

  STableKeyInfo*           pList = NULL;

25,756,889✔

1621

  SArray*                  pMatchList = NULL;

25,756,889✔

1623

  cleanupQueryTableDataCond(&pInfo->base.cond);

25,756,889✔

1625

  pOrgTbInfo = pParam->pOrgTbInfo;

25,756,889✔

1627

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

25,756,889✔

1629

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

25,756,889✔

1630

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

25,756,889✔

1631

  pAPI->metaReaderFn.readerReleaseLock(&orgTable);

25,756,889✔

1632

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

25,756,889✔

1633

  QUERY_CHECK_CODE(code, lino, _return);

25,756,889✔

1634

  switch (orgTable.me.type) {

25,756,889✔

1635

    case TSDB_CHILD_TABLE:

7,913,863✔

1636

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

7,913,863✔

1637

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

7,913,863✔

1638

      pAPI->metaReaderFn.readerReleaseLock(&superTable);

7,913,863✔

1639

      QUERY_CHECK_CODE(code, lino, _return);

7,913,863✔

1640

      schema = &superTable.me.stbEntry.schemaRow;

7,913,863✔

1641

      break;

7,913,863✔

1642

    case TSDB_NORMAL_TABLE:

17,843,026✔

1643

      schema = &orgTable.me.ntbEntry.schemaRow;

17,843,026✔

1644

      break;

17,843,026✔

1651

  pListInfo->oneTableForEachGroup = true;

25,756,889✔

1652

  taosHashClear(pListInfo->map);

25,756,889✔

1653

  taosArrayClear(pListInfo->pTableList);

25,756,889✔

1655

  uint64_t      pUid = orgTable.me.uid;

25,756,889✔

1656

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

25,756,889✔

1657

  int32_t       tableIdx = 0;

25,756,889✔

1658

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

25,756,889✔

1659

  QUERY_CHECK_CODE(code, lino, _return);

25,756,889✔

1660

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

51,513,778✔

1661

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

25,756,889✔

1663

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

25,756,889✔

1664

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

25,756,889✔

1665

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

25,756,889✔

1666

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

25,756,889✔

1667

  if (pInfo->pBlockColMap) {

25,756,889✔

1668

    taosArrayDestroy(pInfo->pBlockColMap);

23,004,254✔

1669

    pInfo->pBlockColMap = NULL;

23,004,254✔

1671

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

25,756,889✔

1672

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

25,756,889✔

1675

  SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID,

25,756,889✔

1679

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

25,756,889✔

1681

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

98,859,995✔

1682

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

73,103,106✔

1683

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

612,142,815✔

1684

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

612,142,815✔

1685

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

73,103,106✔

1686

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

73,103,106✔

1687

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

73,103,106✔

1688

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

73,103,106✔

1689

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

73,103,106✔

1690

        break;

73,103,106✔

1695

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

25,756,889✔

1696

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

25,756,889✔

1701

      pMatchList = pInfo->base.matchInfo.pList;

25,756,889✔

1706

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

124,616,233✔

1707

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

98,859,995✔

1708

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

98,859,995✔

1709

    if (pItem) {

98,859,995✔

1710

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

98,859,995✔

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

98,859,344✔

1717

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

98,859,344✔

1718

      if (i > 0) {

98,859,344✔

1719

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

146,204,910✔

1724

  taosArraySort(pColArray, compareColIdPair);

25,756,238✔

1725

  taosArraySort(pBlockColArray, compareColIdSlotIdPair);

25,756,238✔

1726

  taosArraySort(pInfo->pBlockColMap, compareColIdSlotIdPair);

25,756,238✔

1728

  taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

25,756,238✔

1729

  taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

25,756,238✔

1731

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

25,756,238✔

1732

  QUERY_CHECK_CODE(code, lino, _return);

25,756,238✔

1734

  if (pInfo->pResBlock) {

25,756,238✔

1735

    blockDataDestroy(pInfo->pResBlock);

18,411,196✔

1736

    pInfo->pResBlock = NULL;

18,411,196✔

1739

  if (isNewScanParam(pParam)) {

25,756,238✔

1740

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

7,352,195✔

1741

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

7,352,195✔

1742

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

7,352,195✔

1748

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

18,404,043✔

1750

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

25,756,238✔

1751

  pInfo->currentGroupId = 0;

25,756,238✔

1752

  pInfo->ignoreTag = true;

25,756,238✔

1754

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

25,756,238✔

1755

  QUERY_CHECK_CODE(code, lino, _return);

25,756,238✔

1757

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

25,756,238✔

1758

  if (reader) {

25,756,238✔

1759

    if (isNewScanParam(pParam)) {

18,408,824✔

1760

      pAPI->tsdReader.tsdReaderClose(*reader);

4,781✔

1761

      pInfo->base.dataReader = NULL;

4,781✔

1763

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

4,781✔

1764

      QUERY_CHECK_CODE(code, lino, _return);

4,781✔

1766

      taosRLockLatch(&pTaskInfo->lock);

4,781✔

1767

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

4,781✔

1768

      taosRUnLockLatch(&pTaskInfo->lock);

4,781✔

1769

      QUERY_CHECK_CODE(code, lino, _return);

4,781✔

1771

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

4,781✔

1772

      QUERY_CHECK_CODE(code, lino, _return);

4,781✔

1773

      pInfo->newReader = true;

4,781✔

1775

      pInfo->base.dataReader = *reader;

18,404,043✔

1776

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

18,404,043✔

1777

      QUERY_CHECK_CODE(code, lino, _return);

18,404,043✔

1779

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

18,404,043✔

1780

      pInfo->newReader = false;

18,404,043✔

1781

      pInfo->scanTimes = 0;

18,404,043✔

1784

    pInfo->base.dataReader = NULL;

7,347,414✔

1786

    taosRLockLatch(&pTaskInfo->lock);

7,347,414✔

1787

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

7,347,414✔

1788

    taosRUnLockLatch(&pTaskInfo->lock);

7,347,414✔

1789

    QUERY_CHECK_CODE(code, lino, _return);

7,347,414✔

1791

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

7,347,414✔

1792

    QUERY_CHECK_CODE(code, lino, _return);

7,347,414✔

1793

    pInfo->newReader = true;

7,347,414✔

1795

  pOperator->status = OP_OPENED;

25,756,238✔

1797

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

25,756,238✔

1801

  pInfo->currentGroupId = -1;

25,756,238✔

1802

_return:

25,756,889✔

1803

  if (code) {

25,756,889✔

1804

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

651✔

1806

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

25,756,889✔

1809

  taosArrayDestroy(pColArray);

25,756,889✔

1810

  taosArrayDestroy(pBlockColArray);

25,756,889✔

1811

  pAPI->metaReaderFn.clearReader(&superTable);

25,756,889✔

1812

  pAPI->metaReaderFn.clearReader(&orgTable);

25,756,889✔

1813

  return code;

25,756,889✔

1816

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

235,575,539✔

1817

  int32_t         code = TSDB_CODE_SUCCESS;

235,575,539✔

1818

  int32_t         lino = 0;

235,575,539✔

1819

  STableScanInfo* pInfo = pOperator->info;

235,575,539✔

1820

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

235,592,930✔

1821

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

235,575,517✔

1822

  int32_t         numOfTables = 0;

235,601,480✔

1824

  QRY_PARAM_CHECK(pResult);

235,566,400✔

1826

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

235,558,503✔

1827

  QUERY_CHECK_CODE(code, lino, _end);

235,544,557✔

1829

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

235,544,557✔

1830

    setOperatorCompleted(pOperator);

219,307,625✔

1831

    if (pOperator->dynamicTask) {

219,312,259✔

1832

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

978,700✔

1833

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

977,507✔

1835

    return code;

219,313,561✔

1839

  pOperator->status = OP_OPENED;

16,210,651✔

1840

  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);

16,252,236✔

1842

  int32_t        num = 0;

16,258,363✔

1843

  STableKeyInfo* pList = NULL;

16,257,159✔

1844

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

16,249,977✔

1845

  QUERY_CHECK_CODE(code, lino, _end);

16,262,356✔

1847

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

16,262,356✔

1848

  QUERY_CHECK_CODE(code, lino, _end);

16,254,710✔

1850

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

16,254,710✔

1851

  QUERY_CHECK_CODE(code, lino, _end);

16,263,667✔

1852

  pInfo->scanTimes = 0;

16,263,667✔

1854

  code = doGroupedTableScan(pOperator, pResult);

16,259,743✔

1855

  QUERY_CHECK_CODE(code, lino, _end);

16,276,940✔

1857

  if (*pResult != NULL) {

16,276,940✔

1858

    if (pOperator->dynamicTask) {

13,072,286✔

1859

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

12,812,093✔

1861

    return code;

13,071,149✔

1864

_end:

3,205,781✔

1865

  if (code != TSDB_CODE_SUCCESS) {

3,205,802✔

1869

  return code;

3,205,272✔

1872

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

804,370,904✔

1873

  int32_t         code = TSDB_CODE_SUCCESS;

804,370,904✔

1874

  int32_t         lino = 0;

804,370,904✔

1875

  STableScanInfo* pInfo = pOperator->info;

804,370,904✔

1876

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

804,389,556✔

1877

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

804,376,708✔

1878

  int32_t         num = 0;

804,354,762✔

1879

  STableKeyInfo*  pList = NULL;

804,344,983✔

1880

  SSDataBlock*    pResult = NULL;

804,353,372✔

1881

  const char*     idStr = GET_TASKID(pTaskInfo);

804,403,538✔

1883

  QRY_PARAM_CHECK(pResBlock);

804,357,889✔

1885

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

804,377,901✔

1886

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

224,532,496✔

1891

    taosRLockLatch(&pTaskInfo->lock);

224,511,748✔

1892

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

224,569,382✔

1893

    taosRUnLockLatch(&pTaskInfo->lock);

224,504,104✔

1894

    QUERY_CHECK_CODE(code, lino, _end);

224,535,380✔

1896

    if (pInfo->filesetDelimited) {

224,533,085✔

1897

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

254,332✔

1900

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

224,534,270✔

1905

  code = doGroupedTableScan(pOperator, &pResult);

804,324,231✔

1906

  QUERY_CHECK_CODE(code, lino, _end);

803,250,060✔

1908

  if (pResult != NULL) {

803,250,060✔

1909

    if (pOperator->dynamicTask) {

583,799,955✔

1910

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

2,086,151✔

1913

    *pResBlock = pResult;

583,793,558✔

1914

    return code;

583,660,715✔

1918

    code = startNextGroupScan(pOperator, &pResult);

219,532,037✔

1919

    QUERY_CHECK_CODE(code, lino, _end);

219,662,030✔

1921

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

219,662,030✔

1922

      *pResBlock = pResult;

219,578,171✔

1923

      return code;

219,567,729✔

1927

_end:

2,295✔

1928

  if (code != TSDB_CODE_SUCCESS) {

2,295✔

1929

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

2,295✔

1930

    pTaskInfo->code = code;

2,295✔

1933

  return code;

2,295✔

1936

static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {

25,756,238✔

1937

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

25,756,238✔

1940

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

979,472✔

1941

  int32_t                  code = TSDB_CODE_SUCCESS;

979,472✔

1942

  int32_t                  lino = 0;

979,472✔

1943

  STableScanInfo*          pInfo = pOperator->info;

979,472✔

1944

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

979,472✔

1945

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

979,472✔

1946

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

979,472✔

1948

  code = createTableListInfoFromParam(pOperator);

979,472✔

1949

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

979,472✔

1950

  pOperator->pOperatorGetParam = NULL;

979,472✔

1951

  QUERY_CHECK_CODE(code, lino, _end);

979,472✔

1953

  if (pOperator->status == OP_EXEC_DONE) {

979,472✔

1954

    pInfo->currentGroupId = -1;

23,763✔

1955

    pOperator->status = OP_OPENED;

23,763✔

1956

    SSDataBlock* result = NULL;

23,763✔

1959

      code = startNextGroupScan(pOperator, &result);

23,763✔

1960

      QUERY_CHECK_CODE(code, lino, _end);

23,763✔

1962

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

23,763✔

1963

        (*ppRes) = result;

23,763✔

1964

        return code;

23,763✔

1968

_end:

955,709✔

1969

  if (code) {

955,709✔

1972

  return code;

955,709✔

1975

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

18,413,531✔

1976

  int32_t      code = TSDB_CODE_SUCCESS;

18,413,531✔

1977

  int32_t      lino = 0;

18,413,531✔

1978

  bool         found = false;

18,413,531✔

1980

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

18,413,531✔

1981

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

18,413,531✔

1983

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

88,575,376✔

1984

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

70,161,845✔

1985

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

70,161,845✔

1986

    found = false;

70,161,845✔

1987

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

197,182,140✔

1988

      SColIdSlotIdPair* pColPair = taosArrayGet(pColMap, j);

178,771,689✔

1989

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

178,771,689✔

1990

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

178,771,689✔

1991

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

51,751,394✔

1992

        found = true;

51,751,394✔

1993

        break;

51,751,394✔

1996

    if (!found) {

70,161,845✔

1997

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

18,410,451✔

1998

        pSrcCol->info.slotId = -1;

18,410,451✔

2007

  return code;

18,413,531✔

2013

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

25,756,889✔

2014

  int32_t                  code = TSDB_CODE_SUCCESS;

25,756,889✔

2015

  int32_t                  lino = 0;

25,756,889✔

2016

  STableScanInfo*          pInfo = pOperator->info;

25,756,889✔

2017

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

25,756,889✔

2018

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

25,756,889✔

2019

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

25,756,889✔

2021

  QRY_PARAM_CHECK(ppRes);

25,756,889✔

2022

  qTrace("%s call", __FUNCTION__);

25,756,889✔

2024

  code = createVTableScanInfoFromParam(pOperator);

25,756,889✔

2025

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

25,756,889✔

2026

  pOperator->pOperatorGetParam = NULL;

25,756,889✔

2028

  QUERY_CHECK_CODE(code, lino, _end);

25,756,889✔

2030

  SSDataBlock* result = NULL;

25,756,238✔

2031

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

25,756,238✔

2032

    (*ppRes) = result;

18,540✔

2033

    return code;

18,540✔

2036

  if (pInfo->newReader) {

25,737,698✔

2037

    code = startNextGroupScan(pOperator, &result);

7,352,195✔

2038

    QUERY_CHECK_CODE(code, lino, _end);

7,352,195✔

2040

    code = doGroupedTableScan(pOperator, &result);

18,385,503✔

2041

    QUERY_CHECK_CODE(code, lino, _end);

18,385,503✔

2044

  if (result) {

25,737,698✔

2045

    SSDataBlock* res = NULL;

18,413,531✔

2046

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

18,413,531✔

2047

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

18,413,531✔

2048

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

18,413,531✔

2050

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

18,413,531✔

2051

    QUERY_CHECK_CODE(code, lino, _end);

18,413,531✔

2057

    (*ppRes) = result;

18,413,531✔

2059

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

7,324,167✔

2060

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

7,324,167✔

2062

    blockDataDestroy(pInfo->pResBlock);

7,324,167✔

2063

    pInfo->pResBlock = NULL;

7,324,167✔

2065

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

7,324,167✔

2066

    if (reader) {

7,324,167✔

2067

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

7,324,167✔

2068

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

7,324,167✔

2069

        pInfo->base.dataReader = NULL;

7,324,167✔

2071

      if (pAPI->tsdReader.tsdReaderClose) {

7,324,167✔

2072

        pAPI->tsdReader.tsdReaderClose(*reader);

7,324,167✔

2076

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

7,324,167✔

2077

    QUERY_CHECK_CODE(code, lino, _end);

7,324,167✔

2079

_end:

25,738,349✔

2080

  if (code) {

25,738,349✔

2081

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

651✔

2083

  return code;

25,738,349✔

2086

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

15,715,059✔

2087

  int32_t                  code = TSDB_CODE_SUCCESS;

15,715,059✔

2088

  int32_t                  lino = 0;

15,715,059✔

2089

  STableScanInfo*          pInfo = pOperator->info;

15,715,059✔

2090

  int32_t                  index = 0;

15,715,059✔

2091

  char*                    tagVal = NULL;

15,715,059✔

2093

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

112,342,447✔

2094

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

96,626,801✔

2095

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

96,626,801✔

2097

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

96,625,587✔

2098

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

96,625,587✔

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

96,627,388✔

2117

  return code;

15,715,059✔

2124

static void clearVstbBatchDynamicTableScanInfo(STableScanInfo* pInfo) {

4,301,499✔

2125

  if (pInfo->pBatchColMap) {

4,301,499✔

2126

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

12,860,007✔

2127

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

8,558,508✔

2128

      taosArrayDestroy(pOrgTbInfo->colMap);

8,558,508✔

2130

    taosArrayDestroy(pInfo->pBatchColMap);

4,301,499✔

2131

    pInfo->pBatchColMap = NULL;

4,301,499✔

2133

  if (pInfo->cachedTagList) {

4,301,499✔

2134

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

11,986,600✔

2135

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

10,180,606✔

2136

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

10,180,606✔

2137

        taosMemoryFreeClear(pTagVal->pData);

4,467,390✔

2140

    taosArrayDestroy(pInfo->cachedTagList);

1,805,994✔

2141

    pInfo->cachedTagList = NULL;

1,805,994✔

2143

  if (pInfo->lastColArray) {

4,301,499✔

2144

    taosArrayDestroy(pInfo->lastColArray);

4,301,499✔

2145

    pInfo->lastColArray = NULL;

4,301,499✔

2147

  if (pInfo->lastBlockColArray) {

4,301,499✔

2148

    taosArrayDestroy(pInfo->lastBlockColArray);

4,301,499✔

2149

    pInfo->lastBlockColArray = NULL;

4,301,499✔

2152

  blockDataDestroy(pInfo->pResBlock);

4,301,499✔

2153

  pInfo->pResBlock = NULL;

4,301,499✔

2154

4,301,499✔

2156

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

77,991,456✔

2157

  int32_t                  code = TSDB_CODE_SUCCESS;

77,991,456✔

2158

  int32_t                  lino = 0;

77,991,456✔

2159

  STableScanInfo*          pInfo = pOperator->info;

77,991,456✔

2160

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

78,015,159✔

2161

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

78,009,593✔

2162

  SSDataBlock*             result = NULL;

78,022,724✔

2164

  QRY_PARAM_CHECK(ppRes);

78,022,248✔

2165

  qTrace("%s call", __FUNCTION__);

78,023,511✔

2167

  while (!result) {

155,954,033✔

2168

    code = createVTableScanInfoFromBatchParam(pOperator);

82,261,750✔

2169

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

82,256,466✔

2170

    pOperator->pOperatorGetParam = NULL;

82,248,238✔

2171

    QUERY_CHECK_CODE(code, lino, _end);

82,253,012✔

2173

    if (pInfo->newReader) {

82,253,012✔

2174

      code = startNextGroupScan(pOperator, &result);

8,536,564✔

2175

      QUERY_CHECK_CODE(code, lino, _end);

8,556,311✔

2177

      code = doGroupedTableScan(pOperator, &result);

73,719,667✔

2178

      QUERY_CHECK_CODE(code, lino, _end);

73,695,308✔

2180

    if (!result) {

82,251,619✔

2181

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

8,556,276✔

2183

      pInfo->currentBatchIdx++;

8,556,276✔

2184

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

8,554,497✔

2186

        break;

4,301,499✔

2191

  if (result) {

77,993,782✔

2192

    SSDataBlock* res = NULL;

73,692,283✔

2193

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

73,692,263✔

2194

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

73,716,870✔

2195

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

73,710,930✔

2197

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

73,723,477✔

2198

    QUERY_CHECK_CODE(code, lino, _end);

73,731,932✔

2200

    if (pInfo->cachedTagList) {

73,731,932✔

2201

      code = setTagValFromTagList(pOperator, res);

15,715,059✔

2202

      QUERY_CHECK_CODE(code, lino, _end);

15,715,059✔

2205

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

73,733,059✔

2206

    pInfo->pResBlock = res;

73,731,315✔

2207

    blockDataDestroy(result);

73,731,855✔

2208

    (*ppRes) = res;

73,723,497✔

2210

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

73,722,900✔

2212

    clearVstbBatchDynamicTableScanInfo(pInfo);

4,301,499✔

2213

    setOperatorCompleted(pOperator);

4,301,499✔

2216

_end:

78,027,981✔

2217

  return code;

78,018,419✔

2220

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

104,741,429✔

2221

  int32_t                  code = TSDB_CODE_SUCCESS;

104,741,429✔

2222

  int32_t                  lino = 0;

104,741,429✔

2223

  STableScanInfo*          pInfo = pOperator->info;

104,741,429✔

2224

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

104,763,871✔

2225

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

104,759,959✔

2227

  QRY_PARAM_CHECK(ppRes);

104,756,045✔

2228

  qTrace("%s call", __FUNCTION__);

104,746,215✔

2230

  pOperator->dynamicTask = true;

104,746,215✔

2232

  switch (type) {

104,738,323✔

2233

    case DYN_TYPE_VSTB_WIN_SCAN:

979,472✔

2235

      code = doStbJoinDynamicTableScanNext(pOperator, ppRes);

979,472✔

2236

      QUERY_CHECK_CODE(code, lino, _end);

979,472✔

2237

      break;

979,472✔

2239

    case DYN_TYPE_VSTB_SINGLE_SCAN: {

25,756,889✔

2240

      code = doVstbSingleDynamicTableScanNext(pOperator, ppRes);

25,756,889✔

2241

      QUERY_CHECK_CODE(code, lino, _end);

25,756,889✔

2242

      break;

25,756,238✔

2244

    case DYN_TYPE_VSTB_BATCH_SCAN:

78,001,962✔

2245

      code = doVstbBatchDynamicTableScanNext(pOperator, ppRes);

78,001,962✔

2246

      QUERY_CHECK_CODE(code, lino, _end);

78,013,046✔

2247

      break;

78,013,046✔

2254

_end:

104,749,407✔

2255

  return code;

104,749,407✔

2258

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

914,341,449✔

2259

  int32_t         code = TSDB_CODE_SUCCESS;

914,341,449✔

2260

  int32_t         lino = 0;

914,341,449✔

2261

  STableScanInfo* pInfo = pOperator->info;

914,341,449✔

2262

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

914,405,417✔

2263

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

914,413,752✔

2264

  QRY_PARAM_CHECK(ppRes);

914,403,019✔

2265

  qTrace("%s call", __FUNCTION__);

914,358,529✔

2267

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

914,360,088✔

2268

  QUERY_CHECK_CODE(code, lino, _end);

914,442,765✔

2270

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

914,442,765✔

2271

  if (pGetParam || pInfo->pBatchColMap) {

914,447,161✔

2272

    ETableScanDynType type = pGetParam ?

104,751,888✔

2273

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

104,751,088✔

2276

    code = doDynamicTableScanNext(pOperator, type, ppRes);

104,751,888✔

2277

    QUERY_CHECK_CODE(code, lino, _end);

104,748,193✔

2279

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

104,747,542✔

2280

      return code;

103,807,992✔

2285

  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {

810,635,720✔

2286

    int32_t       numOfTables = 0;

6,285,624✔

2287

    STableKeyInfo tInfo = {0};

6,285,624✔

2288

    pInfo->countState = TABLE_COUNT_STATE_END;

6,285,624✔

2290

    while (1) {

5,869,074✔

2291

      SSDataBlock* result = NULL;

12,154,375✔

2292

      code = doGroupedTableScan(pOperator, &result);

12,152,768✔

2293

      QUERY_CHECK_CODE(code, lino, _end);

12,155,362✔

2295

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

12,155,362✔

2296

        (*ppRes) = result;

6,235,467✔

2297

        return code;

6,238,679✔

2301

      pInfo->currentTable++;

5,919,570✔

2303

      taosRLockLatch(&pTaskInfo->lock);

5,916,340✔

2304

      numOfTables = 0;

5,920,215✔

2305

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

5,920,215✔

2306

      if (code != TSDB_CODE_SUCCESS) {

5,916,984✔

2311

      if (pInfo->currentTable >= numOfTables) {

5,916,984✔

2312

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

46,622✔

2313

        taosRUnLockLatch(&pTaskInfo->lock);

46,622✔

2314

        (*ppRes) = NULL;

46,622✔

2315

        return code;

46,622✔

2318

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

5,867,448✔

2319

      if (!tmp) {

5,871,008✔

2325

      tInfo = *tmp;

5,871,008✔

2326

      taosRUnLockLatch(&pTaskInfo->lock);

5,868,122✔

2328

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

5,872,622✔

2329

      QUERY_CHECK_CODE(code, lino, _end);

5,872,621✔

2330

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

5,872,621✔

2333

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

5,878,977✔

2334

      QUERY_CHECK_CODE(code, lino, _end);

5,869,715✔

2335

      pInfo->scanTimes = 0;

5,869,715✔

2338

    code = groupSeqTableScan(pOperator, ppRes);

804,339,698✔

2339

    QUERY_CHECK_CODE(code, lino, _end);

803,234,531✔

2342

_end:

803,232,236✔

2343

  if (code != TSDB_CODE_SUCCESS) {

803,230,734✔

2344

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

2,946✔

2345

    pTaskInfo->code = code;

2,946✔

2346

    T_LONG_JMP(pTaskInfo->env, code);

2,946✔

2349

  return code;

803,227,788✔

2352

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

1,939,837✔

2353

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

1,939,837✔

2354

  if (!pRecorder) {

1,939,837✔

2357

  STableScanInfo* pTableScanInfo = pOptr->info;

1,939,837✔

2358

  *pRecorder = pTableScanInfo->base.readRecorder;

1,939,837✔

2359

  *pOptrExplain = pRecorder;

1,939,837✔

2360

  *len = sizeof(SFileBlockLoadRecorder);

1,939,837✔

2361

  return 0;

1,939,837✔

2364

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

262,464,599✔

2365

  cleanupQueryTableDataCond(&pBase->cond);

262,464,599✔

2366

  cleanupQueryTableDataCond(&pBase->orgCond);

262,455,634✔

2368

  if (pAPI->tsdReaderClose) {

262,470,539✔

2369

    pAPI->tsdReaderClose(pBase->dataReader);

262,474,647✔

2371

  pBase->dataReader = NULL;

262,408,790✔

2373

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

262,426,827✔

2374

    taosArrayDestroy(pBase->matchInfo.pList);

262,445,412✔

2377

  tableListDestroy(pBase->pTableListInfo);

262,393,800✔

2378

  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);

262,457,751✔

2379

  cleanupExprSupp(&pBase->pseudoSup);

262,434,471✔

2380

262,462,966✔

2382

static void cleanReaderForVTable(STableScanInfo* pInfo){

5,378,087✔

2383

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

5,378,087✔

2384

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

5,379,203✔

2385

    while (pIter != NULL) {

5,402,467✔

2386

      void **reader = pIter;

23,247✔

2387

      if (*reader) {

23,247✔

2388

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

23,247✔

2389

          pInfo->base.dataReader = NULL;

20,875✔

2391

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

23,247✔

2393

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

23,247✔

2396

5,378,711✔

2398

static void destroyTableScanOperatorInfo(void* param) {

235,064,720✔

2399

  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;

235,064,720✔

2400

  blockDataDestroy(pTableScanInfo->pResBlock);

235,064,720✔

2401

  blockDataDestroy(pTableScanInfo->pOrgBlock);

235,096,456✔

2402

  taosHashCleanup(pTableScanInfo->pIgnoreTables);

235,097,433✔

2403

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

235,080,840✔

2404

    cleanReaderForVTable(pTableScanInfo);

5,376,973✔

2405

    taosHashCleanup(pTableScanInfo->readerCache);

5,379,220✔

2407

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

235,081,340✔

2408

  taosArrayDestroy(pTableScanInfo->pBlockColMap);

235,093,686✔

2409

  taosMemoryFreeClear(param);

235,095,136✔

2410

235,083,754✔

2412

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

235,031,699✔

2413

  if (pBlock && dataRequireFlag == FUNC_DATA_REQUIRED_NOT_LOAD) {

235,031,699✔

2414

    int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

7,933,954✔

2415

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

20,090,191✔

2416

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

12,165,874✔

2417

      if (pCol) {

12,161,312✔

2418

        pCol->info.noData = true;

12,161,312✔

2422

235,022,062✔

2424

static int32_t resetTableScanOperatorState(SOperatorInfo* pOper) {

311,782✔

2425

  int32_t         code = TSDB_CODE_SUCCESS;

311,782✔

2426

  STableScanInfo*   pInfo = pOper->info;

311,782✔

2427

  pOper->status = OP_NOT_OPENED;

311,782✔

2429

  pInfo->scanTimes = 0;

311,782✔

2430

  pInfo->currentGroupId = -1;

311,782✔

2431

  pInfo->tableEndIndex = -1;

311,782✔

2432

  pInfo->tableStartIndex = 0;

311,782✔

2433

  pInfo->currentTable = 0;

311,782✔

2434

  pInfo->scanMode = 0;

311,782✔

2435

  pInfo->countState = 0;

311,782✔

2436

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

311,782✔

2438

  if (!pInfo->virtualStableScan) {

311,782✔

2444

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

156,118✔

2445

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

156,118✔

2447

    pInfo->base.dataReader = NULL;

156,118✔

2449

  tableListDestroy(pInfo->base.pTableListInfo);

311,388✔

2450

  pInfo->base.pTableListInfo = tableListCreate();

311,782✔

2451

  if (!pInfo->base.pTableListInfo) {

311,782✔

2455

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

311,782✔

2457

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

311,782✔

2458

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

311,782✔

2459

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

309,241✔

2461

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

309,241✔

2462

    if (code) {

155,070✔

2468

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

311,782✔

2469

  cleanupQueryTableDataCond(&pInfo->base.cond);

311,782✔

2470

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

311,782✔

2471

  if (code) {

311,782✔

2475

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

311,782✔

2476

    cleanupQueryTableDataCond(&pInfo->base.orgCond);

155,664✔

2477

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

155,664✔

2478

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

155,664✔

2481

  pOper->resultInfo.totalRows = 0;

311,782✔

2482

  blockDataEmpty(pInfo->pResBlock);

311,782✔

2483

  blockDataEmpty(pInfo->pOrgBlock);

311,782✔

2484

  taosHashClear(pInfo->pIgnoreTables);

311,782✔

2485

  taosArrayDestroy(pInfo->pBlockColMap);

311,782✔

2486

  pInfo->pBlockColMap = NULL;

311,782✔

2487

  return code;

311,782✔

2490

int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

235,098,166✔

2493

  QRY_PARAM_CHECK(pOptrInfo);

235,098,166✔

2495

  int32_t         code = TSDB_CODE_SUCCESS;

235,105,254✔

2496

  int32_t         lino = 0;

235,105,254✔

2497

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

235,105,254✔

2498

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

235,075,909✔

2499

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

235,078,543✔

2504

  pOperator->pPhyNode = pTableScanNode;

235,078,914✔

2505

  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;

235,084,444✔

2506

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

235,085,588✔

2508

  int32_t numOfCols = 0;

235,085,204✔

2510

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

235,090,789✔

2511

  QUERY_CHECK_CODE(code, lino, _error);

235,054,988✔

2513

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

235,054,988✔

2514

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

235,027,053✔

2515

  QUERY_CHECK_CODE(code, lino, _error);

235,033,183✔

2517

  if (pScanNode->pScanPseudoCols != NULL) {

235,033,183✔

2518

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

53,999,698✔

2519

    pSup->pExprInfo = NULL;

53,991,276✔

2520

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

54,002,128✔

2521

    QUERY_CHECK_CODE(code, lino, _error);

53,982,127✔

2523

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

53,982,127✔

2525

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

53,973,922✔

2528

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

235,056,232✔

2529

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

234,980,670✔

2531

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

235,038,065✔

2532

  pInfo->base.readHandle = *readHandle;

234,924,013✔

2533

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

234,936,617✔

2535

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

235,037,244✔

2536

  pInfo->sample.seed = taosGetTimestampSec();

234,970,636✔

2538

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

235,021,344✔

2539

  initResultSizeInfo(&pOperator->resultInfo, 4096);

235,015,546✔

2540

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

235,024,737✔

2541

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

235,052,794✔

2542

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

235,042,299✔

2544

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

234,930,832✔

2545

  QUERY_CHECK_CODE(code, lino, _error);

235,051,905✔

2547

  pInfo->virtualStableScan = pScanNode->virtualStableScan;

235,051,905✔

2548

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

235,029,523✔

2549

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

5,375,364✔

2550

    pInfo->pResBlock = NULL;

5,377,546✔

2551

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

5,374,199✔

2552

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

5,378,070✔

2555

  if (pInfo->virtualStableScan) {

235,038,410✔

2556

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

5,377,540✔

2557

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

5,380,400✔

2560

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

234,997,728✔

2561

                            pTaskInfo->pStreamRuntimeInfo);

235,041,970✔

2562

  QUERY_CHECK_CODE(code, lino, _error);

234,952,235✔

2564

  pInfo->currentGroupId = -1;

234,928,641✔

2566

  pInfo->tableEndIndex = -1;

234,940,549✔

2567

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

234,898,660✔

2568

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

235,040,732✔

2570

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

234,929,038✔

2572

  pOperator->exprSupp.numOfExprs = numOfCols;

234,950,796✔

2574

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

234,899,543✔

2575

  pInfo->ignoreTag = false;

234,999,672✔

2577

  pInfo->lastBatchIdx = -1;

235,026,600✔

2578

  pInfo->currentBatchIdx = 0;

234,964,252✔

2579

  pInfo->base.pTableListInfo = pTableListInfo;

234,904,710✔

2580

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

234,901,198✔

2590

  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

234,901,198✔

2592

  pOperator->fpSet = createOperatorFpSet(prepareTableScan, doTableScanNext,

234,982,333✔

2598

  setOperatorResetStateFn(pOperator, resetTableScanOperatorState);

234,904,888✔

2600

  pOperator->cost.openCost = 0;

234,952,804✔

2601

  *pOptrInfo = pOperator;

234,987,494✔

2602

  return TSDB_CODE_SUCCESS;

234,868,724✔

2604

_error:

23,594✔

2605

  if (pInfo != NULL) {

23,594✔

2606

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

23,594✔

2607

    destroyTableScanOperatorInfo(pInfo);

23,594✔

2610

  if (pOperator != NULL) {

23,594✔

2611

    pOperator->info = NULL;

23,594✔

2612

    destroyOperator(pOperator);

23,594✔

2614

  pTaskInfo->code = code;

23,594✔

2615

  return code;

23,594✔

2652

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

14,688✔

2653

  int32_t code = TSDB_CODE_SUCCESS;

14,688✔

2654

  int32_t lino = 0;

14,688✔

2655

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

14,688✔

2656

    return code;

1,724✔

2658

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

12,964✔

2659

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

12,964✔

2660

  bool hasUnqualified = false;

12,964✔

2662

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

12,964✔

2663

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

12,964✔

2665

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

12,964✔

2667

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

12,964✔

2670

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

12,964✔

2671

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

12,964✔

2672

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

46,753,860✔

2673

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

46,740,896✔

2674

    void*    data = colDataGetData(pColPk, i);

46,740,896✔

2675

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

46,743,424✔

2676

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

2,528✔

2679

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

2,528✔

2680

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

2,528✔

2681

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

2,528✔

2682

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

2,528✔

2683

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

2,528✔

2684

      taosMemoryFree(tmq);

2,528✔

2686

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

46,738,368✔

2689

    if (!p[i]) {

46,740,896✔

2690

      hasUnqualified = true;

16,614,008✔

2694

  if (hasUnqualified) {

12,964✔

2695

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

4,372✔

2696

    QUERY_CHECK_CODE(code, lino, _end);

4,372✔

2699

_end:

12,964✔

2700

  taosMemoryFree(p);

12,964✔

2701

  if (code != TSDB_CODE_SUCCESS) {

12,964✔

2704

  return code;

12,964✔

2718

static int32_t processBlock(STmqQueryScanInfo* pInfo) {

25,636,369✔

2719

  int32_t         code = TSDB_CODE_SUCCESS;

25,636,369✔

2720

  int32_t         lino = 0;

25,636,369✔

2721

  SOperatorInfo*  pOperator = pInfo->pTmqScanOp;

25,636,369✔

2723

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

25,636,369✔

2724

  QUERY_CHECK_CODE(code, lino, _end);

25,636,369✔

2726

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

25,636,369✔

2727

  QUERY_CHECK_CODE(code, lino, _end);

25,635,659✔

2729

_end:

25,635,659✔

2730

  if (code != TSDB_CODE_SUCCESS) {

25,635,659✔

2733

  return code;

25,636,369✔

2736

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

7,003,821✔

2737

  int32_t code = TSDB_CODE_SUCCESS;

7,003,821✔

2738

  int32_t lino = 0;

7,003,821✔

2739

  SValue  val = {0};

7,003,821✔

2740

  if (hasPrimaryKey) {

7,004,769✔

2741

    code = doBlockDataPrimaryKeyFilter(pBlock, offset);

14,688✔

2742

    if (code != TSDB_CODE_SUCCESS) {

14,688✔

2746

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

14,688✔

2748

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

14,688✔

2749

      return code;

4,056✔

2751

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

10,632✔

2752

    val.type = pColPk->info.type;

10,632✔

2753

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

10,632✔

2754

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

668✔

2757

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

668✔

2758

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

668✔

2759

      val.nData = varDataLen(tmp);

668✔

2760

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

668✔

2762

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

9,964✔

2765

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

7,000,713✔

2767

_end:

7,000,066✔

2768

  if (code != TSDB_CODE_SUCCESS) {

7,000,390✔

2771

  return code;

6,999,119✔

2774

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

25,825,640✔

2775

  int32_t        code = TSDB_CODE_SUCCESS;

25,825,640✔

2776

  int32_t        lino = 0;

25,825,640✔

2777

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

25,825,640✔

2778

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

25,827,004✔

2780

  STmqQueryScanInfo* pInfo = pOperator->info;

25,827,084✔

2781

  const char*      id = GET_TASKID(pTaskInfo);

25,827,394✔

2783

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

25,826,705✔

2785

  if (isTaskKilled(pTaskInfo)) {

25,830,157✔

2790

  blockDataCleanup(pInfo->pRes);

25,828,736✔

2791

  int64_t st = taosGetTimestampMs();

25,828,736✔

2793

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

25,828,736✔

2794

    while (1) {

5,959,365✔

2795

      SSDataBlock* pResult = NULL;

6,285,335✔

2796

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

6,284,980✔

2797

      QUERY_CHECK_CODE(code, lino, _end);

6,284,654✔

2799

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

6,284,654✔

2800

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

6,238,678✔

2801

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

6,239,002✔

2802

        QUERY_CHECK_CODE(code, lino, _end);

6,237,063✔

2803

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

6,237,063✔

2804

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

6,238,033✔

2805

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

6,234,946✔

2806

          QUERY_CHECK_CODE(code, lino, _end);

6,234,625✔

2812

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

6,238,681✔

2813

        (*ppRes) = pInfo->pRes;

273,830✔

2814

        return code;

273,830✔

2816

      int64_t elapsed = taosGetTimestampMs() - st;

5,964,205✔

2817

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

5,964,205✔

2818

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

5,163✔

2821

        (*ppRes) = pInfo->pRes;

5,163✔

2822

        return code;

5,163✔

2826

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

46,622✔

2827

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

46,622✔

2829

    pTSInfo->base.dataReader = NULL;

46,622✔

2830

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

46,622✔

2831

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

46,622✔

2832

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

46,622✔

2833

    if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) {

46,622✔

2834

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

11,540✔

2835

        (*ppRes) = NULL;

4,814✔

2837

        (*ppRes) = pInfo->pRes;

6,726✔

2839

      return 0;

11,540✔

2841

    QUERY_CHECK_CODE(code, lino, _end);

35,082✔

2843

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

35,082✔

2846

  while (1) {

98,166✔

2847

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

76,907,413✔

2851

                                                              pTaskInfo->tmqInfo.sourceExcluded,

25,636,064✔

2854

                                                              pTaskInfo->tmqInfo.enableReplay);

25,636,369✔

2855

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

25,636,014✔

2858

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

25,636,014✔

2861

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

25,636,369✔

2864

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

25,636,369✔

2866

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

25,636,369✔

2868

    int32_t ret = processBlock(pInfo);

25,636,369✔

2869

    if (ret != 0) {

25,636,369✔

2873

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

25,636,369✔

2876

    if (code != 0) {

25,636,369✔

2877

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

25,027,036✔

2878

        (*ppRes) = NULL;

11,860,810✔

2880

        (*ppRes) = pInfo->pRes;

13,166,226✔

2882

      return 0;

25,027,036✔

2885

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

609,333✔

2886

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

103,314✔

2887

      (*ppRes) = pInfo->pRes;

511,167✔

2888

      return 0;

511,167✔

2902

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

336,260✔

2903

  int32_t code = TSDB_CODE_SUCCESS;

336,260✔

2904

  int32_t lino = 0;

336,260✔

2905

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

336,260✔

2906

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

336,260✔

2909

  int32_t size = 0;

336,260✔

2910

  code = tableListGetSize(pTableListInfo, &size);

336,260✔

2911

  QUERY_CHECK_CODE(code, lino, _end);

336,260✔

2912

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

9,918,745✔

2913

    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);

9,582,167✔

2914

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

9,573,931✔

2915

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

9,573,931✔

2916

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

9,582,485✔

2919

  (*ppArrayRes) = tableIdList;

336,578✔

2921

_end:

336,260✔

2922

  if (code != TSDB_CODE_SUCCESS) {

336,260✔

2925

  return code;

336,260✔

2928

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

1,585,683✔

2929

  int32_t        code = TSDB_CODE_SUCCESS;

1,585,683✔

2930

  int32_t        lino = 0;

1,585,683✔

2931

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,585,683✔

2932

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,585,683✔

2934

  STmqRawScanInfo* pInfo = pOperator->info;

1,585,683✔

2935

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

1,585,683✔

2936

  pTaskInfo->tmqInfo.btMetaRsp.batchMetaLen = NULL;

1,585,683✔

2938

  qDebug("tmqsnap doRawScan called");

1,585,683✔

2939

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

1,585,683✔

2940

    bool hasNext = false;

1,579,175✔

2941

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

1,579,175✔

2942

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

1,577,922✔

2943

      if (code != TSDB_CODE_SUCCESS) {

1,576,986✔

2949

    if (pInfo->dataReader && hasNext) {

1,578,239✔

2950

      if (isTaskKilled(pTaskInfo)) {

766,736✔

2955

      SSDataBlock* pBlock = NULL;

766,736✔

2956

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

766,424✔

2957

      QUERY_CHECK_CODE(code, lino, _end);

766,112✔

2959

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

766,112✔

2960

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

766,112✔

2961

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

766,736✔

2962

        QUERY_CHECK_CODE(code, lino, _end);

766,736✔

2963

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

766,736✔

2964

        (*ppRes) = pBlock;

766,736✔

2965

        return code;

766,736✔

2969

    SMetaTableInfo mtInfo = {0};

812,127✔

2970

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

812,127✔

2971

    QUERY_CHECK_CODE(code, lino, _end);

812,127✔

2972

    if (code != 0) {

812,127✔

2976

    STqOffsetVal offset = {0};

812,127✔

2977

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

811,807✔

2978

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

12,836✔

2979

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

12,836✔

2981

      SValue val = {0};

799,291✔

2982

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

799,291✔

2983

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

799,603✔

2986

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

812,439✔

2987

    QUERY_CHECK_CODE(code, lino, _end);

812,439✔

2988

    (*ppRes) = NULL;

812,439✔

2989

    return code;

812,439✔

2990

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

6,508✔

2991

    SSnapContext* sContext = pInfo->sContext;

6,508✔

2992

    int64_t st = taosGetTimestampMs();

6,508✔

2993

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

50,474✔

2994

      void*   data = NULL;

49,770✔

2995

      int32_t dataLen = 0;

49,770✔

2996

      int16_t type = 0;

49,770✔

2997

      int64_t uid = 0;

49,770✔

2998

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

49,770✔

3004

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

49,770✔

3005

        STqOffsetVal offset = {0};

5,804✔

3006

        SValue       val = {0};

5,804✔

3008

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

5,804✔

3009

        QUERY_CHECK_CODE(code, lino, _end);

5,804✔

3010

        break;

5,804✔

3012

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

43,966✔

3013

        SMqMetaRsp tmpMetaRsp = {0};

43,966✔

3014

        tmpMetaRsp.resMsgType = type;

43,966✔

3015

        tmpMetaRsp.metaRspLen = dataLen;

43,966✔

3016

        tmpMetaRsp.metaRsp = data;

43,966✔

3017

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

43,966✔

3018

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

5,238✔

3019

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

5,238✔

3021

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

5,238✔

3022

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

5,238✔

3024

        int32_t  tempRes = TSDB_CODE_SUCCESS;

43,966✔

3025

        uint32_t len = 0;

43,966✔

3026

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

43,966✔

3027

        if (TSDB_CODE_SUCCESS != tempRes) {

43,966✔

3033

        int32_t tLen = sizeof(SMqRspHead) + len;

43,966✔

3034

        void*   tBuf = taosMemoryCalloc(1, tLen);

43,966✔

3035

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

43,966✔

3037

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

43,966✔

3038

        SEncoder encoder = {0};

43,966✔

3039

        tEncoderInit(&encoder, metaBuff, len);

43,966✔

3040

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

43,966✔

3041

        if (tempLen < 0) {

43,966✔

3048

        taosMemoryFreeClear(data);

43,966✔

3049

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

43,966✔

3050

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

43,966✔

3052

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

43,966✔

3053

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

43,966✔

3055

        int64_t elapsed = taosGetTimestampMs() - st;

43,966✔

3056

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

43,966✔

3062

    (*ppRes) = NULL;

6,508✔

3063

    return code;

6,508✔

3077

static void destroyTmqRawScanOperatorInfo(void* param) {

128,869✔

3078

  STmqRawScanInfo* pRawScan = (STmqRawScanInfo*)param;

128,869✔

3079

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

128,869✔

3080

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

128,869✔

3081

  tableListDestroy(pRawScan->pTableListInfo);

128,869✔

3082

  taosMemoryFree(pRawScan);

128,869✔

3083

128,531✔

3088

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

128,869✔

3094

  QRY_PARAM_CHECK(pOptrInfo);

128,869✔

3095

  int32_t code = TSDB_CODE_SUCCESS;

128,869✔

3096

  int32_t lino = 0;

128,869✔

3098

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

128,869✔

3099

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

128,525✔

3100

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

128,525✔

3106

  pInfo->pTableListInfo = tableListCreate();

128,525✔

3107

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

128,869✔

3108

  pInfo->vnode = pHandle->vnode;

128,869✔

3109

  pInfo->pAPI = &pTaskInfo->storageAPI;

128,869✔

3111

  pInfo->sContext = pHandle->sContext;

128,869✔

3112

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

128,869✔

3113

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

128,869✔

3116

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

128,869✔

3118

  *pOptrInfo = pOperator;

128,869✔

3119

  return code;

128,869✔

3131

void destroyTmqScanOperatorInfo(void* param) {

336,260✔

3132

  if (param == NULL) {

336,260✔

3136

  STmqQueryScanInfo* pTmqScan = (STmqQueryScanInfo*)param;

336,260✔

3137

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

336,260✔

3138

    destroyOperator(pTmqScan->pTableScanOp);

335,929✔

3141

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

336,591✔

3142

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

336,260✔

3145

  if (pTmqScan->pTableListInfo) {

336,260✔

3149

  if (pTmqScan->pPseudoExpr) {

336,260✔

3150

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

192,332✔

3151

    taosMemoryFree(pTmqScan->pPseudoExpr);

192,332✔

3154

  taosHashCleanup(pTmqScan->pCol2SlotId);

335,976✔

3155

  pTmqScan->pCol2SlotId = NULL;

336,260✔

3157

  cleanupExprSupp(&pTmqScan->tagCalSup);

336,260✔

3159

  blockDataDestroy(pTmqScan->pRes);

336,260✔

3161

  taosMemoryFree(pTmqScan);

336,260✔

3181

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

336,260✔

3182

  size_t  numOfCols = LIST_LENGTH(pNodeList);

336,260✔

3183

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

2,096,493✔

3184

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

1,759,877✔

3185

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

1,760,596✔

3186

      *pTsSlotId = pNode->slotId;

335,904✔

3189

336,616✔

3191

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

336,260✔

3192

  size_t  numOfCols = LIST_LENGTH(pNodeList);

336,260✔

3193

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

2,096,856✔

3194

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

1,760,596✔

3195

    if (pNode == NULL) {

1,760,596✔

3198

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

1,760,596✔

3199

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

1,760,596✔

3200

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

1,760,596✔

3201

      if (code != TSDB_CODE_SUCCESS) {

1,760,596✔

3206

  return TSDB_CODE_SUCCESS;

336,260✔

3209

int32_t createTmqScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,

335,904✔

3212

  QRY_PARAM_CHECK(pOptrInfo);

335,904✔

3214

  int32_t          code = TSDB_CODE_SUCCESS;

336,260✔

3215

  int32_t          lino = 0;

336,260✔

3216

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

336,260✔

3217

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

336,260✔

3218

  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;

336,260✔

3219

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

336,260✔

3221

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

336,260✔

3226

  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;

336,260✔

3227

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;

335,904✔

3229

  pInfo->pTagCond = pTagCond;

335,904✔

3230

  pInfo->pGroupTags = pTableScanNode->pGroupTags;

336,260✔

3232

  pInfo->primaryTsIndex = -1;

336,260✔

3233

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

336,260✔

3234

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

336,260✔

3235

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

336,260✔

3236

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

335,897✔

3237

  QUERY_CHECK_CODE(code, lino, _error);

336,260✔

3239

  SDataBlockDescNode* pNode = (SDataBlockDescNode*)pDescNode;

336,260✔

3240

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

336,260✔

3246

  if (pTableScanNode->pTags != NULL) {

336,260✔

3258

  SOperatorInfo* pTableScanOp = NULL;

336,260✔

3259

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

336,260✔

3260

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

335,113✔

3265

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

335,632✔

3266

  if (pHandle->version > 0) {

335,977✔

3267

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

334,973✔

3270

  STableKeyInfo* pList = NULL;

335,293✔

3271

  int32_t        num = 0;

335,173✔

3272

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

334,810✔

3273

  QUERY_CHECK_CODE(code, lino, _error);

334,966✔

3275

  if (pHandle->initTableReader) {

334,966✔

3276

    pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;

335,897✔

3277

    pTSInfo->base.dataReader = NULL;

335,654✔

3280

  if (pHandle->initTqReader) {

333,407✔

3281

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

335,614✔

3282

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

336,260✔

3288

  pInfo->pTableScanOp = pTableScanOp;

336,260✔

3290

  pInfo->readHandle = *pHandle;

336,260✔

3291

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

336,260✔

3293

  SArray* tableIdList = NULL;

336,260✔

3294

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

336,260✔

3295

  QUERY_CHECK_CODE(code, lino, _error);

336,260✔

3296

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

336,260✔

3297

  QUERY_CHECK_CODE(code, lino, _error);

336,260✔

3298

  taosArrayDestroy(tableIdList);

336,260✔

3299

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

336,260✔

3302

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

336,260✔

3303

  QUERY_CHECK_CODE(code, lino, _error);

336,260✔

3305

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

336,260✔

3306

  QUERY_CHECK_CODE(code, lino, _error);

336,260✔

3308

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

336,260✔

3309

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

336,260✔

3311

  pInfo->pTmqScanOp = pOperator;

336,260✔

3312

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

336,260✔

3314

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

336,260✔

3316

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

336,260✔

3318

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

336,260✔

3321

  *pOptrInfo = pOperator;

336,260✔

3322

  return code;

336,260✔

3341

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

8,928,304✔

3342

  int32_t        code = TSDB_CODE_SUCCESS;

8,928,304✔

3343

  int32_t        lino = 0;

8,928,304✔

3344

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

8,928,304✔

3345

  STagScanInfo*  pInfo = pOperator->info;

8,929,364✔

3346

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

8,928,834✔

3347

  int32_t        count = pRes->info.rows;

8,928,834✔

3349

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

8,929,364✔

3350

  if (!item) {

8,927,774✔

3357

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

8,927,774✔

3358

  tDecoderClear(&(*mr).coder);

8,929,364✔

3359

  if (code != TSDB_CODE_SUCCESS) {

8,928,834✔

3365

  char str[512];

8,928,730✔

3366

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

39,517,811✔

3367

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

30,588,370✔

3370

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

30,591,318✔

3371

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

3,440,512✔

3372

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

3,440,512✔

3373

      QUERY_CHECK_CODE(code, lino, _end);

3,440,512✔

3374

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

27,148,609✔

3375

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

5,071,210✔

3376

      QUERY_CHECK_CODE(code, lino, _end);

5,071,210✔

3377

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

22,078,459✔

3378

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

15,862✔

3379

      QUERY_CHECK_CODE(code, lino, _end);

15,862✔

3381

      STagVal val = {0};

22,063,657✔

3382

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

22,064,187✔

3383

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

22,064,187✔

3385

      char* data = NULL;

22,064,264✔

3386

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

22,064,264✔

3387

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

22,061,794✔

3389

        data = (char*)p;

1,410✔

3392

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

22,063,734✔

3393

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

22,063,734✔

3394

      QUERY_CHECK_CODE(code, lino, _end);

22,063,127✔

3396

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

22,063,127✔

3398

        taosMemoryFree(data);

7,696,580✔

3403

_end:

8,931,098✔

3404

  if (code != TSDB_CODE_SUCCESS) {

8,929,364✔

3408

    pRes->info.rows++;

8,929,364✔

3411

  return code;

8,928,834✔

3414

static void tagScanFreeUidTag(void* p) {

7,405,850✔

3415

  STUidTagInfo* pInfo = p;

7,405,850✔

3416

  if (pInfo->pTagVal != NULL) {

7,405,850✔

3417

    taosMemoryFree(pInfo->pTagVal);

7,404,772✔

3419

7,405,526✔

3421

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

1,699,448✔

3422

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

1,699,448✔

3423

  if (pColumnData == NULL) {

1,699,441✔

3427

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

1,699,441✔

3428

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

1,699,812✔

3429

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

1,698,792✔

3430

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

1,699,303✔

3432

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

1,696,813✔

3433

  if (code != TSDB_CODE_SUCCESS) {

1,699,448✔

3439

  pParam->columnData = pColumnData;

1,699,448✔

3440

  pParam->colAlloced = true;

1,699,303✔

3441

  return TSDB_CODE_SUCCESS;

1,699,303✔

3444

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

10,536,296✔

3445

  int32_t                code = TSDB_CODE_SUCCESS;

10,536,296✔

3446

  int32_t                lino = 0;

10,536,296✔

3447

  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;

10,536,296✔

3448

  SColumnNode*           pSColumnNode = NULL;

10,536,296✔

3449

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

10,537,312✔

3450

    pSColumnNode = *(SColumnNode**)pNode;

2,964,204✔

3451

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

7,573,325✔

3452

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

524✔

3453

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

524✔

3466

      return DEAL_RES_CONTINUE;

524✔

3469

    return DEAL_RES_CONTINUE;

7,571,932✔

3472

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

2,965,441✔

3473

  if (!data) {

2,963,690✔

3474

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

1,701,165✔

3475

    if (code == TSDB_CODE_DUP_KEY) {

1,703,425✔

3478

    QUERY_CHECK_CODE(code, lino, _end);

1,703,425✔

3479

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

1,703,425✔

3480

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

1,702,045✔

3481

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

1,702,547✔

3482

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

1,702,176✔

3483

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

1,701,674✔

3484

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

1,702,407✔

3486

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

1,262,525✔

3487

    pSColumnNode->slotId = col->slotId;

1,263,032✔

3490

_end:

2,963,324✔

3491

  if (code != TSDB_CODE_SUCCESS) {

2,964,706✔

3492

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

1,016✔

3495

  return DEAL_RES_CONTINUE;

2,963,690✔

3498

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

1,699,812✔

3500

  int32_t code = TSDB_CODE_SUCCESS;

1,699,812✔

3501

  int32_t lino = 0;

1,699,812✔

3502

  int32_t numOfTables = taosArrayGetSize(aUidTags);

1,699,812✔

3503

  SArray* pBlockList = NULL;

1,699,812✔

3505

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

1,699,812✔

3506

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

1,698,428✔

3508

  pBlockList = taosArrayInit(1, POINTER_BYTES);

1,698,428✔

3509

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

1,699,448✔

3511

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

1,699,448✔

3512

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

1,699,448✔

3514

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

1,699,448✔

3516

  SScalarParam output = {0};

1,699,448✔

3517

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

1,699,448✔

3518

  QUERY_CHECK_CODE(code, lino, _end);

1,698,421✔

3520

  gTaskScalarExtra.pStreamInfo = (void*)pStreamRuntimeInfo;

1,698,421✔

3521

  gTaskScalarExtra.pStreamRange = NULL;

1,698,421✔

3522

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

1,698,939✔

3523

  QUERY_CHECK_CODE(code, lino, _end);

1,696,313✔

3525

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

1,696,313✔

3526

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

4,245,511✔

3527

    if (result[i]) {

2,547,739✔

3528

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

2,408,143✔

3529

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

2,408,143✔

3533

  colDataDestroy(output.columnData);

1,699,084✔

3534

  taosMemoryFreeClear(output.columnData);

1,696,315✔

3536

_end:

1,690,755✔

3537

  blockDataDestroy(pResBlock);

1,695,137✔

3538

  taosArrayDestroy(pBlockList);

1,693,196✔

3539

  if (code != TSDB_CODE_SUCCESS) {

1,688,974✔

3542

  return code;

1,688,974✔

3545

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

30,310,703✔

3548

  int32_t code = TSDB_CODE_SUCCESS;

30,310,703✔

3549

  int32_t lino = 0;

30,310,703✔

3550

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

30,310,703✔

3551

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

14,417,019✔

3557

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

14,415,775✔

3558

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

7,202,228✔

3559

      QUERY_CHECK_CODE(code, lino, _end);

7,206,903✔

3560

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

7,208,497✔

3561

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

7,205,384✔

3562

      QUERY_CHECK_CODE(code, lino, _end);

7,207,848✔

3565

    STagVal tagVal = {0};

15,885,884✔

3566

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

15,892,128✔

3567

    if (pUidTagInfo->pTagVal == NULL) {

15,891,470✔

3570

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

15,884,752✔

3572

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

15,892,254✔

3573

        colDataSetNULL(pColInfo, rowIndex);

7,201✔

3574

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

15,885,417✔

3577

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

19,942,858✔

3578

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

4,068,550✔

3581

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

4,073,985✔

3582

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

4,073,260✔

3584

        varDataSetLen(tmp, tagVal.nData);

4,073,260✔

3585

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

4,072,389✔

3586

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

4,069,997✔

3587

        taosMemoryFree(tmp);

4,071,229✔

3588

        QUERY_CHECK_CODE(code, lino, _end);

4,068,019✔

3590

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

11,802,414✔

3591

        QUERY_CHECK_CODE(code, lino, _end);

11,808,893✔

3596

_end:

30,300,749✔

3597

  if (code != TSDB_CODE_SUCCESS) {

30,300,749✔

3600

  return code;

30,296,125✔

3603

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

4,738,688✔

3605

  int32_t       code = TSDB_CODE_SUCCESS;

4,738,688✔

3606

  int32_t       lino = 0;

4,738,688✔

3607

  STagScanInfo* pInfo = pOperator->info;

4,738,688✔

3608

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

4,738,543✔

3609

  if (!ignoreFilterIdx) {

4,735,075✔

3610

    size_t szTables = taosArrayGetSize(aFilterIdxs);

1,695,854✔

3611

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

4,099,833✔

3612

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

2,404,936✔

3613

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);

2,405,155✔

3614

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

2,401,048✔

3615

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

13,723,702✔

3616

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

11,316,171✔

3617

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

11,322,728✔

3618

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

11,322,728✔

3619

        QUERY_CHECK_CODE(code, lino, _end);

11,322,654✔

3623

    size_t szTables = taosArrayGetSize(aUidTags);

3,039,221✔

3624

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

7,903,097✔

3625

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);

4,862,481✔

3626

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

4,862,617✔

3627

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

23,833,080✔

3628

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

18,970,363✔

3629

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

18,972,634✔

3630

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

18,972,634✔

3631

        QUERY_CHECK_CODE(code, lino, _end);

18,970,977✔

3636

_end:

3,040,616✔

3637

  if (code != TSDB_CODE_SUCCESS) {

4,735,513✔

3640

  return code;

4,738,313✔

3643

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

9,449,872✔

3644

  int32_t        code = TSDB_CODE_SUCCESS;

9,449,872✔

3645

  int32_t        lino = 0;

9,449,872✔

3646

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,449,872✔

3647

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

9,451,485✔

3649

  STagScanInfo* pInfo = pOperator->info;

9,452,746✔

3650

  SSDataBlock*  pRes = pInfo->pRes;

9,455,497✔

3652

  QRY_PARAM_CHECK(ppRes);

9,447,726✔

3654

  if (pOperator->status == OP_EXEC_DONE) {

9,449,546✔

3655

    return TSDB_CODE_SUCCESS;

4,637,361✔

3657

  blockDataCleanup(pRes);

4,813,924✔

3659

  if (pInfo->pCtbCursor == NULL) {

4,815,161✔

3660

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

4,816,025✔

3661

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

4,813,765✔

3667

  SArray* aUidTags = pInfo->aUidTags;

4,815,509✔

3668

  SArray* aFilterIdxs = pInfo->aFilterIdxs;

4,816,018✔

3669

  int32_t count = 0;

4,813,258✔

3670

  bool    ctbCursorFinished = false;

4,813,258✔

3671

  while (1) {

90,524✔

3672

    taosArrayClearEx(aUidTags, tagScanFreeUidTag);

4,903,782✔

3673

    taosArrayClear(aFilterIdxs);

4,906,913✔

3675

    int32_t numTables = 0;

4,906,028✔

3676

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

12,319,942✔

3677

      SMCtbCursor* pCur = pInfo->pCtbCursor;

12,318,449✔

3678

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

12,317,576✔

3679

      if (uid == 0) {

12,312,599✔

3680

        ctbCursorFinished = true;

4,906,906✔

3681

        break;

4,906,906✔

3683

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

7,405,693✔

3684

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

7,407,108✔

3685

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

7,406,037✔

3687

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

7,406,037✔

3688

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

7,411,826✔

3689

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

7,411,826✔

3690

      ++numTables;

7,411,826✔

3693

    if (numTables == 0) {

4,906,906✔

3694

      break;

166,340✔

3696

    bool ignoreFilterIdx = true;

4,740,566✔

3697

    if (pInfo->pTagCond != NULL) {

4,740,566✔

3698

      ignoreFilterIdx = false;

1,699,812✔

3699

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

1,699,812✔

3700

      QUERY_CHECK_CODE(code, lino, _end);

1,695,293✔

3702

      ignoreFilterIdx = true;

3,040,247✔

3705

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

4,735,540✔

3706

    QUERY_CHECK_CODE(code, lino, _end);

4,738,684✔

3708

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

4,738,684✔

3710

    if (count != 0) {

4,738,820✔

3711

      break;

4,648,296✔

3715

  if (count > 0) {

4,814,636✔

3716

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

4,647,932✔

3718

  if (count == 0 || ctbCursorFinished) {

4,815,375✔

3719

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

4,815,375✔

3720

    pInfo->pCtbCursor = NULL;

4,810,295✔

3721

    setOperatorCompleted(pOperator);

4,810,295✔

3724

  pRes->info.rows = count;

4,812,260✔

3725

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

4,811,516✔

3726

  if (bLimitReached) {

4,812,882✔

3730

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

4,812,882✔

3731

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

4,815,369✔

3733

_end:

4,808,871✔

3734

  if (code != TSDB_CODE_SUCCESS) {

4,808,871✔

3740

  return code;

4,808,871✔

3743

static int32_t createTagScanTableListInfoFromParam(SOperatorInfo* pOperator) {

3,328,403✔

3744

  STagScanInfo*          pInfo = pOperator->info;

3,328,403✔

3745

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

3,328,403✔

3746

  int32_t                code = 0;

3,328,403✔

3747

  STableListInfo*        pListInfo = pInfo->pTableListInfo;

3,328,403✔

3748

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

3,328,403✔

3749

  tb_uid_t               pUid = pParam->vcUid;

3,328,403✔

3755

  STableKeyInfo info = {.groupId = 0};

3,328,403✔

3757

  int32_t tableIdx = 0;

3,328,403✔

3758

  taosHashClear(pListInfo->map);

3,328,403✔

3759

  taosArrayClear(pListInfo->pTableList);

3,328,403✔

3761

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

3,328,403✔

3768

  info.uid = pUid;

3,328,403✔

3769

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

3,328,403✔

3770

  if (p == NULL) {

3,328,403✔

3774

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

3,328,403✔

3776

  return code;

3,328,403✔

3779

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

6,492,932✔

3780

  int32_t       code = TSDB_CODE_SUCCESS;

6,492,932✔

3781

  int32_t       lino = 0;

6,492,932✔

3782

  STagScanInfo* pInfo = pOperator->info;

6,492,932✔

3783

  if (pOperator->pOperatorGetParam) {

6,492,932✔

3784

    pOperator->resultInfo.totalRows = 0;

3,328,403✔

3785

    pOperator->dynamicTask = true;

3,328,403✔

3786

    pInfo->curPos = 0;

3,328,403✔

3787

    code = createTagScanTableListInfoFromParam(pOperator);

3,328,403✔

3788

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

3,328,403✔

3789

    pOperator->pOperatorGetParam = NULL;

3,328,403✔

3790

    QUERY_CHECK_CODE(code, lino, _end);

3,328,403✔

3792

    if (pOperator->status == OP_EXEC_DONE) {

3,328,403✔

3793

      pOperator->status = OP_OPENED;

1,673,668✔

3797

  if (pOperator->status == OP_EXEC_DONE) {

6,492,402✔

3798

    (*ppRes) = NULL;

1,567,697✔

3799

    return code;

1,567,697✔

3802

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

4,925,765✔

3803

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

4,925,235✔

3805

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

4,924,705✔

3806

  SSDataBlock* pRes = pInfo->pRes;

4,925,765✔

3807

  blockDataCleanup(pRes);

4,925,235✔

3809

  int32_t size = 0;

4,925,765✔

3810

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

4,925,765✔

3811

  QUERY_CHECK_CODE(code, lino, _end);

4,925,765✔

3813

  if (size == 0) {

4,925,765✔

3814

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

19,348✔

3815

    (*ppRes) = NULL;

19,348✔

3816

    return code;

19,348✔

3819

  SMetaReader mr = {0};

4,906,417✔

3820

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

4,906,417✔

3821

  pRes->info.rows = 0;

4,906,417✔

3823

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

13,834,721✔

3824

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

8,928,304✔

3825

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

8,929,364✔

3828

      code = TSDB_CODE_SUCCESS;

8,929,364✔

3830

    QUERY_CHECK_CODE(code, lino, _end);

8,929,364✔

3832

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

8,929,364✔

3833

      setOperatorCompleted(pOperator);

4,906,417✔

3837

  pAPI->metaReaderFn.clearReader(&mr);

4,906,417✔

3838

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

4,905,887✔

3839

  if (bLimitReached) {

4,906,417✔

3844

  if (pOperator->status == OP_EXEC_DONE) {

4,906,417✔

3845

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

4,905,887✔

3848

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

4,906,947✔

3850

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

4,906,417✔

3851

_end:

4,905,887✔

3852

  if (code != TSDB_CODE_SUCCESS) {

4,905,837✔

3857

  return code;

4,905,837✔

3860

static void destroyTagScanOperatorInfo(void* param) {

8,460,363✔

3861

  STagScanInfo* pInfo = (STagScanInfo*)param;

8,460,363✔

3862

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

8,460,363✔

3865

  taosHashCleanup(pInfo->filterCtx.colHash);

8,463,874✔

3866

  taosArrayDestroy(pInfo->filterCtx.cInfoList);

8,448,183✔

3867

  taosArrayDestroy(pInfo->aFilterIdxs);

8,451,485✔

3868

  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);

8,447,546✔

3870

  blockDataDestroy(pInfo->pRes);

8,458,588✔

3871

  taosArrayDestroy(pInfo->matchInfo.pList);

8,459,464✔

3872

  tableListDestroy(pInfo->pTableListInfo);

8,453,371✔

3874

  pInfo->pRes = NULL;

8,458,013✔

3875

  pInfo->pTableListInfo = NULL;

8,459,354✔

3876

  taosMemoryFreeClear(param);

8,459,108✔

3877

8,454,070✔

3879

static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {

5,073✔

3880

  int32_t         code = TSDB_CODE_SUCCESS;

5,073✔

3881

  STagScanInfo*   pInfo = pOper->info;

5,073✔

3883

  pOper->status = OP_NOT_OPENED;

5,073✔

3884

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

5,073✔

3885

  if (pTagScanNode->onlyMetaCtbIdx) {

5,073✔

3886

      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

2,096✔

3887

      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,096✔

3888

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

2,096✔

3889

      pInfo->pCtbCursor = NULL;

2,096✔

3891

      pInfo->curPos = 0;

2,977✔

3894

  return code;

5,073✔

3897

int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,

8,461,673✔

3900

  QRY_PARAM_CHECK(pOptrInfo);

8,461,673✔

3902

  int32_t         code = TSDB_CODE_SUCCESS;

8,464,388✔

3903

  int32_t         lino = 0;

8,464,388✔

3904

  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;

8,464,388✔

3905

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

8,464,388✔

3906

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

8,455,532✔

3907

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

8,453,451✔

3911

  pOperator->pPhyNode = pTagScanNode;

8,453,451✔

3912

  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

8,455,331✔

3914

  int32_t    numOfExprs = 0;

8,457,353✔

3915

  SExprInfo* pExprInfo = NULL;

8,457,384✔

3917

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

8,456,692✔

3918

  QUERY_CHECK_CODE(code, lino, _error);

8,457,985✔

3920

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

8,457,985✔

3921

  QUERY_CHECK_CODE(code, lino, _error);

8,458,541✔

3923

  int32_t num = 0;

8,458,541✔

3924

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

8,458,974✔

3925

  QUERY_CHECK_CODE(code, lino, _error);

8,452,104✔

3927

  pInfo->pTagCond = pTagCond;

8,452,104✔

3928

  pInfo->pTagIndexCond = pTagIndexCond;

8,461,592✔

3929

  pInfo->suid = pPhyNode->suid;

8,456,955✔

3930

  pInfo->pStorageAPI = &pTaskInfo->storageAPI;

8,451,754✔

3932

  pInfo->pTableListInfo = pTableListInfo;

8,459,392✔

3933

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

8,453,596✔

3934

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

8,460,444✔

3936

  pInfo->readHandle = *pReadHandle;

8,449,862✔

3937

  pInfo->curPos = 0;

8,458,557✔

3939

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

8,449,280✔

3940

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

8,449,324✔

3942

  initResultSizeInfo(&pOperator->resultInfo, 4096);

8,454,842✔

3943

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

8,455,866✔

3944

  QUERY_CHECK_CODE(code, lino, _error);

8,464,336✔

3946

  if (pTagScanNode->onlyMetaCtbIdx) {

8,464,336✔

3947

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

4,821,981✔

3948

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

4,822,981✔

3950

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

4,822,227✔

3951

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

4,822,610✔

3953

    pInfo->filterCtx.colHash =

4,820,119✔

3954

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

4,821,739✔

3955

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

4,820,119✔

3957

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

4,819,345✔

3958

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

4,821,972✔

3960

    if (pInfo->pTagCond != NULL) {

4,820,204✔

3961

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

1,702,535✔

3964

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

8,464,322✔

3965

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

8,463,075✔

3967

  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);

8,456,288✔

3969

  *pOptrInfo = pOperator;

8,463,206✔

3970

  return code;

8,455,161✔

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,607,517,217✔

3997

  } else if (rightIdx == -1) {

2,067,692,082✔

3998

    return -1;

484,804✔

4001

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

2,067,207,278✔

4002

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

2,067,401,588✔

4003

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

2,067,850,060✔

4004

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

2,067,850,060✔

4007

  return ret;

2,067,527,994✔

4037

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

3,484,661✔

4038

  int32_t code = TSDB_CODE_SUCCESS;

3,484,661✔

4039

  int32_t lino = 0;

3,484,661✔

4040

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

3,484,661✔

4041

  dst->colList = NULL;

3,484,661✔

4042

  dst->pSlotList = NULL;

3,485,774✔

4044

  if (src->numOfCols > 0) {

3,489,414✔

4045

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

3,465,299✔

4046

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

3,483,940✔

4047

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

3,486,131✔

4049

    if (src->pSlotList != NULL) {

3,485,767✔

4050

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

3,470,836✔

4051

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

3,485,760✔

4052

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

3,483,205✔

4057

  if (code != TSDB_CODE_SUCCESS) {

3,486,131✔

4062

  return code;

3,486,131✔

4065

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

5,631,655✔

4067

  int32_t code = 0;

5,631,655✔

4069

  STableMergeScanInfo*    pInfo = pOperator->info;

5,631,655✔

4070

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

5,632,733✔

4071

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

5,635,309✔

4072

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

5,632,733✔

4073

  const SStorageAPI*      pAPI = &pTaskInfo->storageAPI;

5,632,383✔

4075

  blockDataCleanup(pInput->pReaderBlock);

5,634,581✔

4076

  if (!pInput->bInMemReader) {

5,633,496✔

4077

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

5,439,593✔

4078

                                         (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

2,719,800✔

4079

    if (code != 0) {

2,717,609✔

4084

  pInfo->base.dataReader = pInput->pReader;

5,634,595✔

4086

  while (true) {

4,823✔

4087

    bool hasNext = false;

5,634,987✔

4088

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

5,635,351✔

4089

    if (code != 0) {

5,640,503✔

4095

    if (!hasNext || isTaskKilled(pTaskInfo)) {

5,640,503✔

4096

      if (isTaskKilled(pTaskInfo)) {

3,398,638✔

4102

      *pSubTableHasBlock = false;

3,398,274✔

4103

      break;

3,398,267✔

4106

    if (pInput->tblCond.order == TSDB_ORDER_ASC) {

2,241,865✔

4107

      pInput->tblCond.twindows.skey = pInput->pReaderBlock->info.window.ekey + 1;

2,214,040✔

4109

      pInput->tblCond.twindows.ekey = pInput->pReaderBlock->info.window.skey - 1;

27,825✔

4112

    uint32_t status = 0;

2,241,501✔

4113

    code = loadDataBlock(pOperator, &pInfo->base, pInput->pReaderBlock, &status);

2,241,494✔

4114

    if (code != 0) {

2,241,494✔

4119

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

2,241,494✔

4123

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

2,241,494✔

4124

      continue;

4,823✔

4127

    *pSubTableHasBlock = true;

2,236,671✔

4128

    break;

2,236,671✔

4131

  if (*pSubTableHasBlock) {

5,636,051✔

4132

    pInput->pReaderBlock->info.id.groupId =

4,473,342✔

4133

        tableListGetTableGroupId(pInfo->base.pTableListInfo, pInput->pReaderBlock->info.id.uid);

4,472,971✔

4134

    pOperator->resultInfo.totalRows += pInput->pReaderBlock->info.rows;

2,236,671✔

4136

  if (!pInput->bInMemReader || !*pSubTableHasBlock) {

5,636,422✔

4137

    pAPI->tsdReader.tsdReaderClose(pInput->pReader);

4,051,884✔

4138

    pInput->pReader = NULL;

4,048,594✔

4141

  pInfo->base.dataReader = NULL;

5,632,390✔

4142

  return TSDB_CODE_SUCCESS;

5,633,860✔

4145

static int32_t setGroupStartEndIndex(STableMergeScanInfo* pInfo) {

2,059,610✔

4146

  pInfo->bGroupProcessed = false;

2,059,610✔

4148

  int32_t numOfTables = 0;

2,062,158✔

4149

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

2,052,673✔

4150

  if (code != TSDB_CODE_SUCCESS) {

2,036,615✔

4155

  int32_t i = pInfo->tableStartIndex + 1;

2,036,615✔

4156

  for (; i < numOfTables; ++i) {

3,367,572✔

4157

    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

1,387,678✔

4158

    if (!tableKeyInfo) {

1,386,565✔

4162

    if (tableKeyInfo->groupId != pInfo->groupId) {

1,386,565✔

4163

      break;

58,520✔

4166

  pInfo->tableEndIndex = i - 1;

2,038,414✔

4167

  return TSDB_CODE_SUCCESS;

2,064,314✔

4170

static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {

2,067,275✔

4171

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,464,807✔

4172

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,397,539✔

4173

    if (pInput->rowIdx == -1) {

3,397,539✔

4174

      continue;

1,597,611✔

4177

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,799,935✔

4178

      pInput->rowIdx = 0;

1,799,564✔

4179

      pInput->pageIdx = -1;

1,799,935✔

4182

    pInput->pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,800,306✔

4183

    SColumnInfoData* col = taosArrayGet(pInput->pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

1,799,935✔

4184

    if (!col) {

1,799,193✔

4188

    pInput->aTs = (int64_t*)col->pData;

1,799,193✔

4191

  __merge_compare_fn_t mergeCompareFn = (!pSubTblsInfo->pPkOrderInfo) ? subTblRowCompareTsFn : subTblRowCompareTsPkFn;

2,068,745✔

4192

  return tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, mergeCompareFn);

2,068,745✔

4195

static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {

2,061,073✔

4196

  int32_t code = setGroupStartEndIndex(pInfo);

2,061,073✔

4197

  if (code != TSDB_CODE_SUCCESS) {

2,064,685✔

4201

  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));

2,064,685✔

4202

  if (pSubTblsInfo == NULL) {

2,053,023✔

4205

  pSubTblsInfo->pTsOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);

2,053,023✔

4206

  if (!pSubTblsInfo->pTsOrderInfo) {

2,059,589✔

4210

  if (taosArrayGetSize(pInfo->pSortInfo) == 2) {

2,061,808✔

4217

    pSubTblsInfo->pPkOrderInfo = NULL;

2,056,327✔

4219

  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

2,063,250✔

4220

  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));

2,058,889✔

4221

  if (pSubTblsInfo->aInputs == NULL) {

2,047,570✔

4225

  int32_t bufPageSize = pInfo->bufPageSize;

2,051,196✔

4226

  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;

2,060,338✔

4227

  code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);

2,061,430✔

4228

  if (code != TSDB_CODE_SUCCESS) {

2,051,574✔

4233

  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;

2,051,574✔

4234

  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;

2,064,356✔

4236

  pInfo->pSubTablesMergeInfo = pSubTblsInfo;

2,058,868✔

4237

  return TSDB_CODE_SUCCESS;

2,043,531✔

4250

static int32_t setSubTableCondSuidByUid(STableMergeScanInfo* pInfo, SStorageAPI* pAPI, uint64_t uid,

3,392,800✔

4252

  int32_t     code = TSDB_CODE_SUCCESS;

3,392,800✔

4253

  int32_t     lino = 0;

3,392,800✔

4254

  SMetaReader tableReader = {0};

3,392,800✔

4256

  QUERY_CHECK_NULL(pInfo, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,394,998✔

4257

  QUERY_CHECK_NULL(pAPI, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,394,998✔

4258

  QUERY_CHECK_NULL(pCond, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,394,998✔

4260

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

3,394,998✔

4261

  code = pAPI->metaReaderFn.getTableEntryByUid(&tableReader, uid);

3,394,998✔

4262

  pAPI->metaReaderFn.readerReleaseLock(&tableReader);

3,390,231✔

4263

  QUERY_CHECK_CODE(code, lino, _end);

3,392,079✔

4265

  if (tableReader.me.type == TSDB_CHILD_TABLE) {

3,392,079✔

4266

    pCond->suid = tableReader.me.ctbEntry.suid;

3,238,539✔

4267

  } else if (tableReader.me.type == TSDB_NORMAL_TABLE) {

153,540✔

4268

    pCond->suid = 0;

153,540✔

4274

_end:

3,394,270✔

4275

  if (code != TSDB_CODE_SUCCESS) {

3,387,690✔

4278

  pAPI->metaReaderFn.clearReader(&tableReader);

3,387,690✔

4279

  return code;

3,385,149✔

4282

static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {

2,060,695✔

4283

  int32_t        code = TSDB_CODE_SUCCESS;

2,060,695✔

4284

  int32_t        lino = 0;

2,060,695✔

4285

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

2,060,695✔

4286

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

2,062,886✔

4287

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,046,093✔

4289

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

2,054,500✔

4291

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,458,976✔

4292

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,390,959✔

4293

    pInput->type = SUB_TABLE_MEM_BLOCK;

3,378,898✔

4295

    code = dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);

3,380,004✔

4296

    QUERY_CHECK_CODE(code, lino, _end);

3,391,323✔

4298

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pReaderBlock);

3,391,323✔

4299

    QUERY_CHECK_CODE(code, lino, _end);

3,393,164✔

4301

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pPageBlock);

3,393,164✔

4302

    QUERY_CHECK_CODE(code, lino, _end);

3,397,168✔

4304

    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);

3,397,168✔

4305

    pInput->pKeyInfo = keyInfo;

3,396,811✔

4306

    QUERY_CHECK_NULL(pInput->pKeyInfo, code, lino, _end, terrno);

3,396,811✔

4308

    code = setSubTableCondSuidByUid(pInfo, pAPI, pInput->pKeyInfo->uid, &pInput->tblCond);

3,392,422✔

4309

    QUERY_CHECK_CODE(code, lino, _end);

3,384,393✔

4311

    if (isTaskKilled(pTaskInfo)) {

3,384,393✔

4315

    if (i + 1 < pSubTblsInfo->numInMemReaders) {

3,386,591✔

4316

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

2,659,058✔

4317

                                           (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

1,329,529✔

4318

      QUERY_CHECK_CODE(code, lino, _end);

1,327,674✔

4319

      pInput->bInMemReader = true;

1,327,674✔

4321

      pInput->pReader = NULL;

2,065,448✔

4322

      pInput->bInMemReader = false;

2,065,084✔

4324

    bool hasNext = true;

3,394,221✔

4325

    code = fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);

3,394,592✔

4326

    QUERY_CHECK_CODE(code, lino, _end);

3,398,274✔

4327

    if (!hasNext) {

3,398,274✔

4328

      pInput->rowIdx = -1;

1,597,968✔

4329

      ++pSubTblsInfo->numSubTablesCompleted;

1,597,961✔

4330

      continue;

1,598,339✔

4332

      pInput->rowIdx = 0;

1,800,306✔

4333

      pInput->pageIdx = -1;

1,800,306✔

4337

_end:

2,068,017✔

4338

  if (code != TSDB_CODE_SUCCESS) {

2,067,646✔

4341

  return code;

2,068,010✔

4344

static void adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

2,237,042✔

4345

  int32_t              code = TSDB_CODE_SUCCESS;

2,237,042✔

4346

  int32_t              lino = 0;

2,237,042✔

4347

  SExecTaskInfo*       pTaskInfo = pOperatorInfo->pTaskInfo;

2,237,042✔

4348

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

2,237,042✔

4349

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

2,237,042✔

4350

  bool                 hasNext = true;

2,237,042✔

4351

  code = fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);

2,237,042✔

4352

  QUERY_CHECK_CODE(code, lino, _end);

2,237,042✔

4354

  if (!hasNext) {

2,237,042✔

4355

    pInput->rowIdx = -1;

1,800,306✔

4356

    ++pSubTblsInfo->numSubTablesCompleted;

1,800,306✔

4358

    pInput->rowIdx = 0;

436,736✔

4361

_end:

2,237,042✔

4362

  if (code != TSDB_CODE_SUCCESS) {

2,237,042✔

4367

2,237,042✔

4369

static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

2,147,483,647✔

4370

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

2,147,483,647✔

4371

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

2,147,483,647✔

4373

  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

2,147,483,647✔

4374

  if (pInput->rowIdx < pInputBlock->info.rows - 1) {

2,147,483,647✔

4375

    ++pInput->rowIdx;

2,147,483,647✔

4376

  } else if (pInput->rowIdx == pInputBlock->info.rows - 1) {

2,237,042✔

4377

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

2,237,042✔

4378

      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);

2,237,042✔

4380

    if (pInput->rowIdx != -1) {

2,237,042✔

4381

      SColumnInfoData* col = taosArrayGet(pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

436,736✔

4382

      if (!col) {

436,736✔

4386

      pInput->pInputBlock = pInputBlock;

436,736✔

4387

      pInput->aTs = (int64_t*)col->pData;

436,736✔

4391

  return tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));

2,147,483,647✔

4394

static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {

2,147,483,647✔

4395

  int32_t            code = TSDB_CODE_SUCCESS;

2,147,483,647✔

4396

  int32_t            lino = 0;

2,147,483,647✔

4397

  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

2,147,483,647✔

4398

  SSDataBlock*       pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

2,147,483,647✔

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

99,107,267✔

4410

      QUERY_CHECK_CODE(code, lino, _end);

99,106,154✔

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;

2,147,483,647✔

4420

  pBlock->info.scanFlag = pInputBlock->info.scanFlag;

2,147,483,647✔

4421

  pBlock->info.rows += 1;

2,147,483,647✔

4423

_end:

2,147,483,647✔

4424

  if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

4427

  return code;

2,147,483,647✔

4430

static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,

5,477,364✔

4432

  int32_t                 code = TSDB_CODE_SUCCESS;

5,477,364✔

4433

  int32_t                 lino = 0;

5,477,364✔

4434

  STableMergeScanInfo*    pInfo = pOperator->info;

5,477,364✔

4435

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

5,477,735✔

4436

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

5,478,092✔

4437

  bool                    finished = false;

5,475,894✔

4439

  QRY_PARAM_CHECK(pResBlock);

5,475,894✔

4441

  blockDataCleanup(pBlock);

5,477,728✔

4445

      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {

2,147,483,647✔

4446

        finished = true;

3,386,179✔

4447

        break;

3,386,179✔

4450

      code = appendChosenRowToDataBlock(pSubTblsInfo, pBlock);

2,147,483,647✔

4451

      QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

4453

      code = adjustSubTableForNextRow(pOperator, pSubTblsInfo);

2,147,483,647✔

4454

      QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

4456

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

2,147,483,647✔

4457

        break;

2,091,556✔

4461

    if (isTaskKilled(pTaskInfo)) {

5,477,735✔

4465

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

5,478,463✔

4466

    if (finished || limitReached || pBlock->info.rows > 0) {

5,478,099✔

4471

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

5,478,099✔

4472

    *pResBlock = pBlock;

3,409,354✔

4475

_end:

2,068,745✔

4476

  if (code != TSDB_CODE_SUCCESS) {

5,478,099✔

4480

  return code;

5,478,463✔

4483

static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {

2,061,801✔

4484

  int32_t              code = TSDB_CODE_SUCCESS;

2,061,801✔

4485

  int32_t              lino = 0;

2,061,801✔

4486

  STableMergeScanInfo* pInfo = pOperator->info;

2,061,801✔

4488

  code = initSubTablesMergeInfo(pInfo);

2,064,349✔

4489

  QUERY_CHECK_CODE(code, lino, _end);

2,041,340✔

4491

  code = initSubTableInputs(pOperator, pInfo);

2,041,340✔

4492

  QUERY_CHECK_CODE(code, lino, _end);

2,067,275✔

4494

  code = openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);

2,067,275✔

4495

  QUERY_CHECK_CODE(code, lino, _end);

2,068,738✔

4497

_end:

2,068,738✔

4498

  if (code != TSDB_CODE_SUCCESS) {

2,068,738✔

4501

  return code;

2,068,374✔

4504

static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {

31,842,418✔

4505

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

31,842,418✔

4506

  if (pSubTblsInfo != NULL) {

31,846,443✔

4507

    tMergeTreeDestroy(&pSubTblsInfo->pTree);

2,069,109✔

4509

    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,465,934✔

4510

      STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,398,645✔

4511

      cleanupQueryTableDataCond(&pInput->tblCond);

3,397,917✔

4512

      blockDataDestroy(pInput->pReaderBlock);

3,397,917✔

4513

      blockDataDestroy(pInput->pPageBlock);

3,398,645✔

4514

      taosArrayDestroy(pInput->aBlockPages);

3,398,281✔

4515

      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);

3,398,281✔

4516

      pInput->pReader = NULL;

3,398,281✔

4519

    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);

2,068,381✔

4520

    taosMemoryFree(pSubTblsInfo->aInputs);

2,068,017✔

4522

    taosMemoryFree(pSubTblsInfo);

2,067,653✔

4523

    pInfo->pSubTablesMergeInfo = NULL;

2,066,925✔

4527

31,845,351✔

4529

void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

41,071,187✔

4530

  STableMergeScanInfo* pInfo = pOperator->info;

41,071,187✔

4531

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

41,073,767✔

4532

  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));

41,072,866✔

4534

  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

41,077,595✔

4535

  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;

41,070,785✔

4536

  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;

41,071,708✔

4537

  pInfo->sortExecInfo.loops += sortExecInfo.loops;

41,068,609✔

4538

  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;

41,068,070✔

4539

  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

41,058,550✔

4541

  tsortDestroySortHandle(pInfo->pSortHandle);

41,064,494✔

4542

  pInfo->pSortHandle = NULL;

41,067,081✔

4543

41,063,589✔

4552

static int32_t createVTableMergeScanInfoFromBatchParam(SOperatorInfo* pOperator) {

95,536✔

4553

  int32_t                  code = TSDB_CODE_SUCCESS;

95,536✔

4554

  int32_t                  lino = 0;

95,536✔

4555

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

95,536✔

4556

  STableMergeScanInfo*     pInfo = pOperator->info;

95,536✔

4557

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

95,536✔

4558

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

95,536✔

4559

  SMetaReader              orgTable = {0};

95,536✔

4560

  SOperatorParam*          pGetParam = pOperator->pOperatorGetParam;

95,536✔

4561

  STableScanOperatorParam* pParam = (STableScanOperatorParam*)pGetParam->value;

95,536✔

4563

  if (pParam->paramType != DYN_TYPE_SCAN_PARAM) {

95,536✔

4567

  if (pInfo->pSubTablesMergeInfo) {

95,536✔

4570

  if (pInfo->pSortHandle) {

95,536✔

4573

  if (pInfo->base.dataReader) {

95,536✔

4578

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

95,536✔

4584

  pInfo->numNextDurationBlocks = 0;

95,536✔

4585

  pInfo->nextDurationBlocksIdx = 0;

95,536✔

4586

  pInfo->rtnNextDurationBlocks = false;

95,536✔

4587

  pInfo->bNewFilesetEvent = false;

95,536✔

4588

  pInfo->bNextDurationBlockEvent = false;

95,536✔

4589

  taosHashCleanup(pInfo->mSkipTables);

95,536✔

4590

  pInfo->mSkipTables = NULL;

95,536✔

4592

  if (pInfo->base.orgCond.colList && pInfo->base.orgCond.numOfCols > 0) {

95,536✔

4593

    cleanupQueryTableDataCond(&pInfo->base.cond);

95,536✔

4594

    code = dumpQueryTableCond(&pInfo->base.orgCond, &pInfo->base.cond);

95,536✔

4595

    QUERY_CHECK_CODE(code, lino, _return);

95,536✔

4598

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

95,536✔

4599

    if (pInfo->base.orgCond.colList && pInfo->base.orgCond.numOfCols > 0) {

95,536✔

4600

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

95,536✔

4601

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

95,536✔

4608

  pInfo->tableStartIndex = 0;

95,536✔

4609

  pInfo->tableEndIndex = 0;

95,536✔

4610

  pInfo->hasGroupId = false;

95,536✔

4611

  pInfo->groupId = pParam->groupid;

95,536✔

4612

  pInfo->bGroupProcessed = false;

95,536✔

4613

  pOperator->status = OP_NOT_OPENED;

95,536✔

4615

  pListInfo->oneTableForEachGroup = true;

95,536✔

4616

  pListInfo->numOfOuputGroups = 1;

95,536✔

4617

  taosHashClear(pListInfo->map);

95,536✔

4618

  taosArrayClear(pListInfo->pTableList);

95,536✔

4620

  int32_t tableIdx = 0;

95,536✔

4621

  if (pParam->pBatchTbInfo) {

95,536✔

4622

    int32_t numOfBatchTables = taosArrayGetSize(pParam->pBatchTbInfo);

95,536✔

4623

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

336,082✔

4624

      SOrgTbInfo* pOrgTbInfo = taosArrayGet(pParam->pBatchTbInfo, i);

240,546✔

4625

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

240,546✔

4627

      const char* pTableName = strstr(pOrgTbInfo->tbName, ".");

240,546✔

4628

      pTableName = pTableName ? (pTableName + 1) : pOrgTbInfo->tbName;

240,546✔

4629

      pAPI->metaReaderFn.initReader(&orgTable, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

240,546✔

4630

      code = pAPI->metaReaderFn.getTableEntryByName(&orgTable, pTableName);

240,546✔

4631

      pAPI->metaReaderFn.readerReleaseLock(&orgTable);

240,546✔

4632

      QUERY_CHECK_CODE(code, lino, _return);

240,546✔

4634

      uint64_t uid = orgTable.me.uid;

240,546✔

4635

      code = taosHashPut(pListInfo->map, &uid, sizeof(uid), &tableIdx, sizeof(tableIdx));

240,546✔

4636

      if (code == TSDB_CODE_DUP_KEY) {

240,546✔

4640

      QUERY_CHECK_CODE(code, lino, _return);

240,546✔

4642

      STableKeyInfo keyInfo = {.groupId = pParam->groupid, .uid = uid};

240,546✔

4643

      QUERY_CHECK_NULL(taosArrayPush(pListInfo->pTableList, &keyInfo), code, lino, _return, terrno);

481,092✔

4644

      ++tableIdx;

240,546✔

4646

      qDebug("%s table merge dynamic batch add table name:%s uid:%" PRIu64 " group:%" PRIu64, GET_TASKID(pTaskInfo),

240,546✔

4648

      pAPI->metaReaderFn.clearReader(&orgTable);

240,546✔

4652

  qDebug("%s table merge dynamic batch param consumed, group:%" PRIu64 ", tables:%d", GET_TASKID(pTaskInfo),

95,536✔

4655

_return:

95,536✔

4656

  if (code != TSDB_CODE_SUCCESS) {

95,536✔

4659

  pAPI->metaReaderFn.clearReader(&orgTable);

95,536✔

4660

  return code;

95,536✔

4663

int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

5,448,258✔

4664

  int32_t              code = TSDB_CODE_SUCCESS;

5,448,258✔

4665

  int32_t              lino = 0;

5,448,258✔

4666

  int32_t              tableListSize = 0;

5,448,258✔

4667

  int64_t              st = taosGetTimestampUs();

5,447,530✔

4668

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

5,447,530✔

4669

  STableMergeScanInfo* pInfo = pOperator->info;

5,449,728✔

4671

  QRY_PARAM_CHECK(ppRes);

5,442,777✔

4673

  if (pOperator->status == OP_EXEC_DONE && !pOperator->pOperatorGetParam) {

5,444,261✔

4678

  if (pOperator->pOperatorGetParam) {

5,447,894✔

4679

    code = createVTableMergeScanInfoFromBatchParam(pOperator);

95,536✔

4680

    QUERY_CHECK_CODE(code, lino, _end);

95,536✔

4682

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

95,536✔

4683

    pOperator->pOperatorGetParam = NULL;

95,536✔

4684

    pOperator->dynamicTask = true;

95,536✔

4687

  code = pOperator->fpSet._openFn(pOperator);

5,436,589✔

4688

  QUERY_CHECK_CODE(code, lino, _end);

5,429,666✔

4690

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

5,429,666✔

4691

  QUERY_CHECK_CODE(code, lino, _end);

5,441,706✔

4693

  if (!pInfo->hasGroupId) {

5,441,706✔

4694

    pInfo->hasGroupId = true;

2,033,815✔

4696

    if (tableListSize == 0) {

2,019,563✔

4697

      setOperatorCompleted(pOperator);

35,987✔

4698

      (*ppRes) = NULL;

35,987✔

4699

      return code;

35,987✔

4702

    pInfo->tableStartIndex = 0;

1,983,576✔

4703

    STableKeyInfo* pTmpGpId = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

2,002,560✔

4704

    QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

1,984,332✔

4705

    pInfo->groupId = pTmpGpId->groupId;

1,984,332✔

4706

    code = startSubTablesTableMergeScan(pOperator);

2,002,210✔

4707

    QUERY_CHECK_CODE(code, lino, _end);

2,009,854✔

4710

  SSDataBlock* pBlock = NULL;

5,417,017✔

4711

  while (pInfo->tableStartIndex < tableListSize) {

5,477,364✔

4712

    if (isTaskKilled(pTaskInfo)) {

5,477,364✔

4716

    code = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity, &pBlock);

5,477,000✔

4717

    QUERY_CHECK_CODE(code, lino, _end);

5,478,463✔

4719

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

5,478,463✔

4726

    if (pBlock != NULL) {

5,478,463✔

4727

      pBlock->info.id.groupId = pInfo->groupId;

3,409,354✔

4728

      pOperator->resultInfo.totalRows += pBlock->info.rows;

3,409,354✔

4729

      pInfo->bGroupProcessed = true;

3,409,354✔

4730

      break;

3,409,354✔

4733

      stopSubTablesTableMergeScan(pInfo);

2,069,109✔

4734

      if (pInfo->tableEndIndex >= tableListSize - 1) {

2,067,653✔

4735

        setOperatorCompleted(pOperator);

2,009,133✔

4736

        break;

2,008,041✔

4739

      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

58,520✔

4740

      STableKeyInfo* pTmpGpId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

58,520✔

4741

      QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

58,520✔

4743

      pInfo->groupId = pTmpGpId->groupId;

58,520✔

4744

      code = startSubTablesTableMergeScan(pOperator);

58,520✔

4745

      QUERY_CHECK_CODE(code, lino, _end);

58,520✔

4746

      resetLimitInfoForNextGroup(&pInfo->limitInfo);

58,520✔

4750

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

5,417,395✔

4752

_end:

5,419,943✔

4753

  if (code != TSDB_CODE_SUCCESS) {

5,419,215✔

4758

    (*ppRes) = pBlock;

5,419,215✔

4761

  return code;

5,419,579✔

4764

static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {

3,586,404✔

4765

  int32_t              code = TSDB_CODE_SUCCESS;

3,586,404✔

4766

  int32_t              lino = 0;

3,586,404✔

4767

  SOperatorInfo*       pOperator = (SOperatorInfo*)pTableMergeOpInfo;

3,586,404✔

4768

  STableMergeScanInfo* pInfo = pOperator->info;

3,586,404✔

4769

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,586,404✔

4771

  if (pInfo->mSkipTables == NULL) {

3,586,404✔

4772

    pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,

453,648✔

4774

    QUERY_CHECK_NULL(pInfo->mSkipTables, code, lino, _end, terrno);

453,648✔

4776

  int bSkip = 1;

3,586,404✔

4777

  if (pInfo->mSkipTables != NULL) {

3,586,404✔

4778

    code = taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));

3,586,404✔

4779

    if (code == TSDB_CODE_DUP_KEY) {

3,586,404✔

4782

    QUERY_CHECK_CODE(code, lino, _end);

3,586,404✔

4785

_end:

3,586,404✔

4786

  if (code != TSDB_CODE_SUCCESS) {

3,586,404✔

4791

3,586,404✔

4793

static int32_t doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {

172,441,804✔

4794

  STableMergeScanInfo* pInfo = pOperator->info;

172,441,804✔

4795

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

172,458,475✔

4796

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

172,446,593✔

4797

  SSDataBlock*         pBlock = pInfo->pReaderBlock;

172,436,992✔

4798

  int32_t              code = 0;

172,448,756✔

4799

  bool                 hasNext = false;

172,448,756✔

4800

  STsdbReader*         reader = pInfo->base.dataReader;

172,462,388✔

4802

  code = pAPI->tsdReader.tsdNextDataBlock(reader, &hasNext);

172,450,747✔

4803

  if (code != 0) {

172,441,009✔

4810

  if (!hasNext || isTaskKilled(pTaskInfo)) {

172,441,009✔

4811

    if (isTaskKilled(pTaskInfo)) {

30,730,838✔

4815

    *pFinished = true;

30,725,119✔

4816

    return code;

30,721,963✔

4819

  uint32_t status = 0;

141,715,929✔

4820

  code = loadDataBlock(pOperator, &pInfo->base, pBlock, &status);

141,713,965✔

4822

  if (code != TSDB_CODE_SUCCESS) {

141,718,681✔

4823

    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));

16,833✔

4824

    pTaskInfo->code = code;

16,833✔

4825

    return code;

16,833✔

4828

  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

141,701,848✔

4834

  if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {

141,701,848✔

4835

    *pSkipped = true;

6,495,011✔

4836

    return code;

6,480,022✔

4839

  return code;

135,209,099✔

4842

static int32_t getBlockForTableMergeScan(void* param, SSDataBlock** ppBlock) {

176,082,809✔

4843

  STableMergeScanSortSourceParam* source = param;

176,082,809✔

4845

  SOperatorInfo*       pOperator = source->pOperator;

176,082,809✔

4846

  STableMergeScanInfo* pInfo = pOperator->info;

176,088,175✔

4847

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

176,084,573✔

4848

  SSDataBlock*         pBlock = NULL;

176,089,041✔

4849

  int64_t              st = taosGetTimestampUs();

176,085,312✔

4850

  int32_t              code = TSDB_CODE_SUCCESS;

176,085,312✔

4853

    if (pInfo->rtnNextDurationBlocks) {

191,856,002✔

4854

      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d",

19,404,244✔

4857

      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {

19,408,143✔

4858

        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];

9,295,592✔

4859

        ++pInfo->nextDurationBlocksIdx;

9,295,592✔

4861

        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

19,401,461✔

4862

          blockDataDestroy(pInfo->nextDurationBlocks[i]);

9,294,498✔

4863

          pInfo->nextDurationBlocks[i] = NULL;

9,292,583✔

4866

        pInfo->rtnNextDurationBlocks = false;

10,108,495✔

4867

        pInfo->nextDurationBlocksIdx = 0;

10,108,112✔

4868

        pInfo->numNextDurationBlocks = 0;

10,104,282✔

4869

        continue;

10,106,197✔

4872

      bool bFinished = false;

172,443,761✔

4873

      bool bSkipped = false;

172,451,615✔

4875

      code = doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);

172,460,982✔

4876

      if (code != 0) {

172,433,430✔

4877

        return code;

16,833✔

4880

      pBlock = pInfo->pReaderBlock;

172,416,597✔

4881

      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d",

172,435,106✔

4883

      if (bFinished) {

172,459,205✔

4884

        pInfo->bNewFilesetEvent = false;

30,732,734✔

4885

        break;

30,732,725✔

4888

      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {

141,726,471✔

4889

        if (!bSkipped) {

10,082,089✔

4890

          code = createOneDataBlock(pBlock, true, &pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks]);

9,295,592✔

4891

          if (code) {

9,291,762✔

4896

          ++pInfo->numNextDurationBlocks;

9,291,762✔

4897

          if (pInfo->numNextDurationBlocks > 2) {

9,289,464✔

4905

        if (pInfo->bNewFilesetEvent) {

10,075,578✔

4906

          pInfo->rtnNextDurationBlocks = true;

10,103,461✔

4907

          *ppBlock = NULL;

10,102,312✔

4908

          return code;

10,099,248✔

4911

        if (pInfo->bNextDurationBlockEvent) {

4,596✔

4916

      if (bSkipped) continue;

131,618,372✔

4919

    pBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pBlock->info.id.uid);

135,248,705✔

4921

    pOperator->resultInfo.totalRows += pBlock->info.rows;

135,247,912✔

4922

    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

135,245,256✔

4923

    *ppBlock = pBlock;

135,244,670✔

4925

    return code;

135,242,568✔

4928

  *ppBlock = NULL;

30,731,716✔

4929

  return code;

30,731,333✔

4932

int32_t generateSortByTsPkInfo(SArray* colMatchInfo, int32_t order, SArray** ppSortArray) {

27,355,867✔

4933

  int32_t code = TSDB_CODE_SUCCESS;

27,355,867✔

4934

  int32_t lino = 0;

27,355,867✔

4935

  SArray* pSortInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));

27,355,867✔

4936

  QUERY_CHECK_NULL(pSortInfo, code, lino, _end, terrno);

27,326,907✔

4937

  SBlockOrderInfo biTs = {0};

27,326,907✔

4938

  SBlockOrderInfo biPk = {0};

27,323,527✔

4940

  int32_t tsTargetSlotId = 0;

27,342,212✔

4941

  int32_t pkTargetSlotId = -1;

27,342,212✔

4942

  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {

106,559,345✔

4943

    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);

79,252,674✔

4944

    QUERY_CHECK_NULL(colInfo, code, lino, _end, terrno);

79,253,348✔

4945

    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

79,253,348✔

4946

      tsTargetSlotId = colInfo->dstSlotId;

27,363,920✔

4947

      biTs.order = order;

27,360,306✔

4948

      biTs.slotId = tsTargetSlotId;

27,360,306✔

4949

      biTs.nullFirst = (order == TSDB_ORDER_ASC);

27,360,306✔

4950

      biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, order);

27,360,306✔

4953

    if (colInfo->isPk) {

79,235,244✔

4954

      pkTargetSlotId = colInfo->dstSlotId;

2,440,757✔

4955

      biPk.order = order;

2,441,408✔

4956

      biPk.slotId = pkTargetSlotId;

2,441,408✔

4957

      biPk.nullFirst = (order == TSDB_ORDER_ASC);

2,441,408✔

4958

      biPk.compFn = getKeyComparFunc(colInfo->dataType.type, order);

2,441,408✔

4962

  void* tmp = taosArrayPush(pSortInfo, &biTs);

27,366,634✔

4963

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

27,366,634✔

4964

  if (pkTargetSlotId != -1) {

27,366,634✔

4965

    tmp = taosArrayPush(pSortInfo, &biPk);

2,440,106✔

4966

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

2,440,106✔

4969

  (*ppSortArray) = pSortInfo;

27,366,634✔

4971

_end:

27,348,267✔

4972

  if (code != TSDB_CODE_SUCCESS) {

27,337,513✔

4975

  return code;

27,337,513✔

4978

void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {

31,658,935✔

4979

  STableMergeScanInfo* pTmsInfo = param;

31,658,935✔

4980

  if (type == TSD_READER_NOTIFY_DURATION_START) {

31,658,935✔

4981

    pTmsInfo->bNewFilesetEvent = true;

31,663,424✔

4985

  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);

31,654,098✔

4987

  return;

31,677,831✔

4990

int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

41,058,522✔

4991

  STableMergeScanInfo* pInfo = pOperator->info;

41,058,522✔

4992

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

41,069,575✔

4994

  int32_t code = TSDB_CODE_SUCCESS;

41,059,289✔

4995

  int32_t lino = 0;

41,059,289✔

4996

  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

41,059,289✔

4998

  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));

41,061,873✔

4999

  pInfo->bNewFilesetEvent = false;

41,093,216✔

5000

  pInfo->bNextDurationBlockEvent = false;

41,087,321✔

5002

  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;

41,086,777✔

5003

  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;

41,074,096✔

5005

  pInfo->pSortHandle = NULL;

41,079,178✔

5006

  code = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,

82,162,818✔

5007

                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0, &pInfo->pSortHandle);

41,078,666✔

5008

  if (code) {

41,048,366✔

5012

  if (pInfo->bSortRowId && numOfTable != 1) {

41,048,366✔

5013

    int32_t memSize = 512 * 1024 * 1024;

3,019,088✔

5014

    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);

3,019,088✔

5015

    QUERY_CHECK_CODE(code, lino, _end);

3,020,479✔

5018

  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);

41,053,120✔

5019

  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pOperator);

41,060,127✔

5020

  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);

41,064,928✔

5022

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);

41,059,991✔

5023

  QUERY_CHECK_CODE(code, lino, _end);

41,038,506✔

5025

  STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));

41,038,506✔

5026

  QUERY_CHECK_NULL(param, code, lino, _end, terrno);

41,037,037✔

5027

  param->pOperator = pOperator;

41,037,037✔

5029

  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));

41,037,249✔

5030

  if (ps == NULL) {

41,032,365✔

5035

  ps->param = param;

41,032,365✔

5036

  ps->onlyRef = false;

41,057,678✔

5037

  code = tsortAddSource(pInfo->pSortHandle, ps);

41,062,066✔

5038

  QUERY_CHECK_CODE(code, lino, _end);

41,055,003✔

5040

  if (numOfTable == 1) {

41,055,003✔

5041

    tsortSetSingleTableMerge(pInfo->pSortHandle);

27,524,543✔

5043

    code = tsortOpen(pInfo->pSortHandle);

13,530,460✔

5044

    QUERY_CHECK_CODE(code, lino, _end);

13,533,116✔

5047

_end:

13,524,971✔

5048

  if (code != TSDB_CODE_SUCCESS) {

41,071,157✔

5049

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,145✔

5051

  return code;

41,068,347✔

5054

void startGroupTableMergeScan(SOperatorInfo* pOperator) {

30,961,217✔

5055

  int32_t              code = TSDB_CODE_SUCCESS;

30,961,217✔

5056

  int32_t              lino = 0;

30,961,217✔

5057

  STableMergeScanInfo* pInfo = pOperator->info;

30,961,217✔

5058

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

30,967,597✔

5059

  SReadHandle*         pHandle = &pInfo->base.readHandle;

30,959,785✔

5060

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

30,964,659✔

5061

  qDebug("%s table merge scan start group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

30,936,004✔

5064

    int32_t numOfTables = 0;

30,941,028✔

5065

    code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

30,965,739✔

5066

    QUERY_CHECK_CODE(code, lino, _end);

30,946,903✔

5068

    int32_t i = pInfo->tableStartIndex + 1;

30,946,903✔

5069

    for (; i < numOfTables; ++i) {

139,908,883✔

5070

      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

115,419,012✔

5071

      QUERY_CHECK_NULL(tableKeyInfo, code, lino, _end, terrno);

115,418,512✔

5072

      if (tableKeyInfo->groupId != pInfo->groupId) {

115,418,512✔

5073

        break;

6,446,872✔

5076

    pInfo->tableEndIndex = i - 1;

30,936,743✔

5078

  pInfo->bGroupProcessed = false;

30,970,412✔

5079

  int32_t tableStartIdx = pInfo->tableStartIndex;

30,972,939✔

5080

  int32_t tableEndIdx = pInfo->tableEndIndex;

30,964,867✔

5082

  int32_t        numOfTable = tableEndIdx - tableStartIdx + 1;

30,966,612✔

5083

  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);

30,966,612✔

5084

  code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,

61,928,328✔

5085

                                       (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->mSkipTables);

30,961,222✔

5086

  QUERY_CHECK_CODE(code, lino, _end);

30,952,623✔

5087

  if (pInfo->filesetDelimited) {

30,952,623✔

5088

    pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);

30,960,186✔

5090

  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);

30,962,953✔

5092

  code = startDurationForGroupTableMergeScan(pOperator);

30,944,598✔

5093

  QUERY_CHECK_CODE(code, lino, _end);

30,957,881✔

5095

_end:

30,949,736✔

5096

  if (code != TSDB_CODE_SUCCESS) {

30,960,399✔

5097

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,145✔

5098

    pTaskInfo->code = code;

8,145✔

5099

    T_LONG_JMP(pTaskInfo->env, code);

8,145✔

5101

30,952,254✔

5103

void stopGroupTableMergeScan(SOperatorInfo* pOperator) {

30,963,669✔

5104

  STableMergeScanInfo* pInfo = pOperator->info;

30,963,669✔

5105

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

30,963,669✔

5106

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

30,963,669✔

5108

  stopDurationForGroupTableMergeScan(pOperator);

30,964,178✔

5110

  if (pInfo->base.dataReader != NULL) {

30,954,209✔

5111

    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);

30,957,452✔

5112

    pInfo->base.dataReader = NULL;

30,922,663✔

5114

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

30,931,125✔

5115

    if (pInfo->nextDurationBlocks[i]) {

1,094✔

5116

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

1,094✔

5117

      pInfo->nextDurationBlocks[i] = NULL;

1,094✔

5119

    pInfo->numNextDurationBlocks = 0;

1,094✔

5120

    pInfo->nextDurationBlocksIdx = 0;

1,094✔

5122

  resetLimitInfoForNextGroup(&pInfo->limitInfo);

30,941,956✔

5123

  taosHashCleanup(pInfo->mSkipTables);

30,908,106✔

5124

  pInfo->mSkipTables = NULL;

30,917,652✔

5125

  qDebug("%s table merge scan stop group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

30,922,038✔

5126

30,938,577✔

5130

SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,

83,958,888✔

5132

  STableMergeScanInfo* pInfo = pOperator->info;

83,958,888✔

5133

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

83,965,894✔

5134

  STupleHandle*        pTupleHandle = NULL;

83,969,444✔

5136

  blockDataCleanup(pResBlock);

83,944,551✔

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✔

5146

      if (pTupleHandle == NULL) {

2,147,483,647✔

5147

        break;

67,556,616✔

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,410,024✔

5161

    if (tsortIsClosed(pHandle)) {

83,966,640✔

5166

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);

83,971,598✔

5167

    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,

83,970,376✔

5169

    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {

83,976,539✔

5173

  return (pResBlock->info.rows > 0) ? pResBlock : NULL;

83,976,539✔

5176

int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

70,657,153✔

5177

  if (pOperator->status == OP_EXEC_DONE) {

70,657,153✔

5178

    (*ppRes) = NULL;

45,818✔

5179

    return TSDB_CODE_SUCCESS;

45,818✔

5182

  int32_t              code = TSDB_CODE_SUCCESS;

70,586,998✔

5183

  int32_t              lino = 0;

70,586,998✔

5184

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

70,586,998✔

5185

  STableMergeScanInfo* pInfo = pOperator->info;

70,601,868✔

5187

  code = pOperator->fpSet._openFn(pOperator);

70,611,474✔

5188

  QUERY_CHECK_CODE(code, lino, _end);

70,579,082✔

5190

  int64_t st = taosGetTimestampUs();

70,588,686✔

5192

  int32_t tableListSize = 0;

70,588,686✔

5193

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

70,597,961✔

5194

  QUERY_CHECK_CODE(code, lino, _end);

70,603,422✔

5196

  if (!pInfo->hasGroupId) {

70,603,422✔

5197

    pInfo->hasGroupId = true;

27,704,946✔

5199

    if (tableListSize == 0) {

27,687,154✔

5200

      setOperatorCompleted(pOperator);

3,199,512✔

5201

      (*ppRes) = NULL;

3,199,124✔

5202

      return code;

3,199,124✔

5204

    pInfo->tableStartIndex = 0;

24,487,642✔

5205

    STableKeyInfo* tmp = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

24,505,866✔

5206

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

24,499,374✔

5207

    pInfo->groupId = tmp->groupId;

24,499,374✔

5208

    startGroupTableMergeScan(pOperator);

24,510,408✔

5211

  SSDataBlock* pBlock = NULL;

67,395,216✔

5212

  while (pInfo->tableStartIndex < tableListSize) {

83,948,975✔

5213

    if (isTaskKilled(pTaskInfo)) {

83,962,229✔

5217

    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,

83,961,981✔

5219

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

83,977,928✔

5220

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,878,413✔

5221

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

1,878,413✔

5222

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

1,878,413✔

5224

    if (pBlock != NULL) {

83,976,653✔

5225

      pBlock->info.id.groupId = pInfo->groupId;

42,903,904✔

5226

      pOperator->resultInfo.totalRows += pBlock->info.rows;

42,903,904✔

5227

      pInfo->bGroupProcessed = true;

42,903,120✔

5228

      break;

42,902,969✔

5230

      if (pInfo->bNewFilesetEvent) {

41,072,749✔

5231

        stopDurationForGroupTableMergeScan(pOperator);

10,110,355✔

5232

        code = startDurationForGroupTableMergeScan(pOperator);

10,108,057✔

5233

        QUERY_CHECK_CODE(code, lino, _end);

10,106,887✔

5236

        stopGroupTableMergeScan(pOperator);

30,962,394✔

5237

        if (pInfo->tableEndIndex >= tableListSize - 1) {

30,962,283✔

5238

          setOperatorCompleted(pOperator);

24,513,748✔

5239

          break;

24,515,236✔

5242

        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

6,446,872✔

5243

        STableKeyInfo* tmp = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

6,446,872✔

5244

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

6,446,872✔

5245

        pInfo->groupId = tmp->groupId;

6,446,872✔

5246

        startGroupTableMergeScan(pOperator);

6,446,872✔

5247

        resetLimitInfoForNextGroup(&pInfo->limitInfo);

6,446,461✔

5252

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

67,416,412✔

5254

_end:

67,415,770✔

5255

  if (code != TSDB_CODE_SUCCESS) {

67,415,770✔

5260

    (*ppRes) = pBlock;

67,415,770✔

5263

  return code;

67,413,486✔

5272

void destroyTableMergeScanOperatorInfo(void* param) {

27,368,560✔

5273

  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;

27,368,560✔

5276

  if (pTableScanInfo->base.readerAPI.tsdReaderClose != NULL) {

27,368,560✔

5277

    pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);

27,370,618✔

5278

    pTableScanInfo->base.dataReader = NULL;

27,368,776✔

5281

  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {

27,371,011✔

5288

  tsortDestroySortHandle(pTableScanInfo->pSortHandle);

27,369,603✔

5289

  pTableScanInfo->pSortHandle = NULL;

27,363,536✔

5290

  taosHashCleanup(pTableScanInfo->mSkipTables);

27,363,002✔

5291

  pTableScanInfo->mSkipTables = NULL;

27,366,543✔

5292

  blockDataDestroy(pTableScanInfo->pSortInputBlock);

27,370,651✔

5293

  pTableScanInfo->pSortInputBlock = NULL;

27,364,308✔

5296

  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);

27,365,351✔

5298

  blockDataDestroy(pTableScanInfo->pResBlock);

27,366,681✔

5299

  pTableScanInfo->pResBlock = NULL;

27,367,998✔

5302

  blockDataDestroy(pTableScanInfo->pReaderBlock);

27,367,244✔

5303

  pTableScanInfo->pReaderBlock = NULL;

27,370,106✔

5304

  taosArrayDestroy(pTableScanInfo->pSortInfo);

27,370,106✔

5306

  stopSubTablesTableMergeScan(pTableScanInfo);

27,366,109✔

5308

  taosMemoryFreeClear(param);

27,364,090✔

5309

27,358,706✔

5311

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {

1,256,016✔

5312

  if (pOptr == NULL) {

1,256,016✔

5317

  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));

1,256,016✔

5318

  if (!execInfo) {

1,254,439✔

5321

  STableMergeScanInfo* pInfo = pOptr->info;

1,254,439✔

5322

  execInfo->blockRecorder = pInfo->base.readRecorder;

1,256,007✔

5323

  execInfo->sortExecInfo = pInfo->sortExecInfo;

1,256,390✔

5325

  *pOptrExplain = execInfo;

1,254,840✔

5326

  *len = sizeof(STableMergeScanExecInfo);

1,256,782✔

5328

  return TSDB_CODE_SUCCESS;

1,255,998✔

5331

static int32_t resetTableMergeScanOperatorState(SOperatorInfo* pOper) {

2,406,004✔

5332

  int32_t         code = TSDB_CODE_SUCCESS;

2,406,004✔

5333

  STableMergeScanInfo*   pInfo = pOper->info;

2,406,004✔

5334

  pOper->status = OP_NOT_OPENED;

2,406,004✔

5336

  pInfo->tableEndIndex = 0;

2,406,004✔

5337

  pInfo->tableStartIndex = 0;

2,406,004✔

5338

  pInfo->hasGroupId = false;

2,406,004✔

5339

  if (pInfo->base.readerAPI.tsdReaderClose) {

2,406,004✔

5340

    pInfo->base.readerAPI.tsdReaderClose(pInfo->base.dataReader);

2,406,004✔

5342

  pInfo->base.dataReader = NULL;

2,406,004✔

5343

  pInfo->base.scanFlag = MAIN_SCAN;

2,406,004✔

5345

  pInfo->base.limitInfo = (SLimitInfo){0};

2,406,004✔

5346

  pInfo->base.limitInfo.limit.limit = -1;

2,406,004✔

5347

  pInfo->base.limitInfo.slimit.limit = -1;

2,406,004✔

5349

  tableListDestroy(pInfo->base.pTableListInfo);

2,406,004✔

5350

  pInfo->base.pTableListInfo = tableListCreate();

2,406,004✔

5351

  if (!pInfo->base.pTableListInfo) {

2,406,004✔

5355

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

2,405,826✔

5357

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

2,405,826✔

5358

  code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

4,812,008✔

5360

                                  pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

4,812,008✔

5361

  if (code) {

2,406,004✔

5366

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

2,406,004✔

5367

  cleanupQueryTableDataCond(&pInfo->base.cond);

2,406,004✔

5368

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, &pInfo->base.readHandle, false);

2,406,004✔

5369

  if (code) {

2,405,742✔

5374

  tsortDestroySortHandle(pInfo->pSortHandle);

2,405,742✔

5375

  pInfo->pSortHandle = NULL;

2,405,742✔

5377

  taosHashCleanup(pInfo->mSkipTables);

2,405,742✔

5378

  pInfo->mSkipTables = NULL;

2,406,004✔

5379

  pOper->resultInfo.totalRows = 0;

2,406,004✔

5381

  pInfo->bGroupProcessed = false;

2,406,004✔

5382

  pInfo->bNewFilesetEvent = false;

2,405,742✔

5383

  pInfo->bNextDurationBlockEvent = false;

2,405,480✔

5384

  pInfo->rtnNextDurationBlocks = false;

2,405,742✔

5385

  pInfo->nextDurationBlocksIdx = 0;

2,406,004✔

5386

  pInfo->bSortRowId = false;

2,405,742✔

5388

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

2,406,004✔

5394

  pInfo->numNextDurationBlocks = 0;

2,405,742✔

5396

  stopSubTablesTableMergeScan(pInfo);

2,406,004✔

5397

  return code;

2,405,742✔

5400

int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

27,337,772✔

5403

  QRY_PARAM_CHECK(pOptrInfo);

27,337,772✔

5405

  int32_t              code = TSDB_CODE_SUCCESS;

27,350,044✔

5406

  int32_t              lino = 0;

27,350,044✔

5407

  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));

27,350,044✔

5408

  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

27,303,154✔

5409

  if (pInfo == NULL || pOperator == NULL) {

27,305,621✔

5414

  pOperator->pPhyNode = pTableScanNode;

27,307,171✔

5415

  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

27,308,527✔

5417

  int32_t numOfCols = 0;

27,325,344✔

5418

  code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,

27,334,412✔

5420

  QUERY_CHECK_CODE(code, lino, _error);

27,315,936✔

5422

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle, false);

27,315,936✔

5423

  QUERY_CHECK_CODE(code, lino, _error);

27,325,950✔

5425

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

27,325,950✔

5426

    SExprSupp* pSup = &pInfo->base.pseudoSup;

11,237,613✔

5427

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->pExprInfo, &pSup->numOfExprs);

11,239,272✔

5428

    QUERY_CHECK_CODE(code, lino, _error);

11,226,826✔

5430

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

11,226,826✔

5432

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

11,239,976✔

5435

  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};

27,311,346✔

5437

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

27,285,221✔

5442

  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;

27,285,221✔

5443

  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

27,329,086✔

5444

  pInfo->base.scanFlag = MAIN_SCAN;

27,323,549✔

5445

  pInfo->base.readHandle = *readHandle;

27,321,674✔

5447

  pInfo->readIdx = -1;

27,287,867✔

5449

  pInfo->base.limitInfo.limit.limit = -1;

27,277,053✔

5450

  pInfo->base.limitInfo.slimit.limit = -1;

27,338,068✔

5451

  pInfo->base.pTableListInfo = pTableListInfo;

27,305,703✔

5453

  pInfo->sample.sampleRatio = pTableScanNode->ratio;

27,291,169✔

5454

  pInfo->sample.seed = taosGetTimestampSec();

27,309,703✔

5456

  int32_t sortOrder = pInfo->base.cond.order;

27,334,826✔

5457

  if (pTableScanNode->scan.node.dynamicOp) {

27,324,778✔

5458

    memcpy(&pInfo->base.orgCond, &pInfo->base.cond, sizeof(SQueryTableDataCond));

95,536✔

5459

    sortOrder = pInfo->base.orgCond.order;

95,536✔

5460

    memset(&pInfo->base.cond, 0, sizeof(SQueryTableDataCond));

95,536✔

5463

  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,

27,323,057✔

5464

                            pTaskInfo->pStreamRuntimeInfo);

27,329,429✔

5465

  QUERY_CHECK_CODE(code, lino, _error);

27,292,145✔

5467

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

27,292,145✔

5469

  pInfo->mergeLimit = -1;

27,286,157✔

5470

  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;

27,292,799✔

5471

  if (hasLimit) {

27,312,691✔

5472

    pInfo->mergeLimit = pInfo->limitInfo.limit.offset != -1

4,631,371✔

5473

                            ? pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset

883,007✔

5474

                            : pInfo->limitInfo.limit.limit;

3,194,621✔

5475

    pInfo->mSkipTables = NULL;

2,313,910✔

5478

  initResultSizeInfo(&pOperator->resultInfo, 1024);

27,314,699✔

5479

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

27,314,461✔

5480

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

27,341,614✔

5481

  code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

27,320,172✔

5482

  QUERY_CHECK_CODE(code, lino, _error);

27,359,790✔

5483

  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {

27,359,790✔

5484

    pInfo->bSortRowId = true;

5,393,227✔

5486

    pInfo->bSortRowId = false;

21,967,545✔

5489

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

27,346,566✔

5490

  QUERY_CHECK_CODE(code, lino, _error);

27,363,357✔

5492

  code = generateSortByTsPkInfo(pInfo->base.matchInfo.pList, sortOrder, &pInfo->pSortInfo);

27,363,357✔

5493

  QUERY_CHECK_CODE(code, lino, _error);

27,328,360✔

5495

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pReaderBlock);

27,328,360✔

5496

  QUERY_CHECK_CODE(code, lino, _error);

27,343,779✔

5498

  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;

27,343,779✔

5500

  int32_t  rowSize = pInfo->pResBlock->info.rowSize;

27,352,854✔

5501

  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);

27,351,505✔

5503

  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);

27,299,710✔

5506

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pSortInputBlock);

27,352,920✔

5507

  QUERY_CHECK_CODE(code, lino, _error);

27,348,496✔

5509

  if (!tsExperimental) {

27,348,496✔

5512

    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

27,348,496✔

5515

  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,

27,349,254✔

5517

  pOperator->exprSupp.numOfExprs = numOfCols;

27,328,645✔

5519

  pOperator->fpSet = createOperatorFpSet(

27,341,185✔

5520

      optrDummyOpenFn, (pTableScanNode->paraTablesSort || pTableScanNode->scan.node.dynamicOp) ? doTableMergeScanParaSubTablesNext : doTableMergeScanNext, NULL,

27,349,583✔

5523

  setOperatorResetStateFn(pOperator, resetTableMergeScanOperatorState);

27,308,518✔

5524

  pOperator->cost.openCost = 0;

27,326,570✔

5526

  *pOptrInfo = pOperator;

27,328,318✔

5527

  return code;

27,329,122✔

5566

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {

30,214✔

5567

  if (scanCols != NULL) {

30,214✔

5568

    SNode* pNode = NULL;

15,843✔

5569

    FOREACH(pNode, scanCols) {

37,015✔

5570

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

21,172✔

5573

      STargetNode* targetNode = (STargetNode*)pNode;

21,172✔

5574

      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {

21,172✔

5577

      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);

21,172✔

5578

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

21,172✔

5579

        supp->dbNameSlotId = targetNode->slotId;

10,483✔

5580

      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

10,689✔

5581

        supp->stbNameSlotId = targetNode->slotId;

10,689✔

5585

  return TSDB_CODE_SUCCESS;

30,214✔

5588

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {

30,214✔

5589

  if (pseudoCols != NULL) {

30,214✔

5590

    SNode* pNode = NULL;

30,214✔

5591

    FOREACH(pNode, pseudoCols) {

60,461✔

5592

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

30,214✔

5595

      STargetNode* targetNode = (STargetNode*)pNode;

30,214✔

5596

      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {

30,214✔

5599

      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);

30,247✔

5600

      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {

30,247✔

5601

        supp->tbCountSlotId = targetNode->slotId;

30,247✔

5605

  return TSDB_CODE_SUCCESS;

30,247✔

5608

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {

30,214✔

5609

  if (groupTags != NULL) {

30,214✔

5610

    SNode* pNode = NULL;

15,843✔

5611

    FOREACH(pNode, groupTags) {

37,015✔

5612

      if (nodeType(pNode) != QUERY_NODE_COLUMN) {

21,172✔

5615

      SColumnNode* colNode = (SColumnNode*)pNode;

21,172✔

5616

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

21,172✔

5617

        supp->groupByDbName = true;

10,483✔

5619

      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

21,172✔

5620

        supp->groupByStbName = true;

10,689✔

5624

    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);

14,371✔

5625

    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);

14,404✔

5627

  return TSDB_CODE_SUCCESS;

30,247✔

5630

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,

30,214✔

5632

  int32_t code = 0;

30,214✔

5633

  code = tblCountScanGetInputs(groupTags, tableName, supp);

30,214✔

5634

  if (code != TSDB_CODE_SUCCESS) {

30,247✔

5639

  supp->dbNameSlotId = -1;

30,247✔

5640

  supp->stbNameSlotId = -1;

30,247✔

5641

  supp->tbCountSlotId = -1;

30,247✔

5643

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);

30,214✔

5644

  if (code != TSDB_CODE_SUCCESS) {

30,214✔

5649

  code = tblCountScanGetCountSlotId(pseudoCols, supp);

30,214✔

5650

  if (code != TSDB_CODE_SUCCESS) {

30,214✔

5654

  return code;

30,214✔

5657

int32_t createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,

30,037✔

5659

  QRY_PARAM_CHECK(pOptrInfo);

30,037✔

5661

  int32_t                      code = TSDB_CODE_SUCCESS;

30,070✔

5662

  int32_t                      lino = 0;

30,070✔

5663

  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;

30,070✔

5664

  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));

30,247✔

5665

  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

30,247✔

5666

  if (!pInfo || !pOperator) {

30,214✔

5671

  pInfo->readHandle = *readHandle;

30,214✔

5673

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

30,214✔

5674

  initResultSizeInfo(&pOperator->resultInfo, 1);

30,247✔

5675

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

30,247✔

5676

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

30,247✔

5678

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

30,214✔

5679

  QUERY_CHECK_CODE(code, lino, _error);

30,247✔

5681

  code = getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,

30,247✔

5684

  QUERY_CHECK_CODE(code, lino, _error);

30,247✔

5686

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,

30,247✔

5688

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScanNext, NULL, destoryTableCountScanOperator,

30,247✔

5690

  *pOptrInfo = pOperator;

30,214✔

5691

  return code;

30,247✔

5705

int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,

41,107✔

5707

  int32_t code = TSDB_CODE_SUCCESS;

41,107✔

5708

  int32_t lino = 0;

41,107✔

5709

  if (pSupp->dbNameSlotId != -1) {

41,107✔

5710

    QUERY_CHECK_CONDITION((strlen(dbName) > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

15,834✔

5711

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);

15,834✔

5712

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

15,834✔

5714

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

15,834✔

5715

    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

15,834✔

5717

    varDataSetLen(varDbName, strlen(dbName));

15,834✔

5718

    code = colDataSetVal(colInfoData, 0, varDbName, false);

15,834✔

5719

    QUERY_CHECK_CODE(code, lino, _end);

15,834✔

5722

  if (pSupp->stbNameSlotId != -1) {

41,107✔

5723

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);

20,513✔

5724

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

20,513✔

5725

    if (strlen(stbName) != 0) {

20,513✔

5726

      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

14,806✔

5727

      tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);

14,806✔

5728

      varDataSetLen(varStbName, strlen(stbName));

14,806✔

5729

      code = colDataSetVal(colInfoData, 0, varStbName, false);

14,806✔

5730

      QUERY_CHECK_CODE(code, lino, _end);

14,806✔

5736

  if (pSupp->tbCountSlotId != -1) {

41,074✔

5737

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);

41,074✔

5738

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

41,074✔

5739

    code = colDataSetVal(colInfoData, 0, (char*)&count, false);

41,074✔

5740

    QUERY_CHECK_CODE(code, lino, _end);

41,107✔

5742

  pRes->info.rows = 1;

41,107✔

5744

_end:

41,107✔

5745

  if (code != TSDB_CODE_SUCCESS) {

41,107✔

5748

  return code;

41,074✔

5751

static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {

11,448✔

5752

  STableCountScanSupp* pSupp = &pInfo->supp;

11,448✔

5753

  SSDataBlock*         pRes = pInfo->pRes;

11,448✔

5755

  size_t infodbTableNum;

11,448✔

5756

  getInfosDbMeta(NULL, &infodbTableNum);

11,448✔

5757

  infodbTableNum -= 1;

11,448✔

5758

  size_t perfdbTableNum;

11,448✔

5759

  getPerfDbMeta(NULL, &perfdbTableNum);

11,448✔

5761

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

11,448✔

5762

    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);

9,864✔

5763

    return (pRes->info.rows > 0) ? pRes : NULL;

9,864✔

5765

    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);

1,584✔

5766

    return (pRes->info.rows > 0) ? pRes : NULL;

1,584✔

5770

static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,

1,584✔

5772

  int32_t        code = TSDB_CODE_SUCCESS;

1,584✔

5773

  int32_t        lino = 0;

1,584✔

5774

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,584✔

5775

  if (strlen(pSupp->stbNameFilter) != 0) {

1,584✔

5777

    code = fillTableCountScanDataBlock(pSupp, "", "", 0, pRes);

534✔

5778

    QUERY_CHECK_CODE(code, lino, _end);

534✔

5779

  } else if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {

1,050✔

5780

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

178✔

5781

    QUERY_CHECK_CODE(code, lino, _end);

178✔

5782

  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

872✔

5785

  } else if (strlen(pSupp->dbNameFilter) == 0) {

872✔

5786

    code = fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);

694✔

5787

    QUERY_CHECK_CODE(code, lino, _end);

694✔

5790

    code = fillTableCountScanDataBlock(pSupp, pSupp->dbNameFilter, pSupp->stbNameFilter, 0, pRes);

178✔

5791

    QUERY_CHECK_CODE(code, lino, _end);

178✔

5794

_end:

178✔

5795

  if (code != TSDB_CODE_SUCCESS) {

1,584✔

5800

  setOperatorCompleted(pOperator);

1,584✔

5801

1,584✔

5803

static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,864✔

5806

  int32_t        code = TSDB_CODE_SUCCESS;

9,864✔

5807

  int32_t        lino = 0;

9,864✔

5808

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,864✔

5809

  if (pInfo->currGrpIdx == 0) {

9,864✔

5810

    uint64_t groupId = 0;

3,288✔

5811

    if (pSupp->groupByDbName) {

3,288✔

5812

      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));

2,249✔

5814

      groupId = calcGroupId("", 0);

1,039✔

5817

    pRes->info.id.groupId = groupId;

3,288✔

5818

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

3,288✔

5819

    QUERY_CHECK_CODE(code, lino, _end);

3,288✔

5820

  } else if (pInfo->currGrpIdx == 1) {

6,576✔

5821

    uint64_t groupId = 0;

3,288✔

5822

    if (pSupp->groupByDbName) {

3,288✔

5823

      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));

2,249✔

5825

      groupId = calcGroupId("", 0);

1,039✔

5828

    pRes->info.id.groupId = groupId;

3,288✔

5829

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

3,288✔

5830

    QUERY_CHECK_CODE(code, lino, _end);

3,288✔

5832

    setOperatorCompleted(pOperator);

3,288✔

5834

  pInfo->currGrpIdx++;

9,864✔

5836

_end:

9,864✔

5837

  if (code != TSDB_CODE_SUCCESS) {

9,864✔

5842

9,864✔

5844

static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

71,321✔

5845

  int32_t                      code = TSDB_CODE_SUCCESS;

71,321✔

5846

  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;

71,321✔

5847

  STableCountScanOperatorInfo* pInfo = pOperator->info;

71,354✔

5848

  STableCountScanSupp*         pSupp = &pInfo->supp;

71,354✔

5849

  SSDataBlock*                 pRes = pInfo->pRes;

71,354✔

5851

  blockDataCleanup(pRes);

71,321✔

5852

  QRY_PARAM_CHECK(ppRes);

71,354✔

5854

  if (pOperator->status == OP_EXEC_DONE) {

71,354✔

5855

    return code;

18,522✔

5858

  if (pInfo->readHandle.mnd != NULL) {

52,832✔

5859

    (*ppRes) = buildSysDbTableCount(pOperator, pInfo);

11,448✔

5860

    return code;

11,448✔

5863

  code = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);

41,384✔

5864

  if (code != TSDB_CODE_SUCCESS) {

41,351✔

5869

  if (pRes->info.rows > 0) {

41,351✔

5870

    *ppRes = pRes;

32,914✔

5873

  return code;

41,351✔

5876

static int32_t buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

41,384✔

5878

  int32_t        code = TSDB_CODE_SUCCESS;

41,384✔

5879

  int32_t        lino = 0;

41,384✔

5880

  const char*    db = NULL;

41,384✔

5881

  int32_t        vgId = 0;

41,384✔

5882

  char           dbName[TSDB_DB_NAME_LEN] = {0};

41,384✔

5883

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

41,384✔

5884

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

41,384✔

5887

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);

41,384✔

5888

  SName sn = {0};

41,384✔

5890

  code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

41,384✔

5891

  QUERY_CHECK_CODE(code, lino, _end);

41,351✔

5893

  code = tNameGetDbName(&sn, dbName);

41,351✔

5894

  QUERY_CHECK_CODE(code, lino, _end);

41,351✔

5896

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

41,351✔

5897

    code = buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);

28,597✔

5899

    code = buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);

12,787✔

5902

_end:

41,384✔

5903

  if (code != TSDB_CODE_SUCCESS) {

41,384✔

5908

  return code;

41,384✔

5911

static int32_t buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

28,564✔

5913

  int32_t        code = TSDB_CODE_SUCCESS;

28,564✔

5914

  int32_t        lino = 0;

28,564✔

5915

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

28,564✔

5916

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

28,564✔

5918

  if (pSupp->groupByStbName) {

28,564✔

5919

    if (pInfo->stbUidList == NULL) {

24,446✔

5920

      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));

8,437✔

5921

      QUERY_CHECK_NULL(pInfo->stbUidList, code, lino, _end, terrno);

8,437✔

5922

      code = pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList);

8,437✔

5923

      QUERY_CHECK_CODE(code, lino, _end);

8,437✔

5926

    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {

24,446✔

5927

      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);

14,806✔

5928

      code = buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);

14,806✔

5929

      QUERY_CHECK_CODE(code, lino, _end);

14,806✔

5931

      pInfo->currGrpIdx++;

14,806✔

5932

    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {

9,640✔

5933

      code = buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);

8,437✔

5934

      QUERY_CHECK_CODE(code, lino, _end);

8,437✔

5936

      pInfo->currGrpIdx++;

8,437✔

5938

      setOperatorCompleted(pOperator);

1,203✔

5941

    uint64_t groupId = calcGroupId(dbName, strlen(dbName));

4,118✔

5942

    pRes->info.id.groupId = groupId;

4,118✔

5944

    int64_t dbTableCount = 0;

4,118✔

5945

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);

4,118✔

5946

    code = fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);

4,118✔

5947

    QUERY_CHECK_CODE(code, lino, _end);

4,118✔

5948

    setOperatorCompleted(pOperator);

4,118✔

5951

_end:

28,564✔

5952

  if (code != TSDB_CODE_SUCCESS) {

28,564✔

5956

  return code;

28,564✔

5959

static int32_t buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

12,787✔

5961

  int32_t        code = TSDB_CODE_SUCCESS;

12,787✔

5962

  int32_t        lino = 0;

12,787✔

5963

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

12,787✔

5964

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

12,820✔

5966

  if (strlen(pSupp->dbNameFilter) != 0 && strcmp(dbName, pSupp->dbNameFilter) != 0) {

12,820✔

5970

  } else if (strlen(pSupp->stbNameFilter) != 0) {

12,820✔

5971

    uint64_t uid = 0;

6,276✔

5972

    code = pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid);

6,276✔

5973

    if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

6,276✔

5975

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, 0, pRes);

1,424✔

5976

      QUERY_CHECK_CODE(code, lino, _end);

1,424✔

5978

      QUERY_CHECK_CODE(code, lino, _end);

4,852✔

5980

      int64_t numOfChildTables = 0;

4,852✔

5981

      code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL, NULL);

4,852✔

5982

      QUERY_CHECK_CODE(code, lino, _end);

4,852✔

5984

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes);

4,852✔

5985

      QUERY_CHECK_CODE(code, lino, _end);

4,852✔

5988

    int64_t tbNumVnode = 0;

6,544✔

5989

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);

6,544✔

5990

    code = fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);

6,544✔

5991

    QUERY_CHECK_CODE(code, lino, _end);

6,544✔

5994

_end:

12,820✔

5995

  if (code != TSDB_CODE_SUCCESS) {

12,820✔

6000

  setOperatorCompleted(pOperator);

12,820✔

6001

  return code;

12,820✔

6004

static int32_t buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

8,437✔

6006

  int32_t code = TSDB_CODE_SUCCESS;

8,437✔

6007

  int32_t lino = 0;

8,437✔

6008

  char    fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

8,437✔

6009

  if (pSupp->groupByDbName) {

8,437✔

6010

    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");

4,116✔

6013

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

8,437✔

6014

  pRes->info.id.groupId = groupId;

8,437✔

6016

  int64_t numOfTables = 0;

8,437✔

6017

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);

8,437✔

6019

  if (numOfTables != 0) {

8,437✔

6020

    code = fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);

1,203✔

6023

_end:

7,234✔

6024

  if (code != TSDB_CODE_SUCCESS) {

8,437✔

6027

  return code;

8,437✔

6030

static int32_t buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

14,806✔

6032

  int32_t code = TSDB_CODE_SUCCESS;

14,806✔

6033

  int32_t lino = 0;

14,806✔

6034

  char    stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

14,806✔

6035

  code = pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);

14,806✔

6036

  QUERY_CHECK_CODE(code, lino, _end);

14,806✔

6038

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

14,806✔

6039

  if (pSupp->groupByDbName) {

14,806✔

6040

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));

6,876✔

6042

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));

7,930✔

6045

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

14,806✔

6046

  pRes->info.id.groupId = groupId;

14,806✔

6048

  int64_t ctbNum = 0;

14,806✔

6049

  code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL, NULL);

14,806✔

6050

  QUERY_CHECK_CODE(code, lino, _end);

14,806✔

6051

  code = fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);

14,806✔

6052

  QUERY_CHECK_CODE(code, lino, _end);

14,806✔

6054

_end:

14,806✔

6055

  if (code != TSDB_CODE_SUCCESS) {

14,806✔

6058

  return code;

14,806✔

6061

static void destoryTableCountScanOperator(void* param) {

30,247✔

6062

  STableCountScanOperatorInfo* pTableCountScanInfo = param;

30,247✔

6063

  blockDataDestroy(pTableCountScanInfo->pRes);

30,247✔

6065

  taosArrayDestroy(pTableCountScanInfo->stbUidList);

30,214✔

6066

  taosMemoryFreeClear(param);

30,214✔

6067

30,214✔

taosdata / TDengine / #4985

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4985

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