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

79

bool processBlockWithProbability(const SSampleExecInfo* pInfo) {

590,128,519✔

88

  return true;

590,128,519✔

92

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

67,398✔

93

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

67,398✔

96

67,398✔

98

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

636,176,888✔

99

  int32_t     code = TSDB_CODE_SUCCESS;

636,176,888✔

100

  STimeWindow w = {0};

636,176,888✔

103

  if (pInterval->interval == 0) {

636,467,611✔

104

    *overlap = false;

597,521,528✔

105

    return code;

597,621,804✔

108

  if (order == TSDB_ORDER_ASC) {

39,001,408✔

109

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

36,893,172✔

110

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

36,893,632✔

115

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

36,878,252✔

116

      *overlap = true;

31,961,913✔

117

      return code;

31,964,232✔

121

      getNextTimeWindow(pInterval, &w, order);

4,932,091✔

122

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

4,937,738✔

123

        break;

4,663,292✔

126

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

270,208✔

130

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

270,593✔

131

        *overlap = true;

270,208✔

132

        return code;

270,208✔

136

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

2,108,236✔

137

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

2,116,030✔

142

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

2,111,328✔

143

      *overlap = true;

1,661,411✔

144

      return code;

1,660,905✔

148

      getNextTimeWindow(pInterval, &w, order);

448,746✔

149

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

456,183✔

150

        break;

453,597✔

153

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

2,586✔

157

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

2,586✔

158

        *overlap = true;

2,586✔

159

        return code;

2,586✔

164

  *overlap = false;

5,116,889✔

165

  return code;

5,119,000✔

169

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

167,228,417✔

170

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

167,228,417✔

174

  int64_t buf[2] = {0};

167,256,167✔

175

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

167,225,479✔

177

  STableScanInfo* pTableScanInfo = pOperator->info;

167,255,467✔

179

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

167,172,708✔

182

  if (p1 == NULL) {

167,294,498✔

183

    return NULL;

49,198,587✔

186

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

118,095,911✔

187

  if (NULL == *pPage) {

118,076,657✔

191

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

118,081,307✔

194

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

13,386,219✔

195

  if (NULL == pTableScanInfo->pIgnoreTables) {

13,386,219✔

196

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

1,729,255✔

197

    pTableScanInfo->pIgnoreTables =

1,727,489✔

198

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

1,728,628✔

199

    if (NULL == pTableScanInfo->pIgnoreTables) {

1,727,489✔

204

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

13,387,053✔

206

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

13,382,463✔

210

  return TSDB_CODE_SUCCESS;

13,382,463✔

213

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

700,520,613✔

214

  STableScanInfo* pTableScanInfo = pOperator->info;

700,520,613✔

215

  int32_t         code = TSDB_CODE_SUCCESS;

701,404,386✔

217

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

701,404,386✔

218

    return TSDB_CODE_SUCCESS;

533,871,604✔

221

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

167,237,623✔

223

  SFilePage*  pPage = NULL;

167,264,081✔

224

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

167,250,525✔

226

  if (pRow == NULL) {

167,245,009✔

227

    return TSDB_CODE_SUCCESS;

49,182,139✔

230

  bool notLoadBlock = true;

118,062,870✔

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;

104,685,502✔

239

      break;

104,685,502✔

244

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

117,656,098✔

246

  if (notLoadBlock) {

118,074,598✔

247

    *status = FUNC_DATA_REQUIRED_NOT_LOAD;

13,391,111✔

248

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

13,392,586✔

251

  return code;

118,021,986✔

254

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

9,308,784✔

256

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

9,308,784✔

261

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

9,308,784✔

264

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

148,143,705✔

266

  SStorageAPI* pAPI = &pTaskInfo->storageAPI;

148,143,705✔

267

  bool         allColumnsHaveAgg = true;

148,274,607✔

268

  bool         hasNullSMA = false;

148,323,652✔

269

  if (pLoad != NULL) {

148,246,949✔

270

    *pLoad = false;

148,214,507✔

273

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

148,137,221✔

275

  if (code != TSDB_CODE_SUCCESS) {

148,006,739✔

279

  if (!allColumnsHaveAgg || hasNullSMA) {

148,006,739✔

280

    *pLoad = false;

133,541,730✔

282

    *pLoad = true;

14,465,009✔

285

  return code;

148,051,864✔

288

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

730,327,695✔

290

  int32_t    code = 0;

730,327,695✔

291

  SExprSupp* pSup = &pTableScanInfo->pseudoSup;

730,327,695✔

292

  if (pSup->numOfExprs > 0) {

731,189,125✔

293

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

336,116,611✔

296

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

335,348,606✔

297

      if (pTaskInfo->streamInfo.pState) blockDataCleanup(pBlock);

24✔

302

  return code;

730,734,678✔

305

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

1,123,352,744✔

306

  SLimit*     pLimit = &pLimitInfo->limit;

1,123,352,744✔

307

  const char* id = GET_TASKID(pTaskInfo);

1,123,771,734✔

309

  if (pLimitInfo->remainOffset > 0) {

1,124,164,669✔

310

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

5,364,211✔

311

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

2,618,178✔

312

      blockDataEmpty(pBlock);

2,618,178✔

313

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

2,618,178✔

314

      return false;

2,618,178✔

316

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

2,746,033✔

317

      if (code != TSDB_CODE_SUCCESS) {

2,746,033✔

322

      pLimitInfo->remainOffset = 0;

2,746,033✔

326

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

1,121,536,813✔

328

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

26,959,710✔

329

    blockDataKeepFirstNRows(pBlock, keep);

26,960,113✔

331

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

26,961,204✔

332

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

26,959,802✔

333

    return true;

26,964,174✔

336

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

1,093,849,279✔

337

  return false;

1,094,676,844✔

344

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

731,754,396✔

346

  int32_t        code = TSDB_CODE_SUCCESS;

731,754,396✔

347

  int32_t        lino = 0;

731,754,396✔

348

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

731,754,396✔

349

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

732,165,044✔

350

  bool           loadSMA = false;

731,963,901✔

352

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

731,963,901✔

354

  pCost->totalBlocks += 1;

731,938,596✔

355

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

731,833,512✔

356

  *status = pTableScanInfo->dataBlockLoadFlag;

731,447,558✔

358

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

731,583,789✔

359

    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

94,931,997✔

361

    bool overlap = false;

636,239,567✔

363

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

636,503,441✔

364

    if (ret != TSDB_CODE_SUCCESS) {

635,813,681✔

367

    if (overlap) {

635,813,681✔

368

      (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

33,877,925✔

372

  SDataBlockInfo* pBlockInfo = &pBlock->info;

731,511,726✔

373

  taosMemoryFreeClear(pBlock->pBlockAgg);

731,249,973✔

375

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {

731,775,167✔

382

  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

730,556,463✔

383

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

21,909,930✔

386

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

21,973,537✔

387

    pCost->skipBlocks += 1;

21,958,891✔

388

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

21,955,778✔

389

    return code;

21,955,192✔

390

  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {

708,235,263✔

391

    pCost->loadBlockStatis += 1;

53,357,813✔

392

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

53,293,609✔

393

    bool success = true;

53,293,609✔

394

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

53,311,989✔

395

    if (code) {

53,193,977✔

398

      QUERY_CHECK_CODE(code, lino, _end);

360✔

401

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

53,185,166✔

402

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

5,156,225✔

404

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

5,156,225✔

405

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

5,156,225✔

406

      return code;

5,156,225✔

408

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

48,028,941✔

409

      *status = FUNC_DATA_REQUIRED_DATA_LOAD;

48,204,703✔

413

  if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {

703,831,152✔

420

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

703,923,971✔

421

    bool success = true;

94,942,482✔

422

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

94,984,635✔

423

    if (code) {

94,738,482✔

426

      QUERY_CHECK_CODE(code, lino, _end);

143,278✔

429

    if (success) {

94,881,760✔

430

      size_t size = taosArrayGetSize(pBlock->pDataBlock);

9,308,784✔

431

      bool   keep = false;

9,308,784✔

432

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

9,308,784✔

433

      if (code) {

9,308,784✔

439

      if (!keep) {

9,308,784✔

440

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

3,328,432✔

442

        pCost->filterOutBlocks += 1;

3,328,432✔

443

        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

3,328,432✔

444

        taosMemoryFreeClear(pBlock->pBlockAgg);

3,328,432✔

446

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

3,328,432✔

447

        return TSDB_CODE_SUCCESS;

3,328,432✔

453

  taosMemoryFreeClear(pBlock->pBlockAgg);

701,177,905✔

456

  code = doDynamicPruneDataBlock(pOperator, pBlockInfo, status);

700,787,547✔

457

  if (code) {

700,490,268✔

458

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

28✔

462

  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

700,490,240✔

463

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

13,388,609✔

465

    pCost->skipBlocks += 1;

13,392,589✔

466

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

13,392,074✔

468

    STableScanInfo* p1 = pOperator->info;

13,392,586✔

469

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

13,392,586✔

470

      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;

3,736,285✔

472

      *status = FUNC_DATA_REQUIRED_FILTEROUT;

9,655,789✔

474

    return TSDB_CODE_SUCCESS;

13,392,074✔

477

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

686,481,334✔

478

  pCost->loadBlocks += 1;

687,115,269✔

480

  SSDataBlock* p = NULL;

686,952,438✔

481

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

686,900,568✔

482

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

687,753,361✔

486

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

687,753,361✔

487

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

548,417,310✔

491

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

656,554,617✔

492

    if (code) {

656,462,224✔

498

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

687,180,047✔

500

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

687,745,563✔

501

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

91,580,113✔

502

    QUERY_CHECK_CODE(code, lino, _end);

91,543,491✔

504

    int64_t st = taosGetTimestampUs();

91,503,308✔

505

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

91,494,965✔

506

    pTableScanInfo->readRecorder.filterTime += el;

91,494,965✔

508

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

91,445,401✔

509

      pCost->filterOutBlocks += 1;

18,580,221✔

510

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

18,560,968✔

513

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

72,780,495✔

517

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

687,212,415✔

518

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

687,742,222✔

519

    setOperatorCompleted(pOperator);

3,427,126✔

522

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

687,743,226✔

524

_end:

688,168,742✔

525

  if (code != TSDB_CODE_SUCCESS) {

688,024,997✔

526

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

129,480✔

528

  return code;

687,235,932✔

531

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

67,398✔

532

  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

67,398✔

534

  switchCtxOrder(pCtx, numOfOutput);

67,398✔

535

  pTableScanInfo->cond.order = TSDB_ORDER_DESC;

67,398✔

538

67,398✔

606

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

336,186,983✔

608

  int32_t          code = TSDB_CODE_SUCCESS;

336,186,983✔

609

  int32_t          lino = 0;

336,186,983✔

610

  bool             freeReader = false;

336,186,983✔

611

  LRUHandle*       h = NULL;

336,186,983✔

612

  STableCachedVal  val = {0};

336,186,983✔

613

  SMetaReader      mr = {0};

336,186,983✔

614

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

336,087,481✔

615

  int32_t          insertRet = TAOS_LRU_STATUS_OK;

336,328,831✔

616

  STableCachedVal* pVal = NULL;

336,328,831✔

619

  if (numOfExpr <= 0) {

336,568,150✔

620

    return TSDB_CODE_SUCCESS;

244,854✔

626

  int32_t backupRows = pBlock->info.rows;

336,323,296✔

627

  pBlock->info.rows = rows;

336,257,493✔

633

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

336,253,087✔

634

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

336,183,105✔

635

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

336,488,602✔

636

    if (code != TSDB_CODE_SUCCESS) {

335,940,840✔

651

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

335,940,840✔

653

    val.pName = mr.me.name;

336,347,258✔

654

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

336,347,258✔

656

    freeReader = true;

336,347,258✔

698

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

1,321,543,615✔

699

    const SExprInfo* pExpr1 = &pExpr[j];

985,036,548✔

700

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

984,916,277✔

702

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

985,489,341✔

703

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

985,506,722✔

705

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

985,602,271✔

708

    if (fmIsScanPseudoColumnFunc(functionId)) {

985,770,709✔

709

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

136,459,762✔

710

      if (fType == FUNCTION_TYPE_TBNAME) {

136,470,411✔

711

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

136,470,411✔

712

        QUERY_CHECK_CODE(code, lino, _end);

136,401,758✔

721

      STagVal tagVal = {0};

848,484,391✔

722

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

848,461,249✔

723

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

849,001,387✔

725

      char* data = NULL;

849,133,465✔

726

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

849,133,465✔

727

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

844,045,828✔

729

        data = (char*)p;

5,060,176✔

732

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

848,650,028✔

733

      if (isNullVal) {

848,890,779✔

734

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

4,577,113✔

735

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

844,313,666✔

736

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

843,469,121✔

737

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

843,893,118✔

738

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

291,202,593✔

739

          if (tmp != NULL) {

291,312,814✔

740

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

291,312,814✔

741

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

291,315,854✔

743

            taosMemoryFree(tmp);

291,500,534✔

745

          taosMemoryFree(data);

291,498,301✔

747

        QUERY_CHECK_CODE(code, lino, _end);

843,799,070✔

749

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

3,029,666✔

750

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

2,545,825✔

751

          QUERY_CHECK_CODE(code, lino, _end);

2,547,732✔

758

  pBlock->info.rows = backupRows;

336,507,067✔

760

_end:

336,400,213✔

762

  if (NULL != pVal) {

336,161,415✔

771

  if (freeReader) {

336,161,415✔

772

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

336,106,446✔

774

  if (code != TSDB_CODE_SUCCESS) {

336,441,133✔

777

  return code;

336,218,753✔

780

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

136,421,750✔

782

  int32_t                     code = TSDB_CODE_SUCCESS;

136,421,750✔

783

  int32_t                     lino = 0;

136,421,750✔

784

  struct SScalarFuncExecFuncs fpSet = {0};

136,421,750✔

785

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

136,446,751✔

786

  QUERY_CHECK_CODE(code, lino, _end);

136,478,149✔

788

  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;

136,478,149✔

789

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

136,478,149✔

790

  STR_TO_VARSTR(buf, name)

136,479,506✔

792

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

136,435,353✔

794

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

136,430,046✔

795

  QUERY_CHECK_CODE(code, lino, _end);

136,433,626✔

797

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

136,433,626✔

798

  QUERY_CHECK_CODE(code, lino, _end);

136,431,722✔

800

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

136,431,722✔

801

  SScalarParam param = {.columnData = pColInfoData};

136,439,568✔

803

  if (fpSet.process != NULL) {

136,409,881✔

804

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

136,409,881✔

805

    QUERY_CHECK_CODE(code, lino, _end);

136,445,793✔

810

  colDataDestroy(&infoData);

136,445,793✔

812

_end:

136,436,304✔

813

  if (code != TSDB_CODE_SUCCESS) {

136,459,019✔

816

  return code;

136,459,019✔

885

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

182,161,112✔

886

  int32_t code = TSDB_CODE_SUCCESS;

182,161,112✔

887

  int32_t lino = 0;

182,161,112✔

888

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

182,161,112✔

889

  QUERY_CHECK_CODE(code, lino, _end);

181,726,695✔

891

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

181,726,695✔

892

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

181,903,636✔

893

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

181,803,638✔

894

  taosMemoryFreeClear(pInfo->pResBlock->pBlockAgg);

181,722,328✔

896

  if (!pInfo->needCountEmptyTable) {

182,009,403✔

897

    pInfo->countState = TABLE_COUNT_STATE_END;

163,668,513✔

899

    pInfo->countState = TABLE_COUNT_STATE_SCAN;

18,194,945✔

902

_end:

181,841,373✔

903

  if (code != TSDB_CODE_SUCCESS) {

181,841,373✔

906

  return code;

181,547,834✔

909

void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {

606,274,031✔

910

  if (pInfo->countState == TABLE_COUNT_STATE_END) {

606,274,031✔

911

    return;

489,500,783✔

913

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

117,105,504✔

914

    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;

212,970✔

916

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

116,879,538✔

917

    if (code != TSDB_CODE_SUCCESS) {

116,912,723✔

918

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

52,252,888✔

923

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

47,065,550✔

925

  blockDataEmpty(pBlock);

47,065,550✔

926

  pBlock->info.rows = 1;

47,064,011✔

927

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

47,064,011✔

928

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

47,064,011✔

929

  pBlock->info.blankFill = true;

47,060,420✔

932

  int32_t col_num = blockDataGetNumOfCols(pBlock);

47,058,881✔

933

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

193,896,112✔

934

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

146,837,692✔

939

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

47,058,420✔

941

  return pBlock;

47,060,472✔

944

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

45,216,901✔

945

  STableScanInfo* pTableScanInfo = pOperator->info;

45,216,901✔

947

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

45,221,518✔

949

  pOperator->resultInfo.totalRows++;

45,190,894✔

950

  return pBlock;

45,219,570✔

957

static int32_t prepareTableScan(SOperatorInfo* pOperator) {

766,098,481✔

958

  int32_t code = TSDB_CODE_SUCCESS;

766,098,481✔

959

  int32_t lino = 0;

766,098,481✔

961

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

766,098,481✔

962

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

766,385,840✔

963

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

767,070,366✔

964

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

34,128,734✔

965

         NOTIFY_TYPE_SCAN_PARAM &&

3,042,767✔

966

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

3,042,767✔

967

    STableScanOperatorParam* pNotify =

3,042,767✔

970

    STableScanInfo* pTableScanInfo = pOperator->info;

3,042,767✔

971

    SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

3,042,767✔

972

    SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

3,042,767✔

973

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

3,042,767✔

975

    QUERY_CHECK_CODE(code, lino, _end);

3,042,767✔

977

    if (!pGetParam->reUse) {

3,042,767✔

978

      freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

97,698✔

984

      pGetParam->reUse = false;

2,945,069✔

989

    pOperator->pOperatorGetParam = NULL;

3,042,767✔

992

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

767,135,270✔

993

_end:

88,834,772✔

994

  if (code != TSDB_CODE_SUCCESS) {

767,108,245✔

998

  return code;

767,132,058✔

1001

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

727,437,102✔

1002

  int32_t         code = TSDB_CODE_SUCCESS;

727,437,102✔

1003

  int32_t         lino = 0;

727,437,102✔

1004

  STableScanInfo* pTableScanInfo = pOperator->info;

727,437,102✔

1005

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

727,680,109✔

1006

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

727,114,251✔

1007

  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;

727,304,864✔

1008

  bool            hasNext = false;

727,230,976✔

1009

  int64_t         st = taosGetTimestampUs();

727,183,872✔

1011

  QRY_PARAM_CHECK(ppRes);

727,183,872✔

1012

  pBlock->info.dataLoad = false;

727,205,823✔

1014

  while (true) {

25,229,426✔

1015

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

752,088,896✔

1016

    if (code != TSDB_CODE_SUCCESS) {

751,710,757✔

1018

      QUERY_CHECK_CODE(code, lino, _end);

24✔

1021

    if (!hasNext) {

751,518,813✔

1022

      break;

162,130,062✔

1025

    if (isTaskKilled(pTaskInfo)) {

589,388,751✔

1026

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

16,021✔

1027

      code = pTaskInfo->code;

15,997✔

1028

      QUERY_CHECK_CODE(code, lino, _end);

15,997✔

1031

    if (pOperator->status == OP_EXEC_DONE) {

590,811,621✔

1032

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

58,464✔

1033

      break;

58,464✔

1037

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

590,431,224✔

1038

    if (!processThisBlock) {

590,434,421✔

1042

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

590,434,421✔

1043

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

590,680,180✔

1046

    uint32_t status = 0;

590,985,383✔

1047

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

590,978,906✔

1048

    QUERY_CHECK_CODE(code, lino, _end);

590,096,469✔

1050

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

589,983,629✔

1051

      break;

3,736,285✔

1055

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

586,247,344✔

1056

      continue;

25,229,426✔

1059

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

561,164,415✔

1060

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

561,601,810✔

1062

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

561,546,745✔

1063

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

561,645,304✔

1065

    (*ppRes) = pBlock;

561,615,502✔

1066

    return code;

561,195,231✔

1069

_end:

166,111,165✔

1070

  if (code != TSDB_CODE_SUCCESS) {

166,127,061✔

1071

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

180,743✔

1072

    pTaskInfo->code = code;

180,743✔

1073

    T_LONG_JMP(pTaskInfo->env, code);

128,837✔

1075

  return code;

165,946,318✔

1078

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

775,486,245✔

1079

  int32_t         code = TSDB_CODE_SUCCESS;

775,486,245✔

1080

  int32_t         lino = 0;

775,486,245✔

1081

  STableScanInfo* pTableScanInfo = pOperator->info;

775,486,245✔

1082

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

775,499,804✔

1083

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

775,267,597✔

1085

  QRY_PARAM_CHECK(pBlock);

775,123,975✔

1088

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

775,020,910✔

1089

    return code;

2,807,779✔

1093

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

922,709,657✔

1094

    SSDataBlock* p = NULL;

637,283,023✔

1095

    code = doTableScanImplNext(pOperator, &p);

636,983,431✔

1096

    QUERY_CHECK_CODE(code, lino, _end);

636,737,221✔

1098

    if (p != NULL) {

636,737,221✔

1099

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

486,223,372✔

1100

      *pBlock = p;

486,342,910✔

1101

      return code;

486,370,086✔

1104

    pTableScanInfo->scanTimes += 1;

150,513,849✔

1105

    taosHashClear(pTableScanInfo->pIgnoreTables);

150,579,824✔

1107

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

150,552,394✔

1108

      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

83,305✔

1109

      pTableScanInfo->base.scanFlag = MAIN_SCAN;

83,305✔

1110

      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

83,305✔

1111

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

83,305✔

1114

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

83,305✔

1115

      QUERY_CHECK_CODE(code, lino, _end);

93,392✔

1119

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

285,830,088✔

1120

  if (pTableScanInfo->scanTimes < total) {

285,731,559✔

1121

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

90,112,810✔

1122

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

67,398✔

1123

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

67,398✔

1124

      QUERY_CHECK_CODE(code, lino, _end);

67,398✔

1125

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

67,398✔

1128

    while (pTableScanInfo->scanTimes < total) {

105,401,160✔

1129

      SSDataBlock* p = NULL;

90,125,016✔

1130

      code = doTableScanImplNext(pOperator, &p);

90,098,548✔

1131

      QUERY_CHECK_CODE(code, lino, _end);

90,033,418✔

1133

      if (p != NULL) {

90,033,418✔

1134

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

74,718,759✔

1135

        *pBlock = p;

74,807,610✔

1136

        return code;

74,803,903✔

1139

      pTableScanInfo->scanTimes += 1;

15,314,659✔

1140

      taosHashClear(pTableScanInfo->pIgnoreTables);

15,320,844✔

1142

      if (pTableScanInfo->scanTimes < total) {

15,315,572✔

1148

        QUERY_CHECK_CODE(code, lino, _end);

327✔

1153

  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {

210,926,395✔

1154

    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;

62,822,884✔

1155

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

62,817,223✔

1156

      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {

104,132✔

1169

      int32_t numOfTables = 0;

62,707,024✔

1170

      code = tableListGetSize(pTableListInfo, &numOfTables);

62,707,448✔

1171

      QUERY_CHECK_CODE(code, lino, _end);

62,717,132✔

1173

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

62,717,132✔

1175

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

61,383,750✔

1176

        if (pIte != NULL) {

61,370,900✔

1177

          size_t        keySize = 0;

45,206,128✔

1178

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

45,208,180✔

1179

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

45,211,258✔

1180

          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);

45,213,310✔

1181

          markGroupProcessed(pTableScanInfo, *pGroupId);

45,219,466✔

1182

          *pBlock = getBlockForEmptyTable(pOperator, &info);

45,218,440✔

1184

          return code;

45,214,901✔

1188

    pTableScanInfo->countState = TABLE_COUNT_STATE_END;

17,604,467✔

1191

_end:

148,174,554✔

1192

  if (code != TSDB_CODE_SUCCESS) {

165,743,083✔

1197

  return code;

165,743,083✔

1200

static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {

963,733✔

1201

  STableScanInfo*          pInfo = pOperator->info;

963,733✔

1202

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

964,108✔

1203

  int32_t                  code = 0;

964,108✔

1204

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

964,108✔

1205

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

963,733✔

1206

  int32_t                  num = taosArrayGetSize(pParam->pUidList);

964,108✔

1207

  if (num <= 0) {

964,108✔

1212

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

964,108✔

1216

  if (pParam->tableSeq) {

964,483✔

1217

    pListInfo->oneTableForEachGroup = true;

711,654✔

1218

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

711,654✔

1224

    pListInfo->oneTableForEachGroup = false;

252,454✔

1225

    pListInfo->numOfOuputGroups = 1;

252,454✔

1227

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

964,108✔

1230

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

707,202✔

1231

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

707,202✔

1234

  STableKeyInfo info = {.groupId = 0};

964,108✔

1235

  int32_t       tableIdx = 0;

963,733✔

1236

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

9,000,077✔

1237

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

8,035,969✔

1238

    if (!pUid) {

8,019,753✔

1243

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

8,019,753✔

1244

      if (TSDB_CODE_DUP_KEY == terrno) {

6,729,597✔

1245

        continue;

6,724,487✔

1250

    info.uid = *pUid;

1,312,504✔

1251

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

1,312,504✔

1252

    if (p == NULL) {

1,312,504✔

1256

    tableIdx++;

1,312,504✔

1257

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

1,312,504✔

1260

  return code;

964,108✔

1263

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

166,368,180✔

1265

  const char* idStr = GET_TASKID(pTaskInfo);

166,368,180✔

1266

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

166,477,758✔

1267

  if (code) {

165,578,460✔

1272

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

165,578,460✔

1277

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

331,822,995✔

1278

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

166,136,224✔

1279

  if (code) {

165,892,983✔

1280

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

1,787✔

1283

  return code;

165,478,918✔

1286

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

327,631,427✔

1287

  SColIdPair* node1 = (SColIdPair*)elem1;

327,631,427✔

1288

  SColIdPair* node2 = (SColIdPair*)elem2;

327,631,427✔

1290

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

327,631,427✔

1291

    return 0;

108,798✔

1294

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

327,710,886✔

1297

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

437,603,106✔

1298

  SColIdSlotIdPair* node1 = (SColIdSlotIdPair*)elem1;

437,603,106✔

1299

  SColIdSlotIdPair* node2 = (SColIdSlotIdPair*)elem2;

437,603,106✔

1301

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

437,603,106✔

1302

    return 0;

166,206✔

1305

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

437,606,811✔

1319

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

100,118,146✔

1320

  if (pList == NULL) {

100,118,146✔

1324

  int32_t left = 0;

100,118,146✔

1325

  int32_t right = taosArrayGetSize(pList) - 1;

100,118,146✔

1326

  while (left <= right) {

311,015,121✔

1327

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

311,014,898✔

1328

    SColMatchItem* pItem = taosArrayGet(pList, mid);

311,014,898✔

1329

    if (pItem->colId == colId) {

311,014,675✔

1331

      while (mid > 0) {

100,117,923✔

1332

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

75,200,557✔

1333

        if (pPrev->colId != colId) {

75,200,334✔

1334

          break;

75,200,334✔

1338

      return taosArrayGet(pList, mid);

100,117,700✔

1341

    if (pItem->colId < colId) {

210,896,975✔

1342

      left = mid + 1;

99,023,700✔

1344

      right = mid - 1;

111,873,275✔

1348

  return NULL;

223✔

1351

static bool isNewScanParam(STableScanOperatorParam* pParam) {

42,625,872✔

1352

  return pParam->isNewParam;

42,625,872✔

1355

static int32_t createVTableScanInfoFromBatchParam(SOperatorInfo* pOperator) {

21,251,212✔

1356

  int32_t                  code = 0;

21,251,212✔

1357

  int32_t                  lino = 0;

21,251,212✔

1358

  STableScanInfo*          pInfo = pOperator->info;

21,251,212✔

1359

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

21,252,824✔

1360

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

21,246,870✔

1361

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

21,249,438✔

1362

  SMetaReader              orgTable = {0};

21,249,958✔

1363

  SMetaReader              superTable = {0};

21,279,266✔

1364

  SSchemaWrapper*          schema = NULL;

21,266,184✔

1365

  SArray*                  pColArray = NULL;

21,266,184✔

1366

  SArray*                  pBlockColArray = NULL;

21,266,184✔

1367

  int32_t                  num = 0;

21,266,184✔

1368

  STableKeyInfo*           pList = NULL;

21,264,368✔

1372

  cleanupQueryTableDataCond(&pInfo->base.cond);

21,263,406✔

1374

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

21,261,600✔

1375

  if (pGetParam &&

21,265,788✔

1376

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

5,189,770✔

1377

    STableScanOperatorParam* pParam =

5,202,596✔

1379

    pInfo->pBatchColMap = pParam->pBatchTbInfo;

5,188,088✔

1380

    pParam->pBatchTbInfo = NULL;

5,187,116✔

1381

    pInfo->cachedTimeWindow = pParam->window;

5,175,660✔

1382

    pInfo->currentBatchIdx = 0;

5,190,212✔

1383

    pInfo->lastBatchIdx = -1;

5,183,976✔

1384

    pInfo->cachedTagList = pParam->pTagList;

5,172,548✔

1385

    pParam->pTagList = NULL;

5,185,964✔

1386

    pInfo->cachedGroupId = pParam->groupid;

5,183,898✔

1388

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

5,187,508✔

1390

    taosHashClear(pInfo->pIgnoreTables);

5,187,508✔

1392

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

5,203,216✔

1393

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

5,002,366✔

1394

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

63,290,428✔

1395

        SqlFunctionCtx ctx = pSup->pCtx[i];

58,334,074✔

1396

        ctx.skipDynDataCheck = false;

58,289,180✔

1401

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

21,219,780✔

1402

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

21,279,130✔

1404

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

21,279,130✔

1405

    isNewTable = false;

13,004,760✔

1407

    isNewTable = true;

8,269,218✔

1408

    pInfo->lastBatchIdx = pInfo->currentBatchIdx;

8,269,218✔

1411

  if (isNewTable) {

21,271,378✔

1412

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

8,268,674✔

1413

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

8,272,512✔

1414

    pAPI->metaReaderFn.readerReleaseLock(&orgTable);

8,257,932✔

1415

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

8,240,646✔

1416

    QUERY_CHECK_CODE(code, lino, _return);

8,187,204✔

1417

    switch (orgTable.me.type) {

8,187,204✔

1418

      case TSDB_CHILD_TABLE:

284,104✔

1419

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

284,104✔

1420

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

284,104✔

1421

        pAPI->metaReaderFn.readerReleaseLock(&superTable);

284,684✔

1422

        QUERY_CHECK_CODE(code, lino, _return);

284,684✔

1423

        schema = &superTable.me.stbEntry.schemaRow;

284,684✔

1424

        break;

284,684✔

1425

      case TSDB_NORMAL_TABLE:

7,911,102✔

1426

        schema = &orgTable.me.ntbEntry.schemaRow;

7,911,102✔

1427

        break;

7,911,102✔

1434

    pListInfo->oneTableForEachGroup = true;

8,195,786✔

1435

    taosHashClear(pListInfo->map);

8,209,458✔

1436

    taosArrayClear(pListInfo->pTableList);

8,281,340✔

1438

    uint64_t      pUid = orgTable.me.uid;

8,277,708✔

1439

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

8,277,708✔

1440

    int32_t       tableIdx = 0;

8,278,748✔

1441

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

8,280,300✔

1442

    QUERY_CHECK_CODE(code, lino, _return);

8,280,308✔

1443

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

16,551,274✔

1444

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

8,270,966✔

1446

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

8,270,966✔

1447

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

8,259,534✔

1448

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

8,259,534✔

1449

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

8,259,300✔

1452

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

8,259,300✔

1453

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

8,266,870✔

1455

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

31,810,224✔

1456

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

23,531,652✔

1457

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

187,273,872✔

1458

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

186,964,402✔

1459

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

23,558,366✔

1460

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

23,546,834✔

1461

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

23,546,834✔

1463

          break;

23,524,672✔

1468

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

40,086,752✔

1469

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

31,795,644✔

1470

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

191,487,780✔

1471

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

191,022,290✔

1472

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

191,094,626✔

1473

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

31,821,914✔

1474

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

31,843,280✔

1475

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

31,843,280✔

1477

          break;

31,807,232✔

1482

    taosArraySort(pColArray, compareColIdPair);

8,274,652✔

1483

    taosArraySort(pBlockColArray, compareColIdSlotIdPair);

8,261,696✔

1484

    if (pInfo->pBlockColMap) {

8,271,358✔

1485

      taosArrayDestroy(pInfo->pBlockColMap);

5,718,508✔

1486

      pInfo->pBlockColMap = NULL;

5,717,400✔

1488

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

8,266,076✔

1489

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

8,267,280✔

1492

    taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

8,240,706✔

1493

    taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

8,253,220✔

1495

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

8,274,370✔

1496

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

7,539,978✔

1497

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

93,620,170✔

1498

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

86,118,424✔

1499

        bool           needScan = false;

86,104,110✔

1500

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

173,369,894✔

1501

          SFunctParam pParam = ctx->param[j];

87,260,936✔

1502

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

417,604,882✔

1503

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

349,831,148✔

1504

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

351,297,158✔

1505

              needScan |= true;

21,754,374✔

1506

              break;

21,754,374✔

1510

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

86,134,464✔

1511

        if (!needScan) {

86,212,686✔

1512

          ctx->skipDynDataCheck = true;

64,471,224✔

1514

          ctx->skipDynDataCheck = false;

21,741,462✔

1521

    pColArray = pInfo->lastColArray;

13,002,704✔

1522

    pBlockColArray = pInfo->lastBlockColArray;

13,004,256✔

1525

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

21,268,752✔

1526

  QUERY_CHECK_CODE(code, lino, _return);

21,279,806✔

1528

  if (pInfo->pResBlock) {

21,279,806✔

1529

    blockDataDestroy(pInfo->pResBlock);

16,071,510✔

1530

    pInfo->pResBlock = NULL;

16,069,830✔

1533

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

21,277,716✔

1541

  if (isNewTable) {

21,248,750✔

1542

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

8,246,124✔

1543

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

7,523,530✔

1544

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

7,511,536✔

1546

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

725,364✔

1547

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

718,698✔

1550

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

13,002,626✔

1551

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

1,742,800✔

1553

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

11,257,702✔

1557

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

21,205,260✔

1558

  pInfo->currentGroupId = 0;

21,239,088✔

1559

  pInfo->ignoreTag = true;

21,233,578✔

1561

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

21,209,500✔

1562

  QUERY_CHECK_CODE(code, lino, _return);

21,263,512✔

1564

  if (isNewTable) {

21,263,512✔

1565

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

8,261,312✔

1566

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

5,719,976✔

1568

    pInfo->base.dataReader = NULL;

8,225,208✔

1570

    taosRLockLatch(&pTaskInfo->lock);

8,228,140✔

1571

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

8,281,340✔

1572

    taosRUnLockLatch(&pTaskInfo->lock);

8,208,310✔

1573

    QUERY_CHECK_CODE(code, lino, _return);

8,239,368✔

1575

    pInfo->newReader = true;

8,239,368✔

1576

    if (pInfo->lastColArray) {

8,244,962✔

1577

      taosArrayDestroy(pInfo->lastColArray);

3,066,222✔

1578

      pInfo->lastColArray = NULL;

3,066,222✔

1580

    if (pInfo->lastBlockColArray) {

8,182,960✔

1581

      taosArrayDestroy(pInfo->lastBlockColArray);

3,066,162✔

1582

      pInfo->lastBlockColArray = NULL;

3,067,254✔

1584

    pInfo->lastColArray = pColArray;

8,193,160✔

1585

    pInfo->lastBlockColArray = pBlockColArray;

8,185,548✔

1587

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

13,002,200✔

1588

    QUERY_CHECK_CODE(code, lino, _return);

13,005,800✔

1590

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

13,005,800✔

1591

    pInfo->newReader = false;

13,005,280✔

1592

    pInfo->scanTimes = 0;

13,005,280✔

1595

  pOperator->status = OP_OPENED;

21,165,376✔

1597

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

21,149,712✔

1601

  pInfo->currentGroupId = -1;

21,205,748✔

1602

_return:

21,213,026✔

1603

  if (code) {

21,144,106✔

1606

  pAPI->metaReaderFn.clearReader(&superTable);

21,144,106✔

1607

  pAPI->metaReaderFn.clearReader(&orgTable);

21,195,370✔

1608

  return code;

21,248,508✔

1611

static int32_t createVTableScanInfoFromParam(SOperatorInfo* pOperator) {

24,917,589✔

1612

  int32_t                  code = 0;

24,917,589✔

1613

  int32_t                  lino = 0;

24,917,589✔

1614

  STableScanInfo*          pInfo = pOperator->info;

24,917,589✔

1615

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

24,917,589✔

1616

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

24,917,589✔

1617

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

24,917,589✔

1618

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

24,917,589✔

1619

  SMetaReader              orgTable = {0};

24,917,589✔

1620

  SMetaReader              superTable = {0};

24,917,589✔

1621

  SSchemaWrapper*          schema = NULL;

24,917,589✔

1622

  SArray*                  pColArray = NULL;

24,917,589✔

1623

  SArray*                  pBlockColArray = NULL;

24,917,589✔

1624

  int32_t                  num = 0;

24,917,589✔

1625

  STableKeyInfo*           pList = NULL;

24,917,589✔

1627

  SArray*                  pMatchList = NULL;

24,917,589✔

1629

  cleanupQueryTableDataCond(&pInfo->base.cond);

24,917,589✔

1631

  pOrgTbInfo = pParam->pOrgTbInfo;

24,917,589✔

1633

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

24,917,589✔

1635

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

24,917,589✔

1636

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

24,917,589✔

1637

  pAPI->metaReaderFn.readerReleaseLock(&orgTable);

24,917,589✔

1638

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

24,917,589✔

1639

  QUERY_CHECK_CODE(code, lino, _return);

24,917,589✔

1640

  switch (orgTable.me.type) {

24,917,589✔

1641

    case TSDB_CHILD_TABLE:

6,498,320✔

1642

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

6,498,320✔

1643

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

6,498,320✔

1644

      pAPI->metaReaderFn.readerReleaseLock(&superTable);

6,498,320✔

1645

      QUERY_CHECK_CODE(code, lino, _return);

6,498,320✔

1646

      schema = &superTable.me.stbEntry.schemaRow;

6,498,320✔

1647

      break;

6,498,320✔

1648

    case TSDB_NORMAL_TABLE:

18,419,269✔

1649

      schema = &orgTable.me.ntbEntry.schemaRow;

18,419,269✔

1650

      break;

18,419,269✔

1657

  pListInfo->oneTableForEachGroup = true;

24,917,589✔

1658

  taosHashClear(pListInfo->map);

24,917,589✔

1659

  taosArrayClear(pListInfo->pTableList);

24,917,589✔

1661

  uint64_t      pUid = orgTable.me.uid;

24,917,589✔

1662

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

24,917,589✔

1663

  int32_t       tableIdx = 0;

24,917,589✔

1664

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

24,917,589✔

1665

  QUERY_CHECK_CODE(code, lino, _return);

24,917,589✔

1666

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

49,835,178✔

1667

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

24,917,589✔

1669

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

24,917,589✔

1670

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

24,917,589✔

1671

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

24,917,589✔

1672

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

24,917,589✔

1673

  if (pInfo->pBlockColMap) {

24,917,589✔

1674

    taosArrayDestroy(pInfo->pBlockColMap);

22,424,361✔

1675

    pInfo->pBlockColMap = NULL;

22,424,361✔

1677

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

24,917,589✔

1678

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

24,917,589✔

1681

  SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID,

24,917,589✔

1685

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

24,917,589✔

1687

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

100,118,146✔

1688

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

75,200,557✔

1689

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

636,052,484✔

1690

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

636,052,484✔

1691

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

75,200,557✔

1692

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

75,200,557✔

1693

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

75,200,557✔

1694

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

75,200,557✔

1695

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

75,200,557✔

1696

        break;

75,200,557✔

1701

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

24,917,589✔

1702

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

24,917,589✔

1707

      pMatchList = pInfo->base.matchInfo.pList;

24,917,589✔

1712

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

125,034,884✔

1713

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

100,118,146✔

1714

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

100,118,146✔

1715

    if (pItem) {

100,117,700✔

1716

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

100,117,700✔

1717

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

405✔

1719

        code = TSDB_CODE_VTABLE_COLUMN_TYPE_MISMATCH;

628✔

1720

        goto _return;

628✔

1722

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

100,117,295✔

1723

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

100,117,295✔

1724

      if (i > 0) {

100,117,295✔

1725

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

150,399,412✔

1730

  taosArraySort(pColArray, compareColIdPair);

24,916,738✔

1731

  taosArraySort(pBlockColArray, compareColIdSlotIdPair);

24,916,961✔

1732

  taosArraySort(pInfo->pBlockColMap, compareColIdSlotIdPair);

24,916,961✔

1734

  taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

24,916,961✔

1735

  taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

24,916,961✔

1737

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

24,916,961✔

1738

  QUERY_CHECK_CODE(code, lino, _return);

24,916,961✔

1740

  if (pInfo->pResBlock) {

24,916,961✔

1741

    blockDataDestroy(pInfo->pResBlock);

17,711,231✔

1742

    pInfo->pResBlock = NULL;

17,711,231✔

1745

  if (isNewScanParam(pParam)) {

24,916,961✔

1746

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

7,213,706✔

1747

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

7,213,706✔

1748

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

7,213,706✔

1754

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

17,703,255✔

1756

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

24,916,961✔

1757

  pInfo->currentGroupId = 0;

24,916,961✔

1758

  pInfo->ignoreTag = true;

24,916,961✔

1760

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

24,916,961✔

1761

  QUERY_CHECK_CODE(code, lino, _return);

24,916,738✔

1763

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

24,916,738✔

1764

  if (reader) {

24,916,961✔

1765

    if (isNewScanParam(pParam)) {

17,708,911✔

1766

      pAPI->tsdReader.tsdReaderClose(*reader);

5,656✔

1767

      pInfo->base.dataReader = NULL;

5,656✔

1769

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

5,656✔

1770

      QUERY_CHECK_CODE(code, lino, _return);

5,656✔

1772

      taosRLockLatch(&pTaskInfo->lock);

5,656✔

1773

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

5,656✔

1774

      taosRUnLockLatch(&pTaskInfo->lock);

5,656✔

1775

      QUERY_CHECK_CODE(code, lino, _return);

5,656✔

1777

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

5,656✔

1778

      QUERY_CHECK_CODE(code, lino, _return);

5,656✔

1779

      pInfo->newReader = true;

5,656✔

1781

      pInfo->base.dataReader = *reader;

17,703,255✔

1782

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

17,703,255✔

1783

      QUERY_CHECK_CODE(code, lino, _return);

17,703,255✔

1785

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

17,703,255✔

1786

      pInfo->newReader = false;

17,703,255✔

1787

      pInfo->scanTimes = 0;

17,703,255✔

1790

    pInfo->base.dataReader = NULL;

7,208,050✔

1792

    taosRLockLatch(&pTaskInfo->lock);

7,208,050✔

1793

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

7,208,050✔

1794

    taosRUnLockLatch(&pTaskInfo->lock);

7,208,050✔

1795

    QUERY_CHECK_CODE(code, lino, _return);

7,208,050✔

1797

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

7,208,050✔

1798

    QUERY_CHECK_CODE(code, lino, _return);

7,208,050✔

1799

    pInfo->newReader = true;

7,208,050✔

1801

  pOperator->status = OP_OPENED;

24,916,961✔

1803

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

24,916,961✔

1807

  pInfo->currentGroupId = -1;

24,916,961✔

1808

_return:

24,917,589✔

1809

  if (code) {

24,917,589✔

1810

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

628✔

1812

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

24,917,589✔

1815

  taosArrayDestroy(pColArray);

24,917,589✔

1816

  taosArrayDestroy(pBlockColArray);

24,917,589✔

1817

  pAPI->metaReaderFn.clearReader(&superTable);

24,917,589✔

1818

  pAPI->metaReaderFn.clearReader(&orgTable);

24,917,589✔

1819

  return code;

24,917,589✔

1822

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

162,971,328✔

1823

  int32_t         code = TSDB_CODE_SUCCESS;

162,971,328✔

1824

  int32_t         lino = 0;

162,971,328✔

1825

  STableScanInfo* pInfo = pOperator->info;

162,971,328✔

1826

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

163,003,968✔

1827

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

162,999,649✔

1828

  int32_t         numOfTables = 0;

162,907,421✔

1830

  QRY_PARAM_CHECK(pResult);

162,912,255✔

1832

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

162,897,673✔

1833

  QUERY_CHECK_CODE(code, lino, _end);

163,011,762✔

1835

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

163,011,762✔

1836

    setOperatorCompleted(pOperator);

147,172,882✔

1837

    if (pOperator->dynamicTask) {

147,123,882✔

1838

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

961,491✔

1839

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

961,866✔

1841

    return code;

147,159,200✔

1845

  pOperator->status = OP_OPENED;

15,675,793✔

1846

  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);

15,794,197✔

1848

  int32_t        num = 0;

15,767,099✔

1849

  STableKeyInfo* pList = NULL;

15,767,169✔

1850

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

15,764,919✔

1851

  QUERY_CHECK_CODE(code, lino, _end);

15,805,495✔

1853

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

15,805,495✔

1854

  QUERY_CHECK_CODE(code, lino, _end);

15,795,714✔

1856

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

15,795,714✔

1857

  QUERY_CHECK_CODE(code, lino, _end);

15,759,671✔

1858

  pInfo->scanTimes = 0;

15,759,671✔

1860

  code = doGroupedTableScan(pOperator, pResult);

15,767,039✔

1861

  QUERY_CHECK_CODE(code, lino, _end);

15,842,363✔

1863

  if (*pResult != NULL) {

15,842,363✔

1864

    if (pOperator->dynamicTask) {

12,862,522✔

1865

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

12,616,676✔

1867

    return code;

12,861,498✔

1870

_end:

2,979,841✔

1871

  if (code != TSDB_CODE_SUCCESS) {

2,977,769✔

1875

  return code;

2,979,843✔

1878

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

718,042,437✔

1879

  int32_t         code = TSDB_CODE_SUCCESS;

718,042,437✔

1880

  int32_t         lino = 0;

718,042,437✔

1881

  STableScanInfo* pInfo = pOperator->info;

718,042,437✔

1882

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

717,933,963✔

1883

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

717,982,870✔

1884

  int32_t         num = 0;

717,851,664✔

1885

  STableKeyInfo*  pList = NULL;

717,745,513✔

1886

  SSDataBlock*    pResult = NULL;

717,724,531✔

1887

  const char*     idStr = GET_TASKID(pTaskInfo);

717,832,598✔

1889

  QRY_PARAM_CHECK(pResBlock);

717,905,835✔

1891

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

717,915,584✔

1892

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

150,752,135✔

1897

    taosRLockLatch(&pTaskInfo->lock);

150,863,654✔

1898

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

151,016,584✔

1899

    taosRUnLockLatch(&pTaskInfo->lock);

150,500,419✔

1900

    QUERY_CHECK_CODE(code, lino, _end);

150,752,354✔

1902

    if (pInfo->filesetDelimited) {

150,750,567✔

1903

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

253,172✔

1906

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

150,750,349✔

1911

  code = doGroupedTableScan(pOperator, &pResult);

717,450,149✔

1912

  QUERY_CHECK_CODE(code, lino, _end);

716,807,404✔

1914

  if (pResult != NULL) {

716,807,404✔

1915

    if (pOperator->dynamicTask) {

570,176,238✔

1916

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

2,041,743✔

1919

    *pResBlock = pResult;

570,174,988✔

1920

    return code;

569,441,181✔

1924

    code = startNextGroupScan(pOperator, &pResult);

146,711,079✔

1925

    QUERY_CHECK_CODE(code, lino, _end);

147,565,956✔

1927

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

147,565,956✔

1928

      *pResBlock = pResult;

147,455,267✔

1929

      return code;

147,368,788✔

1933

_end:

1,787✔

1934

  if (code != TSDB_CODE_SUCCESS) {

1,787✔

1935

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

1,787✔

1936

    pTaskInfo->code = code;

1,787✔

1939

  return code;

1,787✔

1942

static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {

24,916,961✔

1943

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

24,916,961✔

1946

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

964,108✔

1947

  int32_t                  code = TSDB_CODE_SUCCESS;

964,108✔

1948

  int32_t                  lino = 0;

964,108✔

1949

  STableScanInfo*          pInfo = pOperator->info;

964,108✔

1950

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

964,108✔

1951

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

964,108✔

1952

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

964,108✔

1954

  code = createTableListInfoFromParam(pOperator);

964,108✔

1955

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

964,108✔

1956

  pOperator->pOperatorGetParam = NULL;

963,725✔

1957

  QUERY_CHECK_CODE(code, lino, _end);

963,350✔

1959

  if (pOperator->status == OP_EXEC_DONE) {

963,350✔

1960

    pInfo->currentGroupId = -1;

23,232✔

1961

    pOperator->status = OP_OPENED;

23,232✔

1962

    SSDataBlock* result = NULL;

23,232✔

1965

      code = startNextGroupScan(pOperator, &result);

23,232✔

1966

      QUERY_CHECK_CODE(code, lino, _end);

23,232✔

1968

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

23,232✔

1969

        (*ppRes) = result;

23,232✔

1970

        return code;

23,232✔

1974

_end:

940,118✔

1975

  if (code) {

940,118✔

1978

  return code;

940,118✔

1981

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

17,712,535✔

1982

  int32_t      code = TSDB_CODE_SUCCESS;

17,712,535✔

1983

  int32_t      lino = 0;

17,712,535✔

1984

  bool         found = false;

17,712,535✔

1986

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

17,712,535✔

1987

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

17,712,535✔

1989

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

88,502,733✔

1990

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

70,790,198✔

1991

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

70,790,198✔

1992

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

203,066,400✔

1993

      SColIdSlotIdPair* pColPair = taosArrayGet(pColMap, j);

185,356,850✔

1994

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

185,356,850✔

1995

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

185,356,850✔

1996

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

53,080,648✔

1997

        found = true;

53,080,648✔

1998

        break;

53,080,648✔

2001

    if (!found) {

70,790,198✔

2002

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

17,709,550✔

2003

        pSrcCol->info.slotId = -1;

17,709,550✔

2012

  return code;

17,712,535✔

2018

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

24,917,589✔

2019

  int32_t                  code = TSDB_CODE_SUCCESS;

24,917,589✔

2020

  int32_t                  lino = 0;

24,917,589✔

2021

  STableScanInfo*          pInfo = pOperator->info;

24,917,589✔

2022

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

24,917,589✔

2023

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

24,917,589✔

2024

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

24,917,589✔

2026

  QRY_PARAM_CHECK(ppRes);

24,917,589✔

2027

  qTrace("%s call", __FUNCTION__);

24,917,589✔

2029

  code = createVTableScanInfoFromParam(pOperator);

24,917,589✔

2030

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

24,917,589✔

2031

  pOperator->pOperatorGetParam = NULL;

24,917,589✔

2033

  QUERY_CHECK_CODE(code, lino, _end);

24,917,589✔

2035

  SSDataBlock* result = NULL;

24,916,961✔

2036

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

24,916,961✔

2037

    (*ppRes) = result;

17,860✔

2038

    return code;

17,860✔

2041

  if (pInfo->newReader) {

24,899,101✔

2042

    code = startNextGroupScan(pOperator, &result);

7,213,706✔

2043

    QUERY_CHECK_CODE(code, lino, _end);

7,213,706✔

2045

    code = doGroupedTableScan(pOperator, &result);

17,685,395✔

2046

    QUERY_CHECK_CODE(code, lino, _end);

17,685,395✔

2049

  if (result) {

24,899,101✔

2050

    SSDataBlock* res = NULL;

17,712,535✔

2051

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

17,712,535✔

2052

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

17,712,535✔

2053

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

17,712,535✔

2055

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

17,712,535✔

2056

    QUERY_CHECK_CODE(code, lino, _end);

17,712,535✔

2062

    (*ppRes) = result;

17,712,535✔

2064

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

7,186,566✔

2065

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

7,186,566✔

2067

    blockDataDestroy(pInfo->pResBlock);

7,186,566✔

2068

    pInfo->pResBlock = NULL;

7,186,566✔

2070

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

7,186,566✔

2071

    if (reader) {

7,186,566✔

2072

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

7,186,566✔

2073

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

7,186,566✔

2074

        pInfo->base.dataReader = NULL;

7,186,566✔

2076

      if (pAPI->tsdReader.tsdReaderClose) {

7,186,566✔

2077

        pAPI->tsdReader.tsdReaderClose(*reader);

7,186,566✔

2081

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

7,186,566✔

2082

    QUERY_CHECK_CODE(code, lino, _end);

7,186,566✔

2084

_end:

24,899,729✔

2085

  if (code) {

24,899,729✔

2086

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

628✔

2088

  return code;

24,899,729✔

2091

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

3,774,272✔

2092

  int32_t                  code = TSDB_CODE_SUCCESS;

3,774,272✔

2093

  int32_t                  lino = 0;

3,774,272✔

2094

  STableScanInfo*          pInfo = pOperator->info;

3,774,272✔

2095

  int32_t                  index = 0;

3,774,272✔

2096

  char*                    tagVal = NULL;

3,774,272✔

2098

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

26,407,648✔

2099

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

22,633,376✔

2100

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

22,633,376✔

2102

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

22,632,856✔

2103

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

22,632,856✔

2105

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

2,147,483,647✔

2106

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

2,147,483,647✔

2107

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

2,147,483,647✔

2108

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

2,147,483,647✔

2110

        varDataSetLen(tagVal, pTagVal->nData);

2,147,483,647✔

2111

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

2,147,483,647✔

2112

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

2,147,483,647✔

2113

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2114

        taosMemoryFreeClear(tagVal);

2,147,483,647✔

2116

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

2,147,483,647✔

2117

        QUERY_CHECK_CODE(code, lino, _end);

2,147,483,647✔

2120

    index++;

22,633,376✔

2122

  return code;

3,774,272✔

2129

static void clearVstbBatchDynamicTableScanInfo(STableScanInfo* pInfo) {

5,211,526✔

2130

  if (pInfo->pBatchColMap) {

5,211,526✔

2131

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

13,490,818✔

2132

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

8,276,732✔

2133

      taosArrayDestroy(pOrgTbInfo->colMap);

8,278,268✔

2135

    taosArrayDestroy(pInfo->pBatchColMap);

5,210,502✔

2136

    pInfo->pBatchColMap = NULL;

5,211,014✔

2138

  if (pInfo->cachedTagList) {

5,209,990✔

2139

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

16,359,552✔

2140

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

13,943,616✔

2141

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

13,944,128✔

2142

        taosMemoryFreeClear(pTagVal->pData);

6,121,472✔

2145

    taosArrayDestroy(pInfo->cachedTagList);

2,418,496✔

2146

    pInfo->cachedTagList = NULL;

2,417,472✔

2148

  if (pInfo->lastColArray) {

5,210,502✔

2149

    taosArrayDestroy(pInfo->lastColArray);

5,209,478✔

2150

    pInfo->lastColArray = NULL;

5,212,550✔

2152

  if (pInfo->lastBlockColArray) {

5,212,038✔

2153

    taosArrayDestroy(pInfo->lastBlockColArray);

5,211,526✔

2154

    pInfo->lastBlockColArray = NULL;

5,209,478✔

2157

  blockDataDestroy(pInfo->pResBlock);

5,209,990✔

2158

  pInfo->pResBlock = NULL;

5,212,038✔

2159

5,212,038✔

2161

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

18,173,170✔

2162

  int32_t                  code = TSDB_CODE_SUCCESS;

18,173,170✔

2163

  int32_t                  lino = 0;

18,173,170✔

2164

  STableScanInfo*          pInfo = pOperator->info;

18,173,170✔

2165

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

18,182,612✔

2166

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

18,181,580✔

2167

  SSDataBlock*             result = NULL;

18,192,504✔

2169

  QRY_PARAM_CHECK(ppRes);

18,194,842✔

2170

  qTrace("%s call", __FUNCTION__);

18,190,776✔

2172

  while (!result) {

34,274,654✔

2173

    code = createVTableScanInfoFromBatchParam(pOperator);

21,267,830✔

2174

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

21,232,920✔

2175

    pOperator->pOperatorGetParam = NULL;

21,220,044✔

2176

    QUERY_CHECK_CODE(code, lino, _end);

21,207,284✔

2178

    if (pInfo->newReader) {

21,207,284✔

2179

      code = startNextGroupScan(pOperator, &result);

8,231,446✔

2180

      QUERY_CHECK_CODE(code, lino, _end);

8,275,524✔

2182

      code = doGroupedTableScan(pOperator, &result);

12,992,618✔

2183

      QUERY_CHECK_CODE(code, lino, _end);

13,005,288✔

2185

    if (!result) {

21,280,812✔

2186

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

8,271,196✔

2188

      pInfo->currentBatchIdx++;

8,271,196✔

2189

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

8,266,818✔

2191

        break;

5,208,966✔

2196

  if (result) {

18,215,790✔

2197

    SSDataBlock* res = NULL;

13,005,800✔

2198

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

13,005,800✔

2199

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

13,003,728✔

2200

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

13,003,224✔

2202

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

13,004,240✔

2203

    QUERY_CHECK_CODE(code, lino, _end);

13,005,800✔

2205

    if (pInfo->cachedTagList) {

13,005,800✔

2206

      code = setTagValFromTagList(pOperator, res);

3,774,272✔

2207

      QUERY_CHECK_CODE(code, lino, _end);

3,774,272✔

2210

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

13,005,800✔

2211

    pInfo->pResBlock = res;

13,005,800✔

2212

    blockDataDestroy(result);

13,005,288✔

2213

    (*ppRes) = res;

13,005,800✔

2215

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

13,005,800✔

2217

    clearVstbBatchDynamicTableScanInfo(pInfo);

5,209,990✔

2218

    setOperatorCompleted(pOperator);

5,211,526✔

2221

_end:

18,217,760✔

2222

  return code;

18,216,224✔

2225

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

44,089,804✔

2226

  int32_t                  code = TSDB_CODE_SUCCESS;

44,089,804✔

2227

  int32_t                  lino = 0;

44,089,804✔

2228

  STableScanInfo*          pInfo = pOperator->info;

44,089,804✔

2229

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

44,093,651✔

2230

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

44,090,232✔

2232

  QRY_PARAM_CHECK(ppRes);

44,087,220✔

2233

  qTrace("%s call", __FUNCTION__);

44,076,923✔

2235

  pOperator->dynamicTask = true;

44,077,298✔

2237

  switch (type) {

44,080,251✔

2238

    case DYN_TYPE_VSTB_WIN_SCAN:

964,108✔

2240

      code = doStbJoinDynamicTableScanNext(pOperator, ppRes);

964,108✔

2241

      QUERY_CHECK_CODE(code, lino, _end);

963,350✔

2242

      break;

963,350✔

2244

    case DYN_TYPE_VSTB_SINGLE_SCAN: {

24,917,589✔

2245

      code = doVstbSingleDynamicTableScanNext(pOperator, ppRes);

24,917,589✔

2246

      QUERY_CHECK_CODE(code, lino, _end);

24,917,589✔

2247

      break;

24,916,961✔

2249

    case DYN_TYPE_VSTB_BATCH_SCAN:

18,198,554✔

2250

      code = doVstbBatchDynamicTableScanNext(pOperator, ppRes);

18,198,554✔

2251

      QUERY_CHECK_CODE(code, lino, _end);

18,215,712✔

2252

      break;

18,215,712✔

2259

_end:

44,096,651✔

2260

  return code;

44,096,651✔

2263

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

766,185,822✔

2264

  int32_t         code = TSDB_CODE_SUCCESS;

766,185,822✔

2265

  int32_t         lino = 0;

766,185,822✔

2266

  STableScanInfo* pInfo = pOperator->info;

766,185,822✔

2267

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

766,666,913✔

2268

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

766,661,731✔

2269

  QRY_PARAM_CHECK(ppRes);

766,517,403✔

2270

  qTrace("%s call", __FUNCTION__);

766,516,258✔

2272

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

766,517,782✔

2273

  QUERY_CHECK_CODE(code, lino, _end);

767,148,035✔

2275

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

767,148,035✔

2276

  if (pGetParam || pInfo->pBatchColMap) {

767,130,812✔

2277

    ETableScanDynType type = pGetParam ?

44,088,549✔

2278

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

44,091,201✔

2281

    code = doDynamicTableScanNext(pOperator, type, ppRes);

44,088,549✔

2282

    QUERY_CHECK_CODE(code, lino, _end);

44,098,128✔

2284

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

44,097,500✔

2285

      return code;

43,159,055✔

2290

  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {

723,907,105✔

2291

    int32_t       numOfTables = 0;

5,931,722✔

2292

    STableKeyInfo tInfo = {0};

5,931,722✔

2293

    pInfo->countState = TABLE_COUNT_STATE_END;

5,931,722✔

2295

    while (1) {

5,524,160✔

2296

      SSDataBlock* result = NULL;

11,455,882✔

2297

      code = doGroupedTableScan(pOperator, &result);

11,454,595✔

2298

      QUERY_CHECK_CODE(code, lino, _end);

11,451,726✔

2300

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

11,451,726✔

2301

        (*ppRes) = result;

5,879,891✔

2302

        return code;

5,888,721✔

2306

      pInfo->currentTable++;

5,568,695✔

2308

      taosRLockLatch(&pTaskInfo->lock);

5,567,459✔

2309

      numOfTables = 0;

5,573,723✔

2310

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

5,573,723✔

2311

      if (code != TSDB_CODE_SUCCESS) {

5,569,340✔

2316

      if (pInfo->currentTable >= numOfTables) {

5,569,340✔

2317

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

43,001✔

2318

        taosRUnLockLatch(&pTaskInfo->lock);

43,001✔

2319

        (*ppRes) = NULL;

43,001✔

2320

        return code;

43,001✔

2323

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

5,514,093✔

2324

      if (!tmp) {

5,522,547✔

2330

      tInfo = *tmp;

5,522,547✔

2331

      taosRUnLockLatch(&pTaskInfo->lock);

5,516,242✔

2333

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

5,529,779✔

2334

      QUERY_CHECK_CODE(code, lino, _end);

5,527,893✔

2335

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

5,527,893✔

2338

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

5,536,409✔

2339

      QUERY_CHECK_CODE(code, lino, _end);

5,519,713✔

2340

      pInfo->scanTimes = 0;

5,519,713✔

2343

    code = groupSeqTableScan(pOperator, ppRes);

717,864,558✔

2344

    QUERY_CHECK_CODE(code, lino, _end);

716,778,649✔

2347

_end:

716,776,862✔

2348

  if (code != TSDB_CODE_SUCCESS) {

716,667,503✔

2349

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

2,415✔

2350

    pTaskInfo->code = code;

2,415✔

2351

    T_LONG_JMP(pTaskInfo->env, code);

2,415✔

2354

  return code;

716,665,088✔

2357

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

1,850,693✔

2358

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

1,850,693✔

2359

  if (!pRecorder) {

1,850,693✔

2362

  STableScanInfo* pTableScanInfo = pOptr->info;

1,850,693✔

2363

  *pRecorder = pTableScanInfo->base.readRecorder;

1,850,693✔

2364

  *pOptrExplain = pRecorder;

1,850,693✔

2365

  *len = sizeof(SFileBlockLoadRecorder);

1,850,693✔

2366

  return 0;

1,850,693✔

2369

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

186,765,800✔

2370

  cleanupQueryTableDataCond(&pBase->cond);

186,765,800✔

2371

  cleanupQueryTableDataCond(&pBase->orgCond);

186,672,327✔

2373

  if (pAPI->tsdReaderClose) {

186,744,148✔

2374

    pAPI->tsdReaderClose(pBase->dataReader);

186,794,639✔

2376

  pBase->dataReader = NULL;

186,587,851✔

2378

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

186,631,987✔

2379

    taosArrayDestroy(pBase->matchInfo.pList);

186,659,173✔

2382

  tableListDestroy(pBase->pTableListInfo);

186,507,837✔

2383

  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);

186,789,112✔

2384

  cleanupExprSupp(&pBase->pseudoSup);

186,651,601✔

2385

186,770,109✔

2387

static void cleanReaderForVTable(STableScanInfo* pInfo){

6,274,876✔

2388

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

6,274,876✔

2389

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

6,275,365✔

2390

    while (pIter != NULL) {

6,297,384✔

2391

      void **reader = pIter;

21,484✔

2392

      if (*reader) {

21,484✔

2393

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

21,484✔

2394

          pInfo->base.dataReader = NULL;

19,164✔

2396

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

21,484✔

2398

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

21,484✔

2401

6,275,855✔

2403

static void destroyTableScanOperatorInfo(void* param) {

162,320,650✔

2404

  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;

162,320,650✔

2405

  blockDataDestroy(pTableScanInfo->pResBlock);

162,320,650✔

2406

  blockDataDestroy(pTableScanInfo->pOrgBlock);

162,286,331✔

2407

  taosHashCleanup(pTableScanInfo->pIgnoreTables);

162,301,215✔

2408

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

162,196,393✔

2409

    cleanReaderForVTable(pTableScanInfo);

6,275,320✔

2410

    taosHashCleanup(pTableScanInfo->readerCache);

6,275,900✔

2412

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

162,207,521✔

2413

  taosArrayDestroy(pTableScanInfo->pBlockColMap);

162,176,032✔

2414

  taosMemoryFreeClear(param);

162,223,655✔

2415

162,209,890✔

2417

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

161,859,671✔

2418

  if (pBlock && dataRequireFlag == FUNC_DATA_REQUIRED_NOT_LOAD) {

161,859,671✔

2419

    int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

7,439,347✔

2420

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

18,677,964✔

2421

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

11,296,835✔

2422

      if (pCol) {

11,323,862✔

2423

        pCol->info.noData = true;

11,325,732✔

2427

161,801,453✔

2429

static int32_t resetTableScanOperatorState(SOperatorInfo* pOper) {

300,624✔

2430

  int32_t         code = TSDB_CODE_SUCCESS;

300,624✔

2431

  STableScanInfo*   pInfo = pOper->info;

300,624✔

2432

  pOper->status = OP_NOT_OPENED;

300,624✔

2434

  pInfo->scanTimes = 0;

300,624✔

2435

  pInfo->currentGroupId = -1;

300,624✔

2436

  pInfo->tableEndIndex = -1;

300,624✔

2437

  pInfo->tableStartIndex = 0;

300,624✔

2438

  pInfo->currentTable = 0;

300,624✔

2439

  pInfo->scanMode = 0;

300,624✔

2440

  pInfo->countState = 0;

300,624✔

2441

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

300,624✔

2443

  if (!pInfo->virtualStableScan) {

300,624✔

2449

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

169,103✔

2450

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

169,103✔

2452

    pInfo->base.dataReader = NULL;

169,103✔

2454

  tableListDestroy(pInfo->base.pTableListInfo);

300,624✔

2455

  pInfo->base.pTableListInfo = tableListCreate();

300,245✔

2456

  if (!pInfo->base.pTableListInfo) {

300,624✔

2460

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

300,624✔

2462

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

300,624✔

2463

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

300,624✔

2464

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

335,435✔

2466

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

335,435✔

2467

    if (code) {

168,087✔

2473

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

300,624✔

2474

  cleanupQueryTableDataCond(&pInfo->base.cond);

300,624✔

2475

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

300,624✔

2476

  if (code) {

300,624✔

2480

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

300,624✔

2481

    cleanupQueryTableDataCond(&pInfo->base.orgCond);

131,521✔

2482

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

131,521✔

2483

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

131,521✔

2486

  pOper->resultInfo.totalRows = 0;

300,624✔

2487

  blockDataEmpty(pInfo->pResBlock);

300,624✔

2488

  blockDataEmpty(pInfo->pOrgBlock);

300,624✔

2489

  taosHashClear(pInfo->pIgnoreTables);

300,624✔

2490

  taosArrayDestroy(pInfo->pBlockColMap);

300,624✔

2491

  pInfo->pBlockColMap = NULL;

300,624✔

2492

  return code;

300,624✔

2495

int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

162,291,368✔

2498

  QRY_PARAM_CHECK(pOptrInfo);

162,291,368✔

2500

  int32_t         code = TSDB_CODE_SUCCESS;

162,311,941✔

2501

  int32_t         lino = 0;

162,311,941✔

2502

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

162,311,941✔

2503

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

162,266,160✔

2504

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

162,170,625✔

2505

    code = terrno;

2,618✔

2509

  pOperator->pPhyNode = pTableScanNode;

162,168,007✔

2510

  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;

162,175,366✔

2511

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

162,186,215✔

2513

  int32_t numOfCols = 0;

162,188,316✔

2515

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

162,224,522✔

2516

  QUERY_CHECK_CODE(code, lino, _error);

161,949,794✔

2518

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

161,949,794✔

2519

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

161,814,898✔

2520

  QUERY_CHECK_CODE(code, lino, _error);

161,243,905✔

2522

  if (pScanNode->pScanPseudoCols != NULL) {

161,243,905✔

2523

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

52,530,691✔

2524

    pSup->pExprInfo = NULL;

52,639,924✔

2525

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

52,613,403✔

2526

    QUERY_CHECK_CODE(code, lino, _error);

52,486,519✔

2528

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

52,486,519✔

2530

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

52,478,782✔

2533

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

161,289,484✔

2534

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

161,380,538✔

2536

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

161,059,718✔

2537

  pInfo->base.readHandle = *readHandle;

160,886,034✔

2538

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

161,057,394✔

2540

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

161,431,625✔

2541

  pInfo->sample.seed = taosGetTimestampSec();

160,901,628✔

2543

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

161,597,155✔

2544

  initResultSizeInfo(&pOperator->resultInfo, 4096);

161,573,620✔

2545

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

160,933,282✔

2546

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

161,664,272✔

2547

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

160,948,398✔

2549

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

160,971,269✔

2550

  QUERY_CHECK_CODE(code, lino, _error);

161,042,356✔

2552

  pInfo->virtualStableScan = pScanNode->virtualStableScan;

161,042,356✔

2553

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

161,078,304✔

2554

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

6,262,787✔

2555

    pInfo->pResBlock = NULL;

6,262,855✔

2556

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

6,254,607✔

2557

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

6,257,114✔

2560

  if (pInfo->virtualStableScan) {

160,952,974✔

2561

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

6,258,951✔

2562

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

6,278,603✔

2565

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

161,144,551✔

2566

                            pTaskInfo->pStreamRuntimeInfo);

160,944,150✔

2567

  QUERY_CHECK_CODE(code, lino, _error);

161,255,658✔

2569

  pInfo->currentGroupId = -1;

161,232,606✔

2571

  pInfo->tableEndIndex = -1;

161,363,896✔

2572

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

161,346,399✔

2573

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

161,276,773✔

2575

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

161,146,676✔

2577

  pOperator->exprSupp.numOfExprs = numOfCols;

161,090,742✔

2579

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

161,223,354✔

2580

  pInfo->ignoreTag = false;

161,335,338✔

2582

  pInfo->lastBatchIdx = -1;

161,176,351✔

2583

  pInfo->currentBatchIdx = 0;

161,047,294✔

2584

  pInfo->base.pTableListInfo = pTableListInfo;

161,061,769✔

2585

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

161,027,039✔

2595

  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

161,027,039✔

2597

  pOperator->fpSet = createOperatorFpSet(prepareTableScan, doTableScanNext,

160,436,162✔

2603

  setOperatorResetStateFn(pOperator, resetTableScanOperatorState);

161,095,058✔

2605

  pOperator->cost.openCost = 0;

161,284,869✔

2606

  *pOptrInfo = pOperator;

161,316,936✔

2607

  return TSDB_CODE_SUCCESS;

160,961,040✔

2609

_error:

23,052✔

2610

  if (pInfo != NULL) {

23,052✔

2611

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

23,052✔

2612

    destroyTableScanOperatorInfo(pInfo);

23,052✔

2615

  if (pOperator != NULL) {

23,052✔

2616

    pOperator->info = NULL;

23,052✔

2617

    destroyOperator(pOperator);

23,052✔

2619

  pTaskInfo->code = code;

23,052✔

2620

  return code;

23,052✔

2657

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

13,910✔

2658

  int32_t code = TSDB_CODE_SUCCESS;

13,910✔

2659

  int32_t lino = 0;

13,910✔

2660

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

13,910✔

2661

    return code;

1,634✔

2663

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

12,276✔

2664

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

12,276✔

2665

  bool hasUnqualified = false;

12,276✔

2667

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

12,276✔

2668

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

12,276✔

2670

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

12,276✔

2672

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

12,276✔

2675

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

12,276✔

2676

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

12,276✔

2677

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

44,267,968✔

2678

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

44,255,692✔

2679

    void*    data = colDataGetData(pColPk, i);

44,255,692✔

2680

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

44,258,076✔

2681

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

2,384✔

2684

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

2,384✔

2685

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

2,384✔

2686

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

2,384✔

2687

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

2,384✔

2688

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

2,384✔

2689

      taosMemoryFree(tmq);

2,384✔

2691

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

44,253,308✔

2694

    if (!p[i]) {

44,255,692✔

2695

      hasUnqualified = true;

15,762,004✔

2699

  if (hasUnqualified) {

12,276✔

2700

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

4,146✔

2701

    QUERY_CHECK_CODE(code, lino, _end);

4,146✔

2704

_end:

12,276✔

2705

  taosMemoryFree(p);

12,276✔

2706

  if (code != TSDB_CODE_SUCCESS) {

12,276✔

2709

  return code;

12,276✔

2723

static int32_t processBlock(SStreamScanInfo* pInfo) {

23,200,376✔

2724

  int32_t         code = TSDB_CODE_SUCCESS;

23,200,376✔

2725

  int32_t         lino = 0;

23,200,376✔

2726

  SOperatorInfo*  pOperator = pInfo->pStreamScanOp;

23,200,376✔

2728

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

23,200,376✔

2729

  QUERY_CHECK_CODE(code, lino, _end);

23,200,376✔

2731

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

23,200,376✔

2732

  QUERY_CHECK_CODE(code, lino, _end);

23,199,686✔

2734

_end:

23,199,686✔

2735

  if (code != TSDB_CODE_SUCCESS) {

23,199,686✔

2738

  return code;

23,200,376✔

2741

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

6,620,005✔

2742

  int32_t code = TSDB_CODE_SUCCESS;

6,620,005✔

2743

  int32_t lino = 0;

6,620,005✔

2744

  SValue  val = {0};

6,620,005✔

2745

  if (hasPrimaryKey) {

6,621,781✔

2746

    code = doBlockDataPrimaryKeyFilter(pBlock, offset);

13,910✔

2747

    if (code != TSDB_CODE_SUCCESS) {

13,910✔

2751

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

13,910✔

2753

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

13,910✔

2754

      return code;

3,848✔

2756

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

10,062✔

2757

    val.type = pColPk->info.type;

10,062✔

2758

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

10,062✔

2759

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

632✔

2762

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

632✔

2763

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

632✔

2764

      val.nData = varDataLen(tmp);

632✔

2765

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

632✔

2767

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

9,430✔

2770

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

6,617,933✔

2772

_end:

6,617,045✔

2773

  if (code != TSDB_CODE_SUCCESS) {

6,616,117✔

2776

  return code;

6,615,212✔

2779

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

23,381,048✔

2780

  int32_t        code = TSDB_CODE_SUCCESS;

23,381,048✔

2781

  int32_t        lino = 0;

23,381,048✔

2782

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

23,381,048✔

2783

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

23,381,358✔

2785

  SStreamScanInfo* pInfo = pOperator->info;

23,380,668✔

2786

  const char*      id = GET_TASKID(pTaskInfo);

23,379,970✔

2788

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

23,378,989✔

2790

  if (isTaskKilled(pTaskInfo)) {

23,382,248✔

2795

  blockDataCleanup(pInfo->pRes);

23,382,066✔

2796

  int64_t st = taosGetTimestampMs();

23,381,021✔

2798

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

23,381,021✔

2799

    while (1) {

5,616,673✔

2800

      SSDataBlock* pResult = NULL;

5,928,296✔

2801

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

5,928,884✔

2802

      QUERY_CHECK_CODE(code, lino, _end);

5,931,407✔

2804

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

5,931,407✔

2805

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

5,888,091✔

2806

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

5,888,406✔

2807

        QUERY_CHECK_CODE(code, lino, _end);

5,887,461✔

2808

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

5,887,461✔

2809

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

5,888,406✔

2810

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

5,884,873✔

2811

          QUERY_CHECK_CODE(code, lino, _end);

5,884,240✔

2817

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

5,888,088✔

2818

        (*ppRes) = pInfo->pRes;

265,131✔

2819

        return code;

265,131✔

2821

      int64_t elapsed = taosGetTimestampMs() - st;

5,622,960✔

2822

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

5,622,960✔

2823

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

5,660✔

2826

        (*ppRes) = pInfo->pRes;

4,079✔

2827

        return code;

4,079✔

2831

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

43,001✔

2832

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

43,001✔

2834

    pTSInfo->base.dataReader = NULL;

43,001✔

2835

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

43,001✔

2836

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

43,001✔

2837

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

43,001✔

2838

    if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) {

43,001✔

2839

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

12,882✔

2840

        (*ppRes) = NULL;

5,215✔

2842

        (*ppRes) = pInfo->pRes;

7,667✔

2844

      return 0;

12,882✔

2846

    QUERY_CHECK_CODE(code, lino, _end);

30,119✔

2848

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

30,119✔

2851

  while (1) {

100,402✔

2852

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

69,598,473✔

2856

                                                              pTaskInfo->streamInfo.sourceExcluded,

23,199,686✔

2859

                                                              pTaskInfo->streamInfo.enableReplay);

23,199,777✔

2860

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

23,199,367✔

2863

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

23,199,367✔

2866

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

23,200,018✔

2869

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

23,198,114✔

2871

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

23,199,796✔

2873

    int32_t ret = processBlock(pInfo);

23,201,096✔

2874

    if (ret != 0) {

23,200,376✔

2878

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

23,200,376✔

2881

    if (code != 0) {

23,200,376✔

2882

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

22,599,954✔

2883

        (*ppRes) = NULL;

11,162,378✔

2885

        (*ppRes) = pInfo->pRes;

11,437,576✔

2887

      return 0;

22,599,954✔

2890

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

600,422✔

2891

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

105,334✔

2892

      (*ppRes) = pInfo->pRes;

500,020✔

2893

      return 0;

500,020✔

2907

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

327,388✔

2908

  int32_t code = TSDB_CODE_SUCCESS;

327,388✔

2909

  int32_t lino = 0;

327,388✔

2910

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

327,388✔

2911

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

329,228✔

2914

  int32_t size = 0;

329,228✔

2915

  code = tableListGetSize(pTableListInfo, &size);

328,880✔

2916

  QUERY_CHECK_CODE(code, lino, _end);

329,228✔

2917

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

9,639,859✔

2918

    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);

9,308,131✔

2919

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

9,280,452✔

2920

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

9,280,452✔

2921

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

9,310,631✔

2924

  (*ppArrayRes) = tableIdList;

331,728✔

2926

_end:

329,228✔

2927

  if (code != TSDB_CODE_SUCCESS) {

329,228✔

2930

  return code;

329,228✔

2933

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

1,484,012✔

2934

  int32_t        code = TSDB_CODE_SUCCESS;

1,484,012✔

2935

  int32_t        lino = 0;

1,484,012✔

2936

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,484,012✔

2937

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,484,012✔

2939

  SStreamRawScanInfo* pInfo = pOperator->info;

1,484,012✔

2940

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

1,484,012✔

2941

  pTaskInfo->streamInfo.btMetaRsp.batchMetaLen = NULL;

1,484,012✔

2943

  qDebug("tmqsnap doRawScan called");

1,484,012✔

2944

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

1,484,012✔

2945

    bool hasNext = false;

1,478,530✔

2946

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

1,478,530✔

2947

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

1,477,678✔

2948

      if (code != TSDB_CODE_SUCCESS) {

1,472,342✔

2954

    if (pInfo->dataReader && hasNext) {

1,473,194✔

2955

      if (isTaskKilled(pTaskInfo)) {

735,134✔

2960

      SSDataBlock* pBlock = NULL;

735,134✔

2961

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

735,134✔

2962

      QUERY_CHECK_CODE(code, lino, _end);

731,580✔

2964

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

731,580✔

2965

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

733,652✔

2966

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

733,950✔

2967

        QUERY_CHECK_CODE(code, lino, _end);

732,172✔

2968

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

732,172✔

2969

        (*ppRes) = pBlock;

735,432✔

2970

        return code;

735,726✔

2974

    SMetaTableInfo mtInfo = {0};

738,060✔

2975

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

742,506✔

2976

    QUERY_CHECK_CODE(code, lino, _end);

741,320✔

2977

    if (code != 0) {

741,320✔

2981

    STqOffsetVal offset = {0};

741,320✔

2982

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

740,434✔

2983

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

9,960✔

2984

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

9,960✔

2986

      SValue val = {0};

732,548✔

2987

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

732,548✔

2988

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

731,370✔

2991

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

742,804✔

2992

    QUERY_CHECK_CODE(code, lino, _end);

742,804✔

2993

    (*ppRes) = NULL;

742,804✔

2994

    return code;

742,804✔

2995

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

5,482✔

2996

    SSnapContext* sContext = pInfo->sContext;

5,482✔

2997

    int64_t st = taosGetTimestampMs();

5,482✔

2998

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

40,734✔

2999

      void*   data = NULL;

40,066✔

3000

      int32_t dataLen = 0;

40,066✔

3001

      int16_t type = 0;

40,066✔

3002

      int64_t uid = 0;

40,066✔

3003

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

40,066✔

3009

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

40,066✔

3010

        STqOffsetVal offset = {0};

4,814✔

3011

        SValue       val = {0};

4,814✔

3013

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

4,814✔

3014

        QUERY_CHECK_CODE(code, lino, _end);

4,814✔

3015

        break;

4,814✔

3017

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

35,252✔

3018

        SMqMetaRsp tmpMetaRsp = {0};

35,252✔

3019

        tmpMetaRsp.resMsgType = type;

35,252✔

3020

        tmpMetaRsp.metaRspLen = dataLen;

35,252✔

3021

        tmpMetaRsp.metaRsp = data;

35,252✔

3022

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

35,252✔

3023

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

4,284✔

3024

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

4,284✔

3026

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

4,284✔

3027

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

4,284✔

3029

        int32_t  tempRes = TSDB_CODE_SUCCESS;

35,252✔

3030

        uint32_t len = 0;

35,252✔

3031

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

35,252✔

3032

        if (TSDB_CODE_SUCCESS != tempRes) {

35,252✔

3038

        int32_t tLen = sizeof(SMqRspHead) + len;

35,252✔

3039

        void*   tBuf = taosMemoryCalloc(1, tLen);

35,252✔

3040

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

35,252✔

3042

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

35,252✔

3043

        SEncoder encoder = {0};

35,252✔

3044

        tEncoderInit(&encoder, metaBuff, len);

35,252✔

3045

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

35,252✔

3046

        if (tempLen < 0) {

35,252✔

3053

        taosMemoryFreeClear(data);

35,252✔

3054

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

35,252✔

3055

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

35,252✔

3057

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

35,252✔

3058

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

35,252✔

3060

        int64_t elapsed = taosGetTimestampMs() - st;

35,252✔

3061

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

35,252✔

3067

    (*ppRes) = NULL;

5,482✔

3068

    return code;

5,482✔

3082

static void destroyTmqRawScanOperatorInfo(void* param) {

122,358✔

3083

  SStreamRawScanInfo* pRawScan = (SStreamRawScanInfo*)param;

122,358✔

3084

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

122,358✔

3085

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

122,062✔

3086

  tableListDestroy(pRawScan->pTableListInfo);

122,032✔

3087

  taosMemoryFree(pRawScan);

122,358✔

3088

122,012✔

3093

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

121,071✔

3099

  QRY_PARAM_CHECK(pOptrInfo);

121,071✔

3100

  int32_t code = TSDB_CODE_SUCCESS;

122,033✔

3101

  int32_t lino = 0;

122,033✔

3103

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

122,033✔

3104

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

121,735✔

3105

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

122,060✔

3111

  pInfo->pTableListInfo = tableListCreate();

122,060✔

3112

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

122,358✔

3113

  pInfo->vnode = pHandle->vnode;

122,358✔

3114

  pInfo->pAPI = &pTaskInfo->storageAPI;

122,358✔

3116

  pInfo->sContext = pHandle->sContext;

122,358✔

3117

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

122,358✔

3118

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

122,358✔

3121

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

122,358✔

3123

  *pOptrInfo = pOperator;

122,358✔

3124

  return code;

122,358✔

3136

void destroyTmqScanOperatorInfo(void* param) {

329,228✔

3137

  if (param == NULL) {

329,228✔

3141

  SStreamScanInfo* pStreamScan = (SStreamScanInfo*)param;

329,228✔

3142

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

329,228✔

3143

    destroyOperator(pStreamScan->pTableScanOp);

329,228✔

3146

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

328,932✔

3147

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

329,228✔

3149

  if (pStreamScan->pVtableMergeHandles) {

329,228✔

3153

  if (pStreamScan->pVtableMergeBuf) {

329,228✔

3157

  if (pStreamScan->pVtableReadyHandles) {

329,228✔

3161

  if (pStreamScan->pTableListInfo) {

329,228✔

3165

  if (pStreamScan->pPseudoExpr) {

329,228✔

3166

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

191,003✔

3167

    taosMemoryFree(pStreamScan->pPseudoExpr);

191,003✔

3170

  taosHashCleanup(pStreamScan->pCol2SlotId);

329,228✔

3171

  pStreamScan->pCol2SlotId = NULL;

329,228✔

3173

  cleanupExprSupp(&pStreamScan->tbnameCalSup);

329,228✔

3174

  cleanupExprSupp(&pStreamScan->tagCalSup);

328,963✔

3176

  blockDataDestroy(pStreamScan->pRes);

328,963✔

3177

  blockDataDestroy(pStreamScan->pDeleteDataRes);

328,582✔

3178

  blockDataDestroy(pStreamScan->pUpdateDataRes);

328,901✔

3179

  blockDataDestroy(pStreamScan->pCreateTbRes);

328,941✔

3180

  taosArrayDestroy(pStreamScan->pBlockLists);

328,901✔

3181

  blockDataDestroy(pStreamScan->pCheckpointRes);

329,228✔

3183

  taosMemoryFree(pStreamScan);

329,228✔

3203

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

329,228✔

3204

  size_t  numOfCols = LIST_LENGTH(pNodeList);

329,228✔

3205

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

2,067,316✔

3206

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

1,738,088✔

3207

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

1,738,088✔

3208

      *pTsSlotId = pNode->slotId;

329,228✔

3211

329,228✔

3213

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

329,228✔

3214

  size_t  numOfCols = LIST_LENGTH(pNodeList);

329,228✔

3215

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

2,067,062✔

3216

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

1,738,088✔

3217

    if (pNode == NULL) {

1,737,773✔

3220

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

1,737,773✔

3221

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

1,738,088✔

3222

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

1,738,088✔

3223

      if (code != TSDB_CODE_SUCCESS) {

1,737,834✔

3228

  return TSDB_CODE_SUCCESS;

328,974✔

3231

int32_t createTmqScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,

329,228✔

3234

  QRY_PARAM_CHECK(pOptrInfo);

329,228✔

3236

  int32_t          code = TSDB_CODE_SUCCESS;

329,228✔

3237

  int32_t          lino = 0;

329,228✔

3238

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

329,228✔

3239

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

329,228✔

3240

  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;

329,228✔

3241

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

329,228✔

3243

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

329,228✔

3248

  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;

329,228✔

3249

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;

329,228✔

3251

  pInfo->pTagCond = pTagCond;

329,228✔

3252

  pInfo->pGroupTags = pTableScanNode->pGroupTags;

329,228✔

3254

  pInfo->primaryTsIndex = -1;

329,228✔

3255

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

329,228✔

3256

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

329,228✔

3257

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

329,228✔

3258

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

328,913✔

3259

  QUERY_CHECK_CODE(code, lino, _error);

329,228✔

3261

  SDataBlockDescNode* pNode = (SDataBlockDescNode*)pDescNode;

329,228✔

3262

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

329,228✔

3268

  pInfo->pPartTbnameSup = NULL;

329,228✔

3269

  if (pTableScanNode->pSubtable != NULL) {

328,913✔

3285

  if (pTableScanNode->pTags != NULL) {

328,913✔

3297

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

328,605✔

3298

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

328,920✔

3300

  SOperatorInfo* pTableScanOp = NULL;

328,920✔

3301

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

328,920✔

3302

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

328,231✔

3303

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

1,295✔

3307

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

326,936✔

3308

  if (pHandle->version > 0) {

327,268✔

3309

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

327,665✔

3312

  STableKeyInfo* pList = NULL;

327,660✔

3313

  int32_t        num = 0;

326,956✔

3314

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

327,345✔

3315

  QUERY_CHECK_CODE(code, lino, _error);

326,282✔

3317

  if (pHandle->initTableReader) {

326,282✔

3318

    pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;

326,893✔

3319

    pTSInfo->base.dataReader = NULL;

327,520✔

3322

  if (pHandle->initTqReader) {

326,011✔

3323

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

326,928✔

3324

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

329,228✔

3326

    pInfo->tqReader = pHandle->tqReader;

48✔

3330

  pInfo->pUpdateInfo = NULL;

328,900✔

3331

  pInfo->pTableScanOp = pTableScanOp;

328,900✔

3333

  pInfo->readHandle = *pHandle;

329,228✔

3334

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

329,228✔

3335

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

328,900✔

3336

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

328,900✔

3337

  pInfo->hasPart = false;

328,331✔

3339

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

328,974✔

3340

  QUERY_CHECK_CODE(code, lino, _error);

327,657✔

3342

  SArray* tableIdList = NULL;

327,657✔

3343

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

327,657✔

3344

  QUERY_CHECK_CODE(code, lino, _error);

328,626✔

3345

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

328,626✔

3346

  QUERY_CHECK_CODE(code, lino, _error);

329,228✔

3347

  taosArrayDestroy(tableIdList);

329,228✔

3348

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

329,228✔

3351

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

328,933✔

3352

  QUERY_CHECK_CODE(code, lino, _error);

328,579✔

3354

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

328,579✔

3355

  QUERY_CHECK_CODE(code, lino, _error);

328,550✔

3357

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

328,550✔

3358

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

328,573✔

3360

  pInfo->groupId = 0;

328,897✔

3361

  pInfo->igCheckGroupId = false;

328,543✔

3362

  pInfo->pStreamScanOp = pOperator;

328,897✔

3363

  pInfo->deleteDataIndex = 0;

328,543✔

3364

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

328,543✔

3365

  QUERY_CHECK_CODE(code, lino, _error);

329,228✔

3367

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

329,228✔

3368

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

329,228✔

3369

  QUERY_CHECK_CODE(code, lino, _error);

329,228✔

3371

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

329,228✔

3372

  pInfo->partitionSup.needCalc = false;

329,228✔

3373

  pInfo->igCheckUpdate = pTableScanNode->igCheckUpdate;

329,228✔

3374

  pInfo->igExpired = pTableScanNode->igExpired;

329,228✔

3375

  pInfo->twAggSup.maxTs = INT64_MIN;

329,228✔

3376

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

329,228✔

3377

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

329,228✔

3378

  pInfo->pFillSup = NULL;

329,228✔

3379

  pInfo->useGetResultRange = false;

329,228✔

3380

  pInfo->pRangeScanRes = NULL;

329,228✔

3382

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

329,228✔

3383

  QUERY_CHECK_CODE(code, lino, _error);

329,228✔

3385

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

329,228✔

3387

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

328,904✔

3389

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

328,904✔

3392

  *pOptrInfo = pOperator;

328,904✔

3393

  return code;

328,904✔

3412

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

10,760,099✔

3413

  int32_t        code = TSDB_CODE_SUCCESS;

10,760,099✔

3414

  int32_t        lino = 0;

10,760,099✔

3415

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

10,760,099✔

3416

  STagScanInfo*  pInfo = pOperator->info;

10,763,306✔

3417

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

10,757,051✔

3418

  int32_t        count = pRes->info.rows;

10,757,587✔

3420

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

10,760,706✔

3421

  if (!item) {

10,753,817✔

3428

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

10,753,817✔

3429

  tDecoderClear(&(*mr).coder);

10,764,288✔

3430

  if (code != TSDB_CODE_SUCCESS) {

10,762,759✔

3436

  char str[512];

10,762,663✔

3437

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

49,454,207✔

3438

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

38,695,639✔

3441

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

38,696,616✔

3442

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

4,537,521✔

3443

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

4,535,907✔

3444

      QUERY_CHECK_CODE(code, lino, _end);

4,536,930✔

3445

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

34,161,864✔

3446

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

6,571,689✔

3447

      QUERY_CHECK_CODE(code, lino, _end);

6,569,555✔

3448

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

27,588,631✔

3449

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

15,321✔

3450

      QUERY_CHECK_CODE(code, lino, _end);

15,321✔

3452

      STagVal val = {0};

27,572,685✔

3453

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

27,573,186✔

3454

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

27,576,290✔

3456

      char* data = NULL;

27,572,798✔

3457

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

27,572,798✔

3458

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

27,571,933✔

3460

        data = (char*)p;

1,374✔

3463

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

27,572,782✔

3464

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

27,571,750✔

3465

      QUERY_CHECK_CODE(code, lino, _end);

27,575,773✔

3467

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

27,575,773✔

3469

        taosMemoryFree(data);

9,924,838✔

3474

_end:

10,767,784✔

3475

  if (code != TSDB_CODE_SUCCESS) {

10,761,644✔

3479

    pRes->info.rows++;

10,761,644✔

3482

  return code;

10,760,582✔

3485

static void tagScanFreeUidTag(void* p) {

7,208,171✔

3486

  STUidTagInfo* pInfo = p;

7,208,171✔

3487

  if (pInfo->pTagVal != NULL) {

7,208,171✔

3488

    taosMemoryFree(pInfo->pTagVal);

7,214,769✔

3490

7,209,070✔

3492

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

1,636,206✔

3493

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

1,636,206✔

3494

  if (pColumnData == NULL) {

1,637,510✔

3498

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

1,637,510✔

3499

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

1,635,702✔

3500

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

1,635,798✔

3501

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

1,630,038✔

3503

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

1,631,236✔

3504

  if (code != TSDB_CODE_SUCCESS) {

1,628,946✔

3510

  pParam->columnData = pColumnData;

1,628,946✔

3511

  pParam->colAlloced = true;

1,632,506✔

3512

  return TSDB_CODE_SUCCESS;

1,635,076✔

3515

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

10,000,989✔

3516

  int32_t                code = TSDB_CODE_SUCCESS;

10,000,989✔

3517

  int32_t                lino = 0;

10,000,989✔

3518

  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;

10,000,989✔

3519

  SColumnNode*           pSColumnNode = NULL;

10,000,989✔

3520

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

10,000,009✔

3521

    pSColumnNode = *(SColumnNode**)pNode;

2,807,776✔

3522

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

7,204,738✔

3523

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

508✔

3524

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

508✔

3537

      return DEAL_RES_CONTINUE;

508✔

3540

    return DEAL_RES_CONTINUE;

7,197,804✔

3543

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

2,809,244✔

3544

  if (!data) {

2,807,532✔

3545

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

1,641,182✔

3546

    if (code == TSDB_CODE_DUP_KEY) {

1,643,382✔

3549

    QUERY_CHECK_CODE(code, lino, _end);

1,643,382✔

3550

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

1,643,382✔

3551

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

1,641,534✔

3552

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

1,640,818✔

3553

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

1,637,969✔

3554

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

1,638,472✔

3555

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

1,642,536✔

3557

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

1,166,350✔

3558

    pSColumnNode->slotId = col->slotId;

1,166,350✔

3561

_end:

2,806,178✔

3562

  if (code != TSDB_CODE_SUCCESS) {

2,804,470✔

3563

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

358✔

3566

  return DEAL_RES_CONTINUE;

2,804,235✔

3569

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

1,640,338✔

3571

  int32_t code = TSDB_CODE_SUCCESS;

1,640,338✔

3572

  int32_t lino = 0;

1,640,338✔

3573

  int32_t numOfTables = taosArrayGetSize(aUidTags);

1,640,338✔

3574

  SArray* pBlockList = NULL;

1,640,702✔

3576

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

1,640,702✔

3577

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

1,632,502✔

3579

  pBlockList = taosArrayInit(1, POINTER_BYTES);

1,632,502✔

3580

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

1,635,220✔

3582

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

1,638,854✔

3583

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

1,638,854✔

3585

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

1,638,854✔

3587

  SScalarParam output = {0};

1,636,922✔

3588

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

1,636,424✔

3589

  QUERY_CHECK_CODE(code, lino, _end);

1,633,722✔

3591

  gTaskScalarExtra.pStreamInfo = (void*)pStreamRuntimeInfo;

1,633,722✔

3592

  gTaskScalarExtra.pStreamRange = NULL;

1,633,722✔

3593

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

1,633,374✔

3594

  QUERY_CHECK_CODE(code, lino, _end);

1,626,966✔

3596

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

1,626,966✔

3597

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

4,080,606✔

3598

    if (result[i]) {

2,446,624✔

3599

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

2,338,119✔

3600

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

2,338,119✔

3604

  colDataDestroy(output.columnData);

1,638,138✔

3605

  taosMemoryFreeClear(output.columnData);

1,631,621✔

3607

_end:

1,627,385✔

3608

  blockDataDestroy(pResBlock);

1,629,726✔

3609

  taosArrayDestroy(pBlockList);

1,622,564✔

3610

  if (code != TSDB_CODE_SUCCESS) {

1,623,040✔

3613

  return code;

1,623,040✔

3616

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

29,520,338✔

3619

  int32_t code = TSDB_CODE_SUCCESS;

29,520,338✔

3620

  int32_t lino = 0;

29,520,338✔

3621

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

29,520,338✔

3622

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

14,113,971✔

3628

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

14,088,726✔

3629

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

7,035,275✔

3630

      QUERY_CHECK_CODE(code, lino, _end);

7,055,639✔

3631

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

7,058,698✔

3632

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

7,047,522✔

3633

      QUERY_CHECK_CODE(code, lino, _end);

7,055,413✔

3636

    STagVal tagVal = {0};

15,393,376✔

3637

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

15,394,138✔

3638

    if (pUidTagInfo->pTagVal == NULL) {

15,410,140✔

3641

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

15,398,013✔

3643

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

15,411,996✔

3644

        colDataSetNULL(pColInfo, rowIndex);

5,923✔

3645

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

15,404,849✔

3648

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

19,211,934✔

3649

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

3,835,105✔

3652

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

3,817,769✔

3653

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

3,812,464✔

3655

        varDataSetLen(tmp, tagVal.nData);

3,812,464✔

3656

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

3,812,845✔

3657

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

3,813,730✔

3658

        taosMemoryFree(tmp);

3,817,977✔

3659

        QUERY_CHECK_CODE(code, lino, _end);

3,817,391✔

3661

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

11,548,860✔

3662

        QUERY_CHECK_CODE(code, lino, _end);

11,587,127✔

3667

_end:

29,528,498✔

3668

  if (code != TSDB_CODE_SUCCESS) {

29,528,498✔

3671

  return code;

29,500,773✔

3674

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

4,608,227✔

3676

  int32_t       code = TSDB_CODE_SUCCESS;

4,608,227✔

3677

  int32_t       lino = 0;

4,608,227✔

3678

  STagScanInfo* pInfo = pOperator->info;

4,608,227✔

3679

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

4,614,596✔

3680

  if (!ignoreFilterIdx) {

4,609,024✔

3681

    size_t szTables = taosArrayGetSize(aFilterIdxs);

1,626,000✔

3682

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

3,958,482✔

3683

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

2,321,460✔

3684

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);

2,324,803✔

3685

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

2,322,913✔

3686

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

13,301,818✔

3687

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

10,960,760✔

3688

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

10,988,503✔

3689

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

10,988,503✔

3690

        QUERY_CHECK_CODE(code, lino, _end);

10,978,905✔

3694

    size_t szTables = taosArrayGetSize(aUidTags);

2,983,024✔

3695

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

7,753,719✔

3696

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);

4,768,988✔

3697

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

4,768,338✔

3698

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

23,275,874✔

3699

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

18,512,005✔

3700

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

18,511,019✔

3701

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

18,511,019✔

3702

        QUERY_CHECK_CODE(code, lino, _end);

18,507,536✔

3707

_end:

2,984,731✔

3708

  if (code != TSDB_CODE_SUCCESS) {

4,621,753✔

3711

  return code;

4,617,159✔

3714

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

9,222,552✔

3715

  int32_t        code = TSDB_CODE_SUCCESS;

9,222,552✔

3716

  int32_t        lino = 0;

9,222,552✔

3717

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,222,552✔

3718

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

9,228,805✔

3720

  STagScanInfo* pInfo = pOperator->info;

9,227,395✔

3721

  SSDataBlock*  pRes = pInfo->pRes;

9,224,364✔

3723

  QRY_PARAM_CHECK(ppRes);

9,221,228✔

3725

  if (pOperator->status == OP_EXEC_DONE) {

9,226,604✔

3726

    return TSDB_CODE_SUCCESS;

4,539,337✔

3728

  blockDataCleanup(pRes);

4,679,165✔

3730

  if (pInfo->pCtbCursor == NULL) {

4,695,113✔

3731

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

4,690,663✔

3732

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

4,695,002✔

3738

  SArray* aUidTags = pInfo->aUidTags;

4,694,666✔

3739

  SArray* aFilterIdxs = pInfo->aFilterIdxs;

4,693,782✔

3740

  int32_t count = 0;

4,694,666✔

3741

  bool    ctbCursorFinished = false;

4,694,666✔

3742

  while (1) {

70,822✔

3743

    taosArrayClearEx(aUidTags, tagScanFreeUidTag);

4,765,479✔

3744

    taosArrayClear(aFilterIdxs);

4,769,579✔

3746

    int32_t numTables = 0;

4,768,551✔

3747

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

11,996,691✔

3748

      SMCtbCursor* pCur = pInfo->pCtbCursor;

11,992,507✔

3749

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

11,990,545✔

3750

      if (uid == 0) {

11,981,045✔

3751

        ctbCursorFinished = true;

4,771,651✔

3752

        break;

4,771,651✔

3754

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

7,209,394✔

3755

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

7,207,307✔

3756

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

7,203,630✔

3758

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

7,203,630✔

3759

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

7,227,541✔

3760

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

7,227,541✔

3761

      ++numTables;

7,227,541✔

3764

    if (numTables == 0) {

4,771,785✔

3765

      break;

146,208✔

3767

    bool ignoreFilterIdx = true;

4,625,577✔

3768

    if (pInfo->pTagCond != NULL) {

4,625,577✔

3769

      ignoreFilterIdx = false;

1,640,338✔

3770

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

1,640,338✔

3771

      QUERY_CHECK_CODE(code, lino, _end);

1,625,345✔

3773

      ignoreFilterIdx = true;

2,985,239✔

3776

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

4,610,584✔

3777

    QUERY_CHECK_CODE(code, lino, _end);

4,616,431✔

3779

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

4,616,431✔

3781

    if (count != 0) {

4,617,881✔

3782

      break;

4,547,068✔

3786

  if (count > 0) {

4,693,276✔

3787

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

4,546,710✔

3789

  if (count == 0 || ctbCursorFinished) {

4,695,184✔

3790

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

4,695,548✔

3791

    pInfo->pCtbCursor = NULL;

4,686,345✔

3792

    setOperatorCompleted(pOperator);

4,689,049✔

3795

  pRes->info.rows = count;

4,689,500✔

3796

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

4,689,366✔

3797

  if (bLimitReached) {

4,686,402✔

3801

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

4,686,402✔

3802

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

4,675,337✔

3804

_end:

4,678,602✔

3805

  if (code != TSDB_CODE_SUCCESS) {

4,678,602✔

3811

  return code;

4,678,602✔

3814

static int32_t createTagScanTableListInfoFromParam(SOperatorInfo* pOperator) {

3,721,053✔

3815

  STagScanInfo*          pInfo = pOperator->info;

3,721,053✔

3816

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

3,721,053✔

3817

  int32_t                code = 0;

3,721,053✔

3818

  STableListInfo*        pListInfo = pInfo->pTableListInfo;

3,721,053✔

3819

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

3,721,053✔

3820

  tb_uid_t               pUid = pParam->vcUid;

3,721,053✔

3826

  STableKeyInfo info = {.groupId = 0};

3,721,053✔

3828

  int32_t tableIdx = 0;

3,721,053✔

3829

  taosHashClear(pListInfo->map);

3,721,053✔

3830

  taosArrayClear(pListInfo->pTableList);

3,721,053✔

3832

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

3,721,053✔

3839

  info.uid = pUid;

3,721,053✔

3840

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

3,721,053✔

3841

  if (p == NULL) {

3,721,053✔

3845

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

3,721,053✔

3847

  return code;

3,721,053✔

3850

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

7,437,258✔

3851

  int32_t       code = TSDB_CODE_SUCCESS;

7,437,258✔

3852

  int32_t       lino = 0;

7,437,258✔

3853

  STagScanInfo* pInfo = pOperator->info;

7,437,258✔

3854

  if (pOperator->pOperatorGetParam) {

7,438,309✔

3855

    pOperator->resultInfo.totalRows = 0;

3,721,053✔

3856

    pOperator->dynamicTask = true;

3,721,053✔

3857

    pInfo->curPos = 0;

3,721,053✔

3858

    code = createTagScanTableListInfoFromParam(pOperator);

3,721,053✔

3859

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

3,721,053✔

3860

    pOperator->pOperatorGetParam = NULL;

3,721,053✔

3861

    QUERY_CHECK_CODE(code, lino, _end);

3,721,053✔

3863

    if (pOperator->status == OP_EXEC_DONE) {

3,721,053✔

3864

      pOperator->status = OP_OPENED;

2,053,202✔

3868

  if (pOperator->status == OP_EXEC_DONE) {

7,440,300✔

3869

    (*ppRes) = NULL;

1,846,071✔

3870

    return code;

1,846,117✔

3873

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

5,591,672✔

3874

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

5,594,275✔

3876

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

5,594,264✔

3877

  SSDataBlock* pRes = pInfo->pRes;

5,591,626✔

3878

  blockDataCleanup(pRes);

5,586,954✔

3880

  int32_t size = 0;

5,594,176✔

3881

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

5,595,258✔

3882

  QUERY_CHECK_CODE(code, lino, _end);

5,586,390✔

3884

  if (size == 0) {

5,586,390✔

3885

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

18,774✔

3886

    (*ppRes) = NULL;

18,774✔

3887

    return code;

18,774✔

3890

  SMetaReader mr = {0};

5,567,616✔

3891

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

5,568,654✔

3892

  pRes->info.rows = 0;

5,576,530✔

3894

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

16,335,151✔

3895

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

10,756,017✔

3896

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

10,759,630✔

3899

      code = TSDB_CODE_SUCCESS;

10,763,721✔

3901

    QUERY_CHECK_CODE(code, lino, _end);

10,759,630✔

3903

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

10,759,630✔

3904

      setOperatorCompleted(pOperator);

5,577,004✔

3908

  pAPI->metaReaderFn.clearReader(&mr);

5,576,394✔

3909

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

5,577,050✔

3910

  if (bLimitReached) {

5,577,050✔

3915

  if (pOperator->status == OP_EXEC_DONE) {

5,577,050✔

3916

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

5,576,538✔

3919

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

5,576,958✔

3921

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

5,577,004✔

3922

_end:

5,576,492✔

3923

  if (code != TSDB_CODE_SUCCESS) {

5,577,004✔

3928

  return code;

5,577,004✔

3931

static void destroyTagScanOperatorInfo(void* param) {

8,561,509✔

3932

  STagScanInfo* pInfo = (STagScanInfo*)param;

8,561,509✔

3933

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

8,561,509✔

3936

  taosHashCleanup(pInfo->filterCtx.colHash);

8,566,414✔

3937

  taosArrayDestroy(pInfo->filterCtx.cInfoList);

8,548,320✔

3938

  taosArrayDestroy(pInfo->aFilterIdxs);

8,553,823✔

3939

  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);

8,555,369✔

3941

  blockDataDestroy(pInfo->pRes);

8,560,993✔

3942

  taosArrayDestroy(pInfo->matchInfo.pList);

8,566,109✔

3943

  tableListDestroy(pInfo->pTableListInfo);

8,561,882✔

3945

  pInfo->pRes = NULL;

8,570,501✔

3946

  pInfo->pTableListInfo = NULL;

8,571,541✔

3947

  taosMemoryFreeClear(param);

8,566,767✔

3948

8,567,440✔

3950

static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {

5,154✔

3951

  int32_t         code = TSDB_CODE_SUCCESS;

5,154✔

3952

  STagScanInfo*   pInfo = pOper->info;

5,154✔

3954

  pOper->status = OP_NOT_OPENED;

5,154✔

3955

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

5,154✔

3956

  if (pTagScanNode->onlyMetaCtbIdx) {

5,154✔

3957

      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

2,032✔

3958

      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

2,032✔

3959

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

2,032✔

3960

      pInfo->pCtbCursor = NULL;

2,032✔

3962

      pInfo->curPos = 0;

3,122✔

3965

  return code;

5,154✔

3968

int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,

8,563,097✔

3971

  QRY_PARAM_CHECK(pOptrInfo);

8,563,097✔

3973

  int32_t         code = TSDB_CODE_SUCCESS;

8,571,548✔

3974

  int32_t         lino = 0;

8,571,548✔

3975

  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;

8,571,548✔

3976

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

8,571,548✔

3977

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

8,555,134✔

3978

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

8,543,893✔

3982

  pOperator->pPhyNode = pTagScanNode;

8,544,391✔

3983

  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

8,546,049✔

3985

  int32_t    numOfExprs = 0;

8,550,785✔

3986

  SExprInfo* pExprInfo = NULL;

8,555,119✔

3988

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

8,550,967✔

3989

  QUERY_CHECK_CODE(code, lino, _error);

8,559,835✔

3991

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

8,559,835✔

3992

  QUERY_CHECK_CODE(code, lino, _error);

8,534,198✔

3994

  int32_t num = 0;

8,534,198✔

3995

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

8,538,659✔

3996

  QUERY_CHECK_CODE(code, lino, _error);

8,542,821✔

3998

  pInfo->pTagCond = pTagCond;

8,542,821✔

3999

  pInfo->pTagIndexCond = pTagIndexCond;

8,546,508✔

4000

  pInfo->suid = pPhyNode->suid;

8,546,520✔

4001

  pInfo->pStorageAPI = &pTaskInfo->storageAPI;

8,547,255✔

4003

  pInfo->pTableListInfo = pTableListInfo;

8,538,049✔

4004

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

8,537,989✔

4005

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

8,560,630✔

4007

  pInfo->readHandle = *pReadHandle;

8,540,091✔

4008

  pInfo->curPos = 0;

8,554,490✔

4010

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

8,526,692✔

4011

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

8,524,438✔

4013

  initResultSizeInfo(&pOperator->resultInfo, 4096);

8,530,655✔

4014

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

8,526,108✔

4015

  QUERY_CHECK_CODE(code, lino, _error);

8,568,763✔

4017

  if (pTagScanNode->onlyMetaCtbIdx) {

8,568,763✔

4018

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

4,699,687✔

4019

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

4,705,050✔

4021

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

4,704,194✔

4022

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

4,703,715✔

4024

    pInfo->filterCtx.colHash =

4,700,950✔

4025

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

4,701,738✔

4026

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

4,702,167✔

4028

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

4,699,978✔

4029

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

4,701,109✔

4031

    if (pInfo->pTagCond != NULL) {

4,694,054✔

4032

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

1,641,768✔

4035

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

8,568,325✔

4036

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

8,571,676✔

4038

  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);

8,559,509✔

4040

  *pOptrInfo = pOperator;

8,570,398✔

4041

  return code;

8,572,959✔

4058

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

241,201,784✔

4059

  int32_t                 left = *(int32_t*)pLeft;

241,201,784✔

4060

  int32_t                 right = *(int32_t*)pRight;

241,204,332✔

4061

  STmsSubTablesMergeInfo* pInfo = (STmsSubTablesMergeInfo*)param;

241,210,156✔

4063

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

241,210,156✔

4064

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

241,224,352✔

4066

  if (leftIdx == -1) {

241,224,716✔

4067

    return 1;

206,513,036✔

4068

  } else if (rightIdx == -1) {

34,711,680✔

4069

    return -1;

343,680✔

4072

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

34,368,000✔

4073

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

34,368,000✔

4074

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

34,368,000✔

4075

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

34,368,000✔

4078

  return ret;

34,368,000✔

4108

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

3,015,958✔

4109

  int32_t code = TSDB_CODE_SUCCESS;

3,015,958✔

4110

  int32_t lino = 0;

3,015,958✔

4111

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

3,015,958✔

4112

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

3,015,958✔

4113

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

3,000,844✔

4114

  for (int i = 0; i < src->numOfCols; i++) {

10,263,404✔

4115

    dst->colList[i] = src->colList[i];

7,301,988✔

4118

_end:

3,009,502✔

4119

  if (code != TSDB_CODE_SUCCESS) {

3,009,502✔

4122

  return code;

2,993,320✔

4125

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

4,564,552✔

4127

  int32_t code = 0;

4,564,552✔

4129

  STableMergeScanInfo*    pInfo = pOperator->info;

4,564,552✔

4130

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

4,562,410✔

4131

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

4,571,068✔

4132

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

4,558,424✔

4133

  const SStorageAPI*      pAPI = &pTaskInfo->storageAPI;

4,570,722✔

4135

  blockDataCleanup(pInput->pReaderBlock);

4,576,510✔

4136

  if (!pInput->bInMemReader) {

4,561,754✔

4137

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

4,550,828✔

4138

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

2,277,034✔

4139

    if (code != 0) {

2,279,188✔

4144

  pInfo->base.dataReader = pInput->pReader;

4,564,982✔

4146

  while (true) {

5,460✔

4147

    bool hasNext = false;

4,570,328✔

4148

    code = pAPI->tsdReader.tsdNextDataBlock(pInfo->base.dataReader, &hasNext);

4,569,248✔

4149

    if (code != 0) {

4,584,136✔

4155

    if (!hasNext || isTaskKilled(pTaskInfo)) {

4,584,136✔

4156

      if (isTaskKilled(pTaskInfo)) {

3,054,432✔

4162

      *pSubTableHasBlock = false;

3,058,722✔

4163

      break;

3,056,932✔

4166

    if (pInput->tblCond.order == TSDB_ORDER_ASC) {

1,530,068✔

4167

      pInput->tblCond.twindows.skey = pInput->pReaderBlock->info.window.ekey + 1;

1,506,408✔

4169

      pInput->tblCond.twindows.ekey = pInput->pReaderBlock->info.window.skey - 1;

23,660✔

4172

    uint32_t status = 0;

1,529,346✔

4173

    code = loadDataBlock(pOperator, &pInfo->base, pInput->pReaderBlock, &status);

1,529,346✔

4174

    if (code != 0) {

1,529,704✔

4179

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

1,529,704✔

4183

    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pInput->pReaderBlock->info.rows == 0) {

1,529,704✔

4184

      continue;

5,460✔

4187

    *pSubTableHasBlock = true;

1,524,608✔

4188

    break;

1,524,608✔

4191

  if (*pSubTableHasBlock) {

4,584,046✔

4192

    pInput->pReaderBlock->info.id.groupId =

3,048,852✔

4193

        tableListGetTableGroupId(pInfo->base.pTableListInfo, pInput->pReaderBlock->info.id.uid);

3,049,216✔

4194

    pOperator->resultInfo.totalRows += pInput->pReaderBlock->info.rows;

1,524,244✔

4196

  if (!pInput->bInMemReader || !*pSubTableHasBlock) {

4,583,694✔

4197

    pAPI->tsdReader.tsdReaderClose(pInput->pReader);

3,436,668✔

4198

    pInput->pReader = NULL;

3,438,106✔

4201

  pInfo->base.dataReader = NULL;

4,580,114✔

4202

  return TSDB_CODE_SUCCESS;

4,578,682✔

4205

static int32_t setGroupStartEndIndex(STableMergeScanInfo* pInfo) {

1,873,550✔

4206

  pInfo->bGroupProcessed = false;

1,873,550✔

4208

  int32_t numOfTables = 0;

1,887,160✔

4209

  int32_t code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

1,837,910✔

4210

  if (code != TSDB_CODE_SUCCESS) {

1,804,198✔

4215

  int32_t i = pInfo->tableStartIndex + 1;

1,804,198✔

4216

  for (; i < numOfTables; ++i) {

2,946,146✔

4217

    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

1,197,696✔

4218

    if (!tableKeyInfo) {

1,192,964✔

4222

    if (tableKeyInfo->groupId != pInfo->groupId) {

1,192,964✔

4223

      break;

56,792✔

4226

  pInfo->tableEndIndex = i - 1;

1,805,242✔

4227

  return TSDB_CODE_SUCCESS;

1,891,790✔

4230

static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {

1,909,518✔

4231

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,957,852✔

4232

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,052,994✔

4233

    if (pInput->rowIdx == -1) {

3,046,556✔

4234

      continue;

1,523,738✔

4237

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,524,608✔

4238

      pInput->rowIdx = 0;

1,523,880✔

4239

      pInput->pageIdx = -1;

1,524,244✔

4242

    pInput->pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

1,524,972✔

4243

    SColumnInfoData* col = taosArrayGet(pInput->pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

1,524,608✔

4244

    if (!col) {

1,524,244✔

4248

    pInput->aTs = (int64_t*)col->pData;

1,524,244✔

4251

  __merge_compare_fn_t mergeCompareFn = (!pSubTblsInfo->pPkOrderInfo) ? subTblRowCompareTsFn : subTblRowCompareTsPkFn;

1,905,938✔

4252

  return tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, mergeCompareFn);

1,907,012✔

4255

static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {

1,871,420✔

4256

  int32_t code = setGroupStartEndIndex(pInfo);

1,871,420✔

4257

  if (code != TSDB_CODE_SUCCESS) {

1,879,504✔

4261

  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));

1,879,504✔

4262

  if (pSubTblsInfo == NULL) {

1,856,294✔

4265

  pSubTblsInfo->pTsOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);

1,856,294✔

4266

  if (!pSubTblsInfo->pTsOrderInfo) {

1,874,266✔

4270

  if (taosArrayGetSize(pInfo->pSortInfo) == 2) {

1,848,400✔

4277

    pSubTblsInfo->pPkOrderInfo = NULL;

1,832,206✔

4279

  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

1,856,288✔

4280

  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));

1,843,746✔

4281

  if (pSubTblsInfo->aInputs == NULL) {

1,814,962✔

4285

  int32_t bufPageSize = pInfo->bufPageSize;

1,835,416✔

4286

  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;

1,836,162✔

4287

  code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);

1,841,914✔

4288

  if (code != TSDB_CODE_SUCCESS) {

1,838,012✔

4293

  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;

1,838,012✔

4294

  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;

1,862,398✔

4296

  pInfo->pSubTablesMergeInfo = pSubTblsInfo;

1,848,370✔

4297

  return TSDB_CODE_SUCCESS;

1,820,320✔

4300

static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {

1,868,162✔

4301

  int32_t        code = TSDB_CODE_SUCCESS;

1,868,162✔

4302

  int32_t        lino = 0;

1,868,162✔

4303

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,868,162✔

4304

  SReadHandle*   pHandle = &pInfo->base.readHandle;

1,870,972✔

4305

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

1,850,196✔

4307

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

1,851,932✔

4309

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,904,550✔

4310

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

2,978,010✔

4311

    pInput->type = SUB_TABLE_MEM_BLOCK;

2,976,422✔

4313

    code = dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);

2,990,760✔

4314

    QUERY_CHECK_CODE(code, lino, _end);

2,979,662✔

4316

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pReaderBlock);

2,979,662✔

4317

    QUERY_CHECK_CODE(code, lino, _end);

3,052,260✔

4319

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pPageBlock);

3,052,260✔

4320

    QUERY_CHECK_CODE(code, lino, _end);

3,058,728✔

4322

    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);

3,058,728✔

4323

    pInput->pKeyInfo = keyInfo;

3,055,840✔

4325

    if (isTaskKilled(pTaskInfo)) {

3,054,044✔

4329

    if (i + 1 < pSubTblsInfo->numInMemReaders) {

3,036,806✔

4330

      code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, keyInfo, 1, pInput->pReaderBlock,

2,279,988✔

4331

                                           (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

1,142,360✔

4332

      QUERY_CHECK_CODE(code, lino, _end);

1,137,992✔

4333

      pInput->bInMemReader = true;

1,137,992✔

4335

      pInput->pReader = NULL;

1,899,822✔

4336

      pInput->bInMemReader = false;

1,900,538✔

4338

    bool hasNext = true;

3,044,986✔

4339

    code = fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);

3,042,444✔

4340

    QUERY_CHECK_CODE(code, lino, _end);

3,055,500✔

4341

    if (!hasNext) {

3,055,500✔

4342

      pInput->rowIdx = -1;

1,530,892✔

4343

      ++pSubTblsInfo->numSubTablesCompleted;

1,531,250✔

4344

      continue;

1,532,688✔

4346

      pInput->rowIdx = 0;

1,524,608✔

4347

      pInput->pageIdx = -1;

1,524,608✔

4351

_end:

1,910,228✔

4352

  if (code != TSDB_CODE_SUCCESS) {

1,910,228✔

4355

  return code;

1,903,062✔

4358

static void adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

1,524,608✔

4359

  int32_t              code = TSDB_CODE_SUCCESS;

1,524,608✔

4360

  int32_t              lino = 0;

1,524,608✔

4361

  SExecTaskInfo*       pTaskInfo = pOperatorInfo->pTaskInfo;

1,524,608✔

4362

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

1,524,608✔

4363

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

1,524,608✔

4364

  bool                 hasNext = true;

1,524,608✔

4365

  code = fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);

1,524,608✔

4366

  QUERY_CHECK_CODE(code, lino, _end);

1,524,608✔

4368

  if (!hasNext) {

1,524,608✔

4369

    pInput->rowIdx = -1;

1,524,608✔

4370

    ++pSubTblsInfo->numSubTablesCompleted;

1,524,608✔

4375

_end:

1,524,608✔

4376

  if (code != TSDB_CODE_SUCCESS) {

1,524,608✔

4381

1,524,608✔

4383

static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

277,033,136✔

4384

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

277,033,136✔

4385

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

277,041,848✔

4387

  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

277,018,320✔

4388

  if (pInput->rowIdx < pInputBlock->info.rows - 1) {

277,115,016✔

4389

    ++pInput->rowIdx;

275,591,088✔

4390

  } else if (pInput->rowIdx == pInputBlock->info.rows - 1) {

1,524,608✔

4391

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

1,524,608✔

4392

      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);

1,524,608✔

4394

    if (pInput->rowIdx != -1) {

1,524,608✔

4405

  return tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));

277,114,604✔

4408

static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {

277,026,832✔

4409

  int32_t            code = TSDB_CODE_SUCCESS;

277,026,832✔

4410

  int32_t            lino = 0;

277,026,832✔

4411

  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

277,026,832✔

4412

  SSDataBlock*       pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

277,010,578✔

4414

  for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {

1,161,421,180✔

4415

    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);

883,791,216✔

4416

    QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);

883,107,296✔

4418

    SColumnInfoData* pSrcColInfo = taosArrayGet(pInputBlock->pDataBlock, i);

883,107,296✔

4419

    QUERY_CHECK_NULL(pSrcColInfo, code, lino, _end, terrno);

881,864,598✔

4420

    bool isNull = colDataIsNull(pSrcColInfo, pInputBlock->info.rows, pInput->rowIdx, NULL);

881,864,598✔

4422

    if (isNull) {

883,199,292✔

4423

      code = colDataSetVal(pColInfo, pBlock->info.rows, NULL, true);

117,874,276✔

4424

      QUERY_CHECK_CODE(code, lino, _end);

117,856,518✔

4426

      if (pSrcColInfo->pData != NULL) {

765,325,016✔

4427

        char* pData = colDataGetData(pSrcColInfo, pInput->rowIdx);

765,488,810✔

4428

        code = colDataSetVal(pColInfo, pBlock->info.rows, pData, false);

767,431,454✔

4429

        QUERY_CHECK_CODE(code, lino, _end);

766,521,068✔

4433

  pBlock->info.dataLoad = 1;

277,181,944✔

4434

  pBlock->info.scanFlag = pInputBlock->info.scanFlag;

277,183,394✔

4435

  pBlock->info.rows += 1;

277,188,776✔

4437

_end:

277,177,880✔

4438

  if (code != TSDB_CODE_SUCCESS) {

277,177,880✔

4441

  return code;

277,121,852✔

4444

static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,

3,082,194✔

4446

  int32_t                 code = TSDB_CODE_SUCCESS;

3,082,194✔

4447

  int32_t                 lino = 0;

3,082,194✔

4448

  STableMergeScanInfo*    pInfo = pOperator->info;

3,082,194✔

4449

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

3,088,292✔

4450

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

3,081,126✔

4451

  bool                    finished = false;

3,083,632✔

4453

  QRY_PARAM_CHECK(pResBlock);

3,083,632✔

4455

  blockDataCleanup(pBlock);

3,082,916✔

4459

      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {

280,190,916✔

4460

        finished = true;

3,094,736✔

4461

        break;

3,094,736✔

4464

      code = appendChosenRowToDataBlock(pSubTblsInfo, pBlock);

277,057,908✔

4465

      QUERY_CHECK_CODE(code, lino, _end);

277,063,902✔

4467

      code = adjustSubTableForNextRow(pOperator, pSubTblsInfo);

277,063,902✔

4468

      QUERY_CHECK_CODE(code, lino, _end);

277,094,712✔

4470

      if (pBlock->info.rows >= capacity) {

277,094,712✔

4475

    if (isTaskKilled(pTaskInfo)) {

3,094,736✔

4479

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pBlock, pTaskInfo);

3,093,662✔

4480

    if (finished || limitReached || pBlock->info.rows > 0) {

3,087,576✔

4485

  if (pBlock->info.rows > 0) {

3,087,576✔

4486

    *pResBlock = pBlock;

1,180,928✔

4489

_end:

1,903,784✔

4490

  if (code != TSDB_CODE_SUCCESS) {

3,084,712✔

4494

  return code;

3,084,354✔

4497

static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {

1,848,406✔

4498

  int32_t              code = TSDB_CODE_SUCCESS;

1,848,406✔

4499

  int32_t              lino = 0;

1,848,406✔

4500

  STableMergeScanInfo* pInfo = pOperator->info;

1,848,406✔

4502

  code = initSubTablesMergeInfo(pInfo);

1,840,242✔

4503

  QUERY_CHECK_CODE(code, lino, _end);

1,825,344✔

4505

  code = initSubTableInputs(pOperator, pInfo);

1,825,344✔

4506

  QUERY_CHECK_CODE(code, lino, _end);

1,904,494✔

4508

  code = openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);

1,904,494✔

4509

  QUERY_CHECK_CODE(code, lino, _end);

1,900,556✔

4511

_end:

1,900,556✔

4512

  if (code != TSDB_CODE_SUCCESS) {

1,900,556✔

4515

  return code;

1,901,636✔

4518

static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {

28,654,434✔

4519

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

28,654,434✔

4520

  if (pSubTblsInfo != NULL) {

28,673,740✔

4521

    tMergeTreeDestroy(&pSubTblsInfo->pTree);

1,905,574✔

4523

    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

4,963,938✔

4524

      STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

3,054,426✔

4525

      taosMemoryFree(pInput->tblCond.colList);

3,050,846✔

4526

      blockDataDestroy(pInput->pReaderBlock);

3,052,272✔

4527

      blockDataDestroy(pInput->pPageBlock);

3,058,364✔

4528

      taosArrayDestroy(pInput->aBlockPages);

3,058,364✔

4529

      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);

3,059,438✔

4530

      pInput->pReader = NULL;

3,057,290✔

4533

    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);

1,913,098✔

4534

    taosMemoryFree(pSubTblsInfo->aInputs);

1,910,586✔

4536

    taosMemoryFree(pSubTblsInfo);

1,905,580✔

4537

    pInfo->pSubTablesMergeInfo = NULL;

1,909,160✔

4541

28,678,042✔

4543

int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

3,058,444✔

4544

  QRY_PARAM_CHECK(ppRes);

3,058,444✔

4546

  int32_t lino = 0;

3,068,134✔

4547

  int32_t tableListSize = 0;

3,068,134✔

4548

  int64_t st = taosGetTimestampUs();

3,037,936✔

4550

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,037,936✔

4551

  STableMergeScanInfo* pInfo = pOperator->info;

3,034,308✔

4553

  if (pOperator->status == OP_EXEC_DONE) {

3,039,756✔

4557

  int32_t code = pOperator->fpSet._openFn(pOperator);

3,030,394✔

4558

  QUERY_CHECK_CODE(code, lino, _end);

3,022,530✔

4560

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

3,022,530✔

4561

  QUERY_CHECK_CODE(code, lino, _end);

3,032,936✔

4563

  if (!pInfo->hasGroupId) {

3,032,936✔

4564

    pInfo->hasGroupId = true;

1,845,194✔

4566

    if (tableListSize == 0) {

1,821,798✔

4567

      setOperatorCompleted(pOperator);

56,420✔

4568

      (*ppRes) = NULL;

56,420✔

4569

      return code;

56,420✔

4572

    pInfo->tableStartIndex = 0;

1,765,378✔

4573

    STableKeyInfo* pTmpGpId = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,785,498✔

4574

    QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

1,777,938✔

4575

    pInfo->groupId = pTmpGpId->groupId;

1,777,938✔

4576

    code = startSubTablesTableMergeScan(pOperator);

1,814,222✔

4577

    QUERY_CHECK_CODE(code, lino, _end);

1,849,486✔

4580

  SSDataBlock* pBlock = NULL;

3,040,092✔

4581

  while (pInfo->tableStartIndex < tableListSize) {

3,087,564✔

4582

    if (isTaskKilled(pTaskInfo)) {

3,087,934✔

4586

    code = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity, &pBlock);

3,085,058✔

4587

    QUERY_CHECK_CODE(code, lino, _end);

3,082,922✔

4589

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

3,082,922✔

4596

    if (pBlock != NULL) {

3,083,996✔

4597

      pBlock->info.id.groupId = pInfo->groupId;

1,180,928✔

4598

      pOperator->resultInfo.totalRows += pBlock->info.rows;

1,180,928✔

4599

      pInfo->bGroupProcessed = true;

1,180,928✔

4600

      break;

1,180,928✔

4603

      stopSubTablesTableMergeScan(pInfo);

1,903,068✔

4604

      if (pInfo->tableEndIndex >= tableListSize - 1) {

1,907,012✔

4605

        setOperatorCompleted(pOperator);

1,848,776✔

4606

        break;

1,848,060✔

4609

      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

57,520✔

4610

      STableKeyInfo* pTmpGpId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

57,520✔

4611

      QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

57,520✔

4613

      pInfo->groupId = pTmpGpId->groupId;

57,520✔

4614

      code = startSubTablesTableMergeScan(pOperator);

57,520✔

4615

      QUERY_CHECK_CODE(code, lino, _end);

57,520✔

4616

      resetLimitInfoForNextGroup(&pInfo->limitInfo);

57,520✔

4620

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

3,031,494✔

4622

_end:

3,035,432✔

4623

  if (code != TSDB_CODE_SUCCESS) {

3,030,062✔

4628

    (*ppRes) = pBlock;

3,030,062✔

4631

  return code;

3,027,198✔

4634

static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {

3,480,823✔

4635

  int32_t              code = TSDB_CODE_SUCCESS;

3,480,823✔

4636

  int32_t              lino = 0;

3,480,823✔

4637

  SOperatorInfo*       pOperator = (SOperatorInfo*)pTableMergeOpInfo;

3,480,823✔

4638

  STableMergeScanInfo* pInfo = pOperator->info;

3,480,823✔

4639

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

3,480,823✔

4641

  if (pInfo->mSkipTables == NULL) {

3,480,823✔

4642

    pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,

276,336✔

4644

    QUERY_CHECK_NULL(pInfo->mSkipTables, code, lino, _end, terrno);

275,934✔

4646

  int bSkip = 1;

3,480,421✔

4647

  if (pInfo->mSkipTables != NULL) {

3,480,421✔

4648

    code = taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));

3,480,421✔

4649

    if (code == TSDB_CODE_DUP_KEY) {

3,480,421✔

4652

    QUERY_CHECK_CODE(code, lino, _end);

3,480,421✔

4655

_end:

3,480,421✔

4656

  if (code != TSDB_CODE_SUCCESS) {

3,480,421✔

4661

3,480,421✔

4663

static int32_t doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {

167,676,661✔

4664

  STableMergeScanInfo* pInfo = pOperator->info;

167,676,661✔

4665

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

167,758,161✔

4666

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

167,684,641✔

4667

  SSDataBlock*         pBlock = pInfo->pReaderBlock;

167,685,176✔

4668

  int32_t              code = 0;

167,683,614✔

4669

  bool                 hasNext = false;

167,683,614✔

4670

  STsdbReader*         reader = pInfo->base.dataReader;

167,672,171✔

4672

  code = pAPI->tsdReader.tsdNextDataBlock(reader, &hasNext);

167,661,498✔

4673

  if (code != 0) {

167,622,392✔

4680

  if (!hasNext || isTaskKilled(pTaskInfo)) {

167,622,392✔

4681

    if (isTaskKilled(pTaskInfo)) {

27,986,619✔

4685

    *pFinished = true;

27,904,468✔

4686

    return code;

27,887,918✔

4689

  uint32_t status = 0;

139,703,955✔

4690

  code = loadDataBlock(pOperator, &pInfo->base, pBlock, &status);

139,724,501✔

4692

  if (code != TSDB_CODE_SUCCESS) {

139,705,201✔

4693

    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));

16,640✔

4694

    pTaskInfo->code = code;

16,640✔

4695

    return code;

16,640✔

4698

  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

139,688,561✔

4704

  if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {

139,688,561✔

4705

    *pSkipped = true;

7,224,524✔

4706

    return code;

7,241,854✔

4709

  return code;

132,475,661✔

4712

static int32_t getBlockForTableMergeScan(void* param, SSDataBlock** ppBlock) {

170,509,310✔

4713

  STableMergeScanSortSourceParam* source = param;

170,509,310✔

4715

  SOperatorInfo*       pOperator = source->pOperator;

170,509,310✔

4716

  STableMergeScanInfo* pInfo = pOperator->info;

170,538,780✔

4717

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

170,578,949✔

4718

  SSDataBlock*         pBlock = NULL;

170,565,825✔

4719

  int64_t              st = taosGetTimestampUs();

170,565,803✔

4720

  int32_t              code = TSDB_CODE_SUCCESS;

170,565,803✔

4723

    if (pInfo->rtnNextDurationBlocks) {

186,892,949✔

4724

      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d",

19,206,122✔

4727

      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {

19,231,218✔

4728

        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];

9,097,456✔

4729

        ++pInfo->nextDurationBlocksIdx;

9,097,081✔

4731

        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

19,220,277✔

4732

          blockDataDestroy(pInfo->nextDurationBlocks[i]);

9,096,394✔

4733

          pInfo->nextDurationBlocks[i] = NULL;

9,090,917✔

4736

        pInfo->rtnNextDurationBlocks = false;

10,128,087✔

4737

        pInfo->nextDurationBlocksIdx = 0;

10,128,087✔

4738

        pInfo->numNextDurationBlocks = 0;

10,125,087✔

4739

        continue;

10,123,587✔

4742

      bool bFinished = false;

167,671,101✔

4743

      bool bSkipped = false;

167,642,620✔

4745

      code = doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);

167,697,262✔

4746

      if (code != 0) {

167,593,328✔

4747

        return code;

16,640✔

4750

      pBlock = pInfo->pReaderBlock;

167,576,688✔

4751

      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d",

167,637,850✔

4753

      if (bFinished) {

167,783,831✔

4754

        pInfo->bNewFilesetEvent = false;

27,929,169✔

4755

        break;

27,929,169✔

4758

      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {

139,854,662✔

4759

        if (!bSkipped) {

10,137,246✔

4760

          code = createOneDataBlock(pBlock, true, &pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks]);

9,096,706✔

4761

          if (code) {

9,084,854✔

4766

          ++pInfo->numNextDurationBlocks;

9,084,854✔

4767

          if (pInfo->numNextDurationBlocks > 2) {

9,078,397✔

4775

        if (pInfo->bNewFilesetEvent) {

10,120,976✔

4776

          pInfo->rtnNextDurationBlocks = true;

10,110,547✔

4777

          *ppBlock = NULL;

10,117,880✔

4778

          return code;

10,106,868✔

4781

        if (pInfo->bNextDurationBlockEvent) {

2,666✔

4786

      if (bSkipped) continue;

129,727,350✔

4789

    pBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pBlock->info.id.uid);

132,620,554✔

4791

    pOperator->resultInfo.totalRows += pBlock->info.rows;

132,621,621✔

4792

    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

132,608,567✔

4793

    *ppBlock = pBlock;

132,608,994✔

4795

    return code;

132,606,418✔

4798

  *ppBlock = NULL;

27,923,681✔

4799

  return code;

27,928,129✔

4802

int32_t generateSortByTsPkInfo(SArray* colMatchInfo, int32_t order, SArray** ppSortArray) {

24,451,782✔

4803

  int32_t code = TSDB_CODE_SUCCESS;

24,451,782✔

4804

  int32_t lino = 0;

24,451,782✔

4805

  SArray* pSortInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));

24,451,782✔

4806

  QUERY_CHECK_NULL(pSortInfo, code, lino, _end, terrno);

24,360,769✔

4807

  SBlockOrderInfo biTs = {0};

24,360,769✔

4808

  SBlockOrderInfo biPk = {0};

24,343,715✔

4810

  int32_t tsTargetSlotId = 0;

24,406,157✔

4811

  int32_t pkTargetSlotId = -1;

24,406,157✔

4812

  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {

93,244,409✔

4813

    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);

69,152,922✔

4814

    QUERY_CHECK_NULL(colInfo, code, lino, _end, terrno);

69,165,039✔

4815

    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

69,165,039✔

4816

      tsTargetSlotId = colInfo->dstSlotId;

24,448,307✔

4817

      biTs.order = order;

24,462,199✔

4818

      biTs.slotId = tsTargetSlotId;

24,462,199✔

4819

      biTs.nullFirst = (order == TSDB_ORDER_ASC);

24,462,199✔

4820

      biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, order);

24,462,199✔

4823

    if (colInfo->isPk) {

68,981,938✔

4824

      pkTargetSlotId = colInfo->dstSlotId;

1,984,933✔

4825

      biPk.order = order;

1,981,518✔

4826

      biPk.slotId = pkTargetSlotId;

1,981,518✔

4827

      biPk.nullFirst = (order == TSDB_ORDER_ASC);

1,981,518✔

4828

      biPk.compFn = getKeyComparFunc(colInfo->dataType.type, order);

1,981,518✔

4832

  void* tmp = taosArrayPush(pSortInfo, &biTs);

24,506,769✔

4833

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

24,506,769✔

4834

  if (pkTargetSlotId != -1) {

24,506,769✔

4835

    tmp = taosArrayPush(pSortInfo, &biPk);

1,993,502✔

4836

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

1,993,502✔

4839

  (*ppSortArray) = pSortInfo;

24,510,889✔

4841

_end:

24,265,062✔

4842

  if (code != TSDB_CODE_SUCCESS) {

24,427,680✔

4845

  return code;

24,427,680✔

4848

void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {

35,644,214✔

4849

  STableMergeScanInfo* pTmsInfo = param;

35,644,214✔

4850

  if (type == TSD_READER_NOTIFY_DURATION_START) {

35,644,214✔

4851

    pTmsInfo->bNewFilesetEvent = true;

35,669,557✔

4855

  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);

35,651,136✔

4857

  return;

35,746,681✔

4860

int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

38,106,248✔

4861

  STableMergeScanInfo* pInfo = pOperator->info;

38,106,248✔

4862

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

38,182,388✔

4864

  int32_t code = TSDB_CODE_SUCCESS;

38,176,652✔

4865

  int32_t lino = 0;

38,176,652✔

4866

  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

38,176,652✔

4868

  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));

38,147,486✔

4869

  pInfo->bNewFilesetEvent = false;

38,358,214✔

4870

  pInfo->bNextDurationBlockEvent = false;

38,264,928✔

4872

  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;

38,257,136✔

4873

  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;

38,263,383✔

4875

  pInfo->pSortHandle = NULL;

38,228,671✔

4876

  code = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,

76,525,203✔

4877

                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0, &pInfo->pSortHandle);

38,241,969✔

4878

  if (code) {

38,120,860✔

4882

  if (pInfo->bSortRowId && numOfTable != 1) {

38,120,860✔

4883

    int32_t memSize = 512 * 1024 * 1024;

3,506,678✔

4884

    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);

3,506,678✔

4885

    QUERY_CHECK_CODE(code, lino, _end);

3,514,388✔

4888

  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);

38,153,734✔

4889

  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pOperator);

38,170,347✔

4890

  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);

38,158,896✔

4892

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);

38,131,826✔

4893

  QUERY_CHECK_CODE(code, lino, _end);

38,111,114✔

4895

  STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));

38,111,114✔

4896

  QUERY_CHECK_NULL(param, code, lino, _end, terrno);

38,116,161✔

4897

  param->pOperator = pOperator;

38,116,161✔

4899

  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));

38,095,082✔

4900

  if (ps == NULL) {

38,082,023✔

4905

  ps->param = param;

38,082,023✔

4906

  ps->onlyRef = false;

38,087,228✔

4907

  code = tsortAddSource(pInfo->pSortHandle, ps);

38,184,104✔

4908

  QUERY_CHECK_CODE(code, lino, _end);

38,212,930✔

4910

  if (numOfTable == 1) {

38,212,930✔

4911

    tsortSetSingleTableMerge(pInfo->pSortHandle);

26,405,565✔

4913

    code = tsortOpen(pInfo->pSortHandle);

11,807,365✔

4914

    QUERY_CHECK_CODE(code, lino, _end);

11,833,897✔

4917

_end:

11,825,577✔

4918

  if (code != TSDB_CODE_SUCCESS) {

38,158,021✔

4919

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,320✔

4921

  return code;

38,163,995✔

4924

void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

38,278,883✔

4925

  STableMergeScanInfo* pInfo = pOperator->info;

38,278,883✔

4926

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

38,279,627✔

4927

  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));

38,284,978✔

4929

  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

38,290,994✔

4930

  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;

38,268,653✔

4931

  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;

38,266,193✔

4932

  pInfo->sortExecInfo.loops += sortExecInfo.loops;

38,272,514✔

4933

  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;

38,267,194✔

4934

  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

38,259,621✔

4936

  tsortDestroySortHandle(pInfo->pSortHandle);

38,250,257✔

4937

  pInfo->pSortHandle = NULL;

38,277,197✔

4938

38,273,626✔

4940

void startGroupTableMergeScan(SOperatorInfo* pOperator) {

28,108,962✔

4941

  int32_t              code = TSDB_CODE_SUCCESS;

28,108,962✔

4942

  int32_t              lino = 0;

28,108,962✔

4943

  STableMergeScanInfo* pInfo = pOperator->info;

28,108,962✔

4944

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,117,486✔

4945

  SReadHandle*         pHandle = &pInfo->base.readHandle;

28,070,881✔

4946

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,033,330✔

4947

  qDebug("%s table merge scan start group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,002,968✔

4950

    int32_t numOfTables = 0;

28,031,373✔

4951

    code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

28,028,942✔

4952

    QUERY_CHECK_CODE(code, lino, _end);

28,087,714✔

4954

    int32_t i = pInfo->tableStartIndex + 1;

28,087,714✔

4955

    for (; i < numOfTables; ++i) {

134,131,710✔

4956

      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

112,347,493✔

4957

      QUERY_CHECK_NULL(tableKeyInfo, code, lino, _end, terrno);

112,344,052✔

4958

      if (tableKeyInfo->groupId != pInfo->groupId) {

112,344,052✔

4959

        break;

6,293,753✔

4962

    pInfo->tableEndIndex = i - 1;

28,077,970✔

4964

  pInfo->bGroupProcessed = false;

28,136,594✔

4965

  int32_t tableStartIdx = pInfo->tableStartIndex;

28,124,513✔

4966

  int32_t tableEndIdx = pInfo->tableEndIndex;

28,120,031✔

4968

  int32_t        numOfTable = tableEndIdx - tableStartIdx + 1;

28,126,494✔

4969

  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);

28,126,494✔

4970

  code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,

56,182,517✔

4971

                                       (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->mSkipTables);

28,084,091✔

4972

  QUERY_CHECK_CODE(code, lino, _end);

28,034,659✔

4973

  if (pInfo->filesetDelimited) {

28,034,659✔

4974

    pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);

28,060,819✔

4976

  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);

28,090,600✔

4978

  code = startDurationForGroupTableMergeScan(pOperator);

27,934,610✔

4979

  QUERY_CHECK_CODE(code, lino, _end);

28,039,115✔

4981

_end:

28,030,795✔

4982

  if (code != TSDB_CODE_SUCCESS) {

28,043,493✔

4983

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

8,320✔

4984

    pTaskInfo->code = code;

8,320✔

4985

    T_LONG_JMP(pTaskInfo->env, code);

8,320✔

4987

28,035,173✔

4989

void stopGroupTableMergeScan(SOperatorInfo* pOperator) {

28,152,353✔

4990

  STableMergeScanInfo* pInfo = pOperator->info;

28,152,353✔

4991

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

28,157,853✔

4992

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

28,158,351✔

4994

  stopDurationForGroupTableMergeScan(pOperator);

28,149,387✔

4996

  if (pInfo->base.dataReader != NULL) {

28,147,239✔

4997

    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);

28,153,888✔

4998

    pInfo->base.dataReader = NULL;

28,041,077✔

5000

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

28,063,198✔

5001

    if (pInfo->nextDurationBlocks[i]) {

1,062✔

5002

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

1,062✔

5003

      pInfo->nextDurationBlocks[i] = NULL;

1,062✔

5005

    pInfo->numNextDurationBlocks = 0;

1,062✔

5006

    pInfo->nextDurationBlocksIdx = 0;

1,062✔

5008

  resetLimitInfoForNextGroup(&pInfo->limitInfo);

28,066,254✔

5009

  taosHashCleanup(pInfo->mSkipTables);

27,981,885✔

5010

  pInfo->mSkipTables = NULL;

27,997,990✔

5011

  qDebug("%s table merge scan stop group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

28,004,657✔

5012

28,146,491✔

5016

SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,

79,185,343✔

5018

  STableMergeScanInfo* pInfo = pOperator->info;

79,185,343✔

5019

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

79,236,108✔

5020

  STupleHandle*        pTupleHandle = NULL;

79,208,944✔

5022

  blockDataCleanup(pResBlock);

79,217,302✔

5025

    while (1) {

2,147,483,647✔

5026

      pTupleHandle = NULL;

2,147,483,647✔

5027

      int32_t code = tsortNextTuple(pHandle, &pTupleHandle);

2,147,483,647✔

5028

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5032

      if (pTupleHandle == NULL) {

2,147,483,647✔

5033

        break;

62,710,832✔

5036

      code = tsortAppendTupleToBlock(pInfo->pSortHandle, pResBlock, pTupleHandle);

2,147,483,647✔

5037

      if (code != TSDB_CODE_SUCCESS) {

2,147,483,647✔

5042

      if (pResBlock->info.rows >= capacity) {

2,147,483,647✔

5043

        break;

16,554,628✔

5047

    if (tsortIsClosed(pHandle)) {

79,265,460✔

5052

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);

79,294,416✔

5053

    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,

79,285,166✔

5055

    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {

79,311,192✔

5059

  return (pResBlock->info.rows > 0) ? pResBlock : NULL;

79,311,192✔

5062

int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

65,835,117✔

5063

  if (pOperator->status == OP_EXEC_DONE) {

65,835,117✔

5064

    (*ppRes) = NULL;

41,915✔

5065

    return TSDB_CODE_SUCCESS;

41,915✔

5068

  int32_t              code = TSDB_CODE_SUCCESS;

65,772,524✔

5069

  int32_t              lino = 0;

65,772,524✔

5070

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

65,772,524✔

5071

  STableMergeScanInfo* pInfo = pOperator->info;

65,794,763✔

5073

  code = pOperator->fpSet._openFn(pOperator);

65,769,310✔

5074

  QUERY_CHECK_CODE(code, lino, _end);

65,726,800✔

5076

  int64_t st = taosGetTimestampUs();

65,773,059✔

5078

  int32_t tableListSize = 0;

65,773,059✔

5079

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

65,772,861✔

5080

  QUERY_CHECK_CODE(code, lino, _end);

65,770,499✔

5082

  if (!pInfo->hasGroupId) {

65,770,499✔

5083

    pInfo->hasGroupId = true;

24,768,665✔

5085

    if (tableListSize == 0) {

24,734,274✔

5086

      setOperatorCompleted(pOperator);

2,982,223✔

5087

      (*ppRes) = NULL;

2,979,744✔

5088

      return code;

2,980,009✔

5090

    pInfo->tableStartIndex = 0;

21,752,051✔

5091

    STableKeyInfo* tmp = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

21,749,765✔

5092

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

21,764,324✔

5093

    pInfo->groupId = tmp->groupId;

21,764,324✔

5094

    startGroupTableMergeScan(pOperator);

21,786,600✔

5097

  SSDataBlock* pBlock = NULL;

62,782,350✔

5098

  while (pInfo->tableStartIndex < tableListSize) {

79,190,287✔

5099

    if (isTaskKilled(pTaskInfo)) {

79,216,962✔

5103

    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,

79,223,604✔

5105

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

79,311,178✔

5106

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

1,845,571✔

5107

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

1,845,571✔

5108

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

1,845,571✔

5110

    if (pBlock != NULL) {

79,310,069✔

5111

      pBlock->info.id.groupId = pInfo->groupId;

41,024,186✔

5112

      pOperator->resultInfo.totalRows += pBlock->info.rows;

41,022,900✔

5113

      pInfo->bGroupProcessed = true;

41,023,190✔

5114

      break;

41,019,326✔

5116

      if (pInfo->bNewFilesetEvent) {

38,285,883✔

5117

        stopDurationForGroupTableMergeScan(pOperator);

10,128,399✔

5118

        code = startDurationForGroupTableMergeScan(pOperator);

10,126,524✔

5119

        QUERY_CHECK_CODE(code, lino, _end);

10,111,719✔

5122

        stopGroupTableMergeScan(pOperator);

28,150,248✔

5123

        if (pInfo->tableEndIndex >= tableListSize - 1) {

28,148,311✔

5124

          setOperatorCompleted(pOperator);

21,843,762✔

5125

          break;

21,843,669✔

5128

        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

6,296,218✔

5129

        STableKeyInfo* tmp = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

6,296,218✔

5130

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

6,296,218✔

5131

        pInfo->groupId = tmp->groupId;

6,296,218✔

5132

        startGroupTableMergeScan(pOperator);

6,296,218✔

5133

        resetLimitInfoForNextGroup(&pInfo->limitInfo);

6,295,705✔

5138

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

62,850,111✔

5140

_end:

62,866,404✔

5141

  if (code != TSDB_CODE_SUCCESS) {

62,868,960✔

5146

    (*ppRes) = pBlock;

62,868,960✔

5149

  return code;

62,857,579✔

5158

void destroyTableMergeScanOperatorInfo(void* param) {

24,547,585✔

5159

  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;

24,547,585✔

5162

  if (pTableScanInfo->base.readerAPI.tsdReaderClose != NULL) {

24,547,585✔

5163

    pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);

24,551,976✔

5164

    pTableScanInfo->base.dataReader = NULL;

24,522,977✔

5167

  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {

24,521,602✔

5174

  tsortDestroySortHandle(pTableScanInfo->pSortHandle);

24,547,572✔

5175

  pTableScanInfo->pSortHandle = NULL;

24,526,310✔

5176

  taosHashCleanup(pTableScanInfo->mSkipTables);

24,528,942✔

5177

  pTableScanInfo->mSkipTables = NULL;

24,511,737✔

5178

  blockDataDestroy(pTableScanInfo->pSortInputBlock);

24,508,221✔

5179

  pTableScanInfo->pSortInputBlock = NULL;

24,548,031✔

5182

  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);

24,549,404✔

5184

  blockDataDestroy(pTableScanInfo->pResBlock);

24,515,208✔

5185

  pTableScanInfo->pResBlock = NULL;

24,553,356✔

5188

  blockDataDestroy(pTableScanInfo->pReaderBlock);

24,553,254✔

5189

  pTableScanInfo->pReaderBlock = NULL;

24,544,742✔

5190

  taosArrayDestroy(pTableScanInfo->pSortInfo);

24,550,170✔

5192

  stopSubTablesTableMergeScan(pTableScanInfo);

24,537,530✔

5194

  taosMemoryFreeClear(param);

24,507,404✔

5195

24,519,395✔

5197

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {

1,223,833✔

5198

  if (pOptr == NULL) {

1,223,833✔

5203

  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));

1,223,833✔

5204

  if (!execInfo) {

1,221,168✔

5207

  STableMergeScanInfo* pInfo = pOptr->info;

1,221,168✔

5208

  execInfo->blockRecorder = pInfo->base.readRecorder;

1,220,777✔

5209

  execInfo->sortExecInfo = pInfo->sortExecInfo;

1,224,216✔

5211

  *pOptrExplain = execInfo;

1,226,865✔

5212

  *len = sizeof(STableMergeScanExecInfo);

1,228,014✔

5214

  return TSDB_CODE_SUCCESS;

1,226,881✔

5217

static int32_t resetTableMergeScanOperatorState(SOperatorInfo* pOper) {

2,218,242✔

5218

  int32_t         code = TSDB_CODE_SUCCESS;

2,218,242✔

5219

  STableMergeScanInfo*   pInfo = pOper->info;

2,218,242✔

5220

  pOper->status = OP_NOT_OPENED;

2,218,242✔

5222

  pInfo->tableEndIndex = 0;

2,218,242✔

5223

  pInfo->tableStartIndex = 0;

2,218,242✔

5224

  pInfo->hasGroupId = false;

2,218,242✔

5225

  if (pInfo->base.readerAPI.tsdReaderClose) {

2,218,242✔

5226

    pInfo->base.readerAPI.tsdReaderClose(pInfo->base.dataReader);

2,218,242✔

5228

  pInfo->base.dataReader = NULL;

2,218,242✔

5229

  pInfo->base.scanFlag = MAIN_SCAN;

2,218,242✔

5231

  pInfo->base.limitInfo = (SLimitInfo){0};

2,218,242✔

5232

  pInfo->base.limitInfo.limit.limit = -1;

2,218,242✔

5233

  pInfo->base.limitInfo.slimit.limit = -1;

2,218,242✔

5235

  tableListDestroy(pInfo->base.pTableListInfo);

2,218,242✔

5236

  pInfo->base.pTableListInfo = tableListCreate();

2,218,242✔

5237

  if (!pInfo->base.pTableListInfo) {

2,218,242✔

5241

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

2,218,242✔

5243

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

2,218,242✔

5244

  code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

4,436,484✔

5246

                                  pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

4,436,249✔

5247

  if (code) {

2,218,242✔

5252

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

2,218,242✔

5253

  cleanupQueryTableDataCond(&pInfo->base.cond);

2,218,242✔

5254

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, &pInfo->base.readHandle, false);

2,218,242✔

5255

  if (code) {

2,218,007✔

5260

  tsortDestroySortHandle(pInfo->pSortHandle);

2,218,007✔

5261

  pInfo->pSortHandle = NULL;

2,218,007✔

5263

  taosHashCleanup(pInfo->mSkipTables);

2,218,007✔

5264

  pInfo->mSkipTables = NULL;

2,218,242✔

5265

  pOper->resultInfo.totalRows = 0;

2,218,242✔

5267

  pInfo->bGroupProcessed = false;

2,218,007✔

5268

  pInfo->bNewFilesetEvent = false;

2,218,242✔

5269

  pInfo->bNextDurationBlockEvent = false;

2,218,242✔

5270

  pInfo->rtnNextDurationBlocks = false;

2,218,007✔

5271

  pInfo->nextDurationBlocksIdx = 0;

2,218,007✔

5272

  pInfo->bSortRowId = false;

2,218,007✔

5274

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

2,218,007✔

5280

  pInfo->numNextDurationBlocks = 0;

2,218,242✔

5282

  stopSubTablesTableMergeScan(pInfo);

2,218,007✔

5283

  return code;

2,218,007✔

5286

int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

24,369,579✔

5289

  QRY_PARAM_CHECK(pOptrInfo);

24,369,579✔

5291

  int32_t              code = TSDB_CODE_SUCCESS;

24,398,250✔

5292

  int32_t              lino = 0;

24,398,250✔

5293

  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));

24,398,250✔

5294

  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

24,319,194✔

5295

  if (pInfo == NULL || pOperator == NULL) {

24,175,405✔

5296

    code = terrno;

1,152✔

5300

  pOperator->pPhyNode = pTableScanNode;

24,174,266✔

5301

  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

24,195,879✔

5303

  int32_t numOfCols = 0;

24,222,641✔

5304

  code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,

24,287,835✔

5306

  QUERY_CHECK_CODE(code, lino, _error);

24,215,714✔

5308

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle, false);

24,215,714✔

5309

  QUERY_CHECK_CODE(code, lino, _error);

24,002,663✔

5311

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

24,002,663✔

5312

    SExprSupp* pSup = &pInfo->base.pseudoSup;

10,345,075✔

5313

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->pExprInfo, &pSup->numOfExprs);

10,319,331✔

5314

    QUERY_CHECK_CODE(code, lino, _error);

10,316,425✔

5316

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

10,316,425✔

5318

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

10,282,474✔

5321

  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};

24,023,208✔

5323

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

24,164,785✔

5328

  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;

24,164,785✔

5329

  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

24,279,102✔

5330

  pInfo->base.scanFlag = MAIN_SCAN;

23,910,000✔

5331

  pInfo->base.readHandle = *readHandle;

24,158,307✔

5333

  pInfo->readIdx = -1;

24,018,304✔

5335

  pInfo->base.limitInfo.limit.limit = -1;

23,986,358✔

5336

  pInfo->base.limitInfo.slimit.limit = -1;

24,203,724✔

5337

  pInfo->base.pTableListInfo = pTableListInfo;

24,093,112✔

5339

  pInfo->sample.sampleRatio = pTableScanNode->ratio;

24,109,742✔

5340

  pInfo->sample.seed = taosGetTimestampSec();

24,053,812✔

5342

  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,

24,232,341✔

5343

                            pTaskInfo->pStreamRuntimeInfo);

24,252,763✔

5344

  QUERY_CHECK_CODE(code, lino, _error);

24,178,383✔

5346

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

24,178,383✔

5348

  pInfo->mergeLimit = -1;

24,132,723✔

5349

  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;

24,143,116✔

5350

  if (hasLimit) {

24,133,254✔

5351

    pInfo->mergeLimit = pInfo->limitInfo.limit.offset != -1

3,810,583✔

5352

                            ? pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset

762,227✔

5353

                            : pInfo->limitInfo.limit.limit;

2,665,900✔

5354

    pInfo->mSkipTables = NULL;

1,910,753✔

5357

  initResultSizeInfo(&pOperator->resultInfo, 1024);

24,134,842✔

5358

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

24,099,987✔

5359

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

24,367,913✔

5360

  code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

24,182,269✔

5361

  QUERY_CHECK_CODE(code, lino, _error);

24,466,661✔

5362

  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {

24,466,661✔

5363

    pInfo->bSortRowId = true;

5,145,335✔

5365

    pInfo->bSortRowId = false;

19,331,209✔

5368

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

24,436,650✔

5369

  QUERY_CHECK_CODE(code, lino, _error);

24,393,217✔

5371

  code = generateSortByTsPkInfo(pInfo->base.matchInfo.pList, pInfo->base.cond.order, &pInfo->pSortInfo);

24,393,217✔

5372

  QUERY_CHECK_CODE(code, lino, _error);

24,283,095✔

5374

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pReaderBlock);

24,283,095✔

5375

  QUERY_CHECK_CODE(code, lino, _error);

24,382,716✔

5377

  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;

24,382,716✔

5379

  int32_t  rowSize = pInfo->pResBlock->info.rowSize;

24,416,260✔

5380

  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);

24,267,229✔

5382

  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);

24,237,025✔

5385

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pSortInputBlock);

24,402,739✔

5386

  QUERY_CHECK_CODE(code, lino, _error);

24,403,368✔

5388

  if (!tsExperimental) {

24,403,368✔

5391

    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

24,403,368✔

5394

  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,

24,407,676✔

5396

  pOperator->exprSupp.numOfExprs = numOfCols;

24,293,916✔

5398

  pOperator->fpSet = createOperatorFpSet(

24,258,610✔

5399

      optrDummyOpenFn, pTableScanNode->paraTablesSort ? doTableMergeScanParaSubTablesNext : doTableMergeScanNext, NULL,

24,331,364✔

5402

  setOperatorResetStateFn(pOperator, resetTableMergeScanOperatorState);

24,239,154✔

5403

  pOperator->cost.openCost = 0;

24,369,341✔

5405

  *pOptrInfo = pOperator;

24,397,142✔

5406

  return code;

24,214,822✔

5445

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {

29,026✔

5446

  if (scanCols != NULL) {

29,026✔

5447

    SNode* pNode = NULL;

15,199✔

5448

    FOREACH(pNode, scanCols) {

35,517✔

5449

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

20,141✔

5452

      STargetNode* targetNode = (STargetNode*)pNode;

20,141✔

5453

      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {

20,141✔

5456

      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);

20,318✔

5457

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

20,318✔

5458

        supp->dbNameSlotId = targetNode->slotId;

10,064✔

5459

      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

10,254✔

5460

        supp->stbNameSlotId = targetNode->slotId;

10,254✔

5464

  return TSDB_CODE_SUCCESS;

29,000✔

5467

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {

29,026✔

5468

  if (pseudoCols != NULL) {

29,026✔

5469

    SNode* pNode = NULL;

29,026✔

5470

    FOREACH(pNode, pseudoCols) {

58,026✔

5471

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

28,869✔

5474

      STargetNode* targetNode = (STargetNode*)pNode;

28,898✔

5475

      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {

28,898✔

5478

      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);

29,049✔

5479

      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {

29,049✔

5480

        supp->tbCountSlotId = targetNode->slotId;

29,020✔

5484

  return TSDB_CODE_SUCCESS;

29,000✔

5487

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {

29,046✔

5488

  if (groupTags != NULL) {

29,046✔

5489

    SNode* pNode = NULL;

15,199✔

5490

    FOREACH(pNode, groupTags) {

35,517✔

5491

      if (nodeType(pNode) != QUERY_NODE_COLUMN) {

20,318✔

5494

      SColumnNode* colNode = (SColumnNode*)pNode;

20,318✔

5495

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

20,318✔

5496

        supp->groupByDbName = true;

10,064✔

5498

      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

20,318✔

5499

        supp->groupByStbName = true;

10,254✔

5503

    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);

13,847✔

5504

    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);

13,847✔

5506

  return TSDB_CODE_SUCCESS;

29,020✔

5509

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,

29,046✔

5511

  int32_t code = 0;

29,046✔

5512

  code = tblCountScanGetInputs(groupTags, tableName, supp);

29,046✔

5513

  if (code != TSDB_CODE_SUCCESS) {

28,843✔

5518

  supp->dbNameSlotId = -1;

28,843✔

5519

  supp->stbNameSlotId = -1;

29,046✔

5520

  supp->tbCountSlotId = -1;

29,075✔

5522

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);

29,075✔

5523

  if (code != TSDB_CODE_SUCCESS) {

29,026✔

5528

  code = tblCountScanGetCountSlotId(pseudoCols, supp);

29,026✔

5529

  if (code != TSDB_CODE_SUCCESS) {

29,000✔

5533

  return code;

29,029✔

5536

int32_t createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,

28,744✔

5538

  QRY_PARAM_CHECK(pOptrInfo);

28,744✔

5540

  int32_t                      code = TSDB_CODE_SUCCESS;

29,075✔

5541

  int32_t                      lino = 0;

29,075✔

5542

  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;

29,075✔

5543

  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));

29,075✔

5544

  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

29,055✔

5545

  if (!pInfo || !pOperator) {

29,075✔

5550

  pInfo->readHandle = *readHandle;

29,075✔

5552

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

29,075✔

5553

  initResultSizeInfo(&pOperator->resultInfo, 1);

29,075✔

5554

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

29,055✔

5555

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

29,075✔

5557

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

29,046✔

5558

  QUERY_CHECK_CODE(code, lino, _error);

29,020✔

5560

  code = getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,

29,020✔

5563

  QUERY_CHECK_CODE(code, lino, _error);

29,029✔

5565

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,

29,029✔

5567

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScanNext, NULL, destoryTableCountScanOperator,

28,869✔

5569

  *pOptrInfo = pOperator;

29,046✔

5570

  return code;

29,020✔

5584

int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,

39,259✔

5586

  int32_t code = TSDB_CODE_SUCCESS;

39,259✔

5587

  int32_t lino = 0;

39,259✔

5588

  if (pSupp->dbNameSlotId != -1) {

39,259✔

5589

    QUERY_CHECK_CONDITION((strlen(dbName) > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

15,176✔

5590

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);

15,176✔

5591

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

15,176✔

5593

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

15,176✔

5594

    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

15,176✔

5596

    varDataSetLen(varDbName, strlen(dbName));

15,176✔

5597

    code = colDataSetVal(colInfoData, 0, varDbName, false);

15,021✔

5598

    QUERY_CHECK_CODE(code, lino, _end);

15,176✔

5601

  if (pSupp->stbNameSlotId != -1) {

39,259✔

5602

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);

19,455✔

5603

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

19,282✔

5604

    if (strlen(stbName) != 0) {

19,282✔

5605

      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

13,953✔

5606

      tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);

13,953✔

5607

      varDataSetLen(varStbName, strlen(stbName));

13,953✔

5608

      code = colDataSetVal(colInfoData, 0, varStbName, false);

13,953✔

5609

      QUERY_CHECK_CODE(code, lino, _end);

14,112✔

5615

  if (pSupp->tbCountSlotId != -1) {

39,447✔

5616

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);

39,389✔

5617

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

39,026✔

5618

    code = colDataSetVal(colInfoData, 0, (char*)&count, false);

39,026✔

5619

    QUERY_CHECK_CODE(code, lino, _end);

39,011✔

5621

  pRes->info.rows = 1;

39,040✔

5623

_end:

39,334✔

5624

  if (code != TSDB_CODE_SUCCESS) {

39,334✔

5627

  return code;

39,075✔

5630

static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {

11,119✔

5631

  STableCountScanSupp* pSupp = &pInfo->supp;

11,119✔

5632

  SSDataBlock*         pRes = pInfo->pRes;

11,119✔

5634

  size_t infodbTableNum;

11,119✔

5635

  getInfosDbMeta(NULL, &infodbTableNum);

11,119✔

5636

  infodbTableNum -= 1;

11,119✔

5637

  size_t perfdbTableNum;

11,119✔

5638

  getPerfDbMeta(NULL, &perfdbTableNum);

11,119✔

5640

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

11,119✔

5641

    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);

9,549✔

5642

    return (pRes->info.rows > 0) ? pRes : NULL;

9,549✔

5644

    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);

1,570✔

5645

    return (pRes->info.rows > 0) ? pRes : NULL;

1,570✔

5649

static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,

1,570✔

5651

  int32_t        code = TSDB_CODE_SUCCESS;

1,570✔

5652

  int32_t        lino = 0;

1,570✔

5653

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

1,570✔

5654

  if (strlen(pSupp->stbNameFilter) != 0) {

1,570✔

5656

    code = fillTableCountScanDataBlock(pSupp, "", "", 0, pRes);

531✔

5657

    QUERY_CHECK_CODE(code, lino, _end);

531✔

5658

  } else if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {

1,039✔

5659

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

177✔

5660

    QUERY_CHECK_CODE(code, lino, _end);

177✔

5661

  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

862✔

5664

  } else if (strlen(pSupp->dbNameFilter) == 0) {

862✔

5665

    code = fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);

685✔

5666

    QUERY_CHECK_CODE(code, lino, _end);

685✔

5669

    code = fillTableCountScanDataBlock(pSupp, pSupp->dbNameFilter, pSupp->stbNameFilter, 0, pRes);

177✔

5670

    QUERY_CHECK_CODE(code, lino, _end);

177✔

5673

_end:

177✔

5674

  if (code != TSDB_CODE_SUCCESS) {

1,570✔

5679

  setOperatorCompleted(pOperator);

1,570✔

5680

1,570✔

5682

static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

9,549✔

5685

  int32_t        code = TSDB_CODE_SUCCESS;

9,549✔

5686

  int32_t        lino = 0;

9,549✔

5687

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

9,549✔

5688

  if (pInfo->currGrpIdx == 0) {

9,549✔

5689

    uint64_t groupId = 0;

3,183✔

5690

    if (pSupp->groupByDbName) {

3,183✔

5691

      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));

2,182✔

5693

      groupId = calcGroupId("", 0);

1,001✔

5696

    pRes->info.id.groupId = groupId;

3,183✔

5697

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

3,183✔

5698

    QUERY_CHECK_CODE(code, lino, _end);

3,183✔

5699

  } else if (pInfo->currGrpIdx == 1) {

6,366✔

5700

    uint64_t groupId = 0;

3,183✔

5701

    if (pSupp->groupByDbName) {

3,183✔

5702

      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));

2,182✔

5704

      groupId = calcGroupId("", 0);

1,001✔

5707

    pRes->info.id.groupId = groupId;

3,183✔

5708

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

3,183✔

5709

    QUERY_CHECK_CODE(code, lino, _end);

3,183✔

5711

    setOperatorCompleted(pOperator);

3,183✔

5713

  pInfo->currGrpIdx++;

9,549✔

5715

_end:

9,549✔

5716

  if (code != TSDB_CODE_SUCCESS) {

9,549✔

5721

9,549✔

5723

static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

68,325✔

5724

  int32_t                      code = TSDB_CODE_SUCCESS;

68,325✔

5725

  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;

68,325✔

5726

  STableCountScanOperatorInfo* pInfo = pOperator->info;

68,484✔

5727

  STableCountScanSupp*         pSupp = &pInfo->supp;

67,971✔

5728

  SSDataBlock*                 pRes = pInfo->pRes;

68,170✔

5730

  blockDataCleanup(pRes);

68,179✔

5731

  QRY_PARAM_CHECK(ppRes);

68,493✔

5733

  if (pOperator->status == OP_EXEC_DONE) {

68,493✔

5734

    return code;

17,818✔

5737

  if (pInfo->readHandle.mnd != NULL) {

50,695✔

5738

    (*ppRes) = buildSysDbTableCount(pOperator, pInfo);

11,119✔

5739

    return code;

11,119✔

5742

  code = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);

39,417✔

5743

  if (code != TSDB_CODE_SUCCESS) {

38,985✔

5748

  if (pRes->info.rows > 0) {

38,985✔

5749

    *ppRes = pRes;

31,444✔

5752

  return code;

39,492✔

5755

static int32_t buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

39,229✔

5757

  int32_t        code = TSDB_CODE_SUCCESS;

39,229✔

5758

  int32_t        lino = 0;

39,229✔

5759

  const char*    db = NULL;

39,229✔

5760

  int32_t        vgId = 0;

39,229✔

5761

  char           dbName[TSDB_DB_NAME_LEN] = {0};

39,388✔

5762

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

39,417✔

5763

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

39,576✔

5766

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);

39,547✔

5767

  SName sn = {0};

39,371✔

5769

  code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

39,371✔

5770

  QUERY_CHECK_CODE(code, lino, _end);

39,400✔

5772

  code = tNameGetDbName(&sn, dbName);

39,400✔

5773

  QUERY_CHECK_CODE(code, lino, _end);

39,576✔

5775

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

39,576✔

5776

    code = buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);

27,270✔

5778

    code = buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);

12,306✔

5781

_end:

38,992✔

5782

  if (code != TSDB_CODE_SUCCESS) {

38,992✔

5787

  return code;

38,992✔

5790

static int32_t buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

26,952✔

5792

  int32_t        code = TSDB_CODE_SUCCESS;

26,952✔

5793

  int32_t        lino = 0;

26,952✔

5794

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

26,952✔

5795

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

27,270✔

5797

  if (pSupp->groupByStbName) {

27,111✔

5798

    if (pInfo->stbUidList == NULL) {

23,328✔

5799

      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));

8,074✔

5800

      QUERY_CHECK_NULL(pInfo->stbUidList, code, lino, _end, terrno);

8,074✔

5801

      code = pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList);

8,074✔

5802

      QUERY_CHECK_CODE(code, lino, _end);

7,897✔

5805

    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {

22,833✔

5806

      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);

14,112✔

5807

      code = buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);

13,936✔

5808

      QUERY_CHECK_CODE(code, lino, _end);

14,112✔

5810

      pInfo->currGrpIdx++;

14,112✔

5811

    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {

9,216✔

5812

      code = buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);

8,074✔

5813

      QUERY_CHECK_CODE(code, lino, _end);

7,915✔

5815

      pInfo->currGrpIdx++;

7,915✔

5817

      setOperatorCompleted(pOperator);

1,142✔

5820

    uint64_t groupId = calcGroupId(dbName, strlen(dbName));

3,942✔

5821

    pRes->info.id.groupId = groupId;

3,942✔

5823

    int64_t dbTableCount = 0;

3,942✔

5824

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);

3,942✔

5825

    code = fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);

3,942✔

5826

    QUERY_CHECK_CODE(code, lino, _end);

3,942✔

5827

    setOperatorCompleted(pOperator);

3,942✔

5830

_end:

27,111✔

5831

  if (code != TSDB_CODE_SUCCESS) {

27,111✔

5835

  return code;

27,111✔

5838

static int32_t buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

12,306✔

5840

  int32_t        code = TSDB_CODE_SUCCESS;

12,306✔

5841

  int32_t        lino = 0;

12,306✔

5842

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

12,306✔

5843

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

12,306✔

5845

  if (strlen(pSupp->dbNameFilter) != 0 && strcmp(dbName, pSupp->dbNameFilter) != 0) {

12,306✔

5849

  } else if (strlen(pSupp->stbNameFilter) != 0) {

12,286✔

5850

    uint64_t uid = 0;

6,048✔

5851

    code = pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid);

6,048✔

5852

    if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

6,048✔

5854

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, 0, pRes);

1,416✔

5855

      QUERY_CHECK_CODE(code, lino, _end);

1,416✔

5857

      QUERY_CHECK_CODE(code, lino, _end);

4,632✔

5859

      int64_t numOfChildTables = 0;

4,632✔

5860

      code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL, NULL);

4,632✔

5861

      QUERY_CHECK_CODE(code, lino, _end);

4,612✔

5863

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes);

4,612✔

5864

      QUERY_CHECK_CODE(code, lino, _end);

4,612✔

5867

    int64_t tbNumVnode = 0;

6,218✔

5868

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);

6,238✔

5869

    code = fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);

6,054✔

5870

    QUERY_CHECK_CODE(code, lino, _end);

6,134✔

5873

_end:

12,211✔

5874

  if (code != TSDB_CODE_SUCCESS) {

12,211✔

5879

  setOperatorCompleted(pOperator);

12,211✔

5880

  return code;

12,027✔

5883

static int32_t buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

7,915✔

5885

  int32_t code = TSDB_CODE_SUCCESS;

7,915✔

5886

  int32_t lino = 0;

7,915✔

5887

  char    fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

7,915✔

5888

  if (pSupp->groupByDbName) {

8,074✔

5889

    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");

3,940✔

5892

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

8,074✔

5893

  pRes->info.id.groupId = groupId;

7,915✔

5895

  int64_t numOfTables = 0;

7,915✔

5896

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);

8,074✔

5898

  if (numOfTables != 0) {

8,074✔

5899

    code = fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);

1,142✔

5902

_end:

6,932✔

5903

  if (code != TSDB_CODE_SUCCESS) {

8,074✔

5906

  return code;

8,074✔

5909

static int32_t buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

13,936✔

5911

  int32_t code = TSDB_CODE_SUCCESS;

13,936✔

5912

  int32_t lino = 0;

13,936✔

5913

  char    stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

13,936✔

5914

  code = pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);

14,112✔

5915

  QUERY_CHECK_CODE(code, lino, _end);

13,935✔

5917

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

13,935✔

5918

  if (pSupp->groupByDbName) {

14,112✔

5919

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));

6,552✔

5921

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));

7,560✔

5924

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

14,112✔

5925

  pRes->info.id.groupId = groupId;

13,953✔

5927

  int64_t ctbNum = 0;

13,953✔

5928

  code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL, NULL);

13,776✔

5929

  QUERY_CHECK_CODE(code, lino, _end);

13,957✔

5930

  code = fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);

13,957✔

5931

  QUERY_CHECK_CODE(code, lino, _end);

13,794✔

5933

_end:

13,794✔

5934

  if (code != TSDB_CODE_SUCCESS) {

13,953✔

5937

  return code;

13,953✔

5940

static void destoryTableCountScanOperator(void* param) {

29,046✔

5941

  STableCountScanOperatorInfo* pTableCountScanInfo = param;

29,046✔

5942

  blockDataDestroy(pTableCountScanInfo->pRes);

29,046✔

5944

  taosArrayDestroy(pTableCountScanInfo->stbUidList);

29,075✔

5945

  taosMemoryFreeClear(param);

29,075✔

5946

29,075✔

taosdata / TDengine / #4973

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4973

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