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

79

bool processBlockWithProbability(const SSampleExecInfo* pInfo) {

640,820,968✔

88

  return true;

640,820,968✔

92

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

69,334✔

93

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

69,334✔

96

69,334✔

98

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

646,601,014✔

99

  int32_t     code = TSDB_CODE_SUCCESS;

646,601,014✔

100

  STimeWindow w = {0};

646,601,014✔

103

  if (pInterval->interval == 0) {

646,732,259✔

104

    *overlap = false;

607,724,271✔

105

    return code;

607,714,194✔

108

  if (order == TSDB_ORDER_ASC) {

39,012,425✔

109

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

37,756,238✔

110

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

37,754,615✔

115

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

37,754,775✔

116

      *overlap = true;

32,934,137✔

117

      return code;

32,933,667✔

121

      getNextTimeWindow(pInterval, &w, order);

4,823,991✔

122

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

4,825,858✔

123

        break;

4,545,281✔

126

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

279,167✔

130

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

279,167✔

131

        *overlap = true;

279,167✔

132

        return code;

279,167✔

136

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

1,256,187✔

137

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

1,256,707✔

142

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

1,256,187✔

143

      *overlap = true;

1,081,261✔

144

      return code;

1,081,261✔

148

      getNextTimeWindow(pInterval, &w, order);

175,446✔

149

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

175,446✔

150

        break;

172,734✔

153

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

2,712✔

157

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

2,712✔

158

        *overlap = true;

2,712✔

159

        return code;

2,712✔

164

  *overlap = false;

4,718,015✔

165

  return code;

4,718,955✔

169

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

159,217,998✔

170

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

159,217,998✔

174

  int64_t buf[2] = {0};

159,224,471✔

175

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

159,224,713✔

177

  STableScanInfo* pTableScanInfo = pOperator->info;

159,219,977✔

179

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

159,203,044✔

182

  if (p1 == NULL) {

159,235,418✔

183

    return NULL;

50,883,287✔

186

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

108,352,131✔

187

  if (NULL == *pPage) {

108,332,109✔

191

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

108,339,519✔

194

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

13,173,968✔

195

  if (NULL == pTableScanInfo->pIgnoreTables) {

13,173,968✔

196

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

1,490,370✔

197

    pTableScanInfo->pIgnoreTables =

1,490,730✔

198

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

1,490,730✔

199

    if (NULL == pTableScanInfo->pIgnoreTables) {

1,490,730✔

204

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

13,175,754✔

206

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

13,173,642✔

210

  return TSDB_CODE_SUCCESS;

13,173,642✔

213

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

727,352,704✔

214

  STableScanInfo* pTableScanInfo = pOperator->info;

727,352,704✔

215

  int32_t         code = TSDB_CODE_SUCCESS;

727,594,525✔

217

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

727,594,525✔

218

    return TSDB_CODE_SUCCESS;

568,379,126✔

221

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

159,212,823✔

223

  SFilePage*  pPage = NULL;

159,236,866✔

224

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

159,238,548✔

226

  if (pRow == NULL) {

159,192,413✔

227

    return TSDB_CODE_SUCCESS;

50,878,120✔

230

  bool notLoadBlock = true;

108,314,293✔

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;

95,171,154✔

239

      break;

95,171,154✔

244

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

108,197,359✔

246

  if (notLoadBlock) {

108,344,662✔

247

    *status = FUNC_DATA_REQUIRED_NOT_LOAD;

13,180,062✔

248

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

13,176,971✔

251

  return code;

108,316,256✔

254

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

9,631,396✔

256

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

9,631,396✔

261

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

9,631,396✔

264

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

187,688,922✔

266

  SStorageAPI* pAPI = &pTaskInfo->storageAPI;

187,688,922✔

267

  bool         allColumnsHaveAgg = true;

187,740,786✔

268

  bool         hasNullSMA = false;

187,754,920✔

269

  if (pLoad != NULL) {

187,737,256✔

270

    *pLoad = false;

187,735,840✔

273

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

187,733,503✔

275

  if (code != TSDB_CODE_SUCCESS) {

187,704,123✔

279

  if (!allColumnsHaveAgg || hasNullSMA) {

187,704,123✔

280

    *pLoad = false;

155,117,582✔

282

    *pLoad = true;

32,586,541✔

285

  return code;

187,710,942✔

288

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

710,557,223✔

290

  int32_t    code = 0;

710,557,223✔

291

  SExprSupp* pSup = &pTableScanInfo->pseudoSup;

710,557,223✔

292

  if (pSup->numOfExprs > 0) {

710,795,633✔

293

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

284,631,612✔

296

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

284,688,514✔

301

  return code;

710,703,635✔

304

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

1,165,186,564✔

305

  SLimit*     pLimit = &pLimitInfo->limit;

1,165,186,564✔

306

  const char* id = GET_TASKID(pTaskInfo);

1,165,270,713✔

308

  if (pLimitInfo->remainOffset > 0) {

1,165,334,786✔

309

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

5,367,874✔

310

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

2,712,359✔

311

      blockDataEmpty(pBlock);

2,712,361✔

312

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

2,712,361✔

313

      return false;

2,712,361✔

315

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

2,655,513✔

316

      if (code != TSDB_CODE_SUCCESS) {

2,655,513✔

321

      pLimitInfo->remainOffset = 0;

2,655,513✔

325

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

1,162,606,194✔

327

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

27,058,752✔

328

    blockDataKeepFirstNRows(pBlock, keep);

27,059,135✔

330

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

27,057,644✔

331

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

27,058,576✔

332

    return true;

27,059,236✔

335

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

1,135,379,926✔

336

  return false;

1,135,500,351✔

339

static bool skipSetTagColumnData(SOperatorInfo* pOperator) {

758,441,110✔

340

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

1,384,658,294✔

341

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

626,089,248✔

344

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

775,354,017✔

346

  int32_t        code = TSDB_CODE_SUCCESS;

775,354,017✔

347

  int32_t        lino = 0;

775,354,017✔

348

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

775,354,017✔

349

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

775,485,100✔

350

  bool           loadSMA = false;

775,424,932✔

352

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

775,424,932✔

354

  pCost->totalBlocks += 1;

775,520,136✔

355

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

775,421,017✔

356

  *status = pTableScanInfo->dataBlockLoadFlag;

775,466,106✔

358

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

775,402,471✔

359

    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

128,574,674✔

361

    bool overlap = false;

646,750,179✔

363

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

646,860,153✔

364

    if (ret != TSDB_CODE_SUCCESS) {

646,480,446✔

367

    if (overlap) {

646,480,446✔

368

      (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

34,295,842✔

372

  SDataBlockInfo* pBlockInfo = &pBlock->info;

775,176,542✔

373

  taosMemoryFreeClear(pBlock->pBlockAgg);

775,316,916✔

375

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {

775,449,203✔

382

  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

775,100,221✔

383

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

21,324,746✔

386

    if (!skipSetTagColumnData(pOperator)) {

21,331,423✔

387

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

21,327,255✔

389

    pCost->skipBlocks += 1;

21,327,255✔

390

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

21,327,255✔

391

    return code;

21,327,255✔

392

  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {

753,750,339✔

393

    pCost->loadBlockStatis += 1;

59,173,938✔

394

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

59,140,523✔

395

    bool success = true;

59,140,523✔

396

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

59,154,895✔

397

    if (code) {

59,144,726✔

403

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

59,118,417✔

404

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

22,928,235✔

406

      if (!skipSetTagColumnData(pOperator)) {

22,929,027✔

407

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

5,806,846✔

409

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

22,951,507✔

410

      return code;

22,943,772✔

412

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

36,190,182✔

413

      *status = FUNC_DATA_REQUIRED_DATA_LOAD;

36,204,000✔

417

  if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {

730,792,037✔

424

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

730,852,677✔

425

    bool success = true;

128,578,488✔

426

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

128,590,013✔

427

    if (code) {

128,552,851✔

430

      QUERY_CHECK_CODE(code, lino, _end);

23,292✔

433

    if (success) {

128,576,143✔

434

      size_t size = taosArrayGetSize(pBlock->pDataBlock);

9,631,396✔

435

      bool   keep = false;

9,631,396✔

436

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

9,631,396✔

437

      if (code) {

9,631,484✔

443

      if (!keep) {

9,631,396✔

444

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

3,477,447✔

446

        pCost->filterOutBlocks += 1;

3,477,447✔

447

        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

3,477,535✔

448

        taosMemoryFreeClear(pBlock->pBlockAgg);

3,477,535✔

450

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

3,477,535✔

451

        return TSDB_CODE_SUCCESS;

3,477,535✔

457

  taosMemoryFreeClear(pBlock->pBlockAgg);

727,534,966✔

460

  code = doDynamicPruneDataBlock(pOperator, pBlockInfo, status);

727,357,225✔

461

  if (code) {

727,419,248✔

466

  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

727,419,248✔

467

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

13,179,627✔

469

    pCost->skipBlocks += 1;

13,183,086✔

470

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

13,179,539✔

472

    STableScanInfo* p1 = pOperator->info;

13,180,062✔

473

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

13,179,702✔

474

      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;

3,945,667✔

476

      *status = FUNC_DATA_REQUIRED_FILTEROUT;

9,233,512✔

478

    return TSDB_CODE_SUCCESS;

13,178,656✔

481

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

714,197,382✔

482

  pCost->loadBlocks += 1;

714,317,414✔

484

  SSDataBlock* p = NULL;

714,309,162✔

485

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

714,407,144✔

486

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

714,458,910✔

490

  if (!skipSetTagColumnData(pOperator)) {

714,458,910✔

491

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

635,913,480✔

492

    QUERY_CHECK_CODE(code, lino, _end);

635,823,878✔

496

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

714,294,044✔

498

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

714,420,324✔

499

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

125,078,590✔

500

    QUERY_CHECK_CODE(code, lino, _end);

125,052,477✔

502

    int64_t st = taosGetTimestampUs();

124,119,440✔

503

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

124,102,440✔

504

    pTableScanInfo->readRecorder.filterTime += el;

124,102,440✔

506

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

124,105,369✔

507

      pCost->filterOutBlocks += 1;

33,697,828✔

508

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

33,696,333✔

511

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

90,386,906✔

515

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

713,388,442✔

516

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

713,450,780✔

517

    setOperatorCompleted(pOperator);

3,372,396✔

520

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

713,450,780✔

522

_end:

714,448,655✔

523

  if (code != TSDB_CODE_SUCCESS) {

714,431,570✔

524

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

955,674✔

526

  return code;

714,330,366✔

529

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

69,334✔

530

  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

69,334✔

532

  switchCtxOrder(pCtx, numOfOutput);

69,334✔

533

  pTableScanInfo->cond.order = TSDB_ORDER_DESC;

69,334✔

536

69,334✔

599

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

284,927,495✔

601

  int32_t          code = TSDB_CODE_SUCCESS;

284,927,495✔

602

  int32_t          lino = 0;

284,927,495✔

603

  bool             freeReader = false;

284,927,495✔

604

  LRUHandle*       h = NULL;

284,927,495✔

605

  STableCachedVal  val = {0};

284,927,495✔

606

  SMetaReader      mr = {0};

284,927,495✔

607

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

284,929,469✔

608

  int32_t          insertRet = TAOS_LRU_STATUS_OK;

284,992,953✔

609

  STableCachedVal* pVal = NULL;

284,992,953✔

612

  if (numOfExpr <= 0) {

285,045,772✔

613

    return TSDB_CODE_SUCCESS;

264,821✔

619

  int32_t backupRows = pBlock->info.rows;

284,780,951✔

620

  pBlock->info.rows = rows;

284,771,667✔

626

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

284,827,882✔

627

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

284,817,983✔

628

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

284,857,119✔

629

    if (code != TSDB_CODE_SUCCESS) {

284,619,552✔

644

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

284,619,552✔

646

    val.pName = mr.me.name;

284,819,131✔

647

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

284,819,131✔

649

    freeReader = true;

284,819,131✔

691

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

884,530,326✔

692

    const SExprInfo* pExpr1 = &pExpr[j];

599,701,330✔

693

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

599,724,511✔

695

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

599,833,091✔

696

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

599,906,114✔

697

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

599,906,114✔

699

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

599,910,993✔

702

    if (fmIsScanPseudoColumnFunc(functionId)) {

599,952,239✔

703

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

95,034,627✔

704

      if (fType == FUNCTION_TYPE_TBNAME) {

95,038,719✔

705

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

95,038,719✔

706

        QUERY_CHECK_CODE(code, lino, _end);

95,026,783✔

715

      STagVal tagVal = {0};

504,642,469✔

716

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

504,677,447✔

717

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

504,894,672✔

719

      char* data = NULL;

504,890,798✔

720

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

504,890,798✔

721

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

499,018,915✔

723

        data = (char*)p;

5,891,834✔

726

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

504,860,638✔

727

      if (isNullVal) {

504,854,632✔

728

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

5,388,835✔

729

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

499,465,797✔

730

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

498,966,344✔

731

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

499,001,886✔

732

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

198,314,428✔

733

          if (tmp != NULL) {

198,385,797✔

734

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

198,385,797✔

735

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

198,405,949✔

737

            taosMemoryFree(tmp);

198,433,986✔

739

          taosMemoryFree(data);

198,434,303✔

741

        QUERY_CHECK_CODE(code, lino, _end);

498,904,047✔

743

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

3,118,765✔

744

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

2,613,430✔

745

          QUERY_CHECK_CODE(code, lino, _end);

2,615,378✔

752

  pBlock->info.rows = backupRows;

284,828,996✔

754

_end:

284,842,075✔

756

  if (NULL != pVal) {

284,837,410✔

765

  if (freeReader) {

284,837,410✔

766

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

284,775,707✔

768

  if (code != TSDB_CODE_SUCCESS) {

284,802,971✔

771

  return code;

284,755,538✔

774

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

95,036,537✔

776

  int32_t                     code = TSDB_CODE_SUCCESS;

95,036,537✔

777

  int32_t                     lino = 0;

95,036,537✔

778

  struct SScalarFuncExecFuncs fpSet = {0};

95,036,537✔

779

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

95,038,907✔

780

  QUERY_CHECK_CODE(code, lino, _end);

95,040,177✔

782

  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;

95,040,177✔

783

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

95,040,177✔

784

  STR_TO_VARSTR(buf, name)

95,041,737✔

786

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

95,024,279✔

788

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

95,036,403✔

789

  QUERY_CHECK_CODE(code, lino, _end);

95,044,465✔

791

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

95,044,465✔

792

  QUERY_CHECK_CODE(code, lino, _end);

95,031,841✔

794

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

95,031,841✔

795

  SScalarParam param = {.columnData = pColInfoData};

95,035,964✔

797

  if (fpSet.process != NULL) {

95,019,379✔

798

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

95,019,379✔

799

    QUERY_CHECK_CODE(code, lino, _end);

95,038,137✔

804

  colDataDestroy(&infoData);

95,038,137✔

806

_end:

95,023,442✔

807

  if (code != TSDB_CODE_SUCCESS) {

95,028,884✔

810

  return code;

95,028,884✔

879

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

256,663,649✔

880

  int32_t code = TSDB_CODE_SUCCESS;

256,663,649✔

881

  int32_t lino = 0;

256,663,649✔

882

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

256,663,649✔

883

  QUERY_CHECK_CODE(code, lino, _end);

256,647,394✔

885

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

256,647,394✔

886

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

256,709,710✔

887

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

256,542,561✔

888

  taosMemoryFreeClear(pInfo->pResBlock->pBlockAgg);

256,682,628✔

890

  if (!pInfo->needCountEmptyTable) {

256,625,296✔

891

    pInfo->countState = TABLE_COUNT_STATE_END;

237,237,677✔

893

    pInfo->countState = TABLE_COUNT_STATE_SCAN;

19,414,199✔

896

_end:

256,639,681✔

897

  if (code != TSDB_CODE_SUCCESS) {

256,639,681✔

900

  return code;

256,644,449✔

903

void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {

639,944,652✔

904

  if (pInfo->countState == TABLE_COUNT_STATE_END) {

639,944,652✔

905

    return;

567,096,207✔

907

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

73,084,400✔

908

    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;

230,206✔

910

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

72,852,019✔

911

    if (code != TSDB_CODE_SUCCESS) {

72,856,812✔

912

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

7,061,707✔

917

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

47,720,176✔

919

  blockDataEmpty(pBlock);

47,720,176✔

920

  pBlock->info.rows = 1;

47,720,176✔

921

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

47,720,176✔

922

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

47,720,176✔

923

  pBlock->info.blankFill = true;

47,720,696✔

926

  int32_t col_num = blockDataGetNumOfCols(pBlock);

47,720,176✔

927

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

197,306,575✔

928

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

149,586,919✔

933

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

47,719,656✔

935

  return pBlock;

47,717,106✔

938

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

45,859,448✔

939

  STableScanInfo* pTableScanInfo = pOperator->info;

45,859,448✔

941

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

45,859,968✔

943

  pOperator->resultInfo.totalRows++;

45,856,898✔

944

  return pBlock;

45,858,408✔

951

static int32_t prepareTableScan(SOperatorInfo* pOperator) {

870,204,941✔

952

  int32_t code = TSDB_CODE_SUCCESS;

870,204,941✔

953

  int32_t lino = 0;

870,204,941✔

955

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

870,204,941✔

956

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

870,274,383✔

957

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

870,395,178✔

958

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

35,952,372✔

959

         NOTIFY_TYPE_SCAN_PARAM &&

3,114,936✔

960

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

3,114,936✔

961

    STableScanOperatorParam* pNotify =

3,114,936✔

964

    STableScanInfo* pTableScanInfo = pOperator->info;

3,114,936✔

965

    SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

3,114,936✔

966

    SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

3,114,936✔

967

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

3,114,936✔

969

    QUERY_CHECK_CODE(code, lino, _end);

3,114,936✔

971

    if (!pGetParam->reUse) {

3,114,936✔

972

      freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

99,886✔

978

      pGetParam->reUse = false;

3,015,050✔

983

    pOperator->pOperatorGetParam = NULL;

3,114,936✔

986

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

870,397,637✔

987

_end:

268,601,707✔

988

  if (code != TSDB_CODE_SUCCESS) {

870,396,850✔

992

  return code;

870,372,210✔

995

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

832,048,807✔

996

  int32_t         code = TSDB_CODE_SUCCESS;

832,048,807✔

997

  int32_t         lino = 0;

832,048,807✔

998

  STableScanInfo* pTableScanInfo = pOperator->info;

832,048,807✔

999

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

832,131,652✔

1000

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

832,066,254✔

1001

  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;

831,972,229✔

1002

  bool            hasNext = false;

832,090,896✔

1003

  int64_t         st = taosGetTimestampUs();

832,064,068✔

1005

  QRY_PARAM_CHECK(ppRes);

832,064,068✔

1006

  pBlock->info.dataLoad = false;

832,094,009✔

1008

  while (true) {

41,769,497✔

1009

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

873,873,842✔

1010

    if (code != TSDB_CODE_SUCCESS) {

873,712,967✔

1011

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

13✔

1015

    if (!hasNext) {

873,599,223✔

1016

      break;

232,750,464✔

1019

    if (isTaskKilled(pTaskInfo)) {

640,848,759✔

1020

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

30,269✔

1021

      code = pTaskInfo->code;

30,269✔

1022

      QUERY_CHECK_CODE(code, lino, _end);

30,269✔

1025

    if (pOperator->status == OP_EXEC_DONE) {

641,137,483✔

1026

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

93,240✔

1027

      break;

93,240✔

1031

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

640,951,503✔

1032

    if (!processThisBlock) {

640,876,014✔

1036

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

640,876,014✔

1037

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

641,019,956✔

1040

    uint32_t status = 0;

641,052,857✔

1041

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

641,128,675✔

1042

    QUERY_CHECK_CODE(code, lino, _end);

640,921,778✔

1044

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

639,983,672✔

1045

      break;

3,945,417✔

1049

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

636,038,255✔

1050

      continue;

41,769,497✔

1053

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

594,353,769✔

1054

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

594,386,145✔

1056

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

594,424,703✔

1057

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

594,426,704✔

1059

    (*ppRes) = pBlock;

594,404,650✔

1060

    return code;

594,283,866✔

1063

_end:

237,768,344✔

1064

  if (code != TSDB_CODE_SUCCESS) {

237,778,737✔

1065

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

981,543✔

1066

    pTaskInfo->code = code;

981,543✔

1067

    T_LONG_JMP(pTaskInfo->env, code);

968,375✔

1069

  return code;

236,797,194✔

1072

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

880,768,243✔

1073

  int32_t         code = TSDB_CODE_SUCCESS;

880,768,243✔

1074

  int32_t         lino = 0;

880,768,243✔

1075

  STableScanInfo* pTableScanInfo = pOperator->info;

880,768,243✔

1076

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

880,854,210✔

1077

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

880,816,646✔

1079

  QRY_PARAM_CHECK(pBlock);

880,779,693✔

1082

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

880,757,606✔

1083

    return code;

2,966,480✔

1087

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

1,098,030,142✔

1088

    SSDataBlock* p = NULL;

777,166,150✔

1089

    code = doTableScanImplNext(pOperator, &p);

777,140,570✔

1090

    QUERY_CHECK_CODE(code, lino, _end);

776,052,189✔

1092

    if (p != NULL) {

776,052,189✔

1093

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

555,811,135✔

1094

      *pBlock = p;

555,820,129✔

1095

      return code;

555,845,304✔

1098

    pTableScanInfo->scanTimes += 1;

220,241,054✔

1099

    taosHashClear(pTableScanInfo->pIgnoreTables);

220,252,164✔

1101

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

220,232,935✔

1102

      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

87,329✔

1103

      pTableScanInfo->base.scanFlag = MAIN_SCAN;

87,329✔

1104

      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

87,329✔

1105

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

87,329✔

1108

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

87,329✔

1109

      QUERY_CHECK_CODE(code, lino, _end);

75,476✔

1113

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

320,909,625✔

1114

  if (pTableScanInfo->scanTimes < total) {

320,910,842✔

1115

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

54,960,305✔

1116

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

69,334✔

1117

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

69,334✔

1118

      QUERY_CHECK_CODE(code, lino, _end);

69,334✔

1119

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

69,334✔

1122

    while (pTableScanInfo->scanTimes < total) {

71,510,627✔

1123

      SSDataBlock* p = NULL;

54,963,915✔

1124

      code = doTableScanImplNext(pOperator, &p);

54,952,230✔

1125

      QUERY_CHECK_CODE(code, lino, _end);

54,917,404✔

1127

      if (p != NULL) {

54,917,404✔

1128

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

38,371,839✔

1129

        *pBlock = p;

38,401,294✔

1130

        return code;

38,400,589✔

1133

      pTableScanInfo->scanTimes += 1;

16,545,565✔

1134

      taosHashClear(pTableScanInfo->pIgnoreTables);

16,546,957✔

1136

      if (pTableScanInfo->scanTimes < total) {

16,546,280✔

1147

  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {

282,479,475✔

1148

    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;

64,515,837✔

1149

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

64,515,219✔

1150

      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {

122,561✔

1151

        pTableScanInfo->countState = TABLE_COUNT_STATE_PROCESSED;

4,088✔

1153

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

4,088✔

1155

        if (NULL == pStart) {

4,088✔

1159

        *pBlock = getBlockForEmptyTable(pOperator, pStart);

4,088✔

1160

        return code;

4,088✔

1163

      int32_t numOfTables = 0;

64,393,408✔

1164

      code = tableListGetSize(pTableListInfo, &numOfTables);

64,393,408✔

1165

      QUERY_CHECK_CODE(code, lino, _end);

64,393,928✔

1167

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

64,393,928✔

1169

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

62,597,530✔

1170

        if (pIte != NULL) {

62,595,642✔

1171

          size_t        keySize = 0;

45,855,360✔

1172

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

45,855,360✔

1173

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

45,855,360✔

1174

          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);

45,855,880✔

1175

          markGroupProcessed(pTableScanInfo, *pGroupId);

45,855,880✔

1176

          *pBlock = getBlockForEmptyTable(pOperator, &info);

45,855,880✔

1178

          return code;

45,854,370✔

1182

    pTableScanInfo->countState = TABLE_COUNT_STATE_END;

18,655,383✔

1185

_end:

217,986,822✔

1186

  if (code != TSDB_CODE_SUCCESS) {

236,630,172✔

1191

  return code;

236,630,172✔

1194

static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {

1,003,281✔

1195

  STableScanInfo*          pInfo = pOperator->info;

1,003,281✔

1196

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

1,003,281✔

1197

  int32_t                  code = 0;

1,003,281✔

1198

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

1,003,281✔

1199

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

1,003,281✔

1200

  int32_t                  num = taosArrayGetSize(pParam->pUidList);

1,003,281✔

1201

  if (num <= 0) {

1,003,281✔

1206

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

1,003,281✔

1210

  if (pParam->tableSeq) {

1,003,677✔

1211

    pListInfo->oneTableForEachGroup = true;

725,796✔

1212

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

725,796✔

1218

    pListInfo->oneTableForEachGroup = false;

277,881✔

1219

    pListInfo->numOfOuputGroups = 1;

277,881✔

1221

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

1,003,677✔

1224

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

721,140✔

1225

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

721,140✔

1228

  STableKeyInfo info = {.groupId = 0};

1,003,677✔

1229

  int32_t       tableIdx = 0;

1,003,677✔

1230

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

9,274,479✔

1231

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

8,270,802✔

1232

    if (!pUid) {

8,269,360✔

1237

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

8,269,360✔

1238

      if (TSDB_CODE_DUP_KEY == terrno) {

6,906,485✔

1239

        continue;

6,907,008✔

1244

    info.uid = *pUid;

1,363,398✔

1245

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

1,363,398✔

1246

    if (p == NULL) {

1,363,271✔

1250

    tableIdx++;

1,363,271✔

1251

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

1,363,271✔

1254

  return code;

1,003,677✔

1257

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

238,515,522✔

1259

  const char* idStr = GET_TASKID(pTaskInfo);

238,515,522✔

1260

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

238,551,195✔

1261

  if (code) {

238,489,456✔

1266

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

238,489,456✔

1271

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

476,994,643✔

1272

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

238,472,138✔

1273

  if (code) {

238,440,715✔

1274

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

2,340✔

1277

  return code;

238,401,244✔

1280

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

343,800,516✔

1281

  SColIdPair* node1 = (SColIdPair*)elem1;

343,800,516✔

1282

  SColIdPair* node2 = (SColIdPair*)elem2;

343,800,516✔

1284

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

343,800,516✔

1285

    return 0;

112,766✔

1288

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

343,698,639✔

1291

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

453,003,825✔

1292

  SColIdSlotIdPair* node1 = (SColIdSlotIdPair*)elem1;

453,003,825✔

1293

  SColIdSlotIdPair* node2 = (SColIdSlotIdPair*)elem2;

453,003,825✔

1295

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

453,003,825✔

1296

    return 0;

172,262✔

1299

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

452,846,622✔

1313

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

101,306,265✔

1314

  if (pList == NULL) {

101,306,265✔

1318

  int32_t left = 0;

101,306,265✔

1319

  int32_t right = taosArrayGetSize(pList) - 1;

101,306,265✔

1320

  while (left <= right) {

312,395,612✔

1321

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

312,395,612✔

1322

    SColMatchItem* pItem = taosArrayGet(pList, mid);

312,395,612✔

1323

    if (pItem->colId == colId) {

312,395,612✔

1325

      while (mid > 0) {

101,306,265✔

1326

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

75,608,876✔

1327

        if (pPrev->colId != colId) {

75,608,876✔

1328

          break;

75,608,876✔

1332

      return taosArrayGet(pList, mid);

101,306,265✔

1335

    if (pItem->colId < colId) {

211,089,347✔

1336

      left = mid + 1;

99,487,305✔

1338

      right = mid - 1;

111,602,042✔

1345

static bool isNewScanParam(STableScanOperatorParam* pParam) {

43,988,664✔

1346

  return pParam->isNewParam;

43,988,664✔

1349

static int32_t createVTableScanInfoFromBatchParam(SOperatorInfo* pOperator) {

87,681,136✔

1350

  int32_t                  code = 0;

87,681,136✔

1351

  int32_t                  lino = 0;

87,681,136✔

1352

  STableScanInfo*          pInfo = pOperator->info;

87,681,136✔

1353

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

87,704,959✔

1354

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

87,704,324✔

1355

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

87,712,462✔

1356

  SMetaReader              orgTable = {0};

87,711,515✔

1357

  SMetaReader              superTable = {0};

87,742,874✔

1358

  SSchemaWrapper*          schema = NULL;

87,717,385✔

1359

  SArray*                  pColArray = NULL;

87,717,385✔

1360

  SArray*                  pBlockColArray = NULL;

87,717,385✔

1361

  int32_t                  num = 0;

87,717,385✔

1362

  STableKeyInfo*           pList = NULL;

87,718,915✔

1366

  cleanupQueryTableDataCond(&pInfo->base.cond);

87,713,057✔

1368

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

87,729,039✔

1369

  if (pGetParam &&

87,732,085✔

1370

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

6,134,475✔

1371

    STableScanOperatorParam* pParam =

6,123,890✔

1373

    pInfo->pBatchColMap = pParam->pBatchTbInfo;

6,139,061✔

1374

    pParam->pBatchTbInfo = NULL;

6,124,359✔

1375

    pInfo->cachedTimeWindow = pParam->window;

6,105,217✔

1376

    pInfo->currentBatchIdx = 0;

6,129,578✔

1377

    pInfo->lastBatchIdx = -1;

6,118,276✔

1378

    pInfo->cachedTagList = pParam->pTagList;

6,101,505✔

1379

    pParam->pTagList = NULL;

6,123,661✔

1380

    pInfo->cachedGroupId = pParam->groupid;

6,118,729✔

1382

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

6,135,292✔

1384

    taosHashClear(pInfo->pIgnoreTables);

6,135,292✔

1386

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

6,129,136✔

1387

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

5,627,232✔

1388

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

70,365,107✔

1389

        SqlFunctionCtx ctx = pSup->pCtx[i];

64,753,553✔

1390

        ctx.skipDynDataCheck = false;

64,736,958✔

1395

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

87,715,358✔

1396

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

87,752,962✔

1398

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

87,752,962✔

1399

    isNewTable = false;

77,364,469✔

1401

    isNewTable = true;

10,380,445✔

1402

    pInfo->lastBatchIdx = pInfo->currentBatchIdx;

10,380,445✔

1405

  if (isNewTable) {

87,718,165✔

1406

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

10,375,587✔

1407

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

10,390,931✔

1408

    pAPI->metaReaderFn.readerReleaseLock(&orgTable);

10,371,788✔

1409

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

10,385,091✔

1410

    QUERY_CHECK_CODE(code, lino, _return);

10,367,517✔

1411

    switch (orgTable.me.type) {

10,367,517✔

1412

      case TSDB_CHILD_TABLE:

591,966✔

1413

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

591,966✔

1414

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

591,966✔

1415

        pAPI->metaReaderFn.readerReleaseLock(&superTable);

591,966✔

1416

        QUERY_CHECK_CODE(code, lino, _return);

591,966✔

1417

        schema = &superTable.me.stbEntry.schemaRow;

591,966✔

1418

        break;

591,966✔

1419

      case TSDB_NORMAL_TABLE:

9,744,046✔

1420

        schema = &orgTable.me.ntbEntry.schemaRow;

9,744,046✔

1421

        break;

9,744,046✔

1422

      default:

31,505✔

1423

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

31,505✔

1428

    pListInfo->oneTableForEachGroup = true;

10,336,012✔

1429

    taosHashClear(pListInfo->map);

10,371,659✔

1430

    taosArrayClear(pListInfo->pTableList);

10,395,997✔

1432

    uint64_t      pUid = orgTable.me.uid;

10,395,474✔

1433

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

10,395,834✔

1434

    int32_t       tableIdx = 0;

10,395,834✔

1435

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

10,396,357✔

1436

    QUERY_CHECK_CODE(code, lino, _return);

10,395,834✔

1437

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

20,787,418✔

1438

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

10,391,584✔

1440

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

10,391,584✔

1441

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

10,388,127✔

1442

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

10,388,127✔

1443

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

10,385,481✔

1446

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

10,385,481✔

1447

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

10,392,565✔

1449

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

38,586,270✔

1450

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

28,193,111✔

1451

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

221,110,402✔

1452

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

220,754,572✔

1453

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

28,202,272✔

1454

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

28,200,025✔

1455

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

28,200,025✔

1457

          break;

28,194,630✔

1462

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

48,980,261✔

1463

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

38,588,282✔

1464

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

225,245,706✔

1465

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

224,985,486✔

1466

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

225,051,117✔

1467

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

38,578,834✔

1468

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

38,593,572✔

1469

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

38,593,572✔

1471

          break;

38,575,708✔

1476

    taosArraySort(pColArray, compareColIdPair);

10,392,071✔

1477

    taosArraySort(pBlockColArray, compareColIdSlotIdPair);

10,384,678✔

1478

    if (pInfo->pBlockColMap) {

10,389,608✔

1479

      taosArrayDestroy(pInfo->pBlockColMap);

7,378,073✔

1480

      pInfo->pBlockColMap = NULL;

7,369,461✔

1482

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

10,379,002✔

1483

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

10,390,702✔

1486

    taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

10,384,351✔

1487

    taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

10,384,715✔

1489

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

10,388,250✔

1490

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

8,765,897✔

1491

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

108,127,352✔

1492

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

99,385,169✔

1493

        bool           needScan = false;

99,385,446✔

1494

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

200,203,246✔

1495

          SFunctParam pParam = ctx->param[j];

100,743,666✔

1496

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

484,138,043✔

1497

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

406,954,397✔

1498

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

407,949,362✔

1499

              needScan |= true;

25,080,407✔

1500

              break;

25,080,407✔

1504

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

99,367,843✔

1505

        if (!needScan) {

99,425,402✔

1506

          ctx->skipDynDataCheck = true;

74,356,900✔

1508

          ctx->skipDynDataCheck = false;

25,068,502✔

1515

    pColArray = pInfo->lastColArray;

77,342,578✔

1516

    pBlockColArray = pInfo->lastBlockColArray;

77,345,686✔

1519

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

87,752,263✔

1520

  QUERY_CHECK_CODE(code, lino, _return);

87,760,638✔

1522

  if (pInfo->pResBlock) {

87,760,638✔

1523

    blockDataDestroy(pInfo->pResBlock);

81,619,592✔

1524

    pInfo->pResBlock = NULL;

81,583,754✔

1527

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

87,740,815✔

1535

  if (isNewTable) {

87,695,958✔

1536

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

10,379,557✔

1537

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

8,918,736✔

1538

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

8,903,121✔

1540

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

1,475,717✔

1541

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

1,463,754✔

1544

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

77,316,401✔

1545

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

1,958,015✔

1547

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

75,363,856✔

1551

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

87,713,142✔

1552

  pInfo->currentGroupId = 0;

87,712,223✔

1553

  pInfo->ignoreTag = true;

87,722,614✔

1555

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

87,700,990✔

1556

  QUERY_CHECK_CODE(code, lino, _return);

87,738,451✔

1558

  if (isNewTable) {

87,738,451✔

1559

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

10,384,196✔

1560

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

7,378,846✔

1562

    pInfo->base.dataReader = NULL;

10,362,983✔

1564

    taosRLockLatch(&pTaskInfo->lock);

10,367,423✔

1565

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

10,395,051✔

1566

    taosRUnLockLatch(&pTaskInfo->lock);

10,352,781✔

1567

    QUERY_CHECK_CODE(code, lino, _return);

10,351,478✔

1569

    pInfo->newReader = true;

10,351,478✔

1570

    if (pInfo->lastColArray) {

10,357,030✔

1571

      taosArrayDestroy(pInfo->lastColArray);

4,250,048✔

1572

      pInfo->lastColArray = NULL;

4,250,571✔

1574

    if (pInfo->lastBlockColArray) {

10,342,291✔

1575

      taosArrayDestroy(pInfo->lastBlockColArray);

4,249,525✔

1576

      pInfo->lastBlockColArray = NULL;

4,249,961✔

1578

    pInfo->lastColArray = pColArray;

10,325,756✔

1579

    pInfo->lastBlockColArray = pBlockColArray;

10,320,585✔

1581

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

77,354,255✔

1582

    QUERY_CHECK_CODE(code, lino, _return);

77,357,253✔

1584

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

77,357,253✔

1585

    pInfo->newReader = false;

77,364,232✔

1586

    pInfo->scanTimes = 0;

77,364,829✔

1589

  pOperator->status = OP_OPENED;

87,728,418✔

1591

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

87,725,137✔

1595

  pInfo->currentGroupId = -1;

87,727,431✔

1596

_return:

87,728,301✔

1597

  if (code) {

87,686,583✔

1600

  pAPI->metaReaderFn.clearReader(&superTable);

87,686,583✔

1601

  pAPI->metaReaderFn.clearReader(&orgTable);

87,696,559✔

1602

  return code;

87,730,422✔

1605

static int32_t createVTableScanInfoFromParam(SOperatorInfo* pOperator) {

25,697,389✔

1606

  int32_t                  code = 0;

25,697,389✔

1607

  int32_t                  lino = 0;

25,697,389✔

1608

  STableScanInfo*          pInfo = pOperator->info;

25,697,389✔

1609

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

25,697,389✔

1610

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

25,697,389✔

1611

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

25,697,389✔

1612

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

25,697,389✔

1613

  SMetaReader              orgTable = {0};

25,697,389✔

1614

  SMetaReader              superTable = {0};

25,697,389✔

1615

  SSchemaWrapper*          schema = NULL;

25,697,389✔

1616

  SArray*                  pColArray = NULL;

25,697,389✔

1617

  SArray*                  pBlockColArray = NULL;

25,697,389✔

1618

  int32_t                  num = 0;

25,697,389✔

1619

  STableKeyInfo*           pList = NULL;

25,697,389✔

1621

  SArray*                  pMatchList = NULL;

25,697,389✔

1623

  cleanupQueryTableDataCond(&pInfo->base.cond);

25,697,389✔

1625

  pOrgTbInfo = pParam->pOrgTbInfo;

25,697,389✔

1627

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

25,697,389✔

1629

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

25,697,389✔

1630

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

25,697,389✔

1631

  pAPI->metaReaderFn.readerReleaseLock(&orgTable);

25,697,389✔

1632

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

25,697,389✔

1633

  QUERY_CHECK_CODE(code, lino, _return);

25,697,389✔

1634

  switch (orgTable.me.type) {

25,697,389✔

1635

    case TSDB_CHILD_TABLE:

7,138,336✔

1636

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

7,138,336✔

1637

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

7,138,336✔

1638

      pAPI->metaReaderFn.readerReleaseLock(&superTable);

7,138,336✔

1639

      QUERY_CHECK_CODE(code, lino, _return);

7,138,336✔

1640

      schema = &superTable.me.stbEntry.schemaRow;

7,138,336✔

1641

      break;

7,138,336✔

1642

    case TSDB_NORMAL_TABLE:

18,559,053✔

1643

      schema = &orgTable.me.ntbEntry.schemaRow;

18,559,053✔

1644

      break;

18,559,053✔

1651

  pListInfo->oneTableForEachGroup = true;

25,697,389✔

1652

  taosHashClear(pListInfo->map);

25,697,389✔

1653

  taosArrayClear(pListInfo->pTableList);

25,697,389✔

1655

  uint64_t      pUid = orgTable.me.uid;

25,697,389✔

1656

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

25,697,389✔

1657

  int32_t       tableIdx = 0;

25,697,389✔

1658

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

25,697,389✔

1659

  QUERY_CHECK_CODE(code, lino, _return);

25,697,389✔

1660

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

51,394,778✔

1661

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

25,697,389✔

1663

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

25,697,389✔

1664

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

25,697,389✔

1665

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

25,697,389✔

1666

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

25,697,389✔

1667

  if (pInfo->pBlockColMap) {

25,697,389✔

1668

    taosArrayDestroy(pInfo->pBlockColMap);

23,025,756✔

1669

    pInfo->pBlockColMap = NULL;

23,025,756✔

1671

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

25,697,389✔

1672

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

25,697,389✔

1675

  SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID,

25,697,389✔

1679

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

25,697,389✔

1681

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

101,306,265✔

1682

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

75,608,876✔

1683

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

636,922,123✔

1684

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

636,922,123✔

1685

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

75,608,876✔

1686

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

75,608,876✔

1687

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

75,608,876✔

1688

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

75,608,876✔

1689

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

75,608,876✔

1690

        break;

75,608,876✔

1695

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

25,697,389✔

1696

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

25,697,389✔

1701

      pMatchList = pInfo->base.matchInfo.pList;

25,697,389✔

1706

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

127,003,002✔

1707

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

101,306,265✔

1708

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

101,306,265✔

1709

    if (pItem) {

101,306,265✔

1710

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

101,306,265✔

1711

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

652✔

1713

        code = TSDB_CODE_VTABLE_COLUMN_TYPE_MISMATCH;

652✔

1714

        goto _return;

652✔

1716

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

101,305,613✔

1717

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

101,305,613✔

1718

      if (i > 0) {

101,305,613✔

1719

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

151,216,448✔

1724

  taosArraySort(pColArray, compareColIdPair);

25,696,737✔

1725

  taosArraySort(pBlockColArray, compareColIdSlotIdPair);

25,696,737✔

1726

  taosArraySort(pInfo->pBlockColMap, compareColIdSlotIdPair);

25,696,737✔

1728

  taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

25,696,737✔

1729

  taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

25,696,737✔

1731

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

25,696,737✔

1732

  QUERY_CHECK_CODE(code, lino, _return);

25,696,737✔

1734

  if (pInfo->pResBlock) {

25,696,737✔

1735

    blockDataDestroy(pInfo->pResBlock);

18,294,291✔

1736

    pInfo->pResBlock = NULL;

18,294,291✔

1739

  if (isNewScanParam(pParam)) {

25,696,737✔

1740

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

7,410,406✔

1741

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

7,410,406✔

1742

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

7,410,406✔

1748

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

18,286,331✔

1750

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

25,696,737✔

1751

  pInfo->currentGroupId = 0;

25,696,737✔

1752

  pInfo->ignoreTag = true;

25,696,737✔

1754

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

25,696,737✔

1755

  QUERY_CHECK_CODE(code, lino, _return);

25,696,737✔

1757

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

25,696,737✔

1758

  if (reader) {

25,696,737✔

1759

    if (isNewScanParam(pParam)) {

18,291,927✔

1760

      pAPI->tsdReader.tsdReaderClose(*reader);

5,596✔

1761

      pInfo->base.dataReader = NULL;

5,596✔

1763

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

5,596✔

1764

      QUERY_CHECK_CODE(code, lino, _return);

5,596✔

1766

      taosRLockLatch(&pTaskInfo->lock);

5,596✔

1767

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

5,596✔

1768

      taosRUnLockLatch(&pTaskInfo->lock);

5,596✔

1769

      QUERY_CHECK_CODE(code, lino, _return);

5,596✔

1771

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

5,596✔

1772

      QUERY_CHECK_CODE(code, lino, _return);

5,596✔

1773

      pInfo->newReader = true;

5,596✔

1775

      pInfo->base.dataReader = *reader;

18,286,331✔

1776

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

18,286,331✔

1777

      QUERY_CHECK_CODE(code, lino, _return);

18,286,331✔

1779

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

18,286,331✔

1780

      pInfo->newReader = false;

18,286,331✔

1781

      pInfo->scanTimes = 0;

18,286,331✔

1784

    pInfo->base.dataReader = NULL;

7,404,810✔

1786

    taosRLockLatch(&pTaskInfo->lock);

7,404,810✔

1787

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

7,404,810✔

1788

    taosRUnLockLatch(&pTaskInfo->lock);

7,404,810✔

1789

    QUERY_CHECK_CODE(code, lino, _return);

7,404,810✔

1791

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

7,404,810✔

1792

    QUERY_CHECK_CODE(code, lino, _return);

7,404,810✔

1793

    pInfo->newReader = true;

7,404,810✔

1795

  pOperator->status = OP_OPENED;

25,696,737✔

1797

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

25,696,737✔

1801

  pInfo->currentGroupId = -1;

25,696,737✔

1802

_return:

25,697,389✔

1803

  if (code) {

25,697,389✔

1804

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

652✔

1806

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

25,697,389✔

1809

  taosArrayDestroy(pColArray);

25,697,389✔

1810

  taosArrayDestroy(pBlockColArray);

25,697,389✔

1811

  pAPI->metaReaderFn.clearReader(&superTable);

25,697,389✔

1812

  pAPI->metaReaderFn.clearReader(&orgTable);

25,697,389✔

1813

  return code;

25,697,389✔

1816

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

233,706,016✔

1817

  int32_t         code = TSDB_CODE_SUCCESS;

233,706,016✔

1818

  int32_t         lino = 0;

233,706,016✔

1819

  STableScanInfo* pInfo = pOperator->info;

233,706,016✔

1820

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

233,696,095✔

1821

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

233,683,838✔

1822

  int32_t         numOfTables = 0;

233,667,253✔

1824

  QRY_PARAM_CHECK(pResult);

233,661,233✔

1826

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

233,682,516✔

1827

  QUERY_CHECK_CODE(code, lino, _end);

233,715,638✔

1829

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

233,715,638✔

1830

    setOperatorCompleted(pOperator);

215,551,569✔

1831

    if (pOperator->dynamicTask) {

215,555,218✔

1832

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

1,001,721✔

1833

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

1,001,721✔

1835

    return code;

215,561,356✔

1839

  pOperator->status = OP_OPENED;

18,058,420✔

1840

  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);

18,142,597✔

1842

  int32_t        num = 0;

18,135,205✔

1843

  STableKeyInfo* pList = NULL;

18,135,368✔

1844

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

18,136,808✔

1845

  QUERY_CHECK_CODE(code, lino, _end);

18,155,273✔

1847

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

18,155,273✔

1848

  QUERY_CHECK_CODE(code, lino, _end);

18,129,786✔

1850

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

18,129,786✔

1851

  QUERY_CHECK_CODE(code, lino, _end);

18,146,952✔

1852

  pInfo->scanTimes = 0;

18,146,952✔

1854

  code = doGroupedTableScan(pOperator, pResult);

18,153,621✔

1855

  QUERY_CHECK_CODE(code, lino, _end);

18,173,981✔

1857

  if (*pResult != NULL) {

18,173,981✔

1858

    if (pOperator->dynamicTask) {

14,838,820✔

1859

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

14,577,762✔

1861

    return code;

14,838,960✔

1864

_end:

3,335,161✔

1865

  if (code != TSDB_CODE_SUCCESS) {

3,334,940✔

1869

  return code;

3,334,940✔

1872

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

754,929,589✔

1873

  int32_t         code = TSDB_CODE_SUCCESS;

754,929,589✔

1874

  int32_t         lino = 0;

754,929,589✔

1875

  STableScanInfo* pInfo = pOperator->info;

754,929,589✔

1876

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

754,960,235✔

1877

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

754,929,017✔

1878

  int32_t         num = 0;

754,958,708✔

1879

  STableKeyInfo*  pList = NULL;

754,956,664✔

1880

  SSDataBlock*    pResult = NULL;

754,924,288✔

1881

  const char*     idStr = GET_TASKID(pTaskInfo);

754,918,975✔

1883

  QRY_PARAM_CHECK(pResBlock);

754,878,330✔

1885

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

754,906,754✔

1886

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

220,713,787✔

1891

    taosRLockLatch(&pTaskInfo->lock);

220,726,949✔

1892

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

220,786,285✔

1893

    taosRUnLockLatch(&pTaskInfo->lock);

220,698,610✔

1894

    QUERY_CHECK_CODE(code, lino, _end);

220,741,699✔

1896

    if (pInfo->filesetDelimited) {

220,739,359✔

1897

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

278,237✔

1900

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

220,654,140✔

1905

  code = doGroupedTableScan(pOperator, &pResult);

754,957,989✔

1906

  QUERY_CHECK_CODE(code, lino, _end);

753,696,621✔

1908

  if (pResult != NULL) {

753,696,621✔

1909

    if (pOperator->dynamicTask) {

538,098,781✔

1910

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

2,104,222✔

1913

    *pResBlock = pResult;

538,089,167✔

1914

    return code;

537,907,238✔

1918

    code = startNextGroupScan(pOperator, &pResult);

215,679,171✔

1919

    QUERY_CHECK_CODE(code, lino, _end);

215,912,863✔

1921

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

215,912,863✔

1922

      *pResBlock = pResult;

215,837,417✔

1923

      return code;

215,827,093✔

1927

_end:

2,340✔

1928

  if (code != TSDB_CODE_SUCCESS) {

2,340✔

1929

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

2,340✔

1930

    pTaskInfo->code = code;

2,340✔

1933

  return code;

2,340✔

1936

static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {

25,696,737✔

1937

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

25,696,737✔

1940

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

1,003,677✔

1941

  int32_t                  code = TSDB_CODE_SUCCESS;

1,003,677✔

1942

  int32_t                  lino = 0;

1,003,677✔

1943

  STableScanInfo*          pInfo = pOperator->info;

1,003,677✔

1944

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

1,003,281✔

1945

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

1,003,281✔

1946

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

1,003,281✔

1948

  code = createTableListInfoFromParam(pOperator);

1,003,677✔

1949

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

1,003,677✔

1950

  pOperator->pOperatorGetParam = NULL;

1,003,677✔

1951

  QUERY_CHECK_CODE(code, lino, _end);

1,003,281✔

1953

  if (pOperator->status == OP_EXEC_DONE) {

1,003,281✔

1954

    pInfo->currentGroupId = -1;

23,748✔

1955

    pOperator->status = OP_OPENED;

23,748✔

1956

    SSDataBlock* result = NULL;

23,748✔

1959

      code = startNextGroupScan(pOperator, &result);

23,748✔

1960

      QUERY_CHECK_CODE(code, lino, _end);

23,748✔

1962

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

23,748✔

1963

        (*ppRes) = result;

23,748✔

1964

        return code;

23,748✔

1968

_end:

979,929✔

1969

  if (code) {

979,533✔

1972

  return code;

979,533✔

1975

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

18,295,787✔

1976

  int32_t      code = TSDB_CODE_SUCCESS;

18,295,787✔

1977

  int32_t      lino = 0;

18,295,787✔

1978

  bool         found = false;

18,295,787✔

1980

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

18,295,787✔

1981

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

18,295,787✔

1983

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

89,992,388✔

1984

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

71,696,601✔

1985

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

71,696,601✔

1986

    found = false;

71,696,601✔

1987

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

203,894,299✔

1988

      SColIdSlotIdPair* pColPair = taosArrayGet(pColMap, j);

185,601,607✔

1989

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

185,601,607✔

1990

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

185,601,607✔

1991

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

53,403,909✔

1992

        found = true;

53,403,909✔

1993

        break;

53,403,909✔

1996

    if (!found) {

71,696,601✔

1997

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

18,292,692✔

1998

        pSrcCol->info.slotId = -1;

18,292,692✔

2007

  return code;

18,295,787✔

2013

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

25,697,389✔

2014

  int32_t                  code = TSDB_CODE_SUCCESS;

25,697,389✔

2015

  int32_t                  lino = 0;

25,697,389✔

2016

  STableScanInfo*          pInfo = pOperator->info;

25,697,389✔

2017

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

25,697,389✔

2018

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

25,697,389✔

2019

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

25,697,389✔

2021

  QRY_PARAM_CHECK(ppRes);

25,697,389✔

2022

  qTrace("%s call", __FUNCTION__);

25,697,389✔

2024

  code = createVTableScanInfoFromParam(pOperator);

25,697,389✔

2025

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

25,697,389✔

2026

  pOperator->pOperatorGetParam = NULL;

25,697,389✔

2028

  QUERY_CHECK_CODE(code, lino, _end);

25,697,389✔

2030

  SSDataBlock* result = NULL;

25,696,737✔

2031

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

25,696,737✔

2032

    (*ppRes) = result;

20,556✔

2033

    return code;

20,556✔

2036

  if (pInfo->newReader) {

25,676,181✔

2037

    code = startNextGroupScan(pOperator, &result);

7,410,406✔

2038

    QUERY_CHECK_CODE(code, lino, _end);

7,410,406✔

2040

    code = doGroupedTableScan(pOperator, &result);

18,265,775✔

2041

    QUERY_CHECK_CODE(code, lino, _end);

18,265,775✔

2044

  if (result) {

25,676,181✔

2045

    SSDataBlock* res = NULL;

18,295,787✔

2046

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

18,295,787✔

2047

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

18,295,787✔

2048

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

18,295,787✔

2050

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

18,295,787✔

2051

    QUERY_CHECK_CODE(code, lino, _end);

18,295,787✔

2057

    (*ppRes) = result;

18,295,787✔

2059

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

7,380,394✔

2060

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

7,380,394✔

2062

    blockDataDestroy(pInfo->pResBlock);

7,380,394✔

2063

    pInfo->pResBlock = NULL;

7,380,394✔

2065

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

7,380,394✔

2066

    if (reader) {

7,380,394✔

2067

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

7,380,394✔

2068

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

7,380,394✔

2069

        pInfo->base.dataReader = NULL;

7,380,394✔

2071

      if (pAPI->tsdReader.tsdReaderClose) {

7,380,394✔

2072

        pAPI->tsdReader.tsdReaderClose(*reader);

7,380,394✔

2076

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

7,380,394✔

2077

    QUERY_CHECK_CODE(code, lino, _end);

7,380,394✔

2079

_end:

25,676,833✔

2080

  if (code) {

25,676,833✔

2081

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

652✔

2083

  return code;

25,676,833✔

2086

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

17,065,876✔

2087

  int32_t                  code = TSDB_CODE_SUCCESS;

17,065,876✔

2088

  int32_t                  lino = 0;

17,065,876✔

2089

  STableScanInfo*          pInfo = pOperator->info;

17,065,876✔

2090

  int32_t                  index = 0;

17,065,876✔

2091

  char*                    tagVal = NULL;

17,065,876✔

2093

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

121,751,739✔

2094

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

104,684,669✔

2095

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

104,684,656✔

2097

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

104,683,462✔

2098

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

104,683,462✔

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

104,685,863✔

2117

  return code;

17,065,876✔

2124

static void clearVstbBatchDynamicTableScanInfo(STableScanInfo* pInfo) {

6,145,426✔

2125

  if (pInfo->pBatchColMap) {

6,145,426✔

2126

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

16,538,740✔

2127

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

10,393,674✔

2128

      taosArrayDestroy(pOrgTbInfo->colMap);

10,392,594✔

2130

    taosArrayDestroy(pInfo->pBatchColMap);

6,143,823✔

2131

    pInfo->pBatchColMap = NULL;

6,144,346✔

2133

  if (pInfo->cachedTagList) {

6,145,623✔

2134

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

18,516,872✔

2135

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

15,747,006✔

2136

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

15,747,726✔

2137

        taosMemoryFreeClear(pTagVal->pData);

6,910,902✔

2140

    taosArrayDestroy(pInfo->cachedTagList);

2,766,986✔

2141

    pInfo->cachedTagList = NULL;

2,766,266✔

2143

  if (pInfo->lastColArray) {

6,145,066✔

2144

    taosArrayDestroy(pInfo->lastColArray);

6,145,426✔

2145

    pInfo->lastColArray = NULL;

6,143,626✔

2147

  if (pInfo->lastBlockColArray) {

6,143,626✔

2148

    taosArrayDestroy(pInfo->lastBlockColArray);

6,144,346✔

2149

    pInfo->lastBlockColArray = NULL;

6,142,906✔

2152

  blockDataDestroy(pInfo->pResBlock);

6,143,626✔

2153

  pInfo->pResBlock = NULL;

6,144,706✔

2154

6,144,706✔

2156

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

83,450,803✔

2157

  int32_t                  code = TSDB_CODE_SUCCESS;

83,450,803✔

2158

  int32_t                  lino = 0;

83,450,803✔

2159

  STableScanInfo*          pInfo = pOperator->info;

83,450,803✔

2160

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

83,479,610✔

2161

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

83,478,620✔

2162

  SSDataBlock*             result = NULL;

83,475,462✔

2164

  QRY_PARAM_CHECK(ppRes);

83,484,359✔

2165

  qTrace("%s call", __FUNCTION__);

83,462,204✔

2167

  while (!result) {

165,023,864✔

2168

    code = createVTableScanInfoFromBatchParam(pOperator);

87,716,317✔

2169

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

87,705,975✔

2170

    pOperator->pOperatorGetParam = NULL;

87,673,862✔

2171

    QUERY_CHECK_CODE(code, lino, _end);

87,698,005✔

2173

    if (pInfo->newReader) {

87,698,005✔

2174

      code = startNextGroupScan(pOperator, &result);

10,354,708✔

2175

      QUERY_CHECK_CODE(code, lino, _end);

10,393,004✔

2177

      code = doGroupedTableScan(pOperator, &result);

77,356,848✔

2178

      QUERY_CHECK_CODE(code, lino, _end);

77,309,759✔

2180

    if (!result) {

87,702,763✔

2181

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

10,390,486✔

2183

      pInfo->currentBatchIdx++;

10,390,486✔

2184

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

10,390,138✔

2186

        break;

6,144,706✔

2191

  if (result) {

83,452,253✔

2192

    SSDataBlock* res = NULL;

77,308,267✔

2193

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

77,321,430✔

2194

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

77,341,404✔

2195

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

77,337,786✔

2197

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

77,337,325✔

2198

    QUERY_CHECK_CODE(code, lino, _end);

77,362,454✔

2200

    if (pInfo->cachedTagList) {

77,362,454✔

2201

      code = setTagValFromTagList(pOperator, res);

17,065,279✔

2202

      QUERY_CHECK_CODE(code, lino, _end);

17,065,876✔

2205

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

77,361,247✔

2206

    pInfo->pResBlock = res;

77,362,441✔

2207

    blockDataDestroy(result);

77,366,036✔

2208

    (*ppRes) = res;

77,361,234✔

2210

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

77,361,831✔

2212

    clearVstbBatchDynamicTableScanInfo(pInfo);

6,143,986✔

2213

    setOperatorCompleted(pOperator);

6,144,706✔

2216

_end:

83,501,054✔

2217

  return code;

83,507,621✔

2220

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

110,167,389✔

2221

  int32_t                  code = TSDB_CODE_SUCCESS;

110,167,389✔

2222

  int32_t                  lino = 0;

110,167,389✔

2223

  STableScanInfo*          pInfo = pOperator->info;

110,167,389✔

2224

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

110,197,450✔

2225

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

110,182,366✔

2227

  QRY_PARAM_CHECK(ppRes);

110,202,860✔

2228

  qTrace("%s call", __FUNCTION__);

110,197,791✔

2230

  pOperator->dynamicTask = true;

110,197,791✔

2232

  switch (type) {

110,171,504✔

2233

    case DYN_TYPE_VSTB_WIN_SCAN:

1,003,677✔

2235

      code = doStbJoinDynamicTableScanNext(pOperator, ppRes);

1,003,677✔

2236

      QUERY_CHECK_CODE(code, lino, _end);

1,003,289✔

2237

      break;

1,003,289✔

2239

    case DYN_TYPE_VSTB_SINGLE_SCAN: {

25,697,389✔

2240

      code = doVstbSingleDynamicTableScanNext(pOperator, ppRes);

25,697,389✔

2241

      QUERY_CHECK_CODE(code, lino, _end);

25,697,389✔

2242

      break;

25,696,737✔

2244

    case DYN_TYPE_VSTB_BATCH_SCAN:

83,470,438✔

2245

      code = doVstbBatchDynamicTableScanNext(pOperator, ppRes);

83,470,438✔

2246

      QUERY_CHECK_CODE(code, lino, _end);

83,481,252✔

2247

      break;

83,481,252✔

2254

_end:

110,181,930✔

2255

  return code;

110,181,930✔

2258

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

870,242,276✔

2259

  int32_t         code = TSDB_CODE_SUCCESS;

870,242,276✔

2260

  int32_t         lino = 0;

870,242,276✔

2261

  STableScanInfo* pInfo = pOperator->info;

870,242,276✔

2262

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

870,390,534✔

2263

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

870,325,764✔

2264

  QRY_PARAM_CHECK(ppRes);

870,399,641✔

2265

  qTrace("%s call", __FUNCTION__);

870,406,444✔

2267

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

870,408,028✔

2268

  QUERY_CHECK_CODE(code, lino, _end);

870,410,108✔

2270

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

870,410,108✔

2271

  if (pGetParam || pInfo->pBatchColMap) {

870,421,022✔

2272

    ETableScanDynType type = pGetParam ?

110,168,743✔

2273

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

110,175,089✔

2276

    code = doDynamicTableScanNext(pOperator, type, ppRes);

110,168,743✔

2277

    QUERY_CHECK_CODE(code, lino, _end);

110,186,804✔

2279

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

110,186,152✔

2280

      return code;

109,222,022✔

2285

  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {

761,197,661✔

2286

    int32_t       numOfTables = 0;

6,285,967✔

2287

    STableKeyInfo tInfo = {0};

6,285,967✔

2288

    pInfo->countState = TABLE_COUNT_STATE_END;

6,285,967✔

2290

    while (1) {

5,870,482✔

2291

      SSDataBlock* result = NULL;

12,156,449✔

2292

      code = doGroupedTableScan(pOperator, &result);

12,157,417✔

2293

      QUERY_CHECK_CODE(code, lino, _end);

12,147,420✔

2295

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

12,147,420✔

2296

        (*ppRes) = result;

6,230,777✔

2297

        return code;

6,242,079✔

2301

      pInfo->currentTable++;

5,914,409✔

2303

      taosRLockLatch(&pTaskInfo->lock);

5,906,634✔

2304

      numOfTables = 0;

5,915,662✔

2305

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

5,915,662✔

2306

      if (code != TSDB_CODE_SUCCESS) {

5,915,661✔

2311

      if (pInfo->currentTable >= numOfTables) {

5,915,661✔

2312

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

43,565✔

2313

        taosRUnLockLatch(&pTaskInfo->lock);

43,565✔

2314

        (*ppRes) = NULL;

43,565✔

2315

        return code;

43,565✔

2318

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

5,863,722✔

2319

      if (!tmp) {

5,870,179✔

2325

      tInfo = *tmp;

5,870,179✔

2326

      taosRUnLockLatch(&pTaskInfo->lock);

5,863,702✔

2328

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

5,872,742✔

2329

      QUERY_CHECK_CODE(code, lino, _end);

5,868,583✔

2330

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

5,868,583✔

2333

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

5,875,935✔

2334

      QUERY_CHECK_CODE(code, lino, _end);

5,865,644✔

2335

      pInfo->scanTimes = 0;

5,865,644✔

2338

    code = groupSeqTableScan(pOperator, ppRes);

754,870,348✔

2339

    QUERY_CHECK_CODE(code, lino, _end);

753,734,590✔

2342

_end:

753,732,250✔

2343

  if (code != TSDB_CODE_SUCCESS) {

753,749,703✔

2344

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

2,992✔

2345

    pTaskInfo->code = code;

2,992✔

2346

    T_LONG_JMP(pTaskInfo->env, code);

2,992✔

2349

  return code;

753,746,711✔

2352

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

1,940,770✔

2353

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

1,940,770✔

2354

  if (!pRecorder) {

1,940,770✔

2357

  STableScanInfo* pTableScanInfo = pOptr->info;

1,940,770✔

2358

  *pRecorder = pTableScanInfo->base.readRecorder;

1,940,770✔

2359

  *pOptrExplain = pRecorder;

1,940,770✔

2360

  *len = sizeof(SFileBlockLoadRecorder);

1,940,770✔

2361

  return 0;

1,940,770✔

2364

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

258,678,726✔

2365

  cleanupQueryTableDataCond(&pBase->cond);

258,678,726✔

2366

  cleanupQueryTableDataCond(&pBase->orgCond);

258,678,770✔

2368

  if (pAPI->tsdReaderClose) {

258,673,734✔

2369

    pAPI->tsdReaderClose(pBase->dataReader);

258,676,783✔

2371

  pBase->dataReader = NULL;

258,624,185✔

2373

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

258,633,514✔

2374

    taosArrayDestroy(pBase->matchInfo.pList);

258,643,624✔

2377

  tableListDestroy(pBase->pTableListInfo);

258,620,002✔

2378

  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);

258,667,922✔

2379

  cleanupExprSupp(&pBase->pseudoSup);

258,653,164✔

2380

258,688,516✔

2382

static void cleanReaderForVTable(STableScanInfo* pInfo){

6,820,575✔

2383

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

6,820,575✔

2384

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

6,819,506✔

2385

    while (pIter != NULL) {

6,843,583✔

2386

      void **reader = pIter;

24,416✔

2387

      if (*reader) {

24,416✔

2388

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

24,416✔

2389

          pInfo->base.dataReader = NULL;

22,052✔

2391

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

24,416✔

2393

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

24,416✔

2396

6,821,193✔

2398

static void destroyTableScanOperatorInfo(void* param) {

232,751,207✔

2399

  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;

232,751,207✔

2400

  blockDataDestroy(pTableScanInfo->pResBlock);

232,751,207✔

2401

  blockDataDestroy(pTableScanInfo->pOrgBlock);

232,740,355✔

2402

  taosHashCleanup(pTableScanInfo->pIgnoreTables);

232,741,247✔

2403

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

232,734,999✔

2404

    cleanReaderForVTable(pTableScanInfo);

6,820,226✔

2405

    taosHashCleanup(pTableScanInfo->readerCache);

6,819,741✔

2407

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

232,739,428✔

2408

  taosArrayDestroy(pTableScanInfo->pBlockColMap);

232,738,396✔

2409

  taosMemoryFreeClear(param);

232,742,717✔

2410

232,708,068✔

2412

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

232,645,486✔

2413

  if (pBlock && dataRequireFlag == FUNC_DATA_REQUIRED_NOT_LOAD) {

232,645,486✔

2414

    int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

7,833,726✔

2415

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

19,771,076✔

2416

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

11,941,900✔

2417

      if (pCol) {

11,940,089✔

2418

        pCol->info.noData = true;

11,941,543✔

2422

232,640,936✔

2424

static int32_t resetTableScanOperatorState(SOperatorInfo* pOper) {

306,901✔

2425

  int32_t         code = TSDB_CODE_SUCCESS;

306,901✔

2426

  STableScanInfo*   pInfo = pOper->info;

306,901✔

2427

  pOper->status = OP_NOT_OPENED;

306,901✔

2429

  pInfo->scanTimes = 0;

306,901✔

2430

  pInfo->currentGroupId = -1;

306,901✔

2431

  pInfo->tableEndIndex = -1;

306,901✔

2432

  pInfo->tableStartIndex = 0;

306,901✔

2433

  pInfo->currentTable = 0;

306,901✔

2434

  pInfo->scanMode = 0;

306,901✔

2435

  pInfo->countState = 0;

306,901✔

2436

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

306,901✔

2438

  if (!pInfo->virtualStableScan) {

306,901✔

2444

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

144,921✔

2445

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

144,921✔

2447

    pInfo->base.dataReader = NULL;

144,921✔

2449

  tableListDestroy(pInfo->base.pTableListInfo);

306,901✔

2450

  pInfo->base.pTableListInfo = tableListCreate();

306,478✔

2451

  if (!pInfo->base.pTableListInfo) {

306,901✔

2455

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

306,901✔

2457

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

306,901✔

2458

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

306,478✔

2459

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

286,429✔

2461

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

286,852✔

2462

    if (code) {

143,865✔

2468

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

306,901✔

2469

  cleanupQueryTableDataCond(&pInfo->base.cond);

306,901✔

2470

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

306,901✔

2471

  if (code) {

306,447✔

2475

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

306,447✔

2476

    cleanupQueryTableDataCond(&pInfo->base.orgCond);

161,980✔

2477

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

161,980✔

2478

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

161,980✔

2481

  pOper->resultInfo.totalRows = 0;

306,901✔

2482

  blockDataEmpty(pInfo->pResBlock);

306,901✔

2483

  blockDataEmpty(pInfo->pOrgBlock);

306,901✔

2484

  taosHashClear(pInfo->pIgnoreTables);

306,901✔

2485

  taosArrayDestroy(pInfo->pBlockColMap);

306,901✔

2486

  pInfo->pBlockColMap = NULL;

306,901✔

2487

  return code;

306,901✔

2490

int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

232,735,171✔

2493

  QRY_PARAM_CHECK(pOptrInfo);

232,735,171✔

2495

  int32_t         code = TSDB_CODE_SUCCESS;

232,746,123✔

2496

  int32_t         lino = 0;

232,746,123✔

2497

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

232,746,123✔

2498

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

232,709,155✔

2499

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

232,706,940✔

2500

    code = terrno;

1,041✔

2504

  pOperator->pPhyNode = pTableScanNode;

232,705,899✔

2505

  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;

232,707,567✔

2506

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

232,710,176✔

2508

  int32_t numOfCols = 0;

232,724,838✔

2510

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

232,727,606✔

2511

  QUERY_CHECK_CODE(code, lino, _error);

232,668,387✔

2513

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

232,668,387✔

2514

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

232,650,356✔

2515

  QUERY_CHECK_CODE(code, lino, _error);

232,611,436✔

2517

  if (pScanNode->pScanPseudoCols != NULL) {

232,611,436✔

2518

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

54,668,111✔

2519

    pSup->pExprInfo = NULL;

54,674,957✔

2520

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

54,683,139✔

2521

    QUERY_CHECK_CODE(code, lino, _error);

54,656,158✔

2523

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

54,656,158✔

2525

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

54,693,797✔

2528

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

232,634,010✔

2529

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

232,551,217✔

2531

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

232,606,756✔

2532

  pInfo->base.readHandle = *readHandle;

232,506,783✔

2533

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

232,564,512✔

2535

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

232,623,039✔

2536

  pInfo->sample.seed = taosGetTimestampSec();

232,681,550✔

2538

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

232,639,668✔

2539

  initResultSizeInfo(&pOperator->resultInfo, 4096);

232,645,085✔

2540

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

232,484,010✔

2541

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

232,647,193✔

2542

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

232,570,525✔

2544

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

232,448,386✔

2545

  QUERY_CHECK_CODE(code, lino, _error);

232,518,286✔

2547

  pInfo->virtualStableScan = pScanNode->virtualStableScan;

232,518,286✔

2548

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

232,538,581✔

2549

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

6,818,784✔

2550

    pInfo->pResBlock = NULL;

6,818,435✔

2551

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

6,818,858✔

2552

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

6,812,847✔

2555

  if (pInfo->virtualStableScan) {

232,513,934✔

2556

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

6,816,122✔

2557

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

6,821,532✔

2560

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

232,660,965✔

2561

                            pTaskInfo->pStreamRuntimeInfo);

232,634,479✔

2562

  QUERY_CHECK_CODE(code, lino, _error);

232,536,489✔

2564

  pInfo->currentGroupId = -1;

232,512,895✔

2566

  pInfo->tableEndIndex = -1;

232,612,622✔

2567

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

232,595,268✔

2568

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

232,498,137✔

2570

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

232,666,553✔

2572

  pOperator->exprSupp.numOfExprs = numOfCols;

232,396,037✔

2574

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

232,560,881✔

2575

  pInfo->ignoreTag = false;

232,635,998✔

2577

  pInfo->lastBatchIdx = -1;

232,555,036✔

2578

  pInfo->currentBatchIdx = 0;

232,499,863✔

2579

  pInfo->base.pTableListInfo = pTableListInfo;

232,625,277✔

2580

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

232,507,665✔

2590

  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

232,507,665✔

2592

  pOperator->fpSet = createOperatorFpSet(prepareTableScan, doTableScanNext,

232,594,665✔

2598

  setOperatorResetStateFn(pOperator, resetTableScanOperatorState);

232,456,491✔

2600

  pOperator->cost.openCost = 0;

232,385,574✔

2601

  *pOptrInfo = pOperator;

232,499,742✔

2602

  return TSDB_CODE_SUCCESS;

232,386,186✔

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

2653

  int32_t code = TSDB_CODE_SUCCESS;

14,344✔

2654

  int32_t lino = 0;

14,344✔

2655

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

14,344✔

2656

    return code;

1,740✔

2658

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

12,604✔

2659

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

12,604✔

2660

  bool hasUnqualified = false;

12,604✔

2662

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

12,604✔

2663

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

12,604✔

2665

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

12,604✔

2667

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

12,604✔

2670

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

12,604✔

2671

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

12,604✔

2672

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

45,263,706✔

2673

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

45,251,102✔

2674

    void*    data = colDataGetData(pColPk, i);

45,251,102✔

2675

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

45,253,534✔

2676

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

2,432✔

2679

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

2,432✔

2680

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

2,432✔

2681

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

2,432✔

2682

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

2,432✔

2683

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

2,432✔

2684

      taosMemoryFree(tmq);

2,432✔

2686

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

45,248,670✔

2689

    if (!p[i]) {

45,251,102✔

2690

      hasUnqualified = true;

16,134,752✔

2694

  if (hasUnqualified) {

12,604✔

2695

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

4,243✔

2696

    QUERY_CHECK_CODE(code, lino, _end);

4,243✔

2699

_end:

12,604✔

2700

  taosMemoryFree(p);

12,604✔

2701

  if (code != TSDB_CODE_SUCCESS) {

12,604✔

2704

  return code;

12,604✔

2718

static int32_t processBlock(STmqQueryScanInfo* pInfo) {

23,913,752✔

2719

  int32_t         code = TSDB_CODE_SUCCESS;

23,913,752✔

2720

  int32_t         lino = 0;

23,913,752✔

2721

  SOperatorInfo*  pOperator = pInfo->pTmqScanOp;

23,913,752✔

2723

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

23,913,752✔

2724

  QUERY_CHECK_CODE(code, lino, _end);

23,913,752✔

2726

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

23,913,752✔

2727

  QUERY_CHECK_CODE(code, lino, _end);

23,913,752✔

2729

_end:

23,913,752✔

2730

  if (code != TSDB_CODE_SUCCESS) {

23,913,752✔

2733

  return code;

23,913,752✔

2736

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

7,022,458✔

2737

  int32_t code = TSDB_CODE_SUCCESS;

7,022,458✔

2738

  int32_t lino = 0;

7,022,458✔

2739

  SValue  val = {0};

7,022,458✔

2740

  if (hasPrimaryKey) {

7,023,092✔

2741

    code = doBlockDataPrimaryKeyFilter(pBlock, offset);

14,344✔

2742

    if (code != TSDB_CODE_SUCCESS) {

14,344✔

2746

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

14,344✔

2748

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

14,344✔

2749

      return code;

3,939✔

2751

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

10,405✔

2752

    val.type = pColPk->info.type;

10,405✔

2753

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

10,405✔

2754

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

663✔

2757

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

663✔

2758

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

663✔

2759

      val.nData = varDataLen(tmp);

663✔

2760

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

663✔

2762

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

9,742✔

2765

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

7,019,153✔

2767

_end:

7,018,196✔

2768

  if (code != TSDB_CODE_SUCCESS) {

7,018,830✔

2771

  return code;

7,018,830✔

2774

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

24,100,499✔

2775

  int32_t        code = TSDB_CODE_SUCCESS;

24,100,499✔

2776

  int32_t        lino = 0;

24,100,499✔

2777

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

24,100,499✔

2778

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

24,103,379✔

2780

  STmqQueryScanInfo* pInfo = pOperator->info;

24,101,940✔

2781

  const char*      id = GET_TASKID(pTaskInfo);

24,101,573✔

2783

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

24,098,756✔

2785

  if (isTaskKilled(pTaskInfo)) {

24,103,043✔

2790

  blockDataCleanup(pInfo->pRes);

24,103,738✔

2791

  int64_t st = taosGetTimestampMs();

24,102,701✔

2793

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

24,102,701✔

2794

    while (1) {

5,962,340✔

2795

      SSDataBlock* pResult = NULL;

6,284,405✔

2796

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

6,284,675✔

2797

      QUERY_CHECK_CODE(code, lino, _end);

6,285,644✔

2799

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

6,285,644✔

2800

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

6,242,402✔

2801

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

6,242,079✔

2802

        QUERY_CHECK_CODE(code, lino, _end);

6,241,433✔

2803

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

6,241,433✔

2804

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

6,242,402✔

2805

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

6,238,463✔

2806

          QUERY_CHECK_CODE(code, lino, _end);

6,238,463✔

2812

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

6,242,402✔

2813

        (*ppRes) = pInfo->pRes;

272,641✔

2814

        return code;

272,641✔

2816

      int64_t elapsed = taosGetTimestampMs() - st;

5,968,469✔

2817

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

5,968,469✔

2818

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

6,452✔

2821

        (*ppRes) = pInfo->pRes;

5,806✔

2822

        return code;

5,806✔

2826

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

43,565✔

2827

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

43,565✔

2829

    pTSInfo->base.dataReader = NULL;

43,565✔

2830

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

43,565✔

2831

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

43,565✔

2832

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

43,565✔

2833

    if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) {

43,565✔

2834

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

10,433✔

2835

        (*ppRes) = NULL;

4,225✔

2837

        (*ppRes) = pInfo->pRes;

6,208✔

2839

      return 0;

10,433✔

2841

    QUERY_CHECK_CODE(code, lino, _end);

33,132✔

2843

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

33,132✔

2846

  while (1) {

98,894✔

2847

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

71,736,651✔

2851

                                                              pTaskInfo->tmqInfo.sourceExcluded,

23,912,675✔

2854

                                                              pTaskInfo->tmqInfo.enableReplay);

23,912,303✔

2855

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

23,912,670✔

2858

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

23,912,670✔

2861

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

23,913,752✔

2864

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

23,913,375✔

2866

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

23,913,375✔

2868

    int32_t ret = processBlock(pInfo);

23,913,375✔

2869

    if (ret != 0) {

23,913,752✔

2873

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

23,913,752✔

2876

    if (code != 0) {

23,913,752✔

2877

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

23,302,317✔

2878

        (*ppRes) = NULL;

10,970,385✔

2880

        (*ppRes) = pInfo->pRes;

12,331,932✔

2882

      return 0;

23,302,317✔

2885

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

611,435✔

2886

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

104,078✔

2887

      (*ppRes) = pInfo->pRes;

512,541✔

2888

      return 0;

512,541✔

2902

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

345,319✔

2903

  int32_t code = TSDB_CODE_SUCCESS;

345,319✔

2904

  int32_t lino = 0;

345,319✔

2905

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

345,319✔

2906

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

345,606✔

2909

  int32_t size = 0;

345,606✔

2910

  code = tableListGetSize(pTableListInfo, &size);

345,606✔

2911

  QUERY_CHECK_CODE(code, lino, _end);

345,606✔

2912

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

10,089,621✔

2913

    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);

9,745,146✔

2914

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

9,736,043✔

2915

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

9,736,043✔

2916

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

9,744,015✔

2919

  (*ppArrayRes) = tableIdList;

344,475✔

2921

_end:

345,606✔

2922

  if (code != TSDB_CODE_SUCCESS) {

345,606✔

2925

  return code;

345,606✔

2928

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

1,612,125✔

2929

  int32_t        code = TSDB_CODE_SUCCESS;

1,612,125✔

2930

  int32_t        lino = 0;

1,612,125✔

2931

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,612,125✔

2932

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,612,125✔

2934

  STmqRawScanInfo* pInfo = pOperator->info;

1,612,125✔

2935

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

1,612,125✔

2936

  pTaskInfo->tmqInfo.btMetaRsp.batchMetaLen = NULL;

1,612,125✔

2938

  qDebug("tmqsnap doRawScan called");

1,612,125✔

2939

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

1,612,125✔

2940

    bool hasNext = false;

1,605,647✔

2941

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

1,605,647✔

2942

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

1,604,405✔

2943

      if (code != TSDB_CODE_SUCCESS) {

1,603,137✔

2949

    if (pInfo->dataReader && hasNext) {

1,604,379✔

2950

      if (isTaskKilled(pTaskInfo)) {

781,653✔

2955

      SSDataBlock* pBlock = NULL;

781,653✔

2956

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

781,336✔

2957

      QUERY_CHECK_CODE(code, lino, _end);

780,702✔

2959

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

780,702✔

2960

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

781,336✔

2961

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

781,970✔

2962

        QUERY_CHECK_CODE(code, lino, _end);

781,336✔

2963

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

781,336✔

2964

        (*ppRes) = pBlock;

782,604✔

2965

        return code;

781,970✔

2969

    SMetaTableInfo mtInfo = {0};

822,726✔

2970

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

823,043✔

2971

    QUERY_CHECK_CODE(code, lino, _end);

823,043✔

2972

    if (code != 0) {

823,043✔

2976

    STqOffsetVal offset = {0};

823,043✔

2977

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

821,458✔

2978

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

12,231✔

2979

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

12,231✔

2981

      SValue val = {0};

810,818✔

2982

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

810,818✔

2983

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

810,184✔

2986

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

823,677✔

2987

    QUERY_CHECK_CODE(code, lino, _end);

823,677✔

2988

    (*ppRes) = NULL;

823,677✔

2989

    return code;

823,677✔

2990

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

6,478✔

2991

    SSnapContext* sContext = pInfo->sContext;

6,478✔

2992

    int64_t st = taosGetTimestampMs();

6,478✔

2993

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

50,134✔

2994

      void*   data = NULL;

49,416✔

2995

      int32_t dataLen = 0;

49,136✔

2996

      int16_t type = 0;

49,136✔

2997

      int64_t uid = 0;

49,136✔

2998

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

49,416✔

3004

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

49,416✔

3005

        STqOffsetVal offset = {0};

5,760✔

3006

        SValue       val = {0};

5,760✔

3008

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

5,760✔

3009

        QUERY_CHECK_CODE(code, lino, _end);

5,760✔

3010

        break;

5,760✔

3012

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

43,656✔

3013

        SMqMetaRsp tmpMetaRsp = {0};

43,376✔

3014

        tmpMetaRsp.resMsgType = type;

43,376✔

3015

        tmpMetaRsp.metaRspLen = dataLen;

43,376✔

3016

        tmpMetaRsp.metaRsp = data;

43,376✔

3017

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

43,376✔

3018

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

5,200✔

3019

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

5,200✔

3021

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

5,200✔

3022

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

5,200✔

3024

        int32_t  tempRes = TSDB_CODE_SUCCESS;

43,376✔

3025

        uint32_t len = 0;

43,376✔

3026

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

43,376✔

3027

        if (TSDB_CODE_SUCCESS != tempRes) {

43,656✔

3033

        int32_t tLen = sizeof(SMqRspHead) + len;

43,656✔

3034

        void*   tBuf = taosMemoryCalloc(1, tLen);

43,656✔

3035

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

43,656✔

3037

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

43,656✔

3038

        SEncoder encoder = {0};

43,656✔

3039

        tEncoderInit(&encoder, metaBuff, len);

43,656✔

3040

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

43,656✔

3041

        if (tempLen < 0) {

43,656✔

3048

        taosMemoryFreeClear(data);

43,656✔

3049

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

43,656✔

3050

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

43,656✔

3052

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

43,656✔

3053

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

43,656✔

3055

        int64_t elapsed = taosGetTimestampMs() - st;

43,656✔

3056

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

43,656✔

3062

    (*ppRes) = NULL;

6,478✔

3063

    return code;

6,478✔

3077

static void destroyTmqRawScanOperatorInfo(void* param) {

129,527✔

3078

  STmqRawScanInfo* pRawScan = (STmqRawScanInfo*)param;

129,527✔

3079

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

129,527✔

3080

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

129,527✔

3081

  tableListDestroy(pRawScan->pTableListInfo);

129,527✔

3082

  taosMemoryFree(pRawScan);

129,527✔

3083

129,527✔

3088

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

129,165✔

3094

  QRY_PARAM_CHECK(pOptrInfo);

129,165✔

3095

  int32_t code = TSDB_CODE_SUCCESS;

129,527✔

3096

  int32_t lino = 0;

129,527✔

3098

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

129,527✔

3099

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

129,527✔

3100

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

129,527✔

3106

  pInfo->pTableListInfo = tableListCreate();

129,527✔

3107

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

129,527✔

3108

  pInfo->vnode = pHandle->vnode;

129,527✔

3109

  pInfo->pAPI = &pTaskInfo->storageAPI;

129,527✔

3111

  pInfo->sContext = pHandle->sContext;

129,527✔

3112

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

129,527✔

3113

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

129,527✔

3116

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

129,527✔

3118

  *pOptrInfo = pOperator;

129,527✔

3119

  return code;

129,527✔

3131

void destroyTmqScanOperatorInfo(void* param) {

345,606✔

3132

  if (param == NULL) {

345,606✔

3136

  STmqQueryScanInfo* pTmqScan = (STmqQueryScanInfo*)param;

345,606✔

3137

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

345,606✔

3138

    destroyOperator(pTmqScan->pTableScanOp);

345,606✔

3141

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

345,326✔

3142

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

345,606✔

3145

  if (pTmqScan->pTableListInfo) {

345,046✔

3149

  if (pTmqScan->pPseudoExpr) {

345,606✔

3150

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

200,766✔

3151

    taosMemoryFree(pTmqScan->pPseudoExpr);

200,766✔

3154

  taosHashCleanup(pTmqScan->pCol2SlotId);

344,966✔

3155

  pTmqScan->pCol2SlotId = NULL;

345,606✔

3157

  cleanupExprSupp(&pTmqScan->tagCalSup);

345,246✔

3159

  blockDataDestroy(pTmqScan->pRes);

345,606✔

3161

  taosMemoryFree(pTmqScan);

345,003✔

3181

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

345,606✔

3182

  size_t  numOfCols = LIST_LENGTH(pNodeList);

345,606✔

3183

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

2,173,720✔

3184

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

1,828,114✔

3185

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

1,828,114✔

3186

      *pTsSlotId = pNode->slotId;

345,606✔

3189

345,606✔

3191

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

345,234✔

3192

  size_t  numOfCols = LIST_LENGTH(pNodeList);

345,234✔

3193

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

2,173,348✔

3194

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

1,827,742✔

3195

    if (pNode == NULL) {

1,827,483✔

3198

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

1,827,483✔

3199

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

1,827,111✔

3200

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

1,827,111✔

3201

      if (code != TSDB_CODE_SUCCESS) {

1,828,114✔

3206

  return TSDB_CODE_SUCCESS;

345,606✔

3209

int32_t createTmqScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,

344,948✔

3212

  QRY_PARAM_CHECK(pOptrInfo);

344,948✔

3214

  int32_t          code = TSDB_CODE_SUCCESS;

345,606✔

3215

  int32_t          lino = 0;

345,606✔

3216

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

345,606✔

3217

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

345,606✔

3218

  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;

345,606✔

3219

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

345,606✔

3221

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

345,606✔

3226

  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;

345,606✔

3227

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;

345,606✔

3229

  pInfo->pTagCond = pTagCond;

345,606✔

3230

  pInfo->pGroupTags = pTableScanNode->pGroupTags;

345,606✔

3232

  pInfo->primaryTsIndex = -1;

345,606✔

3233

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

345,606✔

3234

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

345,606✔

3235

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

345,606✔

3236

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

345,234✔

3237

  QUERY_CHECK_CODE(code, lino, _error);

345,606✔

3239

  SDataBlockDescNode* pNode = (SDataBlockDescNode*)pDescNode;

345,606✔

3240

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

345,606✔

3246

  if (pTableScanNode->pTags != NULL) {

345,234✔

3258

  SOperatorInfo* pTableScanOp = NULL;

345,234✔

3259

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

345,234✔

3260

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

344,574✔

3261

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

286✔

3265

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

344,290✔

3266

  if (pHandle->version > 0) {

345,234✔

3267

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

344,948✔

3270

  STableKeyInfo* pList = NULL;

344,904✔

3271

  int32_t        num = 0;

345,234✔

3272

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

344,948✔

3273

  QUERY_CHECK_CODE(code, lino, _error);

344,273✔

3275

  if (pHandle->initTableReader) {

344,273✔

3276

    pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;

344,273✔

3277

    pTSInfo->base.dataReader = NULL;

343,987✔

3280

  if (pHandle->initTqReader) {

343,615✔

3281

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

344,276✔

3282

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

345,606✔

3284

    pInfo->tqReader = pHandle->tqReader;

2✔

3288

  pInfo->pTableScanOp = pTableScanOp;

345,606✔

3290

  pInfo->readHandle = *pHandle;

345,606✔

3291

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

345,606✔

3293

  SArray* tableIdList = NULL;

345,606✔

3294

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

345,606✔

3295

  QUERY_CHECK_CODE(code, lino, _error);

345,606✔

3296

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

345,606✔

3297

  QUERY_CHECK_CODE(code, lino, _error);

345,606✔

3298

  taosArrayDestroy(tableIdList);

345,606✔

3299

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

345,606✔

3302

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

345,606✔

3303

  QUERY_CHECK_CODE(code, lino, _error);

345,264✔

3305

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

345,264✔

3306

  QUERY_CHECK_CODE(code, lino, _error);

345,606✔

3308

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

345,606✔

3309

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

345,606✔

3311

  pInfo->pTmqScanOp = pOperator;

345,606✔

3312

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

345,606✔

3314

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

345,606✔

3316

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

345,606✔

3318

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

345,606✔

3321

  *pOptrInfo = pOperator;

345,606✔

3322

  return code;

345,606✔

3341

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

11,464,086✔

3342

  int32_t        code = TSDB_CODE_SUCCESS;

11,464,086✔

3343

  int32_t        lino = 0;

11,464,086✔

3344

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

11,464,086✔

3345

  STagScanInfo*  pInfo = pOperator->info;

11,464,321✔

3346

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

11,467,220✔

3347

  int32_t        count = pRes->info.rows;

11,468,266✔

3349

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

11,467,743✔

3350

  if (!item) {

11,459,616✔

3357

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

11,459,616✔

3358

  tDecoderClear(&(*mr).coder);

11,467,023✔

3359

  if (code != TSDB_CODE_SUCCESS) {

11,467,743✔

3365

  char str[512];

11,467,639✔

3366

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

52,464,965✔

3367

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

40,997,586✔

3370

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

40,997,946✔

3371

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

4,822,761✔

3372

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

4,821,355✔

3373

      QUERY_CHECK_CODE(code, lino, _end);

4,822,757✔

3374

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

36,176,678✔

3375

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

7,305,690✔

3376

      QUERY_CHECK_CODE(code, lino, _end);

7,305,690✔

3377

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

28,870,588✔

3378

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

15,978✔

3379

      QUERY_CHECK_CODE(code, lino, _end);

15,978✔

3381

      STagVal val = {0};

28,855,656✔

3382

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

28,856,141✔

3383

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

28,856,342✔

3385

      char* data = NULL;

28,854,453✔

3386

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

28,854,453✔

3387

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

28,851,409✔

3389

        data = (char*)p;

1,437✔

3392

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

28,852,630✔

3393

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

28,850,701✔

3394

      QUERY_CHECK_CODE(code, lino, _end);

28,854,487✔

3396

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

28,854,487✔

3398

        taosMemoryFree(data);

10,436,737✔

3403

_end:

11,470,824✔

3404

  if (code != TSDB_CODE_SUCCESS) {

11,468,751✔

3408

    pRes->info.rows++;

11,468,751✔

3411

  return code;

11,468,747✔

3414

static void tagScanFreeUidTag(void* p) {

7,382,872✔

3415

  STUidTagInfo* pInfo = p;

7,382,872✔

3416

  if (pInfo->pTagVal != NULL) {

7,382,872✔

3417

    taosMemoryFree(pInfo->pTagVal);

7,386,017✔

3419

7,381,353✔

3421

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

1,595,139✔

3422

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

1,595,139✔

3423

  if (pColumnData == NULL) {

1,594,775✔

3427

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

1,594,775✔

3428

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

1,594,635✔

3429

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

1,593,525✔

3430

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

1,592,881✔

3432

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

1,594,766✔

3433

  if (code != TSDB_CODE_SUCCESS) {

1,594,038✔

3439

  pParam->columnData = pColumnData;

1,594,038✔

3440

  pParam->colAlloced = true;

1,594,775✔

3441

  return TSDB_CODE_SUCCESS;

1,593,656✔

3444

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

9,732,139✔

3445

  int32_t                code = TSDB_CODE_SUCCESS;

9,732,139✔

3446

  int32_t                lino = 0;

9,732,139✔

3447

  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;

9,732,139✔

3448

  SColumnNode*           pSColumnNode = NULL;

9,732,139✔

3449

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

9,730,627✔

3450

    pSColumnNode = *(SColumnNode**)pNode;

2,732,450✔

3451

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

7,004,424✔

3452

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

528✔

3453

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

528✔

3466

      return DEAL_RES_CONTINUE;

528✔

3469

    return DEAL_RES_CONTINUE;

7,002,786✔

3472

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

2,732,446✔

3473

  if (!data) {

2,732,184✔

3474

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

1,596,644✔

3475

    if (code == TSDB_CODE_DUP_KEY) {

1,597,792✔

3478

    QUERY_CHECK_CODE(code, lino, _end);

1,597,792✔

3479

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

1,597,792✔

3480

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

1,598,156✔

3481

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

1,596,915✔

3482

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

1,595,543✔

3483

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

1,598,296✔

3484

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

1,597,554✔

3486

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

1,135,540✔

3487

    pSColumnNode->slotId = col->slotId;

1,135,540✔

3490

_end:

2,732,082✔

3491

  if (code != TSDB_CODE_SUCCESS) {

2,732,082✔

3495

  return DEAL_RES_CONTINUE;

2,732,082✔

3498

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

1,595,512✔

3500

  int32_t code = TSDB_CODE_SUCCESS;

1,595,512✔

3501

  int32_t lino = 0;

1,595,512✔

3502

  int32_t numOfTables = taosArrayGetSize(aUidTags);

1,595,512✔

3503

  SArray* pBlockList = NULL;

1,595,512✔

3505

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

1,595,512✔

3506

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

1,594,029✔

3508

  pBlockList = taosArrayInit(1, POINTER_BYTES);

1,594,029✔

3509

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

1,593,496✔

3511

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

1,595,512✔

3512

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

1,595,512✔

3514

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

1,595,512✔

3516

  SScalarParam output = {0};

1,595,139✔

3517

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

1,595,008✔

3518

  QUERY_CHECK_CODE(code, lino, _end);

1,592,788✔

3520

  gTaskScalarExtra.pStreamInfo = (void*)pStreamRuntimeInfo;

1,592,788✔

3521

  gTaskScalarExtra.pStreamRange = NULL;

1,592,788✔

3522

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

1,592,284✔

3523

  QUERY_CHECK_CODE(code, lino, _end);

1,592,162✔

3525

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

1,592,162✔

3526

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

3,983,912✔

3527

    if (result[i]) {

2,391,629✔

3528

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

2,227,907✔

3529

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

2,227,907✔

3533

  colDataDestroy(output.columnData);

1,593,902✔

3534

  taosMemoryFreeClear(output.columnData);

1,592,293✔

3536

_end:

1,589,829✔

3537

  blockDataDestroy(pResBlock);

1,591,079✔

3538

  taosArrayDestroy(pBlockList);

1,591,276✔

3539

  if (code != TSDB_CODE_SUCCESS) {

1,588,701✔

3542

  return code;

1,588,701✔

3545

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

30,055,347✔

3548

  int32_t code = TSDB_CODE_SUCCESS;

30,055,347✔

3549

  int32_t lino = 0;

30,055,347✔

3550

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

30,055,347✔

3551

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

14,330,136✔

3557

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

14,318,982✔

3558

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

7,160,886✔

3559

      QUERY_CHECK_CODE(code, lino, _end);

7,165,186✔

3560

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

7,165,574✔

3561

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

7,161,074✔

3562

      QUERY_CHECK_CODE(code, lino, _end);

7,164,557✔

3565

    STagVal tagVal = {0};

15,724,742✔

3566

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

15,725,769✔

3567

    if (pUidTagInfo->pTagVal == NULL) {

15,724,250✔

3570

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

15,719,285✔

3572

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

15,728,014✔

3573

        colDataSetNULL(pColInfo, rowIndex);

8,205✔

3574

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

15,719,809✔

3577

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

19,712,072✔

3578

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

4,006,620✔

3581

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

3,994,885✔

3582

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

3,994,284✔

3584

        varDataSetLen(tmp, tagVal.nData);

3,994,284✔

3585

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

3,992,670✔

3586

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

3,996,389✔

3587

        taosMemoryFree(tmp);

3,994,436✔

3588

        QUERY_CHECK_CODE(code, lino, _end);

3,995,160✔

3590

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

11,722,405✔

3591

        QUERY_CHECK_CODE(code, lino, _end);

11,721,778✔

3596

_end:

30,058,290✔

3597

  if (code != TSDB_CODE_SUCCESS) {

30,058,290✔

3600

  return code;

30,052,646✔

3603

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

4,722,414✔

3605

  int32_t       code = TSDB_CODE_SUCCESS;

4,722,414✔

3606

  int32_t       lino = 0;

4,722,414✔

3607

  STagScanInfo* pInfo = pOperator->info;

4,722,414✔

3608

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

4,725,798✔

3609

  if (!ignoreFilterIdx) {

4,721,931✔

3610

    size_t szTables = taosArrayGetSize(aFilterIdxs);

1,591,171✔

3611

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

3,818,425✔

3612

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

2,226,658✔

3613

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);

2,227,539✔

3614

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

2,226,518✔

3615

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

12,736,380✔

3616

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

10,508,233✔

3617

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

10,506,190✔

3618

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

10,506,190✔

3619

        QUERY_CHECK_CODE(code, lino, _end);

10,509,862✔

3623

    size_t szTables = taosArrayGetSize(aUidTags);

3,130,760✔

3624

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

8,128,816✔

3625

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);

4,998,288✔

3626

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

4,998,676✔

3627

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

24,540,738✔

3628

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

19,543,436✔

3629

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

19,544,605✔

3630

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

19,544,605✔

3631

        QUERY_CHECK_CODE(code, lino, _end);

19,541,674✔

3636

_end:

3,130,528✔

3637

  if (code != TSDB_CODE_SUCCESS) {

4,722,295✔

3640

  return code;

4,725,406✔

3643

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

9,409,537✔

3644

  int32_t        code = TSDB_CODE_SUCCESS;

9,409,537✔

3645

  int32_t        lino = 0;

9,409,537✔

3646

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,409,537✔

3647

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

9,410,552✔

3649

  STagScanInfo* pInfo = pOperator->info;

9,410,203✔

3650

  SSDataBlock*  pRes = pInfo->pRes;

9,409,541✔

3652

  QRY_PARAM_CHECK(ppRes);

9,409,580✔

3654

  if (pOperator->status == OP_EXEC_DONE) {

9,410,588✔

3655

    return TSDB_CODE_SUCCESS;

4,608,763✔

3657

  blockDataCleanup(pRes);

4,799,556✔

3659

  if (pInfo->pCtbCursor == NULL) {

4,801,539✔

3660

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

4,802,780✔

3661

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

4,802,508✔

3667

  SArray* aUidTags = pInfo->aUidTags;

4,802,144✔

3668

  SArray* aFilterIdxs = pInfo->aFilterIdxs;

4,802,120✔

3669

  int32_t count = 0;

4,802,508✔

3670

  bool    ctbCursorFinished = false;

4,802,508✔

3671

  while (1) {

106,092✔

3672

    taosArrayClearEx(aUidTags, tagScanFreeUidTag);

4,908,600✔

3673

    taosArrayClear(aFilterIdxs);

4,909,869✔

3675

    int32_t numTables = 0;

4,910,257✔

3676

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

12,295,906✔

3677

      SMCtbCursor* pCur = pInfo->pCtbCursor;

12,300,930✔

3678

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

12,299,184✔

3679

      if (uid == 0) {

12,295,695✔

3680

        ctbCursorFinished = true;

4,909,884✔

3681

        break;

4,909,884✔

3683

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

7,385,811✔

3684

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

7,384,641✔

3685

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

7,386,649✔

3687

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

7,386,649✔

3688

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

7,390,557✔

3689

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

7,390,557✔

3690

      ++numTables;

7,390,557✔

3693

    if (numTables == 0) {

4,910,272✔

3694

      break;

183,597✔

3696

    bool ignoreFilterIdx = true;

4,726,675✔

3697

    if (pInfo->pTagCond != NULL) {

4,726,675✔

3698

      ignoreFilterIdx = false;

1,595,512✔

3699

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

1,595,512✔

3700

      QUERY_CHECK_CODE(code, lino, _end);

1,591,649✔

3702

      ignoreFilterIdx = true;

3,130,387✔

3705

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

4,722,036✔

3706

    QUERY_CHECK_CODE(code, lino, _end);

4,725,406✔

3708

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

4,725,406✔

3710

    if (count != 0) {

4,725,018✔

3711

      break;

4,618,926✔

3715

  if (count > 0) {

4,802,523✔

3716

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

4,618,926✔

3718

  if (count == 0 || ctbCursorFinished) {

4,802,911✔

3719

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

4,802,547✔

3720

    pInfo->pCtbCursor = NULL;

4,797,990✔

3721

    setOperatorCompleted(pOperator);

4,800,005✔

3724

  pRes->info.rows = count;

4,798,066✔

3725

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

4,797,702✔

3726

  if (bLimitReached) {

4,800,015✔

3730

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

4,800,015✔

3731

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

4,800,614✔

3733

_end:

4,799,998✔

3734

  if (code != TSDB_CODE_SUCCESS) {

4,799,998✔

3740

  return code;

4,799,998✔

3743

static int32_t createTagScanTableListInfoFromParam(SOperatorInfo* pOperator) {

3,634,861✔

3744

  STagScanInfo*          pInfo = pOperator->info;

3,634,861✔

3745

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

3,634,861✔

3746

  int32_t                code = 0;

3,634,861✔

3747

  STableListInfo*        pListInfo = pInfo->pTableListInfo;

3,634,861✔

3748

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

3,634,861✔

3749

  tb_uid_t               pUid = pParam->vcUid;

3,634,861✔

3755

  STableKeyInfo info = {.groupId = 0};

3,634,861✔

3757

  int32_t tableIdx = 0;

3,634,861✔

3758

  taosHashClear(pListInfo->map);

3,634,861✔

3759

  taosArrayClear(pListInfo->pTableList);

3,634,861✔

3761

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

3,634,861✔

3768

  info.uid = pUid;

3,634,861✔

3769

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

3,634,861✔

3770

  if (p == NULL) {

3,634,861✔

3774

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

3,634,861✔

3776

  return code;

3,634,861✔

3779

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

7,790,919✔

3780

  int32_t       code = TSDB_CODE_SUCCESS;

7,790,919✔

3781

  int32_t       lino = 0;

7,790,919✔

3782

  STagScanInfo* pInfo = pOperator->info;

7,790,919✔

3783

  if (pOperator->pOperatorGetParam) {

7,790,745✔

3784

    pOperator->resultInfo.totalRows = 0;

3,634,861✔

3785

    pOperator->dynamicTask = true;

3,634,861✔

3786

    pInfo->curPos = 0;

3,634,861✔

3787

    code = createTagScanTableListInfoFromParam(pOperator);

3,634,861✔

3788

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

3,634,861✔

3789

    pOperator->pOperatorGetParam = NULL;

3,634,861✔

3790

    QUERY_CHECK_CODE(code, lino, _end);

3,634,861✔

3792

    if (pOperator->status == OP_EXEC_DONE) {

3,634,861✔

3793

      pOperator->status = OP_OPENED;

1,961,589✔

3797

  if (pOperator->status == OP_EXEC_DONE) {

7,789,949✔

3798

    (*ppRes) = NULL;

2,063,157✔

3799

    return code;

2,063,157✔

3802

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

5,726,432✔

3803

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

5,728,361✔

3805

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

5,727,315✔

3806

  SSDataBlock* pRes = pInfo->pRes;

5,728,361✔

3807

  blockDataCleanup(pRes);

5,724,340✔

3809

  int32_t size = 0;

5,727,315✔

3810

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

5,727,838✔

3811

  QUERY_CHECK_CODE(code, lino, _end);

5,724,662✔

3813

  if (size == 0) {

5,724,662✔

3814

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

19,465✔

3815

    (*ppRes) = NULL;

19,465✔

3816

    return code;

19,465✔

3819

  SMetaReader mr = {0};

5,705,197✔

3820

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

5,707,289✔

3821

  pRes->info.rows = 0;

5,709,419✔

3823

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

17,176,677✔

3824

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

11,465,579✔

3825

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

11,467,349✔

3828

      code = TSDB_CODE_SUCCESS;

11,468,789✔

3830

    QUERY_CHECK_CODE(code, lino, _end);

11,467,349✔

3832

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

11,467,349✔

3833

      setOperatorCompleted(pOperator);

5,709,942✔

3837

  pAPI->metaReaderFn.clearReader(&mr);

5,709,004✔

3838

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

5,709,942✔

3839

  if (bLimitReached) {

5,709,942✔

3844

  if (pOperator->status == OP_EXEC_DONE) {

5,709,942✔

3845

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

5,709,942✔

3848

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

5,709,942✔

3850

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

5,709,942✔

3851

_end:

5,709,942✔

3852

  if (code != TSDB_CODE_SUCCESS) {

5,709,942✔

3857

  return code;

5,709,942✔

3860

static void destroyTagScanOperatorInfo(void* param) {

8,907,814✔

3861

  STagScanInfo* pInfo = (STagScanInfo*)param;

8,907,814✔

3862

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

8,907,814✔

3865

  taosHashCleanup(pInfo->filterCtx.colHash);

8,912,169✔

3866

  taosArrayDestroy(pInfo->filterCtx.cInfoList);

8,894,566✔

3867

  taosArrayDestroy(pInfo->aFilterIdxs);

8,899,212✔

3868

  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);

8,899,716✔

3870

  blockDataDestroy(pInfo->pRes);

8,904,360✔

3871

  taosArrayDestroy(pInfo->matchInfo.pList);

8,907,285✔

3872

  tableListDestroy(pInfo->pTableListInfo);

8,901,270✔

3874

  pInfo->pRes = NULL;

8,912,557✔

3875

  pInfo->pTableListInfo = NULL;

8,913,285✔

3876

  taosMemoryFreeClear(param);

8,910,286✔

3877

8,909,199✔

3879

static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {

4,896✔

3880

  int32_t         code = TSDB_CODE_SUCCESS;

4,896✔

3881

  STagScanInfo*   pInfo = pOper->info;

4,896✔

3883

  pOper->status = OP_NOT_OPENED;

4,896✔

3884

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

4,896✔

3885

  if (pTagScanNode->onlyMetaCtbIdx) {

4,896✔

3886

      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

2,112✔

3887

      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,112✔

3888

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

2,112✔

3889

      pInfo->pCtbCursor = NULL;

2,112✔

3891

      pInfo->curPos = 0;

2,784✔

3894

  return code;

4,896✔

3897

int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,

8,911,980✔

3900

  QRY_PARAM_CHECK(pOptrInfo);

8,911,980✔

3902

  int32_t         code = TSDB_CODE_SUCCESS;

8,914,507✔

3903

  int32_t         lino = 0;

8,914,507✔

3904

  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;

8,914,507✔

3905

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

8,914,507✔

3906

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

8,905,587✔

3907

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

8,898,976✔

3908

    code = terrno;

1,078✔

3911

  pOperator->pPhyNode = pTagScanNode;

8,897,898✔

3912

  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

8,899,487✔

3914

  int32_t    numOfExprs = 0;

8,905,512✔

3915

  SExprInfo* pExprInfo = NULL;

8,909,225✔

3917

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

8,904,718✔

3918

  QUERY_CHECK_CODE(code, lino, _error);

8,906,301✔

3920

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

8,906,301✔

3921

  QUERY_CHECK_CODE(code, lino, _error);

8,878,515✔

3923

  int32_t num = 0;

8,878,515✔

3924

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

8,891,835✔

3925

  QUERY_CHECK_CODE(code, lino, _error);

8,899,000✔

3927

  pInfo->pTagCond = pTagCond;

8,899,000✔

3928

  pInfo->pTagIndexCond = pTagIndexCond;

8,899,887✔

3929

  pInfo->suid = pPhyNode->suid;

8,897,052✔

3930

  pInfo->pStorageAPI = &pTaskInfo->storageAPI;

8,907,235✔

3932

  pInfo->pTableListInfo = pTableListInfo;

8,889,761✔

3933

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

8,904,542✔

3934

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

8,908,783✔

3936

  pInfo->readHandle = *pReadHandle;

8,896,432✔

3937

  pInfo->curPos = 0;

8,909,154✔

3939

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

8,886,030✔

3940

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

8,894,464✔

3942

  initResultSizeInfo(&pOperator->resultInfo, 4096);

8,888,441✔

3943

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

8,888,355✔

3944

  QUERY_CHECK_CODE(code, lino, _error);

8,915,067✔

3946

  if (pTagScanNode->onlyMetaCtbIdx) {

8,915,067✔

3947

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

4,807,664✔

3948

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

4,809,452✔

3950

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

4,809,448✔

3951

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

4,809,060✔

3953

    pInfo->filterCtx.colHash =

4,807,414✔

3954

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

4,809,448✔

3955

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

4,808,433✔

3957

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

4,806,653✔

3958

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

4,806,149✔

3960

    if (pInfo->pTagCond != NULL) {

4,805,785✔

3961

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

1,599,168✔

3964

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

8,912,750✔

3965

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

8,912,867✔

3967

  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);

8,906,477✔

3969

  *pOptrInfo = pOperator;

8,907,759✔

3970

  return code;

8,907,529✔

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,179,717,801✔

3997

  } else if (rightIdx == -1) {

1,421,992,850✔

3998

    return -1;

442,186✔

4001

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

1,421,550,664✔

4002

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

1,421,837,736✔

4003

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

1,422,155,982✔

4004

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

1,422,155,982✔

4007

  return ret;

1,422,017,380✔

4037

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

3,365,853✔

4038

  int32_t code = TSDB_CODE_SUCCESS;

3,365,853✔

4039

  int32_t lino = 0;

3,365,853✔

4040

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

3,365,853✔

4041

  dst->colList = NULL;

3,365,853✔

4042

  dst->pSlotList = NULL;

3,362,186✔

4044

  if (src->numOfCols > 0) {

3,363,332✔

4045

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

3,360,020✔

4046

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

3,362,869✔

4047

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

3,341,721✔

4049

    if (src->pSlotList != NULL) {

3,351,576✔

4050

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

3,366,599✔

4051

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

3,360,011✔

4052

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

3,355,643✔

4057

  if (code != TSDB_CODE_SUCCESS) {

3,349,747✔

4062

  return code;

3,365,862✔

4065

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

5,340,595✔

4067

  int32_t code = 0;

5,340,595✔

4069

  STableMergeScanInfo*    pInfo = pOperator->info;

5,340,595✔

4070

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

5,339,858✔

4071

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

5,340,595✔

4072

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

5,343,543✔

4073

  const SStorageAPI*      pAPI = &pTaskInfo->storageAPI;

5,339,130✔

4075

  blockDataCleanup(pInput->pReaderBlock);

5,331,787✔

4076

  if (!pInput->bInMemReader) {

5,335,881✔

4077

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

5,211,129✔

4078

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

2,605,578✔

4079

    if (code != 0) {

2,605,933✔

4084

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

5,337,328✔

4086

  while (true) {

8,952✔

4087

    bool hasNext = false;

5,349,174✔

4088

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

5,349,165✔

4089

    if (code != 0) {

5,354,342✔

4095

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

5,354,342✔

4096

      if (isTaskKilled(pTaskInfo)) {

3,319,118✔

4102

      *pSubTableHasBlock = false;

3,319,118✔

4103

      break;

3,319,473✔

4106

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

2,035,224✔

4107

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

2,009,860✔

4109

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

25,364✔

4112

    uint32_t status = 0;

2,035,224✔

4113

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

2,035,224✔

4114

    if (code != 0) {

2,035,224✔

4119

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

2,035,224✔

4123

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

2,035,224✔

4124

      continue;

8,952✔

4127

    *pSubTableHasBlock = true;

2,025,535✔

4128

    break;

2,025,526✔

4131

  if (*pSubTableHasBlock) {

5,344,289✔

4132

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

4,052,544✔

4133

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

4,051,434✔

4134

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

2,026,272✔

4136

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

5,345,754✔

4137

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

3,893,116✔

4138

    pInput->pReader = NULL;

3,890,932✔

4141

  pInfo->base.dataReader = NULL;

5,343,561✔

4142

  return TSDB_CODE_SUCCESS;

5,342,105✔

4145

static int32_t setGroupStartEndIndex(STableMergeScanInfo* pInfo) {

2,018,049✔

4146

  pInfo->bGroupProcessed = false;

2,018,049✔

4148

  int32_t numOfTables = 0;

2,010,751✔

4149

  int32_t code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

2,005,592✔

4150

  if (code != TSDB_CODE_SUCCESS) {

1,993,908✔

4155

  int32_t i = pInfo->tableStartIndex + 1;

1,993,908✔

4156

  for (; i < numOfTables; ++i) {

3,280,345✔

4157

    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

1,339,559✔

4158

    if (!tableKeyInfo) {

1,335,083✔

4162

    if (tableKeyInfo->groupId != pInfo->groupId) {

1,335,083✔

4163

      break;

58,227✔

4166

  pInfo->tableEndIndex = i - 1;

1,999,013✔

4167

  return TSDB_CODE_SUCCESS;

2,028,659✔

4170

static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {

2,037,404✔

4171

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,351,736✔

4172

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,317,999✔

4173

    if (pInput->rowIdx == -1) {

3,315,069✔

4174

      continue;

1,592,653✔

4177

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,723,872✔

4178

      pInput->rowIdx = 0;

1,724,991✔

4179

      pInput->pageIdx = -1;

1,725,355✔

4182

    pInput->pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,724,973✔

4183

    SColumnInfoData* col = taosArrayGet(pInput->pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

1,724,609✔

4184

    if (!col) {

1,723,872✔

4188

    pInput->aTs = (int64_t*)col->pData;

1,723,872✔

4191

  __merge_compare_fn_t mergeCompareFn = (!pSubTblsInfo->pPkOrderInfo) ? subTblRowCompareTsFn : subTblRowCompareTsPkFn;

2,034,119✔

4192

  return tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, mergeCompareFn);

2,037,768✔

4195

static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {

2,019,523✔

4196

  int32_t code = setGroupStartEndIndex(pInfo);

2,019,523✔

4197

  if (code != TSDB_CODE_SUCCESS) {

2,022,035✔

4201

  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));

2,022,035✔

4202

  if (pSubTblsInfo == NULL) {

2,004,855✔

4205

  pSubTblsInfo->pTsOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);

2,004,855✔

4206

  if (!pSubTblsInfo->pTsOrderInfo) {

2,019,496✔

4210

  if (taosArrayGetSize(pInfo->pSortInfo) == 2) {

1,996,820✔

4217

    pSubTblsInfo->pPkOrderInfo = NULL;

1,990,250✔

4219

  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

2,015,883✔

4220

  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));

2,016,930✔

4221

  if (pSubTblsInfo->aInputs == NULL) {

1,991,670✔

4225

  int32_t bufPageSize = pInfo->bufPageSize;

2,014,746✔

4226

  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;

2,014,000✔

4227

  code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);

2,008,540✔

4228

  if (code != TSDB_CODE_SUCCESS) {

2,000,869✔

4233

  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;

2,000,869✔

4234

  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;

2,018,431✔

4236

  pInfo->pSubTablesMergeInfo = pSubTblsInfo;

2,022,053✔

4237

  return TSDB_CODE_SUCCESS;

1,997,166✔

4250

static int32_t setSubTableCondSuidByUid(STableMergeScanInfo* pInfo, SStorageAPI* pAPI, uint64_t uid,

3,309,218✔

4252

  int32_t     code = TSDB_CODE_SUCCESS;

3,309,218✔

4253

  int32_t     lino = 0;

3,309,218✔

4254

  SMetaReader tableReader = {0};

3,309,218✔

4256

  QUERY_CHECK_NULL(pInfo, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,311,065✔

4257

  QUERY_CHECK_NULL(pAPI, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,311,065✔

4258

  QUERY_CHECK_NULL(pCond, code, lino, _end, TSDB_CODE_INVALID_PARA);

3,311,065✔

4260

  pAPI->metaReaderFn.initReader(&tableReader, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

3,311,065✔

4261

  code = pAPI->metaReaderFn.getTableEntryByUid(&tableReader, uid);

3,312,175✔

4262

  pAPI->metaReaderFn.readerReleaseLock(&tableReader);

3,303,358✔

4263

  QUERY_CHECK_CODE(code, lino, _end);

3,305,942✔

4265

  if (tableReader.me.type == TSDB_CHILD_TABLE) {

3,305,942✔

4266

    pCond->suid = tableReader.me.ctbEntry.suid;

3,200,282✔

4267

  } else if (tableReader.me.type == TSDB_NORMAL_TABLE) {

105,660✔

4268

    pCond->suid = 0;

105,660✔

4274

_end:

3,312,903✔

4275

  if (code != TSDB_CODE_SUCCESS) {

3,298,990✔

4278

  pAPI->metaReaderFn.clearReader(&tableReader);

3,298,990✔

4279

  return code;

3,297,862✔

4282

static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {

2,020,597✔

4283

  int32_t        code = TSDB_CODE_SUCCESS;

2,020,597✔

4284

  int32_t        lino = 0;

2,020,597✔

4285

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

2,020,597✔

4286

  SReadHandle*   pHandle = &pInfo->base.readHandle;

2,005,201✔

4287

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,018,786✔

4289

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

2,014,018✔

4291

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,311,125✔

4292

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,303,421✔

4293

    pInput->type = SUB_TABLE_MEM_BLOCK;

3,282,529✔

4295

    code = dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);

3,290,919✔

4296

    QUERY_CHECK_CODE(code, lino, _end);

3,299,026✔

4298

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pReaderBlock);

3,299,026✔

4299

    QUERY_CHECK_CODE(code, lino, _end);

3,317,653✔

4301

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pPageBlock);

3,317,653✔

4302

    QUERY_CHECK_CODE(code, lino, _end);

3,319,118✔

4304

    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);

3,319,118✔

4305

    pInput->pKeyInfo = keyInfo;

3,314,714✔

4306

    QUERY_CHECK_NULL(pInput->pKeyInfo, code, lino, _end, terrno);

3,312,521✔

4308

    code = setSubTableCondSuidByUid(pInfo, pAPI, pInput->pKeyInfo->uid, &pInput->tblCond);

3,309,600✔

4309

    QUERY_CHECK_CODE(code, lino, _end);

3,297,498✔

4311

    if (isTaskKilled(pTaskInfo)) {

3,297,498✔

4315

    if (i + 1 < pSubTblsInfo->numInMemReaders) {

3,303,731✔

4316

      code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, keyInfo, 1, pInput->pReaderBlock,

2,561,172✔

4317

                                           (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

1,280,586✔

4318

      QUERY_CHECK_CODE(code, lino, _end);

1,277,602✔

4319

      pInput->bInMemReader = true;

1,277,602✔

4321

      pInput->pReader = NULL;

2,029,715✔

4322

      pInput->bInMemReader = false;

2,033,364✔

4324

    bool hasNext = true;

3,307,326✔

4325

    code = fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);

3,309,537✔

4326

    QUERY_CHECK_CODE(code, lino, _end);

3,316,916✔

4327

    if (!hasNext) {

3,316,916✔

4328

      pInput->rowIdx = -1;

1,591,561✔

4329

      ++pSubTblsInfo->numSubTablesCompleted;

1,591,206✔

4330

      continue;

1,593,381✔

4332

      pInput->rowIdx = 0;

1,725,355✔

4333

      pInput->pageIdx = -1;

1,725,728✔

4337

_end:

2,037,777✔

4338

  if (code != TSDB_CODE_SUCCESS) {

2,038,523✔

4341

  return code;

2,037,395✔

4344

static void adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

2,026,272✔

4345

  int32_t              code = TSDB_CODE_SUCCESS;

2,026,272✔

4346

  int32_t              lino = 0;

2,026,272✔

4347

  SExecTaskInfo*       pTaskInfo = pOperatorInfo->pTaskInfo;

2,026,272✔

4348

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

2,026,272✔

4349

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

2,026,272✔

4350

  bool                 hasNext = true;

2,026,272✔

4351

  code = fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);

2,026,272✔

4352

  QUERY_CHECK_CODE(code, lino, _end);

2,026,272✔

4354

  if (!hasNext) {

2,026,272✔

4355

    pInput->rowIdx = -1;

1,725,728✔

4356

    ++pSubTblsInfo->numSubTablesCompleted;

1,725,728✔

4358

    pInput->rowIdx = 0;

300,544✔

4361

_end:

2,026,272✔

4362

  if (code != TSDB_CODE_SUCCESS) {

2,026,272✔

4367

2,026,272✔

4369

static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

1,788,303,027✔

4370

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

1,788,303,027✔

4371

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,788,356,074✔

4373

  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,788,341,929✔

4374

  if (pInput->rowIdx < pInputBlock->info.rows - 1) {

1,788,569,925✔

4375

    ++pInput->rowIdx;

1,786,509,317✔

4376

  } else if (pInput->rowIdx == pInputBlock->info.rows - 1) {

2,026,272✔

4377

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

2,026,272✔

4378

      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);

2,026,272✔

4380

    if (pInput->rowIdx != -1) {

2,026,272✔

4381

      SColumnInfoData* col = taosArrayGet(pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

300,544✔

4382

      if (!col) {

300,544✔

4386

      pInput->pInputBlock = pInputBlock;

300,544✔

4387

      pInput->aTs = (int64_t*)col->pData;

300,544✔

4391

  return tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));

1,788,546,614✔

4394

static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {

1,787,869,821✔

4395

  int32_t            code = TSDB_CODE_SUCCESS;

1,787,869,821✔

4396

  int32_t            lino = 0;

1,787,869,821✔

4397

  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,787,869,821✔

4398

  SSDataBlock*       pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,787,793,361✔

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);

87,064,783✔

4410

      QUERY_CHECK_CODE(code, lino, _end);

87,059,561✔

4412

      if (pSrcColInfo->pData != NULL) {

2,147,483,647✔

4413

        char* pData = colDataGetData(pSrcColInfo, pInput->rowIdx);

2,147,483,647✔

4414

        code = colDataSetVal(pColInfo, pBlock->info.rows, pData, false);

2,147,483,647✔

4415

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

4419

  pBlock->info.dataLoad = 1;

1,786,503,167✔

4420

  pBlock->info.scanFlag = pInputBlock->info.scanFlag;

1,788,394,158✔

4421

  pBlock->info.rows += 1;

1,788,390,629✔

4423

_end:

1,788,386,408✔

4424

  if (code != TSDB_CODE_SUCCESS) {

1,788,386,408✔

4427

  return code;

1,788,363,069✔

4430

static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,

4,761,444✔

4432

  int32_t                 code = TSDB_CODE_SUCCESS;

4,761,444✔

4433

  int32_t                 lino = 0;

4,761,444✔

4434

  STableMergeScanInfo*    pInfo = pOperator->info;

4,761,444✔

4435

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

4,760,707✔

4436

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

4,761,435✔

4437

  bool                    finished = false;

4,758,896✔

4439

  QRY_PARAM_CHECK(pResBlock);

4,758,896✔

4441

  blockDataCleanup(pBlock);

4,761,444✔

4445

      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {

1,789,397,247✔

4446

        finished = true;

3,322,493✔

4447

        break;

3,322,493✔

4450

      code = appendChosenRowToDataBlock(pSubTblsInfo, pBlock);

1,787,901,705✔

4451

      QUERY_CHECK_CODE(code, lino, _end);

1,788,243,874✔

4453

      code = adjustSubTableForNextRow(pOperator, pSubTblsInfo);

1,788,243,874✔

4454

      QUERY_CHECK_CODE(code, lino, _end);

1,787,804,183✔

4456

      if (pBlock->info.rows >= capacity) {

1,787,804,183✔

4457

        break;

1,439,324✔

4461

    if (isTaskKilled(pTaskInfo)) {

4,761,817✔

4465

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pBlock, pTaskInfo);

4,761,817✔

4466

    if (finished || limitReached || pBlock->info.rows > 0) {

4,761,808✔

4471

  if (pBlock->info.rows > 0) {

4,761,808✔

4472

    *pResBlock = pBlock;

2,724,040✔

4475

_end:

2,037,768✔

4476

  if (code != TSDB_CODE_SUCCESS) {

4,761,808✔

4480

  return code;

4,761,808✔

4483

static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {

2,019,869✔

4484

  int32_t              code = TSDB_CODE_SUCCESS;

2,019,869✔

4485

  int32_t              lino = 0;

2,019,869✔

4486

  STableMergeScanInfo* pInfo = pOperator->info;

2,019,869✔

4488

  code = initSubTablesMergeInfo(pInfo);

2,024,610✔

4489

  QUERY_CHECK_CODE(code, lino, _end);

1,992,744✔

4491

  code = initSubTableInputs(pOperator, pInfo);

1,992,744✔

4492

  QUERY_CHECK_CODE(code, lino, _end);

2,036,676✔

4494

  code = openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);

2,036,676✔

4495

  QUERY_CHECK_CODE(code, lino, _end);

2,035,930✔

4497

_end:

2,035,930✔

4498

  if (code != TSDB_CODE_SUCCESS) {

2,035,930✔

4501

  return code;

2,036,303✔

4504

static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {

30,399,644✔

4505

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

30,399,644✔

4506

  if (pSubTblsInfo != NULL) {

30,408,999✔

4507

    tMergeTreeDestroy(&pSubTblsInfo->pTree);

2,036,321✔

4509

    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

5,356,158✔

4510

      STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,319,846✔

4511

      cleanupQueryTableDataCond(&pInput->tblCond);

3,318,026✔

4512

      blockDataDestroy(pInput->pReaderBlock);

3,318,754✔

4513

      blockDataDestroy(pInput->pPageBlock);

3,319,482✔

4514

      taosArrayDestroy(pInput->aBlockPages);

3,319,846✔

4515

      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);

3,319,482✔

4516

      pInput->pReader = NULL;

3,318,745✔

4519

    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);

2,037,413✔

4520

    taosMemoryFree(pSubTblsInfo->aInputs);

2,034,865✔

4522

    taosMemoryFree(pSubTblsInfo);

2,036,321✔

4523

    pInfo->pSubTablesMergeInfo = NULL;

2,035,957✔

4527

30,408,635✔

4529

void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

32,622,786✔

4530

  STableMergeScanInfo* pInfo = pOperator->info;

32,622,786✔

4531

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

32,627,928✔

4532

  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));

32,625,259✔

4534

  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

32,632,679✔

4535

  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;

32,620,429✔

4536

  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;

32,624,491✔

4537

  pInfo->sortExecInfo.loops += sortExecInfo.loops;

32,617,177✔

4538

  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;

32,611,494✔

4539

  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

32,616,559✔

4541

  tsortDestroySortHandle(pInfo->pSortHandle);

32,605,560✔

4542

  pInfo->pSortHandle = NULL;

32,614,052✔

4543

32,616,557✔

4552

static int32_t createVTableMergeScanInfoFromBatchParam(SOperatorInfo* pOperator) {

65,744✔

4553

  int32_t                  code = TSDB_CODE_SUCCESS;

65,744✔

4554

  int32_t                  lino = 0;

65,744✔

4555

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

65,744✔

4556

  STableMergeScanInfo*     pInfo = pOperator->info;

65,744✔

4557

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

65,744✔

4558

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

65,744✔

4559

  SMetaReader              orgTable = {0};

65,744✔

4560

  SOperatorParam*          pGetParam = pOperator->pOperatorGetParam;

65,744✔

4561

  STableScanOperatorParam* pParam = (STableScanOperatorParam*)pGetParam->value;

65,744✔

4563

  if (pParam->paramType != DYN_TYPE_SCAN_PARAM) {

65,744✔

4567

  if (pInfo->pSubTablesMergeInfo) {

65,744✔

4570

  if (pInfo->pSortHandle) {

65,744✔

4573

  if (pInfo->base.dataReader) {

65,744✔

4578

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

65,744✔

4584

  pInfo->numNextDurationBlocks = 0;

65,744✔

4585

  pInfo->nextDurationBlocksIdx = 0;

65,744✔

4586

  pInfo->rtnNextDurationBlocks = false;

65,744✔

4587

  pInfo->bNewFilesetEvent = false;

65,744✔

4588

  pInfo->bNextDurationBlockEvent = false;

65,744✔

4589

  taosHashCleanup(pInfo->mSkipTables);

65,744✔

4590

  pInfo->mSkipTables = NULL;

65,744✔

4592

  if (pInfo->base.orgCond.colList && pInfo->base.orgCond.numOfCols > 0) {

65,744✔

4593

    cleanupQueryTableDataCond(&pInfo->base.cond);

65,744✔

4594

    code = dumpQueryTableCond(&pInfo->base.orgCond, &pInfo->base.cond);

65,744✔

4595

    QUERY_CHECK_CODE(code, lino, _return);

65,744✔

4598

  if (pParam->window.skey == INT64_MAX && pParam->window.ekey == INT64_MIN) {

65,744✔

4599

    if (pInfo->base.orgCond.colList && pInfo->base.orgCond.numOfCols > 0) {

65,744✔

4600

      pInfo->base.cond.twindows.skey = pInfo->base.orgCond.twindows.skey;

65,744✔

4601

      pInfo->base.cond.twindows.ekey = pInfo->base.orgCond.twindows.ekey;

65,744✔

4608

  pInfo->tableStartIndex = 0;

65,744✔

4609

  pInfo->tableEndIndex = 0;

65,744✔

4610

  pInfo->hasGroupId = false;

65,744✔

4611

  pInfo->groupId = pParam->groupid;

65,744✔

4612

  pInfo->bGroupProcessed = false;

65,744✔

4613

  pOperator->status = OP_NOT_OPENED;

65,744✔

4615

  pListInfo->oneTableForEachGroup = true;

65,744✔

4616

  pListInfo->numOfOuputGroups = 1;

65,744✔

4617

  taosHashClear(pListInfo->map);

65,744✔

4618

  taosArrayClear(pListInfo->pTableList);

65,744✔

4620

  int32_t tableIdx = 0;

65,744✔

4621

  if (pParam->pBatchTbInfo) {

65,744✔

4622

    int32_t numOfBatchTables = taosArrayGetSize(pParam->pBatchTbInfo);

65,744✔

4623

    for (int32_t i = 0; i < numOfBatchTables; ++i) {

231,278✔

4624

      SOrgTbInfo* pOrgTbInfo = taosArrayGet(pParam->pBatchTbInfo, i);

165,534✔

4625

      QUERY_CHECK_NULL(pOrgTbInfo, code, lino, _return, terrno);

165,534✔

4627

      const char* pTableName = strstr(pOrgTbInfo->tbName, ".");

165,534✔

4628

      pTableName = pTableName ? (pTableName + 1) : pOrgTbInfo->tbName;

165,534✔

4629

      pAPI->metaReaderFn.initReader(&orgTable, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

165,534✔

4630

      code = pAPI->metaReaderFn.getTableEntryByName(&orgTable, pTableName);

165,534✔

4631

      pAPI->metaReaderFn.readerReleaseLock(&orgTable);

165,534✔

4632

      QUERY_CHECK_CODE(code, lino, _return);

165,534✔

4634

      uint64_t uid = orgTable.me.uid;

165,534✔

4635

      code = taosHashPut(pListInfo->map, &uid, sizeof(uid), &tableIdx, sizeof(tableIdx));

165,534✔

4636

      if (code == TSDB_CODE_DUP_KEY) {

165,534✔

4640

      QUERY_CHECK_CODE(code, lino, _return);

165,534✔

4642

      STableKeyInfo keyInfo = {.groupId = pParam->groupid, .uid = uid};

165,534✔

4643

      QUERY_CHECK_NULL(taosArrayPush(pListInfo->pTableList, &keyInfo), code, lino, _return, terrno);

331,068✔

4644

      ++tableIdx;

165,534✔

4646

      qDebug("%s table merge dynamic batch add table name:%s uid:%" PRIu64 " group:%" PRIu64, GET_TASKID(pTaskInfo),

165,534✔

4648

      pAPI->metaReaderFn.clearReader(&orgTable);

165,534✔

4652

  qDebug("%s table merge dynamic batch param consumed, group:%" PRIu64 ", tables:%d", GET_TASKID(pTaskInfo),

65,744✔

4655

_return:

65,744✔

4656

  if (code != TSDB_CODE_SUCCESS) {

65,744✔

4659

  pAPI->metaReaderFn.clearReader(&orgTable);

65,744✔

4660

  return code;

65,744✔

4663

int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

4,728,195✔

4664

  int32_t              code = TSDB_CODE_SUCCESS;

4,728,195✔

4665

  int32_t              lino = 0;

4,728,195✔

4666

  int32_t              tableListSize = 0;

4,728,195✔

4667

  int64_t              st = taosGetTimestampUs();

4,715,010✔

4668

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

4,715,010✔

4669

  STableMergeScanInfo* pInfo = pOperator->info;

4,710,660✔

4671

  QRY_PARAM_CHECK(ppRes);

4,728,932✔

4673

  if (pOperator->status == OP_EXEC_DONE && !pOperator->pOperatorGetParam) {

4,732,572✔

4678

  if (pOperator->pOperatorGetParam) {

4,715,037✔

4679

    code = createVTableMergeScanInfoFromBatchParam(pOperator);

65,744✔

4680

    QUERY_CHECK_CODE(code, lino, _end);

65,744✔

4682

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

65,744✔

4683

    pOperator->pOperatorGetParam = NULL;

65,744✔

4684

    pOperator->dynamicTask = true;

65,744✔

4687

  code = pOperator->fpSet._openFn(pOperator);

4,731,498✔

4688

  QUERY_CHECK_CODE(code, lino, _end);

4,726,029✔

4690

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

4,726,029✔

4691

  QUERY_CHECK_CODE(code, lino, _end);

4,720,870✔

4693

  if (!pInfo->hasGroupId) {

4,720,870✔

4694

    pInfo->hasGroupId = true;

2,004,146✔

4696

    if (tableListSize == 0) {

1,973,044✔

4697

      setOperatorCompleted(pOperator);

41,030✔

4698

      (*ppRes) = NULL;

41,403✔

4699

      return code;

41,403✔

4702

    pInfo->tableStartIndex = 0;

1,932,014✔

4703

    STableKeyInfo* pTmpGpId = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,929,812✔

4704

    QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

1,930,913✔

4705

    pInfo->groupId = pTmpGpId->groupId;

1,930,913✔

4706

    code = startSubTablesTableMergeScan(pOperator);

1,959,822✔

4707

    QUERY_CHECK_CODE(code, lino, _end);

1,977,703✔

4710

  SSDataBlock* pBlock = NULL;

4,694,063✔

4711

  while (pInfo->tableStartIndex < tableListSize) {

4,760,352✔

4712

    if (isTaskKilled(pTaskInfo)) {

4,758,514✔

4716

    code = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity, &pBlock);

4,758,150✔

4717

    QUERY_CHECK_CODE(code, lino, _end);

4,761,444✔

4719

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

4,761,444✔

4726

    if (pBlock != NULL) {

4,762,181✔

4727

      pBlock->info.id.groupId = pInfo->groupId;

2,724,040✔

4728

      pOperator->resultInfo.totalRows += pBlock->info.rows;

2,724,040✔

4729

      pInfo->bGroupProcessed = true;

2,724,040✔

4730

      break;

2,724,040✔

4733

      stopSubTablesTableMergeScan(pInfo);

2,038,141✔

4734

      if (pInfo->tableEndIndex >= tableListSize - 1) {

2,036,685✔

4735

        setOperatorCompleted(pOperator);

1,978,449✔

4736

        break;

1,977,721✔

4739

      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

58,600✔

4740

      STableKeyInfo* pTmpGpId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

58,600✔

4741

      QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

58,600✔

4743

      pInfo->groupId = pTmpGpId->groupId;

58,600✔

4744

      code = startSubTablesTableMergeScan(pOperator);

58,600✔

4745

      QUERY_CHECK_CODE(code, lino, _end);

58,600✔

4746

      resetLimitInfoForNextGroup(&pInfo->limitInfo);

58,600✔

4750

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

4,701,761✔

4752

_end:

4,702,125✔

4753

  if (code != TSDB_CODE_SUCCESS) {

4,701,761✔

4758

    (*ppRes) = pBlock;

4,701,761✔

4761

  return code;

4,702,489✔

4764

static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {

3,690,799✔

4765

  int32_t              code = TSDB_CODE_SUCCESS;

3,690,799✔

4766

  int32_t              lino = 0;

3,690,799✔

4767

  SOperatorInfo*       pOperator = (SOperatorInfo*)pTableMergeOpInfo;

3,690,799✔

4768

  STableMergeScanInfo* pInfo = pOperator->info;

3,690,799✔

4769

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,691,354✔

4771

  if (pInfo->mSkipTables == NULL) {

3,691,354✔

4772

    pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,

323,656✔

4774

    QUERY_CHECK_NULL(pInfo->mSkipTables, code, lino, _end, terrno);

323,656✔

4776

  int bSkip = 1;

3,691,354✔

4777

  if (pInfo->mSkipTables != NULL) {

3,691,354✔

4778

    code = taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));

3,691,354✔

4779

    if (code == TSDB_CODE_DUP_KEY) {

3,691,354✔

4782

    QUERY_CHECK_CODE(code, lino, _end);

3,691,354✔

4785

_end:

3,691,354✔

4786

  if (code != TSDB_CODE_SUCCESS) {

3,691,354✔

4791

3,691,354✔

4793

static int32_t doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {

161,691,440✔

4794

  STableMergeScanInfo* pInfo = pOperator->info;

161,691,440✔

4795

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

161,698,111✔

4796

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

161,697,609✔

4797

  SSDataBlock*         pBlock = pInfo->pReaderBlock;

161,706,944✔

4798

  int32_t              code = 0;

161,682,689✔

4799

  bool                 hasNext = false;

161,682,689✔

4800

  STsdbReader*         reader = pInfo->base.dataReader;

161,680,692✔

4802

  code = pAPI->tsdReader.tsdNextDataBlock(reader, &hasNext);

161,659,002✔

4803

  if (code != 0) {

161,666,541✔

4810

  if (!hasNext || isTaskKilled(pTaskInfo)) {

161,666,541✔

4811

    if (isTaskKilled(pTaskInfo)) {

29,338,182✔

4815

    *pFinished = true;

29,329,396✔

4816

    return code;

29,313,180✔

4819

  uint32_t status = 0;

132,345,494✔

4820

  code = loadDataBlock(pOperator, &pInfo->base, pBlock, &status);

132,356,432✔

4822

  if (code != TSDB_CODE_SUCCESS) {

132,283,268✔

4823

    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));

17,568✔

4824

    pTaskInfo->code = code;

17,568✔

4825

    return code;

17,568✔

4828

  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

132,265,700✔

4834

  if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {

132,265,700✔

4835

    *pSkipped = true;

5,488,421✔

4836

    return code;

5,569,080✔

4839

  return code;

126,752,210✔

4842

static int32_t getBlockForTableMergeScan(void* param, SSDataBlock** ppBlock) {

159,170,591✔

4843

  STableMergeScanSortSourceParam* source = param;

159,170,591✔

4845

  SOperatorInfo*       pOperator = source->pOperator;

159,170,591✔

4846

  STableMergeScanInfo* pInfo = pOperator->info;

159,183,907✔

4847

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

159,186,498✔

4848

  SSDataBlock*         pBlock = NULL;

159,162,997✔

4849

  int64_t              st = taosGetTimestampUs();

159,178,377✔

4850

  int32_t              code = TSDB_CODE_SUCCESS;

159,178,377✔

4853

    if (pInfo->rtnNextDurationBlocks) {

167,682,789✔

4854

      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d",

5,994,912✔

4857

      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {

6,001,508✔

4858

        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];

2,946,675✔

4859

        ++pInfo->nextDurationBlocksIdx;

2,946,675✔

4861

        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

5,994,594✔

4862

          blockDataDestroy(pInfo->nextDurationBlocks[i]);

2,945,581✔

4863

          pInfo->nextDurationBlocks[i] = NULL;

2,942,865✔

4866

        pInfo->rtnNextDurationBlocks = false;

3,050,177✔

4867

        pInfo->nextDurationBlocksIdx = 0;

3,050,953✔

4868

        pInfo->numNextDurationBlocks = 0;

3,046,297✔

4869

        continue;

3,048,237✔

4872

      bool bFinished = false;

161,690,286✔

4873

      bool bSkipped = false;

161,693,401✔

4875

      code = doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);

161,714,340✔

4876

      if (code != 0) {

161,630,230✔

4877

        return code;

17,568✔

4880

      pBlock = pInfo->pReaderBlock;

161,612,662✔

4881

      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d",

161,640,200✔

4883

      if (bFinished) {

161,714,977✔

4884

        pInfo->bNewFilesetEvent = false;

29,340,846✔

4885

        break;

29,340,846✔

4888

      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {

132,374,131✔

4889

        if (!bSkipped) {

3,054,996✔

4890

          code = createOneDataBlock(pBlock, true, &pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks]);

2,946,675✔

4891

          if (code) {

2,941,243✔

4896

          ++pInfo->numNextDurationBlocks;

2,941,243✔

4897

          if (pInfo->numNextDurationBlocks > 2) {

2,934,647✔

4905

        if (pInfo->bNewFilesetEvent) {

3,050,340✔

4906

          pInfo->rtnNextDurationBlocks = true;

3,041,959✔

4907

          *ppBlock = NULL;

3,046,227✔

4908

          return code;

3,035,723✔

4911

        if (pInfo->bNextDurationBlockEvent) {

388✔

4916

      if (bSkipped) continue;

129,319,702✔

4919

    pBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pBlock->info.id.uid);

126,808,357✔

4921

    pOperator->resultInfo.totalRows += pBlock->info.rows;

126,807,893✔

4922

    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

126,797,268✔

4923

    *ppBlock = pBlock;

126,800,095✔

4925

    return code;

126,797,162✔

4928

  *ppBlock = NULL;

29,340,875✔

4929

  return code;

29,340,875✔

4932

int32_t generateSortByTsPkInfo(SArray* colMatchInfo, int32_t order, SArray** ppSortArray) {

25,927,644✔

4933

  int32_t code = TSDB_CODE_SUCCESS;

25,927,644✔

4934

  int32_t lino = 0;

25,927,644✔

4935

  SArray* pSortInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));

25,927,644✔

4936

  QUERY_CHECK_NULL(pSortInfo, code, lino, _end, terrno);

25,886,768✔

4937

  SBlockOrderInfo biTs = {0};

25,886,768✔

4938

  SBlockOrderInfo biPk = {0};

25,888,044✔

4940

  int32_t tsTargetSlotId = 0;

25,900,315✔

4941

  int32_t pkTargetSlotId = -1;

25,900,315✔

4942

  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {

98,569,607✔

4943

    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);

72,740,804✔

4944

    QUERY_CHECK_NULL(colInfo, code, lino, _end, terrno);

72,747,409✔

4945

    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

72,747,409✔

4946

      tsTargetSlotId = colInfo->dstSlotId;

25,923,707✔

4947

      biTs.order = order;

25,933,041✔

4948

      biTs.slotId = tsTargetSlotId;

25,933,041✔

4949

      biTs.nullFirst = (order == TSDB_ORDER_ASC);

25,933,041✔

4950

      biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, order);

25,933,041✔

4953

    if (colInfo->isPk) {

72,739,118✔

4954

      pkTargetSlotId = colInfo->dstSlotId;

2,223,903✔

4955

      biPk.order = order;

2,223,903✔

4956

      biPk.slotId = pkTargetSlotId;

2,223,903✔

4957

      biPk.nullFirst = (order == TSDB_ORDER_ASC);

2,223,903✔

4958

      biPk.compFn = getKeyComparFunc(colInfo->dataType.type, order);

2,223,903✔

4962

  void* tmp = taosArrayPush(pSortInfo, &biTs);

25,939,746✔

4963

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

25,939,746✔

4964

  if (pkTargetSlotId != -1) {

25,939,746✔

4965

    tmp = taosArrayPush(pSortInfo, &biPk);

2,223,903✔

4966

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

2,223,903✔

4969

  (*ppSortArray) = pSortInfo;

25,939,746✔

4971

_end:

25,921,893✔

4972

  if (code != TSDB_CODE_SUCCESS) {

25,914,061✔

4975

  return code;

25,914,061✔

4978

void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {

4,425,414✔

4979

  STableMergeScanInfo* pTmsInfo = param;

4,425,414✔

4980

  if (type == TSD_READER_NOTIFY_DURATION_START) {

4,425,414✔

4981

    pTmsInfo->bNewFilesetEvent = true;

4,428,906✔

4985

  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);

4,403,165✔

4987

  return;

4,439,206✔

4990

int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

32,580,861✔

4991

  STableMergeScanInfo* pInfo = pOperator->info;

32,580,861✔

4992

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

32,605,516✔

4994

  int32_t code = TSDB_CODE_SUCCESS;

32,596,370✔

4995

  int32_t lino = 0;

32,596,370✔

4996

  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

32,596,370✔

4998

  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));

32,577,856✔

4999

  pInfo->bNewFilesetEvent = false;

32,607,232✔

5000

  pInfo->bNextDurationBlockEvent = false;

32,637,345✔

5002

  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;

32,632,682✔

5003

  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;

32,626,064✔

5005

  pInfo->pSortHandle = NULL;

32,627,068✔

5006

  code = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,

65,271,996✔

5007

                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0, &pInfo->pSortHandle);

32,638,943✔

5008

  if (code) {

32,584,214✔

5012

  if (pInfo->bSortRowId && numOfTable != 1) {

32,584,214✔

5013

    int32_t memSize = 512 * 1024 * 1024;

1,637,552✔

5014

    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);

1,637,552✔

5015

    QUERY_CHECK_CODE(code, lino, _end);

1,635,395✔

5018

  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);

32,591,890✔

5019

  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pOperator);

32,578,963✔

5020

  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);

32,564,411✔

5022

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);

32,561,360✔

5023

  QUERY_CHECK_CODE(code, lino, _end);

32,598,077✔

5025

  STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));

32,598,077✔

5026

  QUERY_CHECK_NULL(param, code, lino, _end, terrno);

32,573,398✔

5027

  param->pOperator = pOperator;

32,573,398✔

5029

  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));

32,583,137✔

5030

  if (ps == NULL) {

32,553,781✔

5035

  ps->param = param;

32,553,781✔

5036

  ps->onlyRef = false;

32,600,953✔

5037

  code = tsortAddSource(pInfo->pSortHandle, ps);

32,606,434✔

5038

  QUERY_CHECK_CODE(code, lino, _end);

32,616,063✔

5040

  if (numOfTable == 1) {

32,616,063✔

5041

    tsortSetSingleTableMerge(pInfo->pSortHandle);

24,129,045✔

5043

    code = tsortOpen(pInfo->pSortHandle);

8,487,018✔

5044

    QUERY_CHECK_CODE(code, lino, _end);

8,496,523✔

5047

_end:

8,487,739✔

5048

  if (code != TSDB_CODE_SUCCESS) {

32,601,840✔

5049

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,784✔

5051

  return code;

32,612,567✔

5054

void startGroupTableMergeScan(SOperatorInfo* pOperator) {

29,569,771✔

5055

  int32_t              code = TSDB_CODE_SUCCESS;

29,569,771✔

5056

  int32_t              lino = 0;

29,569,771✔

5057

  STableMergeScanInfo* pInfo = pOperator->info;

29,569,771✔

5058

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

29,574,248✔

5059

  SReadHandle*         pHandle = &pInfo->base.readHandle;

29,560,795✔

5060

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

29,540,990✔

5061

  qDebug("%s table merge scan start group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

29,567,714✔

5064

    int32_t numOfTables = 0;

29,576,090✔

5065

    code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

29,560,884✔

5066

    QUERY_CHECK_CODE(code, lino, _end);

29,576,920✔

5068

    int32_t i = pInfo->tableStartIndex + 1;

29,576,920✔

5069

    for (; i < numOfTables; ++i) {

139,158,343✔

5070

      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

116,013,690✔

5071

      QUERY_CHECK_NULL(tableKeyInfo, code, lino, _end, terrno);

116,013,739✔

5072

      if (tableKeyInfo->groupId != pInfo->groupId) {

116,013,739✔

5073

        break;

6,431,316✔

5076

    pInfo->tableEndIndex = i - 1;

29,575,969✔

5078

  pInfo->bGroupProcessed = false;

29,584,511✔

5079

  int32_t tableStartIdx = pInfo->tableStartIndex;

29,584,272✔

5080

  int32_t tableEndIdx = pInfo->tableEndIndex;

29,582,056✔

5082

  int32_t        numOfTable = tableEndIdx - tableStartIdx + 1;

29,587,399✔

5083

  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);

29,587,399✔

5084

  code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,

59,147,007✔

5085

                                       (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->mSkipTables);

29,571,554✔

5086

  QUERY_CHECK_CODE(code, lino, _end);

29,546,597✔

5087

  if (pInfo->filesetDelimited) {

29,546,597✔

5088

    pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);

29,546,053✔

5090

  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);

29,570,082✔

5092

  code = startDurationForGroupTableMergeScan(pOperator);

29,530,492✔

5093

  QUERY_CHECK_CODE(code, lino, _end);

29,537,811✔

5095

_end:

29,529,027✔

5096

  if (code != TSDB_CODE_SUCCESS) {

29,540,648✔

5097

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,784✔

5098

    pTaskInfo->code = code;

8,784✔

5099

    T_LONG_JMP(pTaskInfo->env, code);

8,784✔

5101

29,531,864✔

5103

void stopGroupTableMergeScan(SOperatorInfo* pOperator) {

29,573,083✔

5104

  STableMergeScanInfo* pInfo = pOperator->info;

29,573,083✔

5105

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

29,573,632✔

5106

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

29,573,511✔

5108

  stopDurationForGroupTableMergeScan(pOperator);

29,572,503✔

5110

  if (pInfo->base.dataReader != NULL) {

29,565,834✔

5111

    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);

29,571,037✔

5112

    pInfo->base.dataReader = NULL;

29,500,152✔

5114

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

29,505,202✔

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);

29,534,758✔

5123

  taosHashCleanup(pInfo->mSkipTables);

29,490,899✔

5124

  pInfo->mSkipTables = NULL;

29,503,491✔

5125

  qDebug("%s table merge scan stop group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

29,510,058✔

5126

29,570,913✔

5130

SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,

69,829,184✔

5132

  STableMergeScanInfo* pInfo = pOperator->info;

69,829,184✔

5133

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

69,847,084✔

5134

  STupleHandle*        pTupleHandle = NULL;

69,841,176✔

5136

  blockDataCleanup(pResBlock);

69,854,861✔

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;

52,949,461✔

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,894,332✔

5161

    if (tsortIsClosed(pHandle)) {

69,843,793✔

5166

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);

69,851,387✔

5167

    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,

69,850,595✔

5169

    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {

69,868,956✔

5173

  return (pResBlock->info.rows > 0) ? pResBlock : NULL;

69,868,956✔

5176

int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

63,615,299✔

5177

  if (pOperator->status == OP_EXEC_DONE) {

63,615,299✔

5178

    (*ppRes) = NULL;

46,388✔

5179

    return TSDB_CODE_SUCCESS;

46,388✔

5182

  int32_t              code = TSDB_CODE_SUCCESS;

63,538,954✔

5183

  int32_t              lino = 0;

63,538,954✔

5184

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

63,538,954✔

5185

  STableMergeScanInfo* pInfo = pOperator->info;

63,569,968✔

5187

  code = pOperator->fpSet._openFn(pOperator);

63,563,212✔

5188

  QUERY_CHECK_CODE(code, lino, _end);

63,550,413✔

5190

  int64_t st = taosGetTimestampUs();

63,564,220✔

5192

  int32_t tableListSize = 0;

63,564,220✔

5193

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

63,556,151✔

5194

  QUERY_CHECK_CODE(code, lino, _end);

63,560,812✔

5196

  if (!pInfo->hasGroupId) {

63,560,812✔

5197

    pInfo->hasGroupId = true;

26,323,369✔

5199

    if (tableListSize == 0) {

26,315,749✔

5200

      setOperatorCompleted(pOperator);

3,196,862✔

5201

      (*ppRes) = NULL;

3,195,070✔

5202

      return code;

3,193,890✔

5204

    pInfo->tableStartIndex = 0;

23,118,887✔

5205

    STableKeyInfo* tmp = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

23,136,967✔

5206

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

23,114,661✔

5207

    pInfo->groupId = tmp->groupId;

23,114,661✔

5208

    startGroupTableMergeScan(pOperator);

23,135,045✔

5211

  SSDataBlock* pBlock = NULL;

60,341,189✔

5212

  while (pInfo->tableStartIndex < tableListSize) {

69,821,836✔

5213

    if (isTaskKilled(pTaskInfo)) {

69,840,349✔

5217

    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,

69,836,661✔

5219

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

69,868,852✔

5220

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,860,208✔

5221

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

1,860,208✔

5222

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

1,860,208✔

5224

    if (pBlock != NULL) {

69,865,604✔

5225

      pBlock->info.id.groupId = pInfo->groupId;

37,238,341✔

5226

      pOperator->resultInfo.totalRows += pBlock->info.rows;

37,238,180✔

5227

      pInfo->bGroupProcessed = true;

37,237,665✔

5228

      break;

37,236,567✔

5230

      if (pInfo->bNewFilesetEvent) {

32,627,263✔

5231

        stopDurationForGroupTableMergeScan(pOperator);

3,052,435✔

5232

        code = startDurationForGroupTableMergeScan(pOperator);

3,047,779✔

5233

        QUERY_CHECK_CODE(code, lino, _end);

3,049,331✔

5236

        stopGroupTableMergeScan(pOperator);

29,574,440✔

5237

        if (pInfo->tableEndIndex >= tableListSize - 1) {

29,569,830✔

5238

          setOperatorCompleted(pOperator);

23,138,475✔

5239

          break;

23,139,069✔

5242

        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

6,431,316✔

5243

        STableKeyInfo* tmp = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

6,431,316✔

5244

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

6,431,316✔

5245

        pInfo->groupId = tmp->groupId;

6,431,316✔

5246

        startGroupTableMergeScan(pOperator);

6,431,316✔

5247

        resetLimitInfoForNextGroup(&pInfo->limitInfo);

6,431,281✔

5252

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

60,373,832✔

5254

_end:

60,377,635✔

5255

  if (code != TSDB_CODE_SUCCESS) {

60,377,498✔

5260

    (*ppRes) = pBlock;

60,377,498✔

5263

  return code;

60,374,679✔

5272

void destroyTableMergeScanOperatorInfo(void* param) {

25,948,414✔

5273

  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;

25,948,414✔

5276

  if (pTableScanInfo->base.readerAPI.tsdReaderClose != NULL) {

25,948,414✔

5277

    pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);

25,950,661✔

5278

    pTableScanInfo->base.dataReader = NULL;

25,939,028✔

5281

  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {

25,941,507✔

5288

  tsortDestroySortHandle(pTableScanInfo->pSortHandle);

25,941,531✔

5289

  pTableScanInfo->pSortHandle = NULL;

25,940,862✔

5290

  taosHashCleanup(pTableScanInfo->mSkipTables);

25,945,014✔

5291

  pTableScanInfo->mSkipTables = NULL;

25,931,922✔

5292

  blockDataDestroy(pTableScanInfo->pSortInputBlock);

25,932,473✔

5293

  pTableScanInfo->pSortInputBlock = NULL;

25,943,789✔

5296

  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);

25,943,789✔

5298

  blockDataDestroy(pTableScanInfo->pResBlock);

25,941,344✔

5299

  pTableScanInfo->pResBlock = NULL;

25,943,552✔

5302

  blockDataDestroy(pTableScanInfo->pReaderBlock);

25,944,052✔

5303

  pTableScanInfo->pReaderBlock = NULL;

25,939,302✔

5304

  taosArrayDestroy(pTableScanInfo->pSortInfo);

25,942,806✔

5306

  stopSubTablesTableMergeScan(pTableScanInfo);

25,941,676✔

5308

  taosMemoryFreeClear(param);

25,934,938✔

5309

25,935,410✔

5311

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {

1,270,340✔

5312

  if (pOptr == NULL) {

1,270,340✔

5317

  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));

1,270,340✔

5318

  if (!execInfo) {

1,268,756✔

5321

  STableMergeScanInfo* pInfo = pOptr->info;

1,268,756✔

5322

  execInfo->blockRecorder = pInfo->base.readRecorder;

1,267,964✔

5323

  execInfo->sortExecInfo = pInfo->sortExecInfo;

1,267,560✔

5325

  *pOptrExplain = execInfo;

1,270,720✔

5326

  *len = sizeof(STableMergeScanExecInfo);

1,268,344✔

5328

  return TSDB_CODE_SUCCESS;

1,269,936✔

5331

static int32_t resetTableMergeScanOperatorState(SOperatorInfo* pOper) {

2,422,748✔

5332

  int32_t         code = TSDB_CODE_SUCCESS;

2,422,748✔

5333

  STableMergeScanInfo*   pInfo = pOper->info;

2,422,748✔

5334

  pOper->status = OP_NOT_OPENED;

2,422,748✔

5336

  pInfo->tableEndIndex = 0;

2,422,748✔

5337

  pInfo->tableStartIndex = 0;

2,422,748✔

5338

  pInfo->hasGroupId = false;

2,422,748✔

5339

  if (pInfo->base.readerAPI.tsdReaderClose) {

2,422,748✔

5340

    pInfo->base.readerAPI.tsdReaderClose(pInfo->base.dataReader);

2,422,748✔

5342

  pInfo->base.dataReader = NULL;

2,422,748✔

5343

  pInfo->base.scanFlag = MAIN_SCAN;

2,422,748✔

5345

  pInfo->base.limitInfo = (SLimitInfo){0};

2,422,748✔

5346

  pInfo->base.limitInfo.limit.limit = -1;

2,422,748✔

5347

  pInfo->base.limitInfo.slimit.limit = -1;

2,422,748✔

5349

  tableListDestroy(pInfo->base.pTableListInfo);

2,422,748✔

5350

  pInfo->base.pTableListInfo = tableListCreate();

2,422,748✔

5351

  if (!pInfo->base.pTableListInfo) {

2,422,491✔

5355

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

2,422,484✔

5357

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

2,422,484✔

5358

  code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

4,845,496✔

5360

                                  pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

4,845,496✔

5361

  if (code) {

2,422,748✔

5366

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

2,422,748✔

5367

  cleanupQueryTableDataCond(&pInfo->base.cond);

2,422,748✔

5368

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, &pInfo->base.readHandle, false);

2,422,748✔

5369

  if (code) {

2,422,748✔

5374

  tsortDestroySortHandle(pInfo->pSortHandle);

2,422,748✔

5375

  pInfo->pSortHandle = NULL;

2,422,748✔

5377

  taosHashCleanup(pInfo->mSkipTables);

2,422,748✔

5378

  pInfo->mSkipTables = NULL;

2,422,748✔

5379

  pOper->resultInfo.totalRows = 0;

2,422,748✔

5381

  pInfo->bGroupProcessed = false;

2,422,484✔

5382

  pInfo->bNewFilesetEvent = false;

2,422,748✔

5383

  pInfo->bNextDurationBlockEvent = false;

2,422,748✔

5384

  pInfo->rtnNextDurationBlocks = false;

2,422,748✔

5385

  pInfo->nextDurationBlocksIdx = 0;

2,422,484✔

5386

  pInfo->bSortRowId = false;

2,422,484✔

5388

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

2,422,484✔

5394

  pInfo->numNextDurationBlocks = 0;

2,422,748✔

5396

  stopSubTablesTableMergeScan(pInfo);

2,422,748✔

5397

  return code;

2,422,748✔

5400

int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

25,899,622✔

5403

  QRY_PARAM_CHECK(pOptrInfo);

25,899,622✔

5405

  int32_t              code = TSDB_CODE_SUCCESS;

25,915,621✔

5406

  int32_t              lino = 0;

25,915,621✔

5407

  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));

25,915,621✔

5408

  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

25,837,197✔

5409

  if (pInfo == NULL || pOperator == NULL) {

25,843,866✔

5414

  pOperator->pPhyNode = pTableScanNode;

25,855,872✔

5415

  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

25,883,232✔

5417

  int32_t numOfCols = 0;

25,880,198✔

5418

  code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,

25,894,410✔

5420

  QUERY_CHECK_CODE(code, lino, _error);

25,876,176✔

5422

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle, false);

25,876,176✔

5423

  QUERY_CHECK_CODE(code, lino, _error);

25,842,030✔

5425

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

25,842,030✔

5426

    SExprSupp* pSup = &pInfo->base.pseudoSup;

10,909,212✔

5427

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->pExprInfo, &pSup->numOfExprs);

10,896,336✔

5428

    QUERY_CHECK_CODE(code, lino, _error);

10,887,167✔

5430

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

10,887,167✔

5432

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

10,876,298✔

5435

  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};

25,853,059✔

5437

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

25,870,025✔

5442

  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;

25,870,025✔

5443

  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

25,883,147✔

5444

  pInfo->base.scanFlag = MAIN_SCAN;

25,745,020✔

5445

  pInfo->base.readHandle = *readHandle;

25,814,079✔

5447

  pInfo->readIdx = -1;

25,896,064✔

5449

  pInfo->base.limitInfo.limit.limit = -1;

25,773,022✔

5450

  pInfo->base.limitInfo.slimit.limit = -1;

25,855,356✔

5451

  pInfo->base.pTableListInfo = pTableListInfo;

25,815,751✔

5453

  pInfo->sample.sampleRatio = pTableScanNode->ratio;

25,895,249✔

5454

  pInfo->sample.seed = taosGetTimestampSec();

25,835,400✔

5456

  int32_t sortOrder = pInfo->base.cond.order;

25,871,231✔

5457

  if (pTableScanNode->scan.node.dynamicOp) {

25,885,534✔

5458

    memcpy(&pInfo->base.orgCond, &pInfo->base.cond, sizeof(SQueryTableDataCond));

65,744✔

5459

    sortOrder = pInfo->base.orgCond.order;

65,744✔

5460

    memset(&pInfo->base.cond, 0, sizeof(SQueryTableDataCond));

65,744✔

5463

  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,

25,885,023✔

5464

                            pTaskInfo->pStreamRuntimeInfo);

25,857,570✔

5465

  QUERY_CHECK_CODE(code, lino, _error);

25,831,988✔

5467

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

25,831,988✔

5469

  pInfo->mergeLimit = -1;

25,831,133✔

5470

  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;

25,884,248✔

5471

  if (hasLimit) {

25,823,950✔

5472

    pInfo->mergeLimit = pInfo->limitInfo.limit.offset != -1

4,114,780✔

5473

                            ? pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset

807,657✔

5474

                            : pInfo->limitInfo.limit.limit;

2,860,800✔

5475

    pInfo->mSkipTables = NULL;

2,058,317✔

5478

  initResultSizeInfo(&pOperator->resultInfo, 1024);

25,822,027✔

5479

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

25,847,567✔

5480

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

25,904,449✔

5481

  code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

25,846,106✔

5482

  QUERY_CHECK_CODE(code, lino, _error);

25,928,871✔

5483

  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {

25,928,871✔

5484

    pInfo->bSortRowId = true;

5,250,071✔

5486

    pInfo->bSortRowId = false;

20,676,964✔

5489

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

25,909,616✔

5490

  QUERY_CHECK_CODE(code, lino, _error);

25,900,472✔

5492

  code = generateSortByTsPkInfo(pInfo->base.matchInfo.pList, sortOrder, &pInfo->pSortInfo);

25,900,472✔

5493

  QUERY_CHECK_CODE(code, lino, _error);

25,901,039✔

5495

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pReaderBlock);

25,901,039✔

5496

  QUERY_CHECK_CODE(code, lino, _error);

25,908,778✔

5498

  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;

25,908,778✔

5500

  int32_t  rowSize = pInfo->pResBlock->info.rowSize;

25,923,961✔

5501

  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);

25,878,495✔

5503

  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);

25,890,842✔

5506

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pSortInputBlock);

25,931,207✔

5507

  QUERY_CHECK_CODE(code, lino, _error);

25,932,521✔

5509

  if (!tsExperimental) {

25,932,521✔

5512

    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

25,932,521✔

5515

  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,

25,909,815✔

5517

  pOperator->exprSupp.numOfExprs = numOfCols;

25,859,825✔

5519

  pOperator->fpSet = createOperatorFpSet(

25,898,785✔

5520

      optrDummyOpenFn, (pTableScanNode->paraTablesSort || pTableScanNode->scan.node.dynamicOp) ? doTableMergeScanParaSubTablesNext : doTableMergeScanNext, NULL,

25,882,457✔

5523

  setOperatorResetStateFn(pOperator, resetTableMergeScanOperatorState);

25,906,916✔

5524

  pOperator->cost.openCost = 0;

25,863,962✔

5526

  *pOptrInfo = pOperator;

25,869,438✔

5527

  return code;

25,836,871✔

5566

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {

29,993✔

5567

  if (scanCols != NULL) {

29,993✔

5568

    SNode* pNode = NULL;

15,607✔

5569

    FOREACH(pNode, scanCols) {

36,449✔

5570

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

20,842✔

5573

      STargetNode* targetNode = (STargetNode*)pNode;

20,842✔

5574

      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {

20,842✔

5577

      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);

20,842✔

5578

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

20,842✔

5579

        supp->dbNameSlotId = targetNode->slotId;

10,294✔

5580

      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

10,548✔

5581

        supp->stbNameSlotId = targetNode->slotId;

10,548✔

5585

  return TSDB_CODE_SUCCESS;

30,171✔

5588

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {

30,171✔

5589

  if (pseudoCols != NULL) {

30,171✔

5590

    SNode* pNode = NULL;

30,171✔

5591

    FOREACH(pNode, pseudoCols) {

60,342✔

5592

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

30,171✔

5595

      STargetNode* targetNode = (STargetNode*)pNode;

30,171✔

5596

      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {

30,171✔

5599

      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);

30,137✔

5600

      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {

30,171✔

5601

        supp->tbCountSlotId = targetNode->slotId;

30,171✔

5605

  return TSDB_CODE_SUCCESS;

30,171✔

5608

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {

30,171✔

5609

  if (groupTags != NULL) {

30,171✔

5610

    SNode* pNode = NULL;

15,607✔

5611

    FOREACH(pNode, groupTags) {

36,449✔

5612

      if (nodeType(pNode) != QUERY_NODE_COLUMN) {

20,842✔

5615

      SColumnNode* colNode = (SColumnNode*)pNode;

20,842✔

5616

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

20,842✔

5617

        supp->groupByDbName = true;

10,294✔

5619

      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

20,842✔

5620

        supp->groupByStbName = true;

10,548✔

5624

    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);

14,564✔

5625

    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);

14,564✔

5627

  return TSDB_CODE_SUCCESS;

30,171✔

5630

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,

30,171✔

5632

  int32_t code = 0;

30,171✔

5633

  code = tblCountScanGetInputs(groupTags, tableName, supp);

30,171✔

5634

  if (code != TSDB_CODE_SUCCESS) {

30,171✔

5639

  supp->dbNameSlotId = -1;

30,171✔

5640

  supp->stbNameSlotId = -1;

30,171✔

5641

  supp->tbCountSlotId = -1;

30,171✔

5643

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);

30,137✔

5644

  if (code != TSDB_CODE_SUCCESS) {

30,137✔

5649

  code = tblCountScanGetCountSlotId(pseudoCols, supp);

30,137✔

5650

  if (code != TSDB_CODE_SUCCESS) {

30,171✔

5654

  return code;

30,171✔

5657

int32_t createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,

30,171✔

5659

  QRY_PARAM_CHECK(pOptrInfo);

30,171✔

5661

  int32_t                      code = TSDB_CODE_SUCCESS;

30,171✔

5662

  int32_t                      lino = 0;

30,171✔

5663

  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;

30,171✔

5664

  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));

30,171✔

5665

  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

30,171✔

5666

  if (!pInfo || !pOperator) {

30,171✔

5671

  pInfo->readHandle = *readHandle;

30,171✔

5673

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

30,171✔

5674

  initResultSizeInfo(&pOperator->resultInfo, 1);

30,171✔

5675

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

30,171✔

5676

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

30,171✔

5678

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

30,137✔

5679

  QUERY_CHECK_CODE(code, lino, _error);

30,171✔

5681

  code = getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,

30,171✔

5684

  QUERY_CHECK_CODE(code, lino, _error);

30,171✔

5686

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,

30,171✔

5688

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScanNext, NULL, destoryTableCountScanOperator,

30,171✔

5690

  *pOptrInfo = pOperator;

30,171✔

5691

  return code;

30,171✔

5705

int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,

40,518✔

5707

  int32_t code = TSDB_CODE_SUCCESS;

40,518✔

5708

  int32_t lino = 0;

40,518✔

5709

  if (pSupp->dbNameSlotId != -1) {

40,518✔

5710

    QUERY_CHECK_CONDITION((strlen(dbName) > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

15,386✔

5711

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);

15,386✔

5712

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

15,208✔

5714

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

15,208✔

5715

    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

15,208✔

5717

    varDataSetLen(varDbName, strlen(dbName));

15,208✔

5718

    code = colDataSetVal(colInfoData, 0, varDbName, false);

15,208✔

5719

    QUERY_CHECK_CODE(code, lino, _end);

15,208✔

5722

  if (pSupp->stbNameSlotId != -1) {

40,340✔

5723

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);

20,070✔

5724

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

20,248✔

5725

    if (strlen(stbName) != 0) {

20,248✔

5726

      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

14,418✔

5727

      tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);

14,418✔

5728

      varDataSetLen(varStbName, strlen(stbName));

14,240✔

5729

      code = colDataSetVal(colInfoData, 0, varStbName, false);

14,596✔

5730

      QUERY_CHECK_CODE(code, lino, _end);

14,596✔

5736

  if (pSupp->tbCountSlotId != -1) {

40,892✔

5737

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);

40,840✔

5738

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

40,628✔

5739

    code = colDataSetVal(colInfoData, 0, (char*)&count, false);

40,628✔

5740

    QUERY_CHECK_CODE(code, lino, _end);

40,858✔

5742

  pRes->info.rows = 1;

40,892✔

5744

_end:

40,892✔

5745

  if (code != TSDB_CODE_SUCCESS) {

40,892✔

5748

  return code;

40,696✔

5751

static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {

11,341✔

5752

  STableCountScanSupp* pSupp = &pInfo->supp;

11,341✔

5753

  SSDataBlock*         pRes = pInfo->pRes;

11,341✔

5755

  size_t infodbTableNum;

11,341✔

5756

  getInfosDbMeta(NULL, &infodbTableNum);

11,341✔

5757

  infodbTableNum -= 1;

11,341✔

5758

  size_t perfdbTableNum;

11,341✔

5759

  getPerfDbMeta(NULL, &perfdbTableNum);

11,341✔

5761

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

11,341✔

5762

    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);

9,759✔

5763

    return (pRes->info.rows > 0) ? pRes : NULL;

9,759✔

5765

    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);

1,582✔

5766

    return (pRes->info.rows > 0) ? pRes : NULL;

1,582✔

5770

static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,

1,582✔

5772

  int32_t        code = TSDB_CODE_SUCCESS;

1,582✔

5773

  int32_t        lino = 0;

1,582✔

5774

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,582✔

5775

  if (strlen(pSupp->stbNameFilter) != 0) {

1,582✔

5777

    code = fillTableCountScanDataBlock(pSupp, "", "", 0, pRes);

537✔

5778

    QUERY_CHECK_CODE(code, lino, _end);

537✔

5779

  } else if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {

1,045✔

5780

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

179✔

5781

    QUERY_CHECK_CODE(code, lino, _end);

179✔

5782

  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

866✔

5785

  } else if (strlen(pSupp->dbNameFilter) == 0) {

866✔

5786

    code = fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);

687✔

5787

    QUERY_CHECK_CODE(code, lino, _end);

687✔

5790

    code = fillTableCountScanDataBlock(pSupp, pSupp->dbNameFilter, pSupp->stbNameFilter, 0, pRes);

179✔

5791

    QUERY_CHECK_CODE(code, lino, _end);

179✔

5794

_end:

179✔

5795

  if (code != TSDB_CODE_SUCCESS) {

1,582✔

5800

  setOperatorCompleted(pOperator);

1,582✔

5801

1,582✔

5803

static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,759✔

5806

  int32_t        code = TSDB_CODE_SUCCESS;

9,759✔

5807

  int32_t        lino = 0;

9,759✔

5808

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,759✔

5809

  if (pInfo->currGrpIdx == 0) {

9,759✔

5810

    uint64_t groupId = 0;

3,253✔

5811

    if (pSupp->groupByDbName) {

3,253✔

5812

      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));

2,220✔

5814

      groupId = calcGroupId("", 0);

1,033✔

5817

    pRes->info.id.groupId = groupId;

3,253✔

5818

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

3,253✔

5819

    QUERY_CHECK_CODE(code, lino, _end);

3,253✔

5820

  } else if (pInfo->currGrpIdx == 1) {

6,506✔

5821

    uint64_t groupId = 0;

3,253✔

5822

    if (pSupp->groupByDbName) {

3,253✔

5823

      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));

2,220✔

5825

      groupId = calcGroupId("", 0);

1,033✔

5828

    pRes->info.id.groupId = groupId;

3,253✔

5829

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

3,253✔

5830

    QUERY_CHECK_CODE(code, lino, _end);

3,253✔

5832

    setOperatorCompleted(pOperator);

3,253✔

5834

  pInfo->currGrpIdx++;

9,759✔

5836

_end:

9,759✔

5837

  if (code != TSDB_CODE_SUCCESS) {

9,759✔

5842

9,759✔

5844

static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

70,867✔

5845

  int32_t                      code = TSDB_CODE_SUCCESS;

70,867✔

5846

  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;

70,867✔

5847

  STableCountScanOperatorInfo* pInfo = pOperator->info;

70,885✔

5848

  STableCountScanSupp*         pSupp = &pInfo->supp;

71,045✔

5849

  SSDataBlock*                 pRes = pInfo->pRes;

71,063✔

5851

  blockDataCleanup(pRes);

71,063✔

5852

  QRY_PARAM_CHECK(ppRes);

71,045✔

5854

  if (pOperator->status == OP_EXEC_DONE) {

71,045✔

5855

    return code;

18,584✔

5858

  if (pInfo->readHandle.mnd != NULL) {

52,461✔

5859

    (*ppRes) = buildSysDbTableCount(pOperator, pInfo);

11,341✔

5860

    return code;

11,341✔

5863

  code = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);

41,120✔

5864

  if (code != TSDB_CODE_SUCCESS) {

40,746✔

5869

  if (pRes->info.rows > 0) {

40,746✔

5870

    *ppRes = pRes;

32,770✔

5873

  return code;

41,086✔

5876

static int32_t buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

41,082✔

5878

  int32_t        code = TSDB_CODE_SUCCESS;

41,082✔

5879

  int32_t        lino = 0;

41,082✔

5880

  const char*    db = NULL;

41,082✔

5881

  int32_t        vgId = 0;

41,120✔

5882

  char           dbName[TSDB_DB_NAME_LEN] = {0};

41,120✔

5883

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

41,120✔

5884

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

41,120✔

5887

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);

41,120✔

5888

  SName sn = {0};

41,120✔

5890

  code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

41,120✔

5891

  QUERY_CHECK_CODE(code, lino, _end);

40,924✔

5893

  code = tNameGetDbName(&sn, dbName);

40,924✔

5894

  QUERY_CHECK_CODE(code, lino, _end);

41,120✔

5896

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

41,120✔

5897

    code = buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);

28,156✔

5899

    code = buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);

12,964✔

5902

_end:

40,708✔

5903

  if (code != TSDB_CODE_SUCCESS) {

40,708✔

5908

  return code;

40,708✔

5911

static int32_t buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

27,960✔

5913

  int32_t        code = TSDB_CODE_SUCCESS;

27,960✔

5914

  int32_t        lino = 0;

27,960✔

5915

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

27,960✔

5916

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

28,138✔

5918

  if (pSupp->groupByStbName) {

28,138✔

5919

    if (pInfo->stbUidList == NULL) {

23,922✔

5920

      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));

8,316✔

5921

      QUERY_CHECK_NULL(pInfo->stbUidList, code, lino, _end, terrno);

8,316✔

5922

      code = pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList);

8,316✔

5923

      QUERY_CHECK_CODE(code, lino, _end);

8,138✔

5926

    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {

23,922✔

5927

      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);

14,596✔

5928

      code = buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);

14,596✔

5929

      QUERY_CHECK_CODE(code, lino, _end);

14,073✔

5931

      pInfo->currGrpIdx++;

14,073✔

5932

    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {

9,504✔

5933

      code = buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);

8,316✔

5934

      QUERY_CHECK_CODE(code, lino, _end);

8,316✔

5936

      pInfo->currGrpIdx++;

8,316✔

5938

      setOperatorCompleted(pOperator);

1,188✔

5941

    uint64_t groupId = calcGroupId(dbName, strlen(dbName));

4,038✔

5942

    pRes->info.id.groupId = groupId;

4,038✔

5944

    int64_t dbTableCount = 0;

4,038✔

5945

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);

4,038✔

5946

    code = fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);

4,038✔

5947

    QUERY_CHECK_CODE(code, lino, _end);

4,038✔

5948

    setOperatorCompleted(pOperator);

4,038✔

5951

_end:

27,782✔

5952

  if (code != TSDB_CODE_SUCCESS) {

27,782✔

5956

  return code;

27,782✔

5959

static int32_t buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

12,964✔

5961

  int32_t        code = TSDB_CODE_SUCCESS;

12,964✔

5962

  int32_t        lino = 0;

12,964✔

5963

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

12,964✔

5964

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

12,964✔

5966

  if (strlen(pSupp->dbNameFilter) != 0 && strcmp(dbName, pSupp->dbNameFilter) != 0) {

12,964✔

5970

  } else if (strlen(pSupp->stbNameFilter) != 0) {

12,964✔

5971

    uint64_t uid = 0;

6,356✔

5972

    code = pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid);

6,356✔

5973

    if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

6,356✔

5975

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, 0, pRes);

1,432✔

5976

      QUERY_CHECK_CODE(code, lino, _end);

1,398✔

5978

      QUERY_CHECK_CODE(code, lino, _end);

4,924✔

5980

      int64_t numOfChildTables = 0;

4,924✔

5981

      code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL, NULL);

4,924✔

5982

      QUERY_CHECK_CODE(code, lino, _end);

4,924✔

5984

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes);

4,924✔

5985

      QUERY_CHECK_CODE(code, lino, _end);

4,924✔

5988

    int64_t tbNumVnode = 0;

6,608✔

5989

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);

6,608✔

5990

    code = fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);

6,608✔

5991

    QUERY_CHECK_CODE(code, lino, _end);

6,608✔

5994

_end:

12,930✔

5995

  if (code != TSDB_CODE_SUCCESS) {

12,930✔

6000

  setOperatorCompleted(pOperator);

12,930✔

6001

  return code;

12,926✔

6004

static int32_t buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

8,316✔

6006

  int32_t code = TSDB_CODE_SUCCESS;

8,316✔

6007

  int32_t lino = 0;

8,316✔

6008

  char    fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

8,316✔

6009

  if (pSupp->groupByDbName) {

8,316✔

6010

    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");

4,036✔

6013

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

8,316✔

6014

  pRes->info.id.groupId = groupId;

8,316✔

6016

  int64_t numOfTables = 0;

8,316✔

6017

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);

8,316✔

6019

  if (numOfTables != 0) {

8,316✔

6020

    code = fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);

1,188✔

6023

_end:

7,128✔

6024

  if (code != TSDB_CODE_SUCCESS) {

8,316✔

6027

  return code;

8,316✔

6030

static int32_t buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

14,418✔

6032

  int32_t code = TSDB_CODE_SUCCESS;

14,418✔

6033

  int32_t lino = 0;

14,418✔

6034

  char    stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

14,418✔

6035

  code = pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);

14,418✔

6036

  QUERY_CHECK_CODE(code, lino, _end);

14,240✔

6038

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

14,240✔

6039

  if (pSupp->groupByDbName) {

14,240✔

6040

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));

6,752✔

6042

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));

7,844✔

6045

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

14,596✔

6046

  pRes->info.id.groupId = groupId;

14,240✔

6048

  int64_t ctbNum = 0;

14,240✔

6049

  code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL, NULL);

14,596✔

6050

  QUERY_CHECK_CODE(code, lino, _end);

14,251✔

6051

  code = fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);

14,251✔

6052

  QUERY_CHECK_CODE(code, lino, _end);

14,418✔

6054

_end:

14,418✔

6055

  if (code != TSDB_CODE_SUCCESS) {

14,596✔

6058

  return code;

14,596✔

6061

static void destoryTableCountScanOperator(void* param) {

30,171✔

6062

  STableCountScanOperatorInfo* pTableCountScanInfo = param;

30,171✔

6063

  blockDataDestroy(pTableCountScanInfo->pRes);

30,171✔

6065

  taosArrayDestroy(pTableCountScanInfo->stbUidList);

30,153✔

6066

  taosMemoryFreeClear(param);

30,153✔

6067

30,135✔

taosdata / TDengine / #4983

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4983

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