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

79

bool processBlockWithProbability(const SSampleExecInfo* pInfo) {

×

88

  return true;

×

92

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

×

93

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

×

94

    SWITCH_ORDER(pCtx[i].order);

×

96

×

98

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

×

99

  int32_t     code = TSDB_CODE_SUCCESS;

×

100

  STimeWindow w = {0};

×

103

  if (pInterval->interval == 0) {

×

104

    *overlap = false;

×

105

    return code;

×

108

  if (order == TSDB_ORDER_ASC) {

×

109

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

×

110

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

×

111

      qError("w.ekey:%" PRId64 " < pBlockInfo->window.skey:%" PRId64, w.ekey, pBlockInfo->window.skey);

×

112

      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

115

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

×

116

      *overlap = true;

×

117

      return code;

×

121

      getNextTimeWindow(pInterval, &w, order);

×

122

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

×

123

        break;

×

126

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

×

127

        qError("w.ekey:%" PRId64 " <= pBlockInfo->window.ekey:%" PRId64, w.ekey, pBlockInfo->window.ekey);

×

128

        return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

130

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

×

131

        *overlap = true;

×

132

        return code;

×

136

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

×

137

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

×

138

      qError("w.skey:%" PRId64 " > pBlockInfo->window.skey:%" PRId64, w.skey, pBlockInfo->window.ekey);

×

139

      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

142

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

×

143

      *overlap = true;

×

144

      return code;

×

148

      getNextTimeWindow(pInterval, &w, order);

×

149

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

×

150

        break;

×

153

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

×

154

        qError("w.skey:%" PRId64 " >= pBlockInfo->window.skey:%" PRId64, w.skey, pBlockInfo->window.skey);

×

155

        return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

157

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

×

158

        *overlap = true;

×

159

        return code;

×

164

  *overlap = false;

×

165

  return code;

×

169

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

×

170

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

×

171

    return NULL;

×

174

  int64_t buf[2] = {0};

×

175

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

×

177

  STableScanInfo* pTableScanInfo = pOperator->info;

×

179

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

×

182

  if (p1 == NULL) {

×

183

    return NULL;

×

186

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

×

187

  if (NULL == *pPage) {

×

188

    return NULL;

×

191

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

×

194

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

×

195

  if (NULL == pTableScanInfo->pIgnoreTables) {

×

196

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

×

197

    pTableScanInfo->pIgnoreTables =

×

198

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

×

199

    if (NULL == pTableScanInfo->pIgnoreTables) {

×

200

      return terrno;

×

204

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

×

206

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

×

207

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));

×

208

    return tempRes;

×

210

  return TSDB_CODE_SUCCESS;

×

213

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

×

214

  STableScanInfo* pTableScanInfo = pOperator->info;

×

215

  int32_t         code = TSDB_CODE_SUCCESS;

×

217

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

×

218

    return TSDB_CODE_SUCCESS;

×

221

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

×

223

  SFilePage*  pPage = NULL;

×

224

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

×

226

  if (pRow == NULL) {

×

227

    return TSDB_CODE_SUCCESS;

×

230

  bool notLoadBlock = true;

×

231

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

×

232

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

×

234

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

×

236

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

×

237

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

×

238

      notLoadBlock = false;

×

239

      break;

×

244

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

×

246

  if (notLoadBlock) {

×

247

    *status = FUNC_DATA_REQUIRED_NOT_LOAD;

×

248

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

×

251

  return code;

×

254

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

×

256

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

×

257

    *keep = true;

×

258

    return TSDB_CODE_SUCCESS;

×

261

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

×

264

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

×

266

  SStorageAPI* pAPI = &pTaskInfo->storageAPI;

×

267

  bool         allColumnsHaveAgg = true;

×

268

  bool         hasNullSMA = false;

×

269

  if (pLoad != NULL) {

×

270

    *pLoad = false;

×

273

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

×

275

  if (code != TSDB_CODE_SUCCESS) {

×

276

    return code;

×

279

  if (!allColumnsHaveAgg || hasNullSMA) {

×

280

    *pLoad = false;

×

282

    *pLoad = true;

×

285

  return code;

×

288

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

×

290

  int32_t    code = 0;

×

291

  SExprSupp* pSup = &pTableScanInfo->pseudoSup;

×

292

  if (pSup->numOfExprs > 0) {

×

293

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

×

296

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

297

      code = 0;

×

301

  return code;

×

304

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

×

305

  SLimit*     pLimit = &pLimitInfo->limit;

×

306

  const char* id = GET_TASKID(pTaskInfo);

×

308

  if (pLimitInfo->remainOffset > 0) {

×

309

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

×

310

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

×

311

      blockDataEmpty(pBlock);

×

312

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

×

313

      return false;

×

315

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

×

316

      if (code != TSDB_CODE_SUCCESS) {

×

317

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

×

318

        pTaskInfo->code = code;

×

319

        T_LONG_JMP(pTaskInfo->env, code);

×

321

      pLimitInfo->remainOffset = 0;

×

325

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

×

327

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

×

328

    blockDataKeepFirstNRows(pBlock, keep);

×

330

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

×

331

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

×

332

    return true;

×

335

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

×

336

  return false;

×

339

static bool skipSetTagColumnData(SOperatorInfo* pOperator) {

×

340

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

×

341

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

×

344

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

×

346

  int32_t        code = TSDB_CODE_SUCCESS;

×

347

  int32_t        lino = 0;

×

348

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

349

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

350

  bool           loadSMA = false;

×

352

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

×

354

  pCost->totalBlocks += 1;

×

355

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

×

356

  *status = pTableScanInfo->dataBlockLoadFlag;

×

358

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

×

359

    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

×

361

    bool overlap = false;

×

363

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

×

364

    if (ret != TSDB_CODE_SUCCESS) {

×

365

      return ret;

×

367

    if (overlap) {

×

368

      (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;

×

372

  SDataBlockInfo* pBlockInfo = &pBlock->info;

×

373

  taosMemoryFreeClear(pBlock->pBlockAgg);

×

375

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {

×

376

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

×

378

    pCost->filterOutBlocks += 1;

×

379

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

×

380

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

×

381

    return TSDB_CODE_SUCCESS;

×

382

  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

×

383

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

×

386

    if (!skipSetTagColumnData(pOperator)) {

×

387

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

×

389

    pCost->skipBlocks += 1;

×

390

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

×

391

    return code;

×

392

  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {

×

393

    pCost->loadBlockStatis += 1;

×

394

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

×

395

    bool success = true;

×

396

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

×

397

    if (code) {

×

398

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

×

399

      qError("%s failed to retrieve sma info", GET_TASKID(pTaskInfo));

×

400

      QUERY_CHECK_CODE(code, lino, _end);

×

403

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

×

404

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

×

406

      if (!skipSetTagColumnData(pOperator)) {

×

407

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

×

409

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

×

410

      return code;

×

412

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

×

413

      *status = FUNC_DATA_REQUIRED_DATA_LOAD;

×

417

  if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {

×

418

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

×

419

    qError("%s loadDataBlock invalid status:%d", GET_TASKID(pTaskInfo), *status);

×

420

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

424

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

×

425

    bool success = true;

×

426

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

×

427

    if (code) {

×

428

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

×

429

      qError("%s failed to retrieve sma info", GET_TASKID(pTaskInfo));

×

430

      QUERY_CHECK_CODE(code, lino, _end);

×

433

    if (success) {

×

434

      size_t size = taosArrayGetSize(pBlock->pDataBlock);

×

435

      bool   keep = false;

×

436

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

×

437

      if (code) {

×

438

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

×

439

        qError("%s failed to do filter by block sma, code:%s", GET_TASKID(pTaskInfo), tstrerror(code));

×

440

        QUERY_CHECK_CODE(code, lino, _end);

×

443

      if (!keep) {

×

444

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

×

446

        pCost->filterOutBlocks += 1;

×

447

        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

×

448

        taosMemoryFreeClear(pBlock->pBlockAgg);

×

450

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

×

451

        return TSDB_CODE_SUCCESS;

×

457

  taosMemoryFreeClear(pBlock->pBlockAgg);

×

460

  code = doDynamicPruneDataBlock(pOperator, pBlockInfo, status);

×

461

  if (code) {

×

462

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

×

463

    QUERY_CHECK_CODE(code, lino, _end);

×

466

  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {

×

467

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

×

469

    pCost->skipBlocks += 1;

×

470

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

×

472

    STableScanInfo* p1 = pOperator->info;

×

473

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

×

474

      *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT;

×

476

      *status = FUNC_DATA_REQUIRED_FILTEROUT;

×

478

    return TSDB_CODE_SUCCESS;

×

481

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

×

482

  pCost->loadBlocks += 1;

×

484

  SSDataBlock* p = NULL;

×

485

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

×

486

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

×

487

    return code;

×

490

  if (!skipSetTagColumnData(pOperator)) {

×

491

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

×

492

    QUERY_CHECK_CODE(code, lino, _end);

×

496

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

×

498

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

×

499

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

×

500

    QUERY_CHECK_CODE(code, lino, _end);

×

502

    int64_t st = taosGetTimestampUs();

×

503

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

×

504

    pTableScanInfo->readRecorder.filterTime += el;

×

506

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

×

507

      pCost->filterOutBlocks += 1;

×

508

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

×

511

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

×

515

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

×

516

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

×

517

    setOperatorCompleted(pOperator);

×

520

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

×

522

_end:

×

523

  if (code != TSDB_CODE_SUCCESS) {

×

524

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

×

526

  return code;

×

529

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

×

530

  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

×

532

  switchCtxOrder(pCtx, numOfOutput);

×

533

  pTableScanInfo->cond.order = TSDB_ORDER_DESC;

×

536

×

543

static void freeTableCachedVal(void* param) {

×

544

  if (param == NULL) {

×

545

    return;

×

548

  STableCachedVal* pVal = param;

×

549

  taosMemoryFree((void*)pVal->pName);

×

550

  taosMemoryFree(pVal->pTags);

×

551

  taosMemoryFree(pVal);

×

554

static int32_t createTableCacheVal(const SMetaReader* pMetaReader, STableCachedVal** ppResVal) {

×

555

  int32_t          code = TSDB_CODE_SUCCESS;

×

556

  int32_t          lino = 0;

×

557

  STableCachedVal* pVal = taosMemoryMalloc(sizeof(STableCachedVal));

×

558

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

×

560

  pVal->pTags = NULL;

×

561

  pVal->pName = taosStrdup(pMetaReader->me.name);

×

562

  QUERY_CHECK_NULL(pVal->pName, code, lino, _end, terrno);

×

565

  if (pMetaReader->me.type == TSDB_CHILD_TABLE || pMetaReader->me.type == TSDB_VIRTUAL_CHILD_TABLE) {

×

566

    STag* pTag = (STag*)pMetaReader->me.ctbEntry.pTags;

×

567

    pVal->pTags = taosMemoryMalloc(pTag->len);

×

568

    QUERY_CHECK_NULL(pVal->pTags, code, lino, _end, terrno);

×

569

    memcpy(pVal->pTags, pTag, pTag->len);

×

572

  (*ppResVal) = pVal;

×

574

_end:

×

575

  if (code != TSDB_CODE_SUCCESS) {

×

576

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

×

577

    freeTableCachedVal(pVal);

×

579

  return code;

×

583

static void freeCachedMetaItem(const void* key, size_t keyLen, void* value, void* ud) {

×

587

  freeTableCachedVal(value);

×

588

×

590

static void doSetNullValue(SSDataBlock* pBlock, const SExprInfo* pExpr, int32_t numOfExpr) {

×

591

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

×

592

    int32_t dstSlotId = pExpr[j].base.resSchema.slotId;

×

594

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

×

595

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

×

597

×

599

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

×

601

  int32_t          code = TSDB_CODE_SUCCESS;

×

602

  int32_t          lino = 0;

×

603

  bool             freeReader = false;

×

604

  LRUHandle*       h = NULL;

×

605

  STableCachedVal  val = {0};

×

606

  SMetaReader      mr = {0};

×

607

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

×

608

  int32_t          insertRet = TAOS_LRU_STATUS_OK;

×

609

  STableCachedVal* pVal = NULL;

×

612

  if (numOfExpr <= 0) {

×

613

    return TSDB_CODE_SUCCESS;

×

619

  int32_t backupRows = pBlock->info.rows;

×

620

  pBlock->info.rows = rows;

×

626

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

×

627

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

×

628

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

×

629

    if (code != TSDB_CODE_SUCCESS) {

×

631

      if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

632

        qWarn("failed to get table meta, table may have been dropped, uid:0x%" PRIx64 ", code:%s, %s",

×

636

        doSetNullValue(pBlock, pExpr, numOfExpr);

×

638

        qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid, tstrerror(code), idStr);

×

640

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

×

641

      return code;

×

644

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

×

646

    val.pName = mr.me.name;

×

647

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

×

649

    freeReader = true;

×

651

    pCache->metaFetch += 1;

×

653

    h = taosLRUCacheLookup(pCache->pTableMetaEntryCache, &pBlock->info.id.uid, sizeof(pBlock->info.id.uid));

×

654

    if (h == NULL) {

×

655

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

×

656

      freeReader = true;

×

657

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

×

658

      if (code != TSDB_CODE_SUCCESS) {

×

659

        if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

660

          qWarn("failed to get table meta, table may have been dropped, uid:0x%" PRIx64 ", code:%s, %s",

×

663

          doSetNullValue(pBlock, pExpr, numOfExpr);

×

665

          qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid, tstrerror(code),

×

668

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

×

669

        return code;

×

672

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

×

674

      code = createTableCacheVal(&mr, &pVal);

×

675

      QUERY_CHECK_CODE(code, lino, _end);

×

677

      val = *pVal;

×

679

      pCache->cacheHit += 1;

×

680

      STableCachedVal* pValTmp = taosLRUCacheValue(pCache->pTableMetaEntryCache, h);

×

681

      val = *pValTmp;

×

683

      bool bRes = taosLRUCacheRelease(pCache->pTableMetaEntryCache, h, false);

×

684

      qTrace("release LRU cache, res %d", bRes);

×

687

    qDebug("retrieve table meta from cache:%" PRIu64 ", hit:%" PRIu64 " miss:%" PRIu64 ", %s", pCache->metaFetch,

×

691

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

×

692

    const SExprInfo* pExpr1 = &pExpr[j];

×

693

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

×

695

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

×

696

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

×

697

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

×

699

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

×

702

    if (fmIsScanPseudoColumnFunc(functionId)) {

×

703

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

×

704

      if (fType == FUNCTION_TYPE_TBNAME) {

×

705

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

×

706

        QUERY_CHECK_CODE(code, lino, _end);

×

707

      } else if (fType == FUNCTION_TYPE_VGID) {

×

708

        code = setVgIdColData(pBlock, pColInfoData, functionId, pTask->id.vgId);

×

709

        QUERY_CHECK_CODE(code, lino, _end);

×

710

      } else if (fType == FUNCTION_TYPE_VGVER) {

×

711

        code = setVgVerColData(pBlock, pColInfoData, functionId, pBlock->info.version);

×

712

        QUERY_CHECK_CODE(code, lino, _end);

×

715

      STagVal tagVal = {0};

×

716

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

×

717

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

×

719

      char* data = NULL;

×

720

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

×

721

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

×

723

        data = (char*)p;

×

726

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

×

727

      if (isNullVal) {

×

728

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

×

729

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

×

730

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

×

731

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

×

732

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

×

733

          if (tmp != NULL) {

×

734

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

×

735

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

×

737

            taosMemoryFree(tmp);

×

739

          taosMemoryFree(data);

×

741

        QUERY_CHECK_CODE(code, lino, _end);

×

743

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

×

744

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

×

745

          QUERY_CHECK_CODE(code, lino, _end);

×

752

  pBlock->info.rows = backupRows;

×

754

_end:

×

756

  if (NULL != pVal) {

×

757

    insertRet =

×

758

        taosLRUCacheInsert(pCache->pTableMetaEntryCache, &pBlock->info.id.uid, sizeof(uint64_t), pVal,

×

760

    if (insertRet != TAOS_LRU_STATUS_OK) {

×

761

      qWarn("failed to put meta into lru cache, code:%d, %s", insertRet, idStr);

×

765

  if (freeReader) {

×

766

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

×

768

  if (code != TSDB_CODE_SUCCESS) {

×

769

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

×

771

  return code;

×

774

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

×

776

  int32_t                     code = TSDB_CODE_SUCCESS;

×

777

  int32_t                     lino = 0;

×

778

  struct SScalarFuncExecFuncs fpSet = {0};

×

779

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

×

780

  QUERY_CHECK_CODE(code, lino, _end);

×

782

  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;

×

783

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

×

784

  STR_TO_VARSTR(buf, name)

×

786

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

×

788

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

×

789

  QUERY_CHECK_CODE(code, lino, _end);

×

791

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

×

792

  QUERY_CHECK_CODE(code, lino, _end);

×

794

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

×

795

  SScalarParam param = {.columnData = pColInfoData};

×

797

  if (fpSet.process != NULL) {

×

798

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

×

799

    QUERY_CHECK_CODE(code, lino, _end);

×

801

    qError("failed to get the corresponding callback function, functionId:%d", functionId);

×

804

  colDataDestroy(&infoData);

×

806

_end:

×

807

  if (code != TSDB_CODE_SUCCESS) {

×

808

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

×

810

  return code;

×

813

int32_t setVgIdColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, int32_t vgId) {

×

814

  int32_t                     code = TSDB_CODE_SUCCESS;

×

815

  int32_t                     lino = 0;

×

816

  struct SScalarFuncExecFuncs fpSet = {0};

×

817

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

×

818

  QUERY_CHECK_CODE(code, lino, _end);

×

820

  SColumnInfoData infoData = createColumnInfoData(pColInfoData->info.type, pColInfoData->info.bytes, 1);

×

822

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

×

823

  QUERY_CHECK_CODE(code, lino, _end);

×

825

  code = colDataSetVal(&infoData, 0, (const char*)&vgId, false);

×

826

  QUERY_CHECK_CODE(code, lino, _end);

×

828

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

×

829

  SScalarParam param = {.columnData = pColInfoData};

×

831

  if (fpSet.process != NULL) {

×

832

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

×

833

    QUERY_CHECK_CODE(code, lino, _end);

×

835

    qError("failed to get the corresponding callback function, functionId:%d", functionId);

×

838

_end:

×

839

  colDataDestroy(&infoData);

×

840

  if (code != TSDB_CODE_SUCCESS) {

×

841

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

×

843

  return code;

×

846

int32_t setVgVerColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, int64_t vgVer) {

×

847

  int32_t                     code = TSDB_CODE_SUCCESS;

×

848

  int32_t                     lino = 0;

×

849

  struct SScalarFuncExecFuncs fpSet = {0};

×

850

  code = fmGetScalarFuncExecFuncs(functionId, &fpSet);

×

851

  QUERY_CHECK_CODE(code, lino, _end);

×

853

  SColumnInfoData infoData = createColumnInfoData(pColInfoData->info.type, pColInfoData->info.bytes, 1);

×

855

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

×

856

  QUERY_CHECK_CODE(code, lino, _end);

×

858

  code = colDataSetVal(&infoData, 0, (const char*)&vgVer, false);

×

859

  QUERY_CHECK_CODE(code, lino, _end);

×

861

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

×

862

  SScalarParam param = {.columnData = pColInfoData};

×

864

  if (fpSet.process != NULL) {

×

865

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

×

866

    QUERY_CHECK_CODE(code, lino, _end);

×

868

    qError("failed to get the corresponding callback function, functionId:%d", functionId);

×

871

_end:

×

872

  colDataDestroy(&infoData);

×

873

  if (code != TSDB_CODE_SUCCESS) {

×

874

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

×

876

  return code;

×

879

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

×

880

  int32_t code = TSDB_CODE_SUCCESS;

×

881

  int32_t lino = 0;

×

882

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

×

883

  QUERY_CHECK_CODE(code, lino, _end);

×

885

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

×

886

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

×

887

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

×

888

  taosMemoryFreeClear(pInfo->pResBlock->pBlockAgg);

×

890

  if (!pInfo->needCountEmptyTable) {

×

891

    pInfo->countState = TABLE_COUNT_STATE_END;

×

893

    pInfo->countState = TABLE_COUNT_STATE_SCAN;

×

896

_end:

×

897

  if (code != TSDB_CODE_SUCCESS) {

×

898

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

×

900

  return code;

×

903

void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {

×

904

  if (pInfo->countState == TABLE_COUNT_STATE_END) {

×

905

    return;

×

907

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

×

908

    pInfo->countState = TABLE_COUNT_STATE_PROCESSED;

×

910

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

×

911

    if (code != TSDB_CODE_SUCCESS) {

×

912

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

×

917

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

×

919

  blockDataEmpty(pBlock);

×

920

  pBlock->info.rows = 1;

×

921

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

×

922

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

×

923

  pBlock->info.blankFill = true;

×

926

  int32_t col_num = blockDataGetNumOfCols(pBlock);

×

927

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

×

928

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

×

933

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

×

935

  return pBlock;

×

938

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

×

939

  STableScanInfo* pTableScanInfo = pOperator->info;

×

941

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

×

943

  pOperator->resultInfo.totalRows++;

×

944

  return pBlock;

×

951

static int32_t prepareTableScan(SOperatorInfo* pOperator) {

×

952

  int32_t code = TSDB_CODE_SUCCESS;

×

953

  int32_t lino = 0;

×

955

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

×

956

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

×

957

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

×

958

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

×

959

         NOTIFY_TYPE_SCAN_PARAM &&

×

960

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

×

961

    STableScanOperatorParam* pNotify =

×

964

    STableScanInfo* pTableScanInfo = pOperator->info;

×

965

    SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

×

966

    SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

×

967

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

×

969

    QUERY_CHECK_CODE(code, lino, _end);

×

971

    if (!pGetParam->reUse) {

×

972

      freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

×

978

      pGetParam->reUse = false;

×

983

    pOperator->pOperatorGetParam = NULL;

×

986

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

×

987

_end:

×

988

  if (code != TSDB_CODE_SUCCESS) {

×

989

    qError("%s, %s failed at line %d since %s",

×

992

  return code;

×

995

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

×

996

  int32_t         code = TSDB_CODE_SUCCESS;

×

997

  int32_t         lino = 0;

×

998

  STableScanInfo* pTableScanInfo = pOperator->info;

×

999

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

×

1000

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

×

1001

  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;

×

1002

  bool            hasNext = false;

×

1003

  int64_t         st = taosGetTimestampUs();

×

1005

  QRY_PARAM_CHECK(ppRes);

×

1006

  pBlock->info.dataLoad = false;

×

1008

  while (true) {

×

1009

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

×

1010

    if (code != TSDB_CODE_SUCCESS) {

×

1011

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

×

1012

      QUERY_CHECK_CODE(code, lino, _end);

×

1015

    if (!hasNext) {

×

1016

      break;

×

1019

    if (isTaskKilled(pTaskInfo)) {

×

1020

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

×

1021

      code = pTaskInfo->code;

×

1022

      QUERY_CHECK_CODE(code, lino, _end);

×

1025

    if (pOperator->status == OP_EXEC_DONE) {

×

1026

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

×

1027

      break;

×

1031

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

×

1032

    if (!processThisBlock) {

×

1033

      continue;

×

1036

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

×

1037

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

×

1040

    uint32_t status = 0;

×

1041

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

×

1042

    QUERY_CHECK_CODE(code, lino, _end);

×

1044

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

×

1045

      break;

×

1049

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

×

1050

      continue;

×

1053

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

×

1054

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

×

1056

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

×

1057

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

×

1059

    (*ppRes) = pBlock;

×

1060

    return code;

×

1063

_end:

×

1064

  if (code != TSDB_CODE_SUCCESS) {

×

1065

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

×

1066

    pTaskInfo->code = code;

×

1067

    T_LONG_JMP(pTaskInfo->env, code);

×

1069

  return code;

×

1072

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

×

1073

  int32_t         code = TSDB_CODE_SUCCESS;

×

1074

  int32_t         lino = 0;

×

1075

  STableScanInfo* pTableScanInfo = pOperator->info;

×

1076

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

×

1077

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

×

1079

  QRY_PARAM_CHECK(pBlock);

×

1082

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

×

1083

    return code;

×

1087

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

×

1088

    SSDataBlock* p = NULL;

×

1089

    code = doTableScanImplNext(pOperator, &p);

×

1090

    QUERY_CHECK_CODE(code, lino, _end);

×

1092

    if (p != NULL) {

×

1093

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

×

1094

      *pBlock = p;

×

1095

      return code;

×

1098

    pTableScanInfo->scanTimes += 1;

×

1099

    taosHashClear(pTableScanInfo->pIgnoreTables);

×

1101

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

×

1102

      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

×

1103

      pTableScanInfo->base.scanFlag = MAIN_SCAN;

×

1104

      pTableScanInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

×

1105

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

×

1108

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

×

1109

      QUERY_CHECK_CODE(code, lino, _end);

×

1113

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

×

1114

  if (pTableScanInfo->scanTimes < total) {

×

1115

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

×

1116

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

×

1117

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

×

1118

      QUERY_CHECK_CODE(code, lino, _end);

×

1119

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

×

1122

    while (pTableScanInfo->scanTimes < total) {

×

1123

      SSDataBlock* p = NULL;

×

1124

      code = doTableScanImplNext(pOperator, &p);

×

1125

      QUERY_CHECK_CODE(code, lino, _end);

×

1127

      if (p != NULL) {

×

1128

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

×

1129

        *pBlock = p;

×

1130

        return code;

×

1133

      pTableScanInfo->scanTimes += 1;

×

1134

      taosHashClear(pTableScanInfo->pIgnoreTables);

×

1136

      if (pTableScanInfo->scanTimes < total) {

×

1137

        setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);

×

1138

        pTableScanInfo->base.scanFlag = MAIN_SCAN;

×

1140

        qDebug("%s start to repeat descending order scan data blocks", GET_TASKID(pTaskInfo));

×

1141

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

×

1142

        QUERY_CHECK_CODE(code, lino, _end);

×

1147

  if (pTableScanInfo->countState < TABLE_COUNT_STATE_END) {

×

1148

    STableListInfo* pTableListInfo = pTableScanInfo->base.pTableListInfo;

×

1149

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

×

1150

      if (pTableScanInfo->countState < TABLE_COUNT_STATE_PROCESSED) {

×

1151

        pTableScanInfo->countState = TABLE_COUNT_STATE_PROCESSED;

×

1153

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

×

1155

        if (NULL == pStart) {

×

1156

          return code;

×

1159

        *pBlock = getBlockForEmptyTable(pOperator, pStart);

×

1160

        return code;

×

1163

      int32_t numOfTables = 0;

×

1164

      code = tableListGetSize(pTableListInfo, &numOfTables);

×

1165

      QUERY_CHECK_CODE(code, lino, _end);

×

1167

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

×

1169

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

×

1170

        if (pIte != NULL) {

×

1171

          size_t        keySize = 0;

×

1172

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

×

1173

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

×

1174

          taosHashCancelIterate(pTableListInfo->remainGroups, pIte);

×

1175

          markGroupProcessed(pTableScanInfo, *pGroupId);

×

1176

          *pBlock = getBlockForEmptyTable(pOperator, &info);

×

1178

          return code;

×

1182

    pTableScanInfo->countState = TABLE_COUNT_STATE_END;

×

1185

_end:

×

1186

  if (code != TSDB_CODE_SUCCESS) {

×

1187

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

×

1188

    pTaskInfo->code = code;

×

1191

  return code;

×

1194

static int32_t createTableListInfoFromParam(SOperatorInfo* pOperator) {

×

1195

  STableScanInfo*          pInfo = pOperator->info;

×

1196

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1197

  int32_t                  code = 0;

×

1198

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

×

1199

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

×

1200

  int32_t                  num = taosArrayGetSize(pParam->pUidList);

×

1201

  if (num <= 0) {

×

1202

    qError("empty table scan uid list");

×

1203

    return TSDB_CODE_INVALID_PARA;

×

1206

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

×

1210

  if (pParam->tableSeq) {

×

1211

    pListInfo->oneTableForEachGroup = true;

×

1212

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

×

1213

      taosHashClear(pListInfo->map);

×

1214

      taosArrayClear(pListInfo->pTableList);

×

1215

      pOperator->status = OP_EXEC_DONE;

×

1218

    pListInfo->oneTableForEachGroup = false;

×

1219

    pListInfo->numOfOuputGroups = 1;

×

1221

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

×

1224

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

×

1225

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

×

1228

  STableKeyInfo info = {.groupId = 0};

×

1229

  int32_t       tableIdx = 0;

×

1230

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

×

1231

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

×

1232

    if (!pUid) {

×

1233

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

1234

      return terrno;

×

1237

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

×

1238

      if (TSDB_CODE_DUP_KEY == terrno) {

×

1239

        continue;

×

1241

      return terrno;

×

1244

    info.uid = *pUid;

×

1245

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

×

1246

    if (p == NULL) {

×

1247

      return terrno;

×

1250

    tableIdx++;

×

1251

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

×

1254

  return code;

×

1257

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

×

1259

  const char* idStr = GET_TASKID(pTaskInfo);

×

1260

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

×

1261

  if (code) {

×

1262

    qError("%s failed to init groupScan Info, code:%s at line:%d", idStr, tstrerror(code), __LINE__);

×

1263

    return code;

×

1266

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

×

1267

    qError("%s tsdb reader should be null", idStr);

×

1268

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1271

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

×

1272

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

×

1273

  if (code) {

×

1274

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

×

1277

  return code;

×

1280

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

×

1281

  SColIdPair* node1 = (SColIdPair*)elem1;

×

1282

  SColIdPair* node2 = (SColIdPair*)elem2;

×

1284

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

×

1285

    return 0;

×

1288

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

×

1291

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

×

1292

  SColIdSlotIdPair* node1 = (SColIdSlotIdPair*)elem1;

×

1293

  SColIdSlotIdPair* node2 = (SColIdSlotIdPair*)elem2;

×

1295

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

×

1296

    return 0;

×

1299

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

×

1302

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

×

1303

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

×

1304

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

×

1306

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

×

1307

    return 0;

×

1310

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

×

1313

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

×

1314

  if (pList == NULL) {

×

1315

    return NULL;

×

1318

  int32_t left = 0;

×

1319

  int32_t right = taosArrayGetSize(pList) - 1;

×

1320

  while (left <= right) {

×

1321

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

×

1322

    SColMatchItem* pItem = taosArrayGet(pList, mid);

×

1323

    if (pItem->colId == colId) {

×

1325

      while (mid > 0) {

×

1326

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

×

1327

        if (pPrev->colId != colId) {

×

1328

          break;

×

1330

        mid--;

×

1332

      return taosArrayGet(pList, mid);

×

1335

    if (pItem->colId < colId) {

×

1336

      left = mid + 1;

×

1338

      right = mid - 1;

×

1342

  return NULL;

×

1345

static bool isNewScanParam(STableScanOperatorParam* pParam) {

×

1346

  return pParam->isNewParam;

×

1349

static int32_t createVTableScanInfoFromBatchParam(SOperatorInfo* pOperator) {

×

1350

  int32_t                  code = 0;

×

1351

  int32_t                  lino = 0;

×

1352

  STableScanInfo*          pInfo = pOperator->info;

×

1353

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1354

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

×

1355

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

×

1356

  SMetaReader              orgTable = {0};

×

1357

  SMetaReader              superTable = {0};

×

1358

  SSchemaWrapper*          schema = NULL;

×

1359

  SArray*                  pColArray = NULL;

×

1360

  SArray*                  pBlockColArray = NULL;

×

1361

  int32_t                  num = 0;

×

1362

  STableKeyInfo*           pList = NULL;

×

1366

  cleanupQueryTableDataCond(&pInfo->base.cond);

×

1368

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

×

1369

  if (pGetParam &&

×

1370

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

×

1371

    STableScanOperatorParam* pParam =

×

1373

    pInfo->pBatchColMap = pParam->pBatchTbInfo;

×

1374

    pParam->pBatchTbInfo = NULL;

×

1375

    pInfo->cachedTimeWindow = pParam->window;

×

1376

    pInfo->currentBatchIdx = 0;

×

1377

    pInfo->lastBatchIdx = -1;

×

1378

    pInfo->cachedTagList = pParam->pTagList;

×

1379

    pParam->pTagList = NULL;

×

1380

    pInfo->cachedGroupId = pParam->groupid;

×

1382

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

×

1384

    taosHashClear(pInfo->pIgnoreTables);

×

1386

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

×

1387

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

×

1388

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

×

1389

        SqlFunctionCtx ctx = pSup->pCtx[i];

×

1390

        ctx.skipDynDataCheck = false;

×

1395

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

×

1396

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

×

1398

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

×

1399

    isNewTable = false;

×

1401

    isNewTable = true;

×

1402

    pInfo->lastBatchIdx = pInfo->currentBatchIdx;

×

1405

  if (isNewTable) {

×

1406

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

×

1407

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

×

1408

    pAPI->metaReaderFn.readerReleaseLock(&orgTable);

×

1409

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

×

1410

    QUERY_CHECK_CODE(code, lino, _return);

×

1411

    switch (orgTable.me.type) {

×

1412

      case TSDB_CHILD_TABLE:

×

1413

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

×

1414

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

×

1415

        pAPI->metaReaderFn.readerReleaseLock(&superTable);

×

1416

        QUERY_CHECK_CODE(code, lino, _return);

×

1417

        schema = &superTable.me.stbEntry.schemaRow;

×

1418

        break;

×

1419

      case TSDB_NORMAL_TABLE:

×

1420

        schema = &orgTable.me.ntbEntry.schemaRow;

×

1421

        break;

×

1422

      default:

×

1423

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

×

1424

        return TSDB_CODE_INVALID_PARA;

×

1428

    pListInfo->oneTableForEachGroup = true;

×

1429

    taosHashClear(pListInfo->map);

×

1430

    taosArrayClear(pListInfo->pTableList);

×

1432

    uint64_t      pUid = orgTable.me.uid;

×

1433

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

×

1434

    int32_t       tableIdx = 0;

×

1435

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

×

1436

    QUERY_CHECK_CODE(code, lino, _return);

×

1437

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

×

1438

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

×

1440

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

×

1441

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

×

1442

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

×

1443

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

×

1446

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

×

1447

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

×

1449

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

×

1450

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

×

1451

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

×

1452

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

×

1453

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

×

1454

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

×

1455

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

×

1457

          break;

×

1462

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

×

1463

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

×

1464

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

×

1465

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

×

1466

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

×

1467

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

×

1468

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

×

1469

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

×

1471

          break;

×

1476

    taosArraySort(pColArray, compareColIdPair);

×

1477

    taosArraySort(pBlockColArray, compareColIdSlotIdPair);

×

1478

    if (pInfo->pBlockColMap) {

×

1479

      taosArrayDestroy(pInfo->pBlockColMap);

×

1480

      pInfo->pBlockColMap = NULL;

×

1482

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

×

1483

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

×

1486

    taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

×

1487

    taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

×

1489

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

×

1490

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

×

1491

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

×

1492

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

×

1493

        bool           needScan = false;

×

1494

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

×

1495

          SFunctParam pParam = ctx->param[j];

×

1496

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

×

1497

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

×

1498

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

×

1499

              needScan |= true;

×

1500

              break;

×

1504

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

×

1505

        if (!needScan) {

×

1506

          ctx->skipDynDataCheck = true;

×

1508

          ctx->skipDynDataCheck = false;

×

1515

    pColArray = pInfo->lastColArray;

×

1516

    pBlockColArray = pInfo->lastBlockColArray;

×

1519

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

×

1520

  QUERY_CHECK_CODE(code, lino, _return);

×

1522

  if (pInfo->pResBlock) {

×

1523

    blockDataDestroy(pInfo->pResBlock);

×

1524

    pInfo->pResBlock = NULL;

×

1527

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

×

1535

  if (isNewTable) {

×

1536

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

×

1537

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

×

1538

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

×

1540

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

×

1541

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

×

1544

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

×

1545

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

×

1547

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

×

1551

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

×

1552

  pInfo->currentGroupId = 0;

×

1553

  pInfo->ignoreTag = true;

×

1555

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

×

1556

  QUERY_CHECK_CODE(code, lino, _return);

×

1558

  if (isNewTable) {

×

1559

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

×

1560

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

×

1562

    pInfo->base.dataReader = NULL;

×

1564

    taosRLockLatch(&pTaskInfo->lock);

×

1565

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

×

1566

    taosRUnLockLatch(&pTaskInfo->lock);

×

1567

    QUERY_CHECK_CODE(code, lino, _return);

×

1569

    pInfo->newReader = true;

×

1570

    if (pInfo->lastColArray) {

×

1571

      taosArrayDestroy(pInfo->lastColArray);

×

1572

      pInfo->lastColArray = NULL;

×

1574

    if (pInfo->lastBlockColArray) {

×

1575

      taosArrayDestroy(pInfo->lastBlockColArray);

×

1576

      pInfo->lastBlockColArray = NULL;

×

1578

    pInfo->lastColArray = pColArray;

×

1579

    pInfo->lastBlockColArray = pBlockColArray;

×

1581

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

×

1582

    QUERY_CHECK_CODE(code, lino, _return);

×

1584

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

×

1585

    pInfo->newReader = false;

×

1586

    pInfo->scanTimes = 0;

×

1589

  pOperator->status = OP_OPENED;

×

1591

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

×

1592

    pOperator->resultInfo.capacity = pInfo->pResBlock->info.capacity;

×

1595

  pInfo->currentGroupId = -1;

×

1596

_return:

×

1597

  if (code) {

×

1598

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

×

1600

  pAPI->metaReaderFn.clearReader(&superTable);

×

1601

  pAPI->metaReaderFn.clearReader(&orgTable);

×

1602

  return code;

×

1605

static int32_t createVTableScanInfoFromParam(SOperatorInfo* pOperator) {

×

1606

  int32_t                  code = 0;

×

1607

  int32_t                  lino = 0;

×

1608

  STableScanInfo*          pInfo = pOperator->info;

×

1609

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1610

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

×

1611

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

×

1612

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

×

1613

  SMetaReader              orgTable = {0};

×

1614

  SMetaReader              superTable = {0};

×

1615

  SSchemaWrapper*          schema = NULL;

×

1616

  SArray*                  pColArray = NULL;

×

1617

  SArray*                  pBlockColArray = NULL;

×

1618

  int32_t                  num = 0;

×

1619

  STableKeyInfo*           pList = NULL;

×

1621

  SArray*                  pMatchList = NULL;

×

1623

  cleanupQueryTableDataCond(&pInfo->base.cond);

×

1625

  pOrgTbInfo = pParam->pOrgTbInfo;

×

1627

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

×

1629

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

×

1630

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

×

1631

  pAPI->metaReaderFn.readerReleaseLock(&orgTable);

×

1632

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

×

1633

  QUERY_CHECK_CODE(code, lino, _return);

×

1634

  switch (orgTable.me.type) {

×

1635

    case TSDB_CHILD_TABLE:

×

1636

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

×

1637

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

×

1638

      pAPI->metaReaderFn.readerReleaseLock(&superTable);

×

1639

      QUERY_CHECK_CODE(code, lino, _return);

×

1640

      schema = &superTable.me.stbEntry.schemaRow;

×

1641

      break;

×

1642

    case TSDB_NORMAL_TABLE:

×

1643

      schema = &orgTable.me.ntbEntry.schemaRow;

×

1644

      break;

×

1645

    default:

×

1646

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

×

1647

      return TSDB_CODE_INVALID_PARA;

×

1651

  pListInfo->oneTableForEachGroup = true;

×

1652

  taosHashClear(pListInfo->map);

×

1653

  taosArrayClear(pListInfo->pTableList);

×

1655

  uint64_t      pUid = orgTable.me.uid;

×

1656

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

×

1657

  int32_t       tableIdx = 0;

×

1658

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

×

1659

  QUERY_CHECK_CODE(code, lino, _return);

×

1660

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

×

1661

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

×

1663

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

×

1664

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

×

1665

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

×

1666

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

×

1667

  if (pInfo->pBlockColMap) {

×

1668

    taosArrayDestroy(pInfo->pBlockColMap);

×

1669

    pInfo->pBlockColMap = NULL;

×

1671

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

×

1672

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

×

1675

  SColIdPair tsPair = {.vtbColId = PRIMARYKEY_TIMESTAMP_COL_ID,

×

1679

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

×

1681

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

×

1682

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

×

1683

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

×

1684

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

×

1685

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

×

1686

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

×

1687

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

×

1688

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

×

1689

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

×

1690

        break;

×

1695

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

×

1696

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

×

1697

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

×

1698

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

×

1699

      taosArraySort(pMatchList, compareColMatchItemByColId);

×

1701

      pMatchList = pInfo->base.matchInfo.pList;

×

1706

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

×

1707

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

×

1708

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

×

1709

    if (pItem) {

×

1710

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

×

1711

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

×

1713

        code = TSDB_CODE_VTABLE_COLUMN_TYPE_MISMATCH;

×

1714

        goto _return;

×

1716

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

×

1717

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

×

1718

      if (i > 0) {

×

1719

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

×

1724

  taosArraySort(pColArray, compareColIdPair);

×

1725

  taosArraySort(pBlockColArray, compareColIdSlotIdPair);

×

1726

  taosArraySort(pInfo->pBlockColMap, compareColIdSlotIdPair);

×

1728

  taosArrayRemoveDuplicate(pColArray, compareColIdPair, NULL);

×

1729

  taosArrayRemoveDuplicate(pBlockColArray, compareColIdSlotIdPair, NULL);

×

1731

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

×

1732

  QUERY_CHECK_CODE(code, lino, _return);

×

1734

  if (pInfo->pResBlock) {

×

1735

    blockDataDestroy(pInfo->pResBlock);

×

1736

    pInfo->pResBlock = NULL;

×

1739

  if (isNewScanParam(pParam)) {

×

1740

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

×

1741

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

×

1742

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

×

1744

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

×

1745

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

×

1748

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

×

1750

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

×

1751

  pInfo->currentGroupId = 0;

×

1752

  pInfo->ignoreTag = true;

×

1754

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

×

1755

  QUERY_CHECK_CODE(code, lino, _return);

×

1757

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

×

1758

  if (reader) {

×

1759

    if (isNewScanParam(pParam)) {

×

1760

      pAPI->tsdReader.tsdReaderClose(*reader);

×

1761

      pInfo->base.dataReader = NULL;

×

1763

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

×

1764

      QUERY_CHECK_CODE(code, lino, _return);

×

1766

      taosRLockLatch(&pTaskInfo->lock);

×

1767

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

×

1768

      taosRUnLockLatch(&pTaskInfo->lock);

×

1769

      QUERY_CHECK_CODE(code, lino, _return);

×

1771

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

×

1772

      QUERY_CHECK_CODE(code, lino, _return);

×

1773

      pInfo->newReader = true;

×

1775

      pInfo->base.dataReader = *reader;

×

1776

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

×

1777

      QUERY_CHECK_CODE(code, lino, _return);

×

1779

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

×

1780

      pInfo->newReader = false;

×

1781

      pInfo->scanTimes = 0;

×

1784

    pInfo->base.dataReader = NULL;

×

1786

    taosRLockLatch(&pTaskInfo->lock);

×

1787

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

×

1788

    taosRUnLockLatch(&pTaskInfo->lock);

×

1789

    QUERY_CHECK_CODE(code, lino, _return);

×

1791

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

×

1792

    QUERY_CHECK_CODE(code, lino, _return);

×

1793

    pInfo->newReader = true;

×

1795

  pOperator->status = OP_OPENED;

×

1797

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

×

1798

    pOperator->resultInfo.capacity = pInfo->pResBlock->info.capacity;

×

1801

  pInfo->currentGroupId = -1;

×

1802

_return:

×

1803

  if (code) {

×

1804

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

×

1806

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

×

1807

    taosArrayDestroy(pMatchList);

×

1809

  taosArrayDestroy(pColArray);

×

1810

  taosArrayDestroy(pBlockColArray);

×

1811

  pAPI->metaReaderFn.clearReader(&superTable);

×

1812

  pAPI->metaReaderFn.clearReader(&orgTable);

×

1813

  return code;

×

1816

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

×

1817

  int32_t         code = TSDB_CODE_SUCCESS;

×

1818

  int32_t         lino = 0;

×

1819

  STableScanInfo* pInfo = pOperator->info;

×

1820

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

×

1821

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

×

1822

  int32_t         numOfTables = 0;

×

1824

  QRY_PARAM_CHECK(pResult);

×

1826

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

×

1827

  QUERY_CHECK_CODE(code, lino, _end);

×

1829

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

×

1830

    setOperatorCompleted(pOperator);

×

1831

    if (pOperator->dynamicTask) {

×

1832

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

×

1833

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

×

1835

    return code;

×

1839

  pOperator->status = OP_OPENED;

×

1840

  resetLimitInfoForNextGroup(&pInfo->base.limitInfo);

×

1842

  int32_t        num = 0;

×

1843

  STableKeyInfo* pList = NULL;

×

1844

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

×

1845

  QUERY_CHECK_CODE(code, lino, _end);

×

1847

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

×

1848

  QUERY_CHECK_CODE(code, lino, _end);

×

1850

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

×

1851

  QUERY_CHECK_CODE(code, lino, _end);

×

1852

  pInfo->scanTimes = 0;

×

1854

  code = doGroupedTableScan(pOperator, pResult);

×

1855

  QUERY_CHECK_CODE(code, lino, _end);

×

1857

  if (*pResult != NULL) {

×

1858

    if (pOperator->dynamicTask) {

×

1859

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

×

1861

    return code;

×

1864

_end:

×

1865

  if (code != TSDB_CODE_SUCCESS) {

×

1866

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

×

1867

    pTaskInfo->code = code;

×

1869

  return code;

×

1872

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

×

1873

  int32_t         code = TSDB_CODE_SUCCESS;

×

1874

  int32_t         lino = 0;

×

1875

  STableScanInfo* pInfo = pOperator->info;

×

1876

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

×

1877

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

×

1878

  int32_t         num = 0;

×

1879

  STableKeyInfo*  pList = NULL;

×

1880

  SSDataBlock*    pResult = NULL;

×

1881

  const char*     idStr = GET_TASKID(pTaskInfo);

×

1883

  QRY_PARAM_CHECK(pResBlock);

×

1885

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

×

1886

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

×

1887

      setOperatorCompleted(pOperator);

×

1888

      return code;

×

1891

    taosRLockLatch(&pTaskInfo->lock);

×

1892

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

×

1893

    taosRUnLockLatch(&pTaskInfo->lock);

×

1894

    QUERY_CHECK_CODE(code, lino, _end);

×

1896

    if (pInfo->filesetDelimited) {

×

1897

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

×

1900

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

×

1901

      pOperator->resultInfo.capacity = pInfo->pResBlock->info.capacity;

×

1905

  code = doGroupedTableScan(pOperator, &pResult);

×

1906

  QUERY_CHECK_CODE(code, lino, _end);

×

1908

  if (pResult != NULL) {

×

1909

    if (pOperator->dynamicTask) {

×

1910

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

×

1913

    *pResBlock = pResult;

×

1914

    return code;

×

1918

    code = startNextGroupScan(pOperator, &pResult);

×

1919

    QUERY_CHECK_CODE(code, lino, _end);

×

1921

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

×

1922

      *pResBlock = pResult;

×

1923

      return code;

×

1927

_end:

×

1928

  if (code != TSDB_CODE_SUCCESS) {

×

1929

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

×

1930

    pTaskInfo->code = code;

×

1933

  return code;

×

1936

static bool isEmptyQueryTimeWindow(STimeWindow* pWindow) {

×

1937

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

×

1940

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

×

1941

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1942

  int32_t                  lino = 0;

×

1943

  STableScanInfo*          pInfo = pOperator->info;

×

1944

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1945

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

×

1946

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

×

1948

  code = createTableListInfoFromParam(pOperator);

×

1949

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

×

1950

  pOperator->pOperatorGetParam = NULL;

×

1951

  QUERY_CHECK_CODE(code, lino, _end);

×

1953

  if (pOperator->status == OP_EXEC_DONE) {

×

1954

    pInfo->currentGroupId = -1;

×

1955

    pOperator->status = OP_OPENED;

×

1956

    SSDataBlock* result = NULL;

×

1959

      code = startNextGroupScan(pOperator, &result);

×

1960

      QUERY_CHECK_CODE(code, lino, _end);

×

1962

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

×

1963

        (*ppRes) = result;

×

1964

        return code;

×

1968

_end:

×

1969

  if (code) {

×

1970

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

×

1972

  return code;

×

1975

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

×

1976

  int32_t      code = TSDB_CODE_SUCCESS;

×

1977

  int32_t      lino = 0;

×

1978

  bool         found = false;

×

1980

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

×

1981

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

×

1983

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

×

1984

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

×

1985

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

×

1986

    found = false;

×

1987

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

×

1988

      SColIdSlotIdPair* pColPair = taosArrayGet(pColMap, j);

×

1989

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

×

1990

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

×

1991

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

×

1992

        found = true;

×

1993

        break;

×

1996

    if (!found) {

×

1997

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

×

1998

        pSrcCol->info.slotId = -1;

×

2000

        qError("cannot find vstb slotId for org colId:%d (pColMap size:%d)", pSrcCol->info.colId,

×

2002

        code = TSDB_CODE_QRY_VSTB_SLOTID_NOT_FOUND;

×

2007

  return code;

×

2008

_return:

×

2009

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

×

2010

  return code;

×

2013

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

×

2014

  int32_t                  code = TSDB_CODE_SUCCESS;

×

2015

  int32_t                  lino = 0;

×

2016

  STableScanInfo*          pInfo = pOperator->info;

×

2017

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

2018

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

×

2019

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

×

2021

  QRY_PARAM_CHECK(ppRes);

×

2022

  qTrace("%s call", __FUNCTION__);

×

2024

  code = createVTableScanInfoFromParam(pOperator);

×

2025

  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

×

2026

  pOperator->pOperatorGetParam = NULL;

×

2028

  QUERY_CHECK_CODE(code, lino, _end);

×

2030

  SSDataBlock* result = NULL;

×

2031

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

×

2032

    (*ppRes) = result;

×

2033

    return code;

×

2036

  if (pInfo->newReader) {

×

2037

    code = startNextGroupScan(pOperator, &result);

×

2038

    QUERY_CHECK_CODE(code, lino, _end);

×

2040

    code = doGroupedTableScan(pOperator, &result);

×

2041

    QUERY_CHECK_CODE(code, lino, _end);

×

2044

  if (result) {

×

2045

    SSDataBlock* res = NULL;

×

2046

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

×

2047

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

×

2048

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

×

2050

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

×

2051

    QUERY_CHECK_CODE(code, lino, _end);

×

2057

    (*ppRes) = result;

×

2059

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

×

2060

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

×

2062

    blockDataDestroy(pInfo->pResBlock);

×

2063

    pInfo->pResBlock = NULL;

×

2065

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

×

2066

    if (reader) {

×

2067

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

×

2068

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

×

2069

        pInfo->base.dataReader = NULL;

×

2071

      if (pAPI->tsdReader.tsdReaderClose) {

×

2072

        pAPI->tsdReader.tsdReaderClose(*reader);

×

2076

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

×

2077

    QUERY_CHECK_CODE(code, lino, _end);

×

2079

_end:

×

2080

  if (code) {

×

2081

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

×

2083

  return code;

×

2086

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

×

2087

  int32_t                  code = TSDB_CODE_SUCCESS;

×

2088

  int32_t                  lino = 0;

×

2089

  STableScanInfo*          pInfo = pOperator->info;

×

2090

  int32_t                  index = 0;

×

2091

  char*                    tagVal = NULL;

×

2093

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

×

2094

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

×

2095

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

×

2097

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

×

2098

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

×

2100

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

×

2101

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

×

2102

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

×

2103

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

×

2105

        varDataSetLen(tagVal, pTagVal->nData);

×

2106

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

×

2107

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

×

2108

        QUERY_CHECK_CODE(code, lino, _end);

×

2109

        taosMemoryFreeClear(tagVal);

×

2111

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

×

2112

        QUERY_CHECK_CODE(code, lino, _end);

×

2115

    index++;

×

2117

  return code;

×

2118

_end:

×

2119

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

×

2120

  taosMemoryFreeClear(tagVal);

×

2121

  return code;

×

2124

static void clearVstbBatchDynamicTableScanInfo(STableScanInfo* pInfo) {

×

2125

  if (pInfo->pBatchColMap) {

×

2126

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

×

2127

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

×

2128

      taosArrayDestroy(pOrgTbInfo->colMap);

×

2130

    taosArrayDestroy(pInfo->pBatchColMap);

×

2131

    pInfo->pBatchColMap = NULL;

×

2133

  if (pInfo->cachedTagList) {

×

2134

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

×

2135

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

×

2136

      if (IS_VAR_DATA_TYPE(pTagVal->type)) {

×

2137

        taosMemoryFreeClear(pTagVal->pData);

×

2140

    taosArrayDestroy(pInfo->cachedTagList);

×

2141

    pInfo->cachedTagList = NULL;

×

2143

  if (pInfo->lastColArray) {

×

2144

    taosArrayDestroy(pInfo->lastColArray);

×

2145

    pInfo->lastColArray = NULL;

×

2147

  if (pInfo->lastBlockColArray) {

×

2148

    taosArrayDestroy(pInfo->lastBlockColArray);

×

2149

    pInfo->lastBlockColArray = NULL;

×

2152

  blockDataDestroy(pInfo->pResBlock);

×

2153

  pInfo->pResBlock = NULL;

×

2154

×

2156

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

×

2157

  int32_t                  code = TSDB_CODE_SUCCESS;

×

2158

  int32_t                  lino = 0;

×

2159

  STableScanInfo*          pInfo = pOperator->info;

×

2160

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

2161

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

×

2162

  SSDataBlock*             result = NULL;

×

2164

  QRY_PARAM_CHECK(ppRes);

×

2165

  qTrace("%s call", __FUNCTION__);

×

2167

  while (!result) {

×

2168

    code = createVTableScanInfoFromBatchParam(pOperator);

×

2169

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

×

2170

    pOperator->pOperatorGetParam = NULL;

×

2171

    QUERY_CHECK_CODE(code, lino, _end);

×

2173

    if (pInfo->newReader) {

×

2174

      code = startNextGroupScan(pOperator, &result);

×

2175

      QUERY_CHECK_CODE(code, lino, _end);

×

2177

      code = doGroupedTableScan(pOperator, &result);

×

2178

      QUERY_CHECK_CODE(code, lino, _end);

×

2180

    if (!result) {

×

2181

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

×

2183

      pInfo->currentBatchIdx++;

×

2184

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

×

2186

        break;

×

2191

  if (result) {

×

2192

    SSDataBlock* res = NULL;

×

2193

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

×

2194

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

×

2195

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

×

2197

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

×

2198

    QUERY_CHECK_CODE(code, lino, _end);

×

2200

    if (pInfo->cachedTagList) {

×

2201

      code = setTagValFromTagList(pOperator, res);

×

2202

      QUERY_CHECK_CODE(code, lino, _end);

×

2205

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

×

2206

    pInfo->pResBlock = res;

×

2207

    blockDataDestroy(result);

×

2208

    (*ppRes) = res;

×

2210

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

×

2212

    clearVstbBatchDynamicTableScanInfo(pInfo);

×

2213

    setOperatorCompleted(pOperator);

×

2216

_end:

×

2217

  return code;

×

2220

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

×

2221

  int32_t                  code = TSDB_CODE_SUCCESS;

×

2222

  int32_t                  lino = 0;

×

2223

  STableScanInfo*          pInfo = pOperator->info;

×

2224

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

2225

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

×

2227

  QRY_PARAM_CHECK(ppRes);

×

2228

  qTrace("%s call", __FUNCTION__);

×

2230

  pOperator->dynamicTask = true;

×

2232

  switch (type) {

×

2233

    case DYN_TYPE_VSTB_WIN_SCAN:

×

2235

      code = doStbJoinDynamicTableScanNext(pOperator, ppRes);

×

2236

      QUERY_CHECK_CODE(code, lino, _end);

×

2237

      break;

×

2239

    case DYN_TYPE_VSTB_SINGLE_SCAN: {

×

2240

      code = doVstbSingleDynamicTableScanNext(pOperator, ppRes);

×

2241

      QUERY_CHECK_CODE(code, lino, _end);

×

2242

      break;

×

2244

    case DYN_TYPE_VSTB_BATCH_SCAN:

×

2245

      code = doVstbBatchDynamicTableScanNext(pOperator, ppRes);

×

2246

      QUERY_CHECK_CODE(code, lino, _end);

×

2247

      break;

×

2248

    default:

×

2249

      code = TSDB_CODE_INVALID_PARA;

×

2250

      QUERY_CHECK_CODE(code, lino, _end);

×

2251

      break;

×

2254

_end:

×

2255

  return code;

×

2258

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

×

2259

  int32_t         code = TSDB_CODE_SUCCESS;

×

2260

  int32_t         lino = 0;

×

2261

  STableScanInfo* pInfo = pOperator->info;

×

2262

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

×

2263

  SStorageAPI*    pAPI = &pTaskInfo->storageAPI;

×

2264

  QRY_PARAM_CHECK(ppRes);

×

2265

  qTrace("%s call", __FUNCTION__);

×

2267

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

×

2268

  QUERY_CHECK_CODE(code, lino, _end);

×

2270

  SOperatorParam* pGetParam = pOperator->pOperatorGetParam;

×

2271

  if (pGetParam || pInfo->pBatchColMap) {

×

2272

    ETableScanDynType type = pGetParam ?

×

2273

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

×

2276

    code = doDynamicTableScanNext(pOperator, type, ppRes);

×

2277

    QUERY_CHECK_CODE(code, lino, _end);

×

2279

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

×

2280

      return code;

×

2285

  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {

×

2286

    int32_t       numOfTables = 0;

×

2287

    STableKeyInfo tInfo = {0};

×

2288

    pInfo->countState = TABLE_COUNT_STATE_END;

×

2290

    while (1) {

×

2291

      SSDataBlock* result = NULL;

×

2292

      code = doGroupedTableScan(pOperator, &result);

×

2293

      QUERY_CHECK_CODE(code, lino, _end);

×

2295

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

×

2296

        (*ppRes) = result;

×

2297

        return code;

×

2301

      pInfo->currentTable++;

×

2303

      taosRLockLatch(&pTaskInfo->lock);

×

2304

      numOfTables = 0;

×

2305

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

×

2306

      if (code != TSDB_CODE_SUCCESS) {

×

2307

        taosRUnLockLatch(&pTaskInfo->lock);

×

2308

        TSDB_CHECK_CODE(code, lino, _end);

×

2311

      if (pInfo->currentTable >= numOfTables) {

×

2312

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

×

2313

        taosRUnLockLatch(&pTaskInfo->lock);

×

2314

        (*ppRes) = NULL;

×

2315

        return code;

×

2318

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

×

2319

      if (!tmp) {

×

2320

        taosRUnLockLatch(&pTaskInfo->lock);

×

2321

        (*ppRes) = NULL;

×

2322

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

×

2325

      tInfo = *tmp;

×

2326

      taosRUnLockLatch(&pTaskInfo->lock);

×

2328

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

×

2329

      QUERY_CHECK_CODE(code, lino, _end);

×

2330

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

×

2333

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

×

2334

      QUERY_CHECK_CODE(code, lino, _end);

×

2335

      pInfo->scanTimes = 0;

×

2338

    code = groupSeqTableScan(pOperator, ppRes);

×

2339

    QUERY_CHECK_CODE(code, lino, _end);

×

2342

_end:

×

2343

  if (code != TSDB_CODE_SUCCESS) {

×

2344

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

×

2345

    pTaskInfo->code = code;

×

2346

    T_LONG_JMP(pTaskInfo->env, code);

×

2349

  return code;

×

2352

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

×

2353

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

×

2354

  if (!pRecorder) {

×

2355

    return terrno;

×

2357

  STableScanInfo* pTableScanInfo = pOptr->info;

×

2358

  *pRecorder = pTableScanInfo->base.readRecorder;

×

2359

  *pOptrExplain = pRecorder;

×

2360

  *len = sizeof(SFileBlockLoadRecorder);

×

2361

  return 0;

×

2364

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

×

2365

  cleanupQueryTableDataCond(&pBase->cond);

×

2366

  cleanupQueryTableDataCond(&pBase->orgCond);

×

2368

  if (pAPI->tsdReaderClose) {

×

2369

    pAPI->tsdReaderClose(pBase->dataReader);

×

2371

  pBase->dataReader = NULL;

×

2373

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

×

2374

    taosArrayDestroy(pBase->matchInfo.pList);

×

2377

  tableListDestroy(pBase->pTableListInfo);

×

2378

  taosLRUCacheCleanup(pBase->metaCache.pTableMetaEntryCache);

×

2379

  cleanupExprSupp(&pBase->pseudoSup);

×

2380

×

2382

static void cleanReaderForVTable(STableScanInfo* pInfo){

×

2383

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

×

2384

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

×

2385

    while (pIter != NULL) {

×

2386

      void **reader = pIter;

×

2387

      if (*reader) {

×

2388

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

×

2389

          pInfo->base.dataReader = NULL;

×

2391

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

×

2393

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

×

2396

×

2398

static void destroyTableScanOperatorInfo(void* param) {

×

2399

  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;

×

2400

  blockDataDestroy(pTableScanInfo->pResBlock);

×

2401

  blockDataDestroy(pTableScanInfo->pOrgBlock);

×

2402

  taosHashCleanup(pTableScanInfo->pIgnoreTables);

×

2403

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

×

2404

    cleanReaderForVTable(pTableScanInfo);

×

2405

    taosHashCleanup(pTableScanInfo->readerCache);

×

2407

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

×

2408

  taosArrayDestroy(pTableScanInfo->pBlockColMap);

×

2409

  taosMemoryFreeClear(param);

×

2410

×

2412

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

×

2413

  if (pBlock && dataRequireFlag == FUNC_DATA_REQUIRED_NOT_LOAD) {

×

2414

    int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

×

2415

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

×

2416

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

×

2417

      if (pCol) {

×

2418

        pCol->info.noData = true;

×

2422

×

2424

static int32_t resetTableScanOperatorState(SOperatorInfo* pOper) {

×

2425

  int32_t         code = TSDB_CODE_SUCCESS;

×

2426

  STableScanInfo*   pInfo = pOper->info;

×

2427

  pOper->status = OP_NOT_OPENED;

×

2429

  pInfo->scanTimes = 0;

×

2430

  pInfo->currentGroupId = -1;

×

2431

  pInfo->tableEndIndex = -1;

×

2432

  pInfo->tableStartIndex = 0;

×

2433

  pInfo->currentTable = 0;

×

2434

  pInfo->scanMode = 0;

×

2435

  pInfo->countState = 0;

×

2436

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

×

2438

  if (!pInfo->virtualStableScan) {

×

2444

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

×

2445

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

×

2447

    pInfo->base.dataReader = NULL;

×

2449

  tableListDestroy(pInfo->base.pTableListInfo);

×

2450

  pInfo->base.pTableListInfo = tableListCreate();

×

2451

  if (!pInfo->base.pTableListInfo) {

×

2452

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2453

    return terrno;

×

2455

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

×

2457

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

×

2458

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

×

2459

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

×

2461

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

×

2462

    if (code) {

×

2463

      qError("%s failed to createScanTableListInfo, code:%s", __func__, tstrerror(code));

×

2464

      return code;

×

2468

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

×

2469

  cleanupQueryTableDataCond(&pInfo->base.cond);

×

2470

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

×

2471

  if (code) {

×

2472

    qError("%s failed to initQueryTableDataCond, code:%s", __func__, tstrerror(code));

×

2473

    return code;

×

2475

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

×

2476

    cleanupQueryTableDataCond(&pInfo->base.orgCond);

×

2477

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

×

2478

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

×

2481

  pOper->resultInfo.totalRows = 0;

×

2482

  blockDataEmpty(pInfo->pResBlock);

×

2483

  blockDataEmpty(pInfo->pOrgBlock);

×

2484

  taosHashClear(pInfo->pIgnoreTables);

×

2485

  taosArrayDestroy(pInfo->pBlockColMap);

×

2486

  pInfo->pBlockColMap = NULL;

×

2487

  return code;

×

2490

int32_t createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

×

2493

  QRY_PARAM_CHECK(pOptrInfo);

×

2495

  int32_t         code = TSDB_CODE_SUCCESS;

×

2496

  int32_t         lino = 0;

×

2497

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

×

2498

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

×

2499

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

×

2500

    code = terrno;

×

2501

    goto _error;

×

2504

  pOperator->pPhyNode = pTableScanNode;

×

2505

  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;

×

2506

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

×

2508

  int32_t numOfCols = 0;

×

2510

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

×

2511

  QUERY_CHECK_CODE(code, lino, _error);

×

2513

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

×

2514

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

×

2515

  QUERY_CHECK_CODE(code, lino, _error);

×

2517

  if (pScanNode->pScanPseudoCols != NULL) {

×

2518

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

×

2519

    pSup->pExprInfo = NULL;

×

2520

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

×

2521

    QUERY_CHECK_CODE(code, lino, _error);

×

2523

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

×

2525

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

×

2528

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

×

2529

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

×

2531

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

×

2532

  pInfo->base.readHandle = *readHandle;

×

2533

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

×

2535

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

×

2536

  pInfo->sample.seed = taosGetTimestampSec();

×

2538

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

×

2539

  initResultSizeInfo(&pOperator->resultInfo, 4096);

×

2540

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

×

2541

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

×

2542

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

×

2544

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

×

2545

  QUERY_CHECK_CODE(code, lino, _error);

×

2547

  pInfo->virtualStableScan = pScanNode->virtualStableScan;

×

2548

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

×

2549

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

×

2550

    pInfo->pResBlock = NULL;

×

2551

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

×

2552

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

×

2555

  if (pInfo->virtualStableScan) {

×

2556

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

×

2557

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

×

2560

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

×

2561

                            pTaskInfo->pStreamRuntimeInfo);

×

2562

  QUERY_CHECK_CODE(code, lino, _error);

×

2564

  pInfo->currentGroupId = -1;

×

2566

  pInfo->tableEndIndex = -1;

×

2567

  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;

×

2568

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

×

2570

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

×

2572

  pOperator->exprSupp.numOfExprs = numOfCols;

×

2574

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

×

2575

  pInfo->ignoreTag = false;

×

2577

  pInfo->lastBatchIdx = -1;

×

2578

  pInfo->currentBatchIdx = 0;

×

2579

  pInfo->base.pTableListInfo = pTableListInfo;

×

2580

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

×

2581

    pInfo->base.metaCache.pTableMetaEntryCache = taosLRUCacheInit(tsMetaEntryCacheSize, -1, .5);

×

2582

    if (pInfo->base.metaCache.pTableMetaEntryCache == NULL) {

×

2583

      code = terrno;

×

2584

      QUERY_CHECK_CODE(code, lino, _error);

×

2587

    taosLRUCacheSetStrictCapacity(pInfo->base.metaCache.pTableMetaEntryCache, false);

×

2590

  pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

×

2592

  pOperator->fpSet = createOperatorFpSet(prepareTableScan, doTableScanNext,

×

2598

  setOperatorResetStateFn(pOperator, resetTableScanOperatorState);

×

2600

  pOperator->cost.openCost = 0;

×

2601

  *pOptrInfo = pOperator;

×

2602

  return TSDB_CODE_SUCCESS;

×

2604

_error:

×

2605

  if (pInfo != NULL) {

×

2606

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

×

2607

    destroyTableScanOperatorInfo(pInfo);

×

2610

  if (pOperator != NULL) {

×

2611

    pOperator->info = NULL;

×

2612

    destroyOperator(pOperator);

×

2614

  pTaskInfo->code = code;

×

2615

  return code;

×

2618

int32_t createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {

×

2619

  QRY_PARAM_CHECK(pOptrInfo);

×

2621

  int32_t         code = 0;

×

2622

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

×

2623

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

×

2624

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

×

2625

    code = terrno;

×

2626

    goto _end;

×

2629

  pInfo->base.dataReader = pReadHandle;

×

2632

  setOperatorInfo(pOperator, "TableSeqScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN, false, OP_NOT_OPENED,

×

2634

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableScanImplNext, NULL, NULL, optrDefaultBufFn, NULL,

×

2636

  *pOptrInfo = pOperator;

×

2637

  return code;

×

2639

_end:

×

2640

  if (pInfo != NULL) {

×

2641

    taosMemoryFree(pInfo);

×

2644

  if (pOperator != NULL) {

×

2645

    taosMemoryFree(pOperator);

×

2648

  pTaskInfo->code = code;

×

2649

  return code;

×

2652

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

×

2653

  int32_t code = TSDB_CODE_SUCCESS;

×

2654

  int32_t lino = 0;

×

2655

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

×

2656

    return code;

×

2658

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

×

2659

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

×

2660

  bool hasUnqualified = false;

×

2662

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

×

2663

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

×

2665

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

×

2667

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

×

2670

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

×

2671

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

×

2672

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

×

2673

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

×

2674

    void*    data = colDataGetData(pColPk, i);

×

2675

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

×

2676

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

×

2677

        QUERY_CHECK_CODE(code = TSDB_CODE_BLOB_NOT_SUPPORT_PRIMARY_KEY, lino, _end);

×

2679

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

×

2680

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

×

2681

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

×

2682

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

×

2683

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

×

2684

      taosMemoryFree(tmq);

×

2686

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

×

2689

    if (!p[i]) {

×

2690

      hasUnqualified = true;

×

2694

  if (hasUnqualified) {

×

2695

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

×

2696

    QUERY_CHECK_CODE(code, lino, _end);

×

2699

_end:

×

2700

  taosMemoryFree(p);

×

2701

  if (code != TSDB_CODE_SUCCESS) {

×

2702

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

×

2704

  return code;

×

2707

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

×

2708

  int32_t p1 = *(int32_t*)param1;

×

2709

  int32_t p2 = *(int32_t*)param2;

×

2711

  if (p1 == p2) {

×

2712

    return 0;

×

2714

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

×

2718

static int32_t processBlock(STmqQueryScanInfo* pInfo) {

×

2719

  int32_t         code = TSDB_CODE_SUCCESS;

×

2720

  int32_t         lino = 0;

×

2721

  SOperatorInfo*  pOperator = pInfo->pTmqScanOp;

×

2723

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

×

2724

  QUERY_CHECK_CODE(code, lino, _end);

×

2726

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

×

2727

  QUERY_CHECK_CODE(code, lino, _end);

×

2729

_end:

×

2730

  if (code != TSDB_CODE_SUCCESS) {

×

2731

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

×

2733

  return code;

×

2736

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

×

2737

  int32_t code = TSDB_CODE_SUCCESS;

×

2738

  int32_t lino = 0;

×

2739

  SValue  val = {0};

×

2740

  if (hasPrimaryKey) {

×

2741

    code = doBlockDataPrimaryKeyFilter(pBlock, offset);

×

2742

    if (code != TSDB_CODE_SUCCESS) {

×

2743

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

×

2744

      return code;

×

2746

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

×

2748

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

×

2749

      return code;

×

2751

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

×

2752

    val.type = pColPk->info.type;

×

2753

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

×

2754

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

×

2755

        return TSDB_CODE_BLOB_NOT_SUPPORT_PRIMARY_KEY;

×

2757

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

×

2758

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

×

2759

      val.nData = varDataLen(tmp);

×

2760

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

×

2762

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

×

2765

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

×

2767

_end:

×

2768

  if (code != TSDB_CODE_SUCCESS) {

×

2769

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

×

2771

  return code;

×

2774

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

×

2775

  int32_t        code = TSDB_CODE_SUCCESS;

×

2776

  int32_t        lino = 0;

×

2777

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

2778

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

2780

  STmqQueryScanInfo* pInfo = pOperator->info;

×

2781

  const char*      id = GET_TASKID(pTaskInfo);

×

2783

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

×

2785

  if (isTaskKilled(pTaskInfo)) {

×

2786

    (*ppRes) = NULL;

×

2787

    return pTaskInfo->code;

×

2790

  blockDataCleanup(pInfo->pRes);

×

2791

  int64_t st = taosGetTimestampMs();

×

2793

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

×

2794

    while (1) {

×

2795

      SSDataBlock* pResult = NULL;

×

2796

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

×

2797

      QUERY_CHECK_CODE(code, lino, _end);

×

2799

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

×

2800

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

×

2801

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

×

2802

        QUERY_CHECK_CODE(code, lino, _end);

×

2803

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

×

2804

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

×

2805

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

×

2806

          QUERY_CHECK_CODE(code, lino, _end);

×

2812

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

×

2813

        (*ppRes) = pInfo->pRes;

×

2814

        return code;

×

2816

      int64_t elapsed = taosGetTimestampMs() - st;

×

2817

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

×

2818

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

×

2819

          terrno = TSDB_CODE_TMQ_FETCH_TIMEOUT ;

×

2821

        (*ppRes) = pInfo->pRes;

×

2822

        return code;

×

2826

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

×

2827

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

×

2829

    pTSInfo->base.dataReader = NULL;

×

2830

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

×

2831

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

×

2832

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

×

2833

    if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) {

×

2834

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

×

2835

        (*ppRes) = NULL;

×

2837

        (*ppRes) = pInfo->pRes;

×

2839

      return 0;

×

2841

    QUERY_CHECK_CODE(code, lino, _end);

×

2843

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

×

2846

  while (1) {

×

2847

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

×

2851

                                                              pTaskInfo->tmqInfo.sourceExcluded,

×

2854

                                                              pTaskInfo->tmqInfo.enableReplay);

×

2855

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

×

2856

      goto _end;

×

2858

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

×

2861

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

×

2864

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

×

2866

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

×

2868

    int32_t ret = processBlock(pInfo);

×

2869

    if (ret != 0) {

×

2870

      code = ret;

×

2871

      goto _end;

×

2873

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

×

2876

    if (code != 0) {

×

2877

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

×

2878

        (*ppRes) = NULL;

×

2880

        (*ppRes) = pInfo->pRes;

×

2882

      return 0;

×

2885

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

×

2886

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

×

2887

      (*ppRes) = pInfo->pRes;

×

2888

      return 0;

×

2892

_end:

×

2893

  if (code != TSDB_CODE_SUCCESS) {

×

2894

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

×

2895

    pTaskInfo->code = code;

×

2896

    T_LONG_JMP(pTaskInfo->env, code);

×

2898

  (*ppRes) = NULL;

×

2899

  return code;

×

2902

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

×

2903

  int32_t code = TSDB_CODE_SUCCESS;

×

2904

  int32_t lino = 0;

×

2905

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

×

2906

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

×

2909

  int32_t size = 0;

×

2910

  code = tableListGetSize(pTableListInfo, &size);

×

2911

  QUERY_CHECK_CODE(code, lino, _end);

×

2912

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

×

2913

    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);

×

2914

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

×

2915

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

×

2916

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

×

2919

  (*ppArrayRes) = tableIdList;

×

2921

_end:

×

2922

  if (code != TSDB_CODE_SUCCESS) {

×

2923

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

×

2925

  return code;

×

2928

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

×

2929

  int32_t        code = TSDB_CODE_SUCCESS;

×

2930

  int32_t        lino = 0;

×

2931

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

2932

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

2934

  STmqRawScanInfo* pInfo = pOperator->info;

×

2935

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

×

2936

  pTaskInfo->tmqInfo.btMetaRsp.batchMetaLen = NULL;

×

2938

  qDebug("tmqsnap doRawScan called");

×

2939

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

×

2940

    bool hasNext = false;

×

2941

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

×

2942

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

×

2943

      if (code != TSDB_CODE_SUCCESS) {

×

2944

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

×

2945

        QUERY_CHECK_CODE(code, lino, _end);

×

2949

    if (pInfo->dataReader && hasNext) {

×

2950

      if (isTaskKilled(pTaskInfo)) {

×

2951

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

×

2952

        return code;

×

2955

      SSDataBlock* pBlock = NULL;

×

2956

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

×

2957

      QUERY_CHECK_CODE(code, lino, _end);

×

2959

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

×

2960

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

×

2961

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

×

2962

        QUERY_CHECK_CODE(code, lino, _end);

×

2963

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

×

2964

        (*ppRes) = pBlock;

×

2965

        return code;

×

2969

    SMetaTableInfo mtInfo = {0};

×

2970

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

×

2971

    QUERY_CHECK_CODE(code, lino, _end);

×

2972

    if (code != 0) {

×

2974

      QUERY_CHECK_CODE(code, lino, _end);

×

2976

    STqOffsetVal offset = {0};

×

2977

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

×

2978

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

×

2979

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

×

2981

      SValue val = {0};

×

2982

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

×

2983

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

×

2986

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

×

2987

    QUERY_CHECK_CODE(code, lino, _end);

×

2988

    (*ppRes) = NULL;

×

2989

    return code;

×

2990

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

×

2991

    SSnapContext* sContext = pInfo->sContext;

×

2992

    int64_t st = taosGetTimestampMs();

×

2993

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

×

2994

      void*   data = NULL;

×

2995

      int32_t dataLen = 0;

×

2996

      int16_t type = 0;

×

2997

      int64_t uid = 0;

×

2998

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

×

2999

        qError("tmqsnap getTableInfoFromSnapshot error");

×

3000

        taosMemoryFreeClear(data);

×

3001

        break;

×

3004

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

×

3005

        STqOffsetVal offset = {0};

×

3006

        SValue       val = {0};

×

3007

        tqOffsetResetToData(&offset, 0, INT64_MIN, val);

×

3008

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

×

3009

        QUERY_CHECK_CODE(code, lino, _end);

×

3010

        break;

×

3012

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

×

3013

        SMqMetaRsp tmpMetaRsp = {0};

×

3014

        tmpMetaRsp.resMsgType = type;

×

3015

        tmpMetaRsp.metaRspLen = dataLen;

×

3016

        tmpMetaRsp.metaRsp = data;

×

3017

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

×

3018

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

×

3019

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

×

3021

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

×

3022

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

×

3024

        int32_t  tempRes = TSDB_CODE_SUCCESS;

×

3025

        uint32_t len = 0;

×

3026

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

×

3027

        if (TSDB_CODE_SUCCESS != tempRes) {

×

3028

          qError("tmqsnap tEncodeMqMetaRsp error");

×

3029

          taosMemoryFreeClear(data);

×

3030

          break;

×

3033

        int32_t tLen = sizeof(SMqRspHead) + len;

×

3034

        void*   tBuf = taosMemoryCalloc(1, tLen);

×

3035

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

×

3037

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

×

3038

        SEncoder encoder = {0};

×

3039

        tEncoderInit(&encoder, metaBuff, len);

×

3040

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

×

3041

        if (tempLen < 0) {

×

3042

          qError("tmqsnap tEncodeMqMetaRsp error");

×

3043

          tEncoderClear(&encoder);

×

3044

          taosMemoryFreeClear(tBuf);

×

3045

          taosMemoryFreeClear(data);

×

3046

          break;

×

3048

        taosMemoryFreeClear(data);

×

3049

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

×

3050

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

×

3052

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

×

3053

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

×

3055

        int64_t elapsed = taosGetTimestampMs() - st;

×

3056

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

×

3062

    (*ppRes) = NULL;

×

3063

    return code;

×

3066

_end:

×

3067

  if (code != TSDB_CODE_SUCCESS) {

×

3068

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

×

3069

    pTaskInfo->code = code;

×

3070

    T_LONG_JMP(pTaskInfo->env, code);

×

3073

  (*ppRes) = NULL;

×

3074

  return code;

×

3077

static void destroyTmqRawScanOperatorInfo(void* param) {

×

3078

  STmqRawScanInfo* pRawScan = (STmqRawScanInfo*)param;

×

3079

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

×

3080

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

×

3081

  tableListDestroy(pRawScan->pTableListInfo);

×

3082

  taosMemoryFree(pRawScan);

×

3083

×

3088

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

×

3094

  QRY_PARAM_CHECK(pOptrInfo);

×

3095

  int32_t code = TSDB_CODE_SUCCESS;

×

3096

  int32_t lino = 0;

×

3098

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

×

3099

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

×

3100

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

×

3101

    code = terrno;

×

3102

    lino = __LINE__;

×

3103

    goto _end;

×

3106

  pInfo->pTableListInfo = tableListCreate();

×

3107

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

×

3108

  pInfo->vnode = pHandle->vnode;

×

3109

  pInfo->pAPI = &pTaskInfo->storageAPI;

×

3111

  pInfo->sContext = pHandle->sContext;

×

3112

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

×

3113

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

×

3116

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

×

3118

  *pOptrInfo = pOperator;

×

3119

  return code;

×

3121

_end:

×

3122

  if (code != TSDB_CODE_SUCCESS) {

×

3123

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

×

3125

  taosMemoryFree(pInfo);

×

3126

  taosMemoryFree(pOperator);

×

3127

  pTaskInfo->code = code;

×

3128

  return code;

×

3131

void destroyTmqScanOperatorInfo(void* param) {

×

3132

  if (param == NULL) {

×

3133

    return;

×

3136

  STmqQueryScanInfo* pTmqScan = (STmqQueryScanInfo*)param;

×

3137

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

×

3138

    destroyOperator(pTmqScan->pTableScanOp);

×

3141

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

×

3142

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

×

3145

  if (pTmqScan->pTableListInfo) {

×

3146

    tableListDestroy(pTmqScan->pTableListInfo);

×

3147

    pTmqScan->pTableListInfo = NULL;

×

3149

  if (pTmqScan->pPseudoExpr) {

×

3150

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

×

3151

    taosMemoryFree(pTmqScan->pPseudoExpr);

×

3154

  taosHashCleanup(pTmqScan->pCol2SlotId);

×

3155

  pTmqScan->pCol2SlotId = NULL;

×

3157

  cleanupExprSupp(&pTmqScan->tagCalSup);

×

3159

  blockDataDestroy(pTmqScan->pRes);

×

3161

  taosMemoryFree(pTmqScan);

×

3164

int32_t addPrimaryKeyCol(SSDataBlock* pBlock, uint8_t type, int32_t bytes) {

×

3165

  int32_t code = TSDB_CODE_SUCCESS;

×

3166

  int32_t lino = 0;

×

3167

  pBlock->info.rowSize += bytes;

×

3168

  SColumnInfoData infoData = {0};

×

3169

  infoData.info.type = type;

×

3170

  infoData.info.bytes = bytes;

×

3171

  void* tmp = taosArrayPush(pBlock->pDataBlock, &infoData);

×

3172

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

×

3174

_end:

×

3175

  if (code != TSDB_CODE_SUCCESS) {

×

3176

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

×

3178

  return code;

×

3181

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

×

3182

  size_t  numOfCols = LIST_LENGTH(pNodeList);

×

3183

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

×

3184

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

×

3185

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

×

3186

      *pTsSlotId = pNode->slotId;

×

3189

×

3191

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

×

3192

  size_t  numOfCols = LIST_LENGTH(pNodeList);

×

3193

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

×

3194

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

×

3195

    if (pNode == NULL) {

×

3196

      return TSDB_CODE_INVALID_PARA;

×

3198

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

×

3199

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

×

3200

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

×

3201

      if (code != TSDB_CODE_SUCCESS) {

×

3202

        return code;

×

3206

  return TSDB_CODE_SUCCESS;

×

3209

int32_t createTmqScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,

×

3212

  QRY_PARAM_CHECK(pOptrInfo);

×

3214

  int32_t          code = TSDB_CODE_SUCCESS;

×

3215

  int32_t          lino = 0;

×

3216

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

×

3217

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

×

3218

  SStorageAPI*     pAPI = &pTaskInfo->storageAPI;

×

3219

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

×

3221

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

×

3222

    code = terrno;

×

3223

    goto _error;

×

3226

  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;

×

3227

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;

×

3229

  pInfo->pTagCond = pTagCond;

×

3230

  pInfo->pGroupTags = pTableScanNode->pGroupTags;

×

3232

  pInfo->primaryTsIndex = -1;

×

3233

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

×

3234

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

×

3235

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

×

3236

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

×

3237

  QUERY_CHECK_CODE(code, lino, _error);

×

3239

  SDataBlockDescNode* pNode = (SDataBlockDescNode*)pDescNode;

×

3240

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

×

3241

    qError("%s column num mismatch between desc and scan node", __func__);

×

3242

    code = TSDB_CODE_TMQ_INVALID_MSG;

×

3243

    goto _error;

×

3246

  if (pTableScanNode->pTags != NULL) {

×

3248

    SExprInfo* pTagExpr = createExpr(pTableScanNode->pTags, &numOfTags);

×

3249

    if (pTagExpr == NULL) {

×

3250

      goto _error;

×

3252

    code = initExprSupp(&pInfo->tagCalSup, pTagExpr, numOfTags, &pTaskInfo->storageAPI.functionStore);

×

3253

    if (code != 0) {

×

3254

      goto _error;

×

3258

  SOperatorInfo* pTableScanOp = NULL;

×

3259

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

×

3260

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

×

3261

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

×

3262

    goto _error;

×

3265

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

×

3266

  if (pHandle->version > 0) {

×

3267

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

×

3270

  STableKeyInfo* pList = NULL;

×

3271

  int32_t        num = 0;

×

3272

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

×

3273

  QUERY_CHECK_CODE(code, lino, _error);

×

3275

  if (pHandle->initTableReader) {

×

3276

    pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;

×

3277

    pTSInfo->base.dataReader = NULL;

×

3280

  if (pHandle->initTqReader) {

×

3281

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

×

3282

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

×

3284

    pInfo->tqReader = pHandle->tqReader;

×

3285

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

×

3288

  pInfo->pTableScanOp = pTableScanOp;

×

3290

  pInfo->readHandle = *pHandle;

×

3291

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

×

3293

  SArray* tableIdList = NULL;

×

3294

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

×

3295

  QUERY_CHECK_CODE(code, lino, _error);

×

3296

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

×

3297

  QUERY_CHECK_CODE(code, lino, _error);

×

3298

  taosArrayDestroy(tableIdList);

×

3299

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

×

3302

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

×

3303

  QUERY_CHECK_CODE(code, lino, _error);

×

3305

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

×

3306

  QUERY_CHECK_CODE(code, lino, _error);

×

3308

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

×

3309

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

×

3311

  pInfo->pTmqScanOp = pOperator;

×

3312

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

×

3314

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

×

3316

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

×

3318

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

×

3321

  *pOptrInfo = pOperator;

×

3322

  return code;

×

3324

_error:

×

3325

  if (pInfo != NULL && pInfo->pTableScanOp != NULL) {

×

3326

    STableScanInfo* p = (STableScanInfo*)pInfo->pTableScanOp->info;

×

3327

    if (p != NULL) {

×

3328

      p->base.pTableListInfo = NULL;

×

3330

    destroyTmqScanOperatorInfo(pInfo);

×

3333

  if (pOperator != NULL) {

×

3334

    pOperator->info = NULL;

×

3335

    destroyOperator(pOperator);

×

3337

  pTaskInfo->code = code;

×

3338

  return code;

×

3341

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

×

3342

  int32_t        code = TSDB_CODE_SUCCESS;

×

3343

  int32_t        lino = 0;

×

3344

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

3345

  STagScanInfo*  pInfo = pOperator->info;

×

3346

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

×

3347

  int32_t        count = pRes->info.rows;

×

3349

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

×

3350

  if (!item) {

×

3351

    qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", item->uid, tstrerror(terrno),

×

3353

    tDecoderClear(&(*mr).coder);

×

3354

    goto _end;

×

3357

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

×

3358

  tDecoderClear(&(*mr).coder);

×

3359

  if (code != TSDB_CODE_SUCCESS) {

×

3360

    qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", item->uid, tstrerror(terrno),

×

3362

    goto _end;

×

3366

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

×

3367

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

×

3370

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

×

3371

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

×

3372

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

×

3373

      QUERY_CHECK_CODE(code, lino, _end);

×

3374

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

×

3375

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

×

3376

      QUERY_CHECK_CODE(code, lino, _end);

×

3377

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

×

3378

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

×

3379

      QUERY_CHECK_CODE(code, lino, _end);

×

3381

      STagVal val = {0};

×

3382

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

×

3383

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

×

3385

      char* data = NULL;

×

3386

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

×

3387

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

×

3389

        data = (char*)p;

×

3392

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

×

3393

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

×

3394

      QUERY_CHECK_CODE(code, lino, _end);

×

3396

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

×

3398

        taosMemoryFree(data);

×

3403

_end:

×

3404

  if (code != TSDB_CODE_SUCCESS) {

×

3405

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

×

3406

    pTaskInfo->code = code;

×

3408

    pRes->info.rows++;

×

3411

  return code;

×

3414

static void tagScanFreeUidTag(void* p) {

×

3415

  STUidTagInfo* pInfo = p;

×

3416

  if (pInfo->pTagVal != NULL) {

×

3417

    taosMemoryFree(pInfo->pTagVal);

×

3419

×

3421

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

×

3422

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

×

3423

  if (pColumnData == NULL) {

×

3424

    return terrno;

×

3427

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

×

3428

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

×

3429

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

×

3430

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

×

3432

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

×

3433

  if (code != TSDB_CODE_SUCCESS) {

×

3434

    terrno = code;

×

3435

    taosMemoryFree(pColumnData);

×

3436

    return terrno;

×

3439

  pParam->columnData = pColumnData;

×

3440

  pParam->colAlloced = true;

×

3441

  return TSDB_CODE_SUCCESS;

×

3444

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

×

3445

  int32_t                code = TSDB_CODE_SUCCESS;

×

3446

  int32_t                lino = 0;

×

3447

  STagScanFilterContext* pCtx = (STagScanFilterContext*)pContext;

×

3448

  SColumnNode*           pSColumnNode = NULL;

×

3449

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

×

3450

    pSColumnNode = *(SColumnNode**)pNode;

×

3451

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

×

3452

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

×

3453

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

×

3454

      pSColumnNode = NULL;

×

3455

      pCtx->code = nodesMakeNode(QUERY_NODE_COLUMN, (SNode**)&pSColumnNode);

×

3456

      if (NULL == pSColumnNode) {

×

3457

        return DEAL_RES_ERROR;

×

3459

      pSColumnNode->colId = -1;

×

3460

      pSColumnNode->colType = COLUMN_TYPE_TBNAME;

×

3461

      pSColumnNode->node.resType.type = TSDB_DATA_TYPE_VARCHAR;

×

3462

      pSColumnNode->node.resType.bytes = TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE;

×

3463

      nodesDestroyNode(*pNode);

×

3464

      *pNode = (SNode*)pSColumnNode;

×

3466

      return DEAL_RES_CONTINUE;

×

3469

    return DEAL_RES_CONTINUE;

×

3472

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

×

3473

  if (!data) {

×

3474

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

×

3475

    if (code == TSDB_CODE_DUP_KEY) {

×

3476

      code = TSDB_CODE_SUCCESS;

×

3478

    QUERY_CHECK_CODE(code, lino, _end);

×

3479

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

×

3480

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

×

3481

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

×

3482

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

×

3483

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

×

3484

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

×

3486

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

×

3487

    pSColumnNode->slotId = col->slotId;

×

3490

_end:

×

3491

  if (code != TSDB_CODE_SUCCESS) {

×

3492

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

×

3493

    return DEAL_RES_ERROR;

×

3495

  return DEAL_RES_CONTINUE;

×

3498

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

×

3500

  int32_t code = TSDB_CODE_SUCCESS;

×

3501

  int32_t lino = 0;

×

3502

  int32_t numOfTables = taosArrayGetSize(aUidTags);

×

3503

  SArray* pBlockList = NULL;

×

3505

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

×

3506

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

×

3508

  pBlockList = taosArrayInit(1, POINTER_BYTES);

×

3509

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

×

3511

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

×

3512

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

×

3514

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

×

3516

  SScalarParam output = {0};

×

3517

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

×

3518

  QUERY_CHECK_CODE(code, lino, _end);

×

3520

  gTaskScalarExtra.pStreamInfo = (void*)pStreamRuntimeInfo;

×

3521

  gTaskScalarExtra.pStreamRange = NULL;

×

3522

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

×

3523

  QUERY_CHECK_CODE(code, lino, _end);

×

3525

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

×

3526

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

×

3527

    if (result[i]) {

×

3528

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

×

3529

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

×

3533

  colDataDestroy(output.columnData);

×

3534

  taosMemoryFreeClear(output.columnData);

×

3536

_end:

×

3537

  blockDataDestroy(pResBlock);

×

3538

  taosArrayDestroy(pBlockList);

×

3539

  if (code != TSDB_CODE_SUCCESS) {

×

3540

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

×

3542

  return code;

×

3545

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

×

3548

  int32_t code = TSDB_CODE_SUCCESS;

×

3549

  int32_t lino = 0;

×

3550

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

×

3551

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

×

3552

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

×

3553

      STR_TO_VARSTR(str, "ctbidx");

×

3555

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

×

3556

      QUERY_CHECK_CODE(code, lino, _end);

×

3557

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

×

3558

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

×

3559

      QUERY_CHECK_CODE(code, lino, _end);

×

3560

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

×

3561

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

×

3562

      QUERY_CHECK_CODE(code, lino, _end);

×

3565

    STagVal tagVal = {0};

×

3566

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

×

3567

    if (pUidTagInfo->pTagVal == NULL) {

×

3568

      colDataSetNULL(pColInfo, rowIndex);

×

3570

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

×

3572

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

×

3573

        colDataSetNULL(pColInfo, rowIndex);

×

3574

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

×

3575

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

×

3576

        QUERY_CHECK_CODE(code, lino, _end);

×

3577

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

×

3578

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

×

3579

          QUERY_CHECK_CODE(code = TSDB_CODE_BLOB_NOT_SUPPORT_TAG, lino, _end);

×

3581

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

×

3582

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

×

3584

        varDataSetLen(tmp, tagVal.nData);

×

3585

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

×

3586

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

×

3587

        taosMemoryFree(tmp);

×

3588

        QUERY_CHECK_CODE(code, lino, _end);

×

3590

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

×

3591

        QUERY_CHECK_CODE(code, lino, _end);

×

3596

_end:

×

3597

  if (code != TSDB_CODE_SUCCESS) {

×

3598

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

×

3600

  return code;

×

3603

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

×

3605

  int32_t       code = TSDB_CODE_SUCCESS;

×

3606

  int32_t       lino = 0;

×

3607

  STagScanInfo* pInfo = pOperator->info;

×

3608

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

×

3609

  if (!ignoreFilterIdx) {

×

3610

    size_t szTables = taosArrayGetSize(aFilterIdxs);

×

3611

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

×

3612

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

×

3613

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, idx);

×

3614

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

×

3615

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

×

3616

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

×

3617

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

×

3618

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

×

3619

        QUERY_CHECK_CODE(code, lino, _end);

×

3623

    size_t szTables = taosArrayGetSize(aUidTags);

×

3624

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

×

3625

      STUidTagInfo* pUidTagInfo = taosArrayGet(aUidTags, i);

×

3626

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

×

3627

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

×

3628

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

×

3629

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

×

3630

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

×

3631

        QUERY_CHECK_CODE(code, lino, _end);

×

3636

_end:

×

3637

  if (code != TSDB_CODE_SUCCESS) {

×

3638

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

×

3640

  return code;

×

3643

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

×

3644

  int32_t        code = TSDB_CODE_SUCCESS;

×

3645

  int32_t        lino = 0;

×

3646

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

3647

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

3649

  STagScanInfo* pInfo = pOperator->info;

×

3650

  SSDataBlock*  pRes = pInfo->pRes;

×

3652

  QRY_PARAM_CHECK(ppRes);

×

3654

  if (pOperator->status == OP_EXEC_DONE) {

×

3655

    return TSDB_CODE_SUCCESS;

×

3657

  blockDataCleanup(pRes);

×

3659

  if (pInfo->pCtbCursor == NULL) {

×

3660

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

×

3661

    QUERY_CHECK_NULL(pInfo->pCtbCursor, code, lino, _end, terrno);

×

3663

    code = pAPI->metaFn.resumeCtbCursor(pInfo->pCtbCursor, 0);

×

3664

    QUERY_CHECK_CODE(code, lino, _end);

×

3667

  SArray* aUidTags = pInfo->aUidTags;

×

3668

  SArray* aFilterIdxs = pInfo->aFilterIdxs;

×

3669

  int32_t count = 0;

×

3670

  bool    ctbCursorFinished = false;

×

3671

  while (1) {

×

3672

    taosArrayClearEx(aUidTags, tagScanFreeUidTag);

×

3673

    taosArrayClear(aFilterIdxs);

×

3675

    int32_t numTables = 0;

×

3676

    while (numTables < pOperator->resultInfo.capacity) {

×

3677

      SMCtbCursor* pCur = pInfo->pCtbCursor;

×

3678

      tb_uid_t     uid = pAPI->metaFn.ctbCursorNext(pInfo->pCtbCursor);

×

3679

      if (uid == 0) {

×

3680

        ctbCursorFinished = true;

×

3681

        break;

×

3683

      STUidTagInfo info = {.uid = uid, .pTagVal = pCur->pVal};

×

3684

      info.pTagVal = taosMemoryMalloc(pCur->vLen);

×

3685

      QUERY_CHECK_NULL(info.pTagVal, code, lino, _end, terrno);

×

3687

      memcpy(info.pTagVal, pCur->pVal, pCur->vLen);

×

3688

      void* tmp = taosArrayPush(aUidTags, &info);

×

3689

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

3690

      ++numTables;

×

3693

    if (numTables == 0) {

×

3694

      break;

×

3696

    bool ignoreFilterIdx = true;

×

3697

    if (pInfo->pTagCond != NULL) {

×

3698

      ignoreFilterIdx = false;

×

3699

      code = tagScanFilterByTagCond(aUidTags, pInfo->pTagCond, aFilterIdxs, pInfo->readHandle.vnode, pAPI, pInfo, pTaskInfo->pStreamRuntimeInfo);

×

3700

      QUERY_CHECK_CODE(code, lino, _end);

×

3702

      ignoreFilterIdx = true;

×

3705

    code = tagScanFillResultBlock(pOperator, pRes, aUidTags, aFilterIdxs, ignoreFilterIdx, pAPI);

×

3706

    QUERY_CHECK_CODE(code, lino, _end);

×

3708

    count = ignoreFilterIdx ? taosArrayGetSize(aUidTags) : taosArrayGetSize(aFilterIdxs);

×

3710

    if (count != 0) {

×

3711

      break;

×

3715

  if (count > 0) {

×

3716

    pAPI->metaFn.pauseCtbCursor(pInfo->pCtbCursor);

×

3718

  if (count == 0 || ctbCursorFinished) {

×

3719

    pAPI->metaFn.closeCtbCursor(pInfo->pCtbCursor);

×

3720

    pInfo->pCtbCursor = NULL;

×

3721

    setOperatorCompleted(pOperator);

×

3724

  pRes->info.rows = count;

×

3725

  bool bLimitReached = applyLimitOffset(&pInfo->limitInfo, pRes, pTaskInfo);

×

3726

  if (bLimitReached) {

×

3727

    setOperatorCompleted(pOperator);

×

3730

  pOperator->resultInfo.totalRows += pRes->info.rows;

×

3731

  (*ppRes) = (pRes->info.rows == 0) ? NULL : pInfo->pRes;

×

3733

_end:

×

3734

  if (code != TSDB_CODE_SUCCESS) {

×

3735

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

3736

    pTaskInfo->code = code;

×

3737

    T_LONG_JMP(pTaskInfo->env, code);

×

3740

  return code;

×

3743

static int32_t createTagScanTableListInfoFromParam(SOperatorInfo* pOperator) {

×

3744

  STagScanInfo*          pInfo = pOperator->info;

×

3745

  SExecTaskInfo*         pTaskInfo = pOperator->pTaskInfo;

×

3746

  int32_t                code = 0;

×

3747

  STableListInfo*        pListInfo = pInfo->pTableListInfo;

×

3748

  STagScanOperatorParam* pParam = (STagScanOperatorParam*)pOperator->pOperatorGetParam->value;

×

3749

  tb_uid_t               pUid = pParam->vcUid;

×

3755

  STableKeyInfo info = {.groupId = 0};

×

3757

  int32_t tableIdx = 0;

×

3758

  taosHashClear(pListInfo->map);

×

3759

  taosArrayClear(pListInfo->pTableList);

×

3761

  if (taosHashPut(pListInfo->map, &pUid, sizeof(uint64_t), &tableIdx, sizeof(int32_t))) {

×

3762

    if (TSDB_CODE_DUP_KEY == terrno) {

×

3764

      return terrno;

×

3768

  info.uid = pUid;

×

3769

  void* p = taosArrayPush(pListInfo->pTableList, &info);

×

3770

  if (p == NULL) {

×

3771

    return terrno;

×

3774

  qDebug("add dynamic table scan uid:%" PRIu64 ", %s", info.uid, GET_TASKID(pTaskInfo));

×

3776

  return code;

×

3779

static int32_t doTagScanFromMetaEntryNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

×

3780

  int32_t       code = TSDB_CODE_SUCCESS;

×

3781

  int32_t       lino = 0;

×

3782

  STagScanInfo* pInfo = pOperator->info;

×

3783

  if (pOperator->pOperatorGetParam) {

×

3784

    pOperator->resultInfo.totalRows = 0;

×

3785

    pOperator->dynamicTask = true;

×

3786

    pInfo->curPos = 0;

×

3787

    code = createTagScanTableListInfoFromParam(pOperator);

×

3788

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

×

3789

    pOperator->pOperatorGetParam = NULL;

×

3790

    QUERY_CHECK_CODE(code, lino, _end);

×

3792

    if (pOperator->status == OP_EXEC_DONE) {

×

3793

      pOperator->status = OP_OPENED;

×

3797

  if (pOperator->status == OP_EXEC_DONE) {

×

3798

    (*ppRes) = NULL;

×

3799

    return code;

×

3802

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

3803

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

3805

  SExprInfo*   pExprInfo = &pOperator->exprSupp.pExprInfo[0];

×

3806

  SSDataBlock* pRes = pInfo->pRes;

×

3807

  blockDataCleanup(pRes);

×

3809

  int32_t size = 0;

×

3810

  code = tableListGetSize(pInfo->pTableListInfo, &size);

×

3811

  QUERY_CHECK_CODE(code, lino, _end);

×

3813

  if (size == 0) {

×

3814

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

×

3815

    (*ppRes) = NULL;

×

3816

    return code;

×

3819

  SMetaReader mr = {0};

×

3820

  pAPI->metaReaderFn.initReader(&mr, pInfo->readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

×

3821

  pRes->info.rows = 0;

×

3823

  while (pInfo->curPos < size && pRes->info.rows < pOperator->resultInfo.capacity) {

×

3824

    code = doTagScanOneTable(pOperator, pRes, &mr, &pTaskInfo->storageAPI);

×

3825

    if (code != TSDB_CODE_OUT_OF_MEMORY && code != TSDB_CODE_QRY_REACH_QMEM_THRESHOLD &&

×

3828

      code = TSDB_CODE_SUCCESS;

×

3830

    QUERY_CHECK_CODE(code, lino, _end);

×

3832

    if (++pInfo->curPos >= size) {

×

3833

      setOperatorCompleted(pOperator);

×

3837

  pAPI->metaReaderFn.clearReader(&mr);

×

3838

  bool bLimitReached = applyLimitOffset(&pInfo->limitInfo, pRes, pTaskInfo);

×

3839

  if (bLimitReached) {

×

3840

    setOperatorCompleted(pOperator);

×

3844

  if (pOperator->status == OP_EXEC_DONE) {

×

3845

    setTaskStatus(pTaskInfo, TASK_COMPLETED);

×

3848

  pOperator->resultInfo.totalRows += pRes->info.rows;

×

3850

  (*ppRes) = (pRes->info.rows == 0) ? NULL : pInfo->pRes;

×

3851

_end:

×

3852

  if (code != TSDB_CODE_SUCCESS) {

×

3853

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

3854

    pTaskInfo->code = code;

×

3855

    T_LONG_JMP(pTaskInfo->env, code);

×

3857

  return code;

×

3860

static void destroyTagScanOperatorInfo(void* param) {

×

3861

  STagScanInfo* pInfo = (STagScanInfo*)param;

×

3862

  if (pInfo->pCtbCursor != NULL && pInfo->pStorageAPI != NULL) {

×

3863

    pInfo->pStorageAPI->metaFn.closeCtbCursor(pInfo->pCtbCursor);

×

3865

  taosHashCleanup(pInfo->filterCtx.colHash);

×

3866

  taosArrayDestroy(pInfo->filterCtx.cInfoList);

×

3867

  taosArrayDestroy(pInfo->aFilterIdxs);

×

3868

  taosArrayDestroyEx(pInfo->aUidTags, tagScanFreeUidTag);

×

3870

  blockDataDestroy(pInfo->pRes);

×

3871

  taosArrayDestroy(pInfo->matchInfo.pList);

×

3872

  tableListDestroy(pInfo->pTableListInfo);

×

3874

  pInfo->pRes = NULL;

×

3875

  pInfo->pTableListInfo = NULL;

×

3876

  taosMemoryFreeClear(param);

×

3877

×

3879

static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {

×

3880

  int32_t         code = TSDB_CODE_SUCCESS;

×

3881

  STagScanInfo*   pInfo = pOper->info;

×

3883

  pOper->status = OP_NOT_OPENED;

×

3884

  STagScanPhysiNode* pTagScanNode = (STagScanPhysiNode*)pOper->pPhyNode;

×

3885

  if (pTagScanNode->onlyMetaCtbIdx) {

×

3886

      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;

×

3887

      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

3888

      pAPI->metaFn.closeCtbCursor(pInfo->pCtbCursor);

×

3889

      pInfo->pCtbCursor = NULL;

×

3891

      pInfo->curPos = 0;

×

3894

  return code;

×

3897

int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,

×

3900

  QRY_PARAM_CHECK(pOptrInfo);

×

3902

  int32_t         code = TSDB_CODE_SUCCESS;

×

3903

  int32_t         lino = 0;

×

3904

  SScanPhysiNode* pPhyNode = (SScanPhysiNode*)pTagScanNode;

×

3905

  STagScanInfo*   pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo));

×

3906

  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

×

3907

  if (pInfo == NULL || pOperator == NULL) {

×

3908

    code = terrno;

×

3909

    goto _error;

×

3911

  pOperator->pPhyNode = pTagScanNode;

×

3912

  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

×

3914

  int32_t    numOfExprs = 0;

×

3915

  SExprInfo* pExprInfo = NULL;

×

3917

  code = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &pExprInfo, &numOfExprs);

×

3918

  QUERY_CHECK_CODE(code, lino, _error);

×

3920

  code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs, &pTaskInfo->storageAPI.functionStore);

×

3921

  QUERY_CHECK_CODE(code, lino, _error);

×

3923

  int32_t num = 0;

×

3924

  code = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);

×

3925

  QUERY_CHECK_CODE(code, lino, _error);

×

3927

  pInfo->pTagCond = pTagCond;

×

3928

  pInfo->pTagIndexCond = pTagIndexCond;

×

3929

  pInfo->suid = pPhyNode->suid;

×

3930

  pInfo->pStorageAPI = &pTaskInfo->storageAPI;

×

3932

  pInfo->pTableListInfo = pTableListInfo;

×

3933

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

×

3934

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

×

3936

  pInfo->readHandle = *pReadHandle;

×

3937

  pInfo->curPos = 0;

×

3939

  initLimitInfo(pPhyNode->node.pLimit, pPhyNode->node.pSlimit, &pInfo->limitInfo);

×

3940

  setOperatorInfo(pOperator, "TagScanOperator", QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN, false, OP_NOT_OPENED, pInfo,

×

3942

  initResultSizeInfo(&pOperator->resultInfo, 4096);

×

3943

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

×

3944

  QUERY_CHECK_CODE(code, lino, _error);

×

3946

  if (pTagScanNode->onlyMetaCtbIdx) {

×

3947

    pInfo->aUidTags = taosArrayInit(pOperator->resultInfo.capacity, sizeof(STUidTagInfo));

×

3948

    QUERY_CHECK_NULL(pInfo->aUidTags, code, lino, _error, terrno);

×

3950

    pInfo->aFilterIdxs = taosArrayInit(pOperator->resultInfo.capacity, sizeof(int32_t));

×

3951

    QUERY_CHECK_NULL(pInfo->aFilterIdxs, code, lino, _error, terrno);

×

3953

    pInfo->filterCtx.colHash =

×

3954

        taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);

×

3955

    QUERY_CHECK_NULL(pInfo->filterCtx.colHash, code, lino, _error, terrno);

×

3957

    pInfo->filterCtx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));

×

3958

    QUERY_CHECK_NULL(pInfo->filterCtx.cInfoList, code, lino, _error, terrno);

×

3960

    if (pInfo->pTagCond != NULL) {

×

3961

      nodesRewriteExprPostOrder(&pTagCond, tagScanRewriteTagColumn, (void*)&pInfo->filterCtx);

×

3964

  __optr_fn_t tagScanNextFn = (pTagScanNode->onlyMetaCtbIdx) ? doTagScanFromCtbIdxNext : doTagScanFromMetaEntryNext;

×

3965

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, tagScanNextFn, NULL, destroyTagScanOperatorInfo,

×

3967

  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);

×

3969

  *pOptrInfo = pOperator;

×

3970

  return code;

×

3972

_error:

×

3973

  if (pInfo) {

×

3974

    pInfo->pTableListInfo = NULL;

×

3977

  if (pInfo != NULL) destroyTagScanOperatorInfo(pInfo);

×

3978

  if (pOperator != NULL) {

×

3979

    pOperator->info = NULL;

×

3980

    destroyOperator(pOperator);

×

3982

  return code;

×

3987

static int32_t subTblRowCompareTsFn(const void* pLeft, const void* pRight, void* param) {

×

3988

  int32_t                 left = *(int32_t*)pLeft;

×

3989

  int32_t                 right = *(int32_t*)pRight;

×

3990

  STmsSubTablesMergeInfo* pInfo = (STmsSubTablesMergeInfo*)param;

×

3992

  int32_t leftIdx = pInfo->aInputs[left].rowIdx;

×

3993

  int32_t rightIdx = pInfo->aInputs[right].rowIdx;

×

3995

  if (leftIdx == -1) {

×

3996

    return 1;

×

3997

  } else if (rightIdx == -1) {

×

3998

    return -1;

×

4001

  int64_t leftTs = pInfo->aInputs[left].aTs[leftIdx];

×

4002

  int64_t rightTs = pInfo->aInputs[right].aTs[rightIdx];

×

4003

  int32_t ret = leftTs > rightTs ? 1 : ((leftTs < rightTs) ? -1 : 0);

×

4004

  if (pInfo->pTsOrderInfo->order == TSDB_ORDER_DESC) {

×

4005

    ret = -1 * ret;

×

4007

  return ret;

×

4010

static int32_t subTblRowCompareTsPkFn(const void* pLeft, const void* pRight, void* param) {

×

4011

  int32_t                 left = *(int32_t*)pLeft;

×

4012

  int32_t                 right = *(int32_t*)pRight;

×

4013

  STmsSubTablesMergeInfo* pInfo = (STmsSubTablesMergeInfo*)param;

×

4015

  int32_t leftIdx = pInfo->aInputs[left].rowIdx;

×

4016

  int32_t rightIdx = pInfo->aInputs[right].rowIdx;

×

4018

  if (leftIdx == -1) {

×

4019

    return 1;

×

4020

  } else if (rightIdx == -1) {

×

4021

    return -1;

×

4024

  int64_t leftTs = pInfo->aInputs[left].aTs[leftIdx];

×

4025

  int64_t rightTs = pInfo->aInputs[right].aTs[rightIdx];

×

4026

  int32_t ret = leftTs > rightTs ? 1 : ((leftTs < rightTs) ? -1 : 0);

×

4027

  if (pInfo->pTsOrderInfo->order == TSDB_ORDER_DESC) {

×

4028

    ret = -1 * ret;

×

4030

  if (ret == 0 && pInfo->pPkOrderInfo) {

×

4031

    ret = tsortComparBlockCell(pInfo->aInputs[left].pInputBlock, pInfo->aInputs[right].pInputBlock, leftIdx, rightIdx,

×

4032

                               pInfo->pPkOrderInfo);

×

4034

  return ret;

×

4037

int32_t dumpQueryTableCond(const SQueryTableDataCond* src, SQueryTableDataCond* dst) {

×

4038

  int32_t code = TSDB_CODE_SUCCESS;

×

4039

  int32_t lino = 0;

×

4040

  memcpy((void*)dst, (void*)src, sizeof(SQueryTableDataCond));

×

4041

  dst->colList = NULL;

×

4042

  dst->pSlotList = NULL;

×

4044

  if (src->numOfCols > 0) {

×

4045

    dst->colList = taosMemoryCalloc(src->numOfCols, sizeof(SColumnInfo));

×

4046

    QUERY_CHECK_NULL(dst->colList, code, lino, _end, terrno);

×

4047

    memcpy(dst->colList, src->colList, src->numOfCols * sizeof(SColumnInfo));

×

4049

    if (src->pSlotList != NULL) {

×

4050

      dst->pSlotList = taosMemoryMalloc(src->numOfCols * sizeof(int32_t));

×

4051

      QUERY_CHECK_NULL(dst->pSlotList, code, lino, _end, terrno);

×

4052

      memcpy(dst->pSlotList, src->pSlotList, src->numOfCols * sizeof(int32_t));

×

4056

_end:

×

4057

  if (code != TSDB_CODE_SUCCESS) {

×

4058

    taosMemoryFreeClear(dst->colList);

×

4059

    taosMemoryFreeClear(dst->pSlotList);

×

4060

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

4062

  return code;

×

4065

static int32_t fetchNextSubTableBlockFromReader(SOperatorInfo* pOperator, STmsSubTableInput* pInput,

×

4067

  int32_t code = 0;

×

4069

  STableMergeScanInfo*    pInfo = pOperator->info;

×

4070

  SReadHandle*            pHandle = &pInfo->base.readHandle;

×

4071

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

×

4072

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

×

4073

  const SStorageAPI*      pAPI = &pTaskInfo->storageAPI;

×

4075

  blockDataCleanup(pInput->pReaderBlock);

×

4076

  if (!pInput->bInMemReader) {

×

4077

    code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, pInput->pKeyInfo, 1, pInput->pReaderBlock,

×

4078

                                         (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

×

4079

    if (code != 0) {

×

4080

      return code;

×

4084

  pInfo->base.dataReader = pInput->pReader;

×

4086

  while (true) {

×

4087

    bool hasNext = false;

×

4088

    code = pAPI->tsdReader.tsdNextDataBlock(pInfo->base.dataReader, &hasNext);

×

4089

    if (code != 0) {

×

4090

      pAPI->tsdReader.tsdReaderReleaseDataBlock(pInfo->base.dataReader);

×

4091

      pInfo->base.dataReader = NULL;

×

4092

      return code;

×

4095

    if (!hasNext || isTaskKilled(pTaskInfo)) {

×

4096

      if (isTaskKilled(pTaskInfo)) {

×

4097

        pAPI->tsdReader.tsdReaderReleaseDataBlock(pInfo->base.dataReader);

×

4098

        pInfo->base.dataReader = NULL;

×

4099

        return code;

×

4102

      *pSubTableHasBlock = false;

×

4103

      break;

×

4106

    if (pInput->tblCond.order == TSDB_ORDER_ASC) {

×

4107

      pInput->tblCond.twindows.skey = pInput->pReaderBlock->info.window.ekey + 1;

×

4109

      pInput->tblCond.twindows.ekey = pInput->pReaderBlock->info.window.skey - 1;

×

4112

    uint32_t status = 0;

×

4113

    code = loadDataBlock(pOperator, &pInfo->base, pInput->pReaderBlock, &status);

×

4114

    if (code != 0) {

×

4115

      pInfo->base.dataReader = NULL;

×

4116

      return code;

×

4119

    if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

×

4120

      *pSubTableHasBlock = false;

×

4121

      break;

×

4123

    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pInput->pReaderBlock->info.rows == 0) {

×

4124

      continue;

×

4127

    *pSubTableHasBlock = true;

×

4128

    break;

×

4131

  if (*pSubTableHasBlock) {

×

4132

    pInput->pReaderBlock->info.id.groupId =

×

4133

        tableListGetTableGroupId(pInfo->base.pTableListInfo, pInput->pReaderBlock->info.id.uid);

×

4134

    pOperator->resultInfo.totalRows += pInput->pReaderBlock->info.rows;

×

4136

  if (!pInput->bInMemReader || !*pSubTableHasBlock) {

×

4137

    pAPI->tsdReader.tsdReaderClose(pInput->pReader);

×

4138

    pInput->pReader = NULL;

×

4141

  pInfo->base.dataReader = NULL;

×

4142

  return TSDB_CODE_SUCCESS;

×

4145

static int32_t setGroupStartEndIndex(STableMergeScanInfo* pInfo) {

×

4146

  pInfo->bGroupProcessed = false;

×

4148

  int32_t numOfTables = 0;

×

4149

  int32_t code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

×

4150

  if (code != TSDB_CODE_SUCCESS) {

×

4151

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

4152

    return code;

×

4155

  int32_t i = pInfo->tableStartIndex + 1;

×

4156

  for (; i < numOfTables; ++i) {

×

4157

    STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

×

4158

    if (!tableKeyInfo) {

×

4159

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

4160

      return terrno;

×

4162

    if (tableKeyInfo->groupId != pInfo->groupId) {

×

4163

      break;

×

4166

  pInfo->tableEndIndex = i - 1;

×

4167

  return TSDB_CODE_SUCCESS;

×

4170

static int32_t openSubTablesMergeSort(STmsSubTablesMergeInfo* pSubTblsInfo) {

×

4171

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

×

4172

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

×

4173

    if (pInput->rowIdx == -1) {

×

4174

      continue;

×

4177

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

×

4178

      pInput->rowIdx = 0;

×

4179

      pInput->pageIdx = -1;

×

4182

    pInput->pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

×

4183

    SColumnInfoData* col = taosArrayGet(pInput->pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

×

4184

    if (!col) {

×

4185

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

4186

      return terrno;

×

4188

    pInput->aTs = (int64_t*)col->pData;

×

4191

  __merge_compare_fn_t mergeCompareFn = (!pSubTblsInfo->pPkOrderInfo) ? subTblRowCompareTsFn : subTblRowCompareTsPkFn;

×

4192

  return tMergeTreeCreate(&pSubTblsInfo->pTree, pSubTblsInfo->numSubTables, pSubTblsInfo, mergeCompareFn);

×

4195

static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {

×

4196

  int32_t code = setGroupStartEndIndex(pInfo);

×

4197

  if (code != TSDB_CODE_SUCCESS) {

×

4198

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

4199

    return code;

×

4201

  STmsSubTablesMergeInfo* pSubTblsInfo = taosMemoryCalloc(1, sizeof(STmsSubTablesMergeInfo));

×

4202

  if (pSubTblsInfo == NULL) {

×

4203

    return terrno;

×

4205

  pSubTblsInfo->pTsOrderInfo = taosArrayGet(pInfo->pSortInfo, 0);

×

4206

  if (!pSubTblsInfo->pTsOrderInfo) {

×

4207

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

4208

    return terrno;

×

4210

  if (taosArrayGetSize(pInfo->pSortInfo) == 2) {

×

4211

    pSubTblsInfo->pPkOrderInfo = taosArrayGet(pInfo->pSortInfo, 1);

×

4212

    if (!pSubTblsInfo->pPkOrderInfo) {

×

4213

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

4214

      return terrno;

×

4217

    pSubTblsInfo->pPkOrderInfo = NULL;

×

4219

  pSubTblsInfo->numSubTables = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

×

4220

  pSubTblsInfo->aInputs = taosMemoryCalloc(pSubTblsInfo->numSubTables, sizeof(STmsSubTableInput));

×

4221

  if (pSubTblsInfo->aInputs == NULL) {

×

4222

    taosMemoryFree(pSubTblsInfo);

×

4223

    return terrno;

×

4225

  int32_t bufPageSize = pInfo->bufPageSize;

×

4226

  int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;

×

4227

  code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);

×

4228

  if (code != TSDB_CODE_SUCCESS) {

×

4229

    taosMemoryFree(pSubTblsInfo->aInputs);

×

4230

    taosMemoryFree(pSubTblsInfo);

×

4231

    return code;

×

4233

  pSubTblsInfo->numTableBlocksInMem = pSubTblsInfo->numSubTables;

×

4234

  pSubTblsInfo->numInMemReaders = pSubTblsInfo->numSubTables;

×

4236

  pInfo->pSubTablesMergeInfo = pSubTblsInfo;

×

4237

  return TSDB_CODE_SUCCESS;

×

4250

static int32_t setSubTableCondSuidByUid(STableMergeScanInfo* pInfo, SStorageAPI* pAPI, uint64_t uid,

×

4252

  int32_t     code = TSDB_CODE_SUCCESS;

×

4253

  int32_t     lino = 0;

×

4254

  SMetaReader tableReader = {0};

×

4256

  QUERY_CHECK_NULL(pInfo, code, lino, _end, TSDB_CODE_INVALID_PARA);

×

4257

  QUERY_CHECK_NULL(pAPI, code, lino, _end, TSDB_CODE_INVALID_PARA);

×

4258

  QUERY_CHECK_NULL(pCond, code, lino, _end, TSDB_CODE_INVALID_PARA);

×

4260

  pAPI->metaReaderFn.initReader(&tableReader, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

×

4261

  code = pAPI->metaReaderFn.getTableEntryByUid(&tableReader, uid);

×

4262

  pAPI->metaReaderFn.readerReleaseLock(&tableReader);

×

4263

  QUERY_CHECK_CODE(code, lino, _end);

×

4265

  if (tableReader.me.type == TSDB_CHILD_TABLE) {

×

4266

    pCond->suid = tableReader.me.ctbEntry.suid;

×

4267

  } else if (tableReader.me.type == TSDB_NORMAL_TABLE) {

×

4268

    pCond->suid = 0;

×

4270

    code = TSDB_CODE_INVALID_PARA;

×

4271

    QUERY_CHECK_CODE(code, lino, _end);

×

4274

_end:

×

4275

  if (code != TSDB_CODE_SUCCESS) {

×

4276

    qError("%s failed at line %d since %s, uid:%" PRIu64, __func__, lino, tstrerror(code), uid);

×

4278

  pAPI->metaReaderFn.clearReader(&tableReader);

×

4279

  return code;

×

4282

static int32_t initSubTableInputs(SOperatorInfo* pOperator, STableMergeScanInfo* pInfo) {

×

4283

  int32_t        code = TSDB_CODE_SUCCESS;

×

4284

  int32_t        lino = 0;

×

4285

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

4286

  SReadHandle*   pHandle = &pInfo->base.readHandle;

×

4287

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

4289

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

×

4291

  for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

×

4292

    STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

×

4293

    pInput->type = SUB_TABLE_MEM_BLOCK;

×

4295

    code = dumpQueryTableCond(&pInfo->base.cond, &pInput->tblCond);

×

4296

    QUERY_CHECK_CODE(code, lino, _end);

×

4298

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pReaderBlock);

×

4299

    QUERY_CHECK_CODE(code, lino, _end);

×

4301

    code = createOneDataBlock(pInfo->pResBlock, false, &pInput->pPageBlock);

×

4302

    QUERY_CHECK_CODE(code, lino, _end);

×

4304

    STableKeyInfo* keyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i + pInfo->tableStartIndex);

×

4305

    pInput->pKeyInfo = keyInfo;

×

4306

    QUERY_CHECK_NULL(pInput->pKeyInfo, code, lino, _end, terrno);

×

4308

    code = setSubTableCondSuidByUid(pInfo, pAPI, pInput->pKeyInfo->uid, &pInput->tblCond);

×

4309

    QUERY_CHECK_CODE(code, lino, _end);

×

4311

    if (isTaskKilled(pTaskInfo)) {

×

4312

      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);

×

4315

    if (i + 1 < pSubTblsInfo->numInMemReaders) {

×

4316

      code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInput->tblCond, keyInfo, 1, pInput->pReaderBlock,

×

4317

                                           (void**)&pInput->pReader, GET_TASKID(pTaskInfo), NULL);

×

4318

      QUERY_CHECK_CODE(code, lino, _end);

×

4319

      pInput->bInMemReader = true;

×

4321

      pInput->pReader = NULL;

×

4322

      pInput->bInMemReader = false;

×

4324

    bool hasNext = true;

×

4325

    code = fetchNextSubTableBlockFromReader(pOperator, pInput, &hasNext);

×

4326

    QUERY_CHECK_CODE(code, lino, _end);

×

4327

    if (!hasNext) {

×

4328

      pInput->rowIdx = -1;

×

4329

      ++pSubTblsInfo->numSubTablesCompleted;

×

4330

      continue;

×

4332

      pInput->rowIdx = 0;

×

4333

      pInput->pageIdx = -1;

×

4337

_end:

×

4338

  if (code != TSDB_CODE_SUCCESS) {

×

4339

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

4341

  return code;

×

4344

static void adjustSubTableFromMemBlock(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

×

4345

  int32_t              code = TSDB_CODE_SUCCESS;

×

4346

  int32_t              lino = 0;

×

4347

  SExecTaskInfo*       pTaskInfo = pOperatorInfo->pTaskInfo;

×

4348

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

×

4349

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

×

4350

  bool                 hasNext = true;

×

4351

  code = fetchNextSubTableBlockFromReader(pOperatorInfo, pInput, &hasNext);

×

4352

  QUERY_CHECK_CODE(code, lino, _end);

×

4354

  if (!hasNext) {

×

4355

    pInput->rowIdx = -1;

×

4356

    ++pSubTblsInfo->numSubTablesCompleted;

×

4358

    pInput->rowIdx = 0;

×

4361

_end:

×

4362

  if (code != TSDB_CODE_SUCCESS) {

×

4363

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

4364

    pTaskInfo->code = code;

×

4365

    T_LONG_JMP(pTaskInfo->env, code);

×

4367

×

4369

static int32_t adjustSubTableForNextRow(SOperatorInfo* pOperatorInfo, STmsSubTablesMergeInfo* pSubTblsInfo) {

×

4370

  STableMergeScanInfo* pInfo = pOperatorInfo->info;

×

4371

  STmsSubTableInput*   pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

×

4373

  SSDataBlock* pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

×

4374

  if (pInput->rowIdx < pInputBlock->info.rows - 1) {

×

4375

    ++pInput->rowIdx;

×

4376

  } else if (pInput->rowIdx == pInputBlock->info.rows - 1) {

×

4377

    if (pInput->type == SUB_TABLE_MEM_BLOCK) {

×

4378

      adjustSubTableFromMemBlock(pOperatorInfo, pSubTblsInfo);

×

4380

    if (pInput->rowIdx != -1) {

×

4381

      SColumnInfoData* col = taosArrayGet(pInputBlock->pDataBlock, pSubTblsInfo->pTsOrderInfo->slotId);

×

4382

      if (!col) {

×

4383

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

4384

        return terrno;

×

4386

      pInput->pInputBlock = pInputBlock;

×

4387

      pInput->aTs = (int64_t*)col->pData;

×

4391

  return tMergeTreeAdjust(pSubTblsInfo->pTree, tMergeTreeGetAdjustIndex(pSubTblsInfo->pTree));

×

4394

static int32_t appendChosenRowToDataBlock(STmsSubTablesMergeInfo* pSubTblsInfo, SSDataBlock* pBlock) {

×

4395

  int32_t            code = TSDB_CODE_SUCCESS;

×

4396

  int32_t            lino = 0;

×

4397

  STmsSubTableInput* pInput = pSubTblsInfo->aInputs + tMergeTreeGetChosenIndex(pSubTblsInfo->pTree);

×

4398

  SSDataBlock*       pInputBlock = (pInput->type == SUB_TABLE_MEM_BLOCK) ? pInput->pReaderBlock : pInput->pPageBlock;

×

4400

  for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {

×

4401

    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);

×

4402

    QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);

×

4404

    SColumnInfoData* pSrcColInfo = taosArrayGet(pInputBlock->pDataBlock, i);

×

4405

    QUERY_CHECK_NULL(pSrcColInfo, code, lino, _end, terrno);

×

4406

    bool isNull = colDataIsNull(pSrcColInfo, pInputBlock->info.rows, pInput->rowIdx, NULL);

×

4408

    if (isNull) {

×

4409

      code = colDataSetVal(pColInfo, pBlock->info.rows, NULL, true);

×

4410

      QUERY_CHECK_CODE(code, lino, _end);

×

4412

      if (pSrcColInfo->pData != NULL) {

×

4413

        char* pData = colDataGetData(pSrcColInfo, pInput->rowIdx);

×

4414

        code = colDataSetVal(pColInfo, pBlock->info.rows, pData, false);

×

4415

        QUERY_CHECK_CODE(code, lino, _end);

×

4419

  pBlock->info.dataLoad = 1;

×

4420

  pBlock->info.scanFlag = pInputBlock->info.scanFlag;

×

4421

  pBlock->info.rows += 1;

×

4423

_end:

×

4424

  if (code != TSDB_CODE_SUCCESS) {

×

4425

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

4427

  return code;

×

4430

static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,

×

4432

  int32_t                 code = TSDB_CODE_SUCCESS;

×

4433

  int32_t                 lino = 0;

×

4434

  STableMergeScanInfo*    pInfo = pOperator->info;

×

4435

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

×

4436

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

×

4437

  bool                    finished = false;

×

4439

  QRY_PARAM_CHECK(pResBlock);

×

4441

  blockDataCleanup(pBlock);

×

4443

  while (true) {

×

4445

      if (pSubTblsInfo->numSubTablesCompleted >= pSubTblsInfo->numSubTables) {

×

4446

        finished = true;

×

4447

        break;

×

4450

      code = appendChosenRowToDataBlock(pSubTblsInfo, pBlock);

×

4451

      QUERY_CHECK_CODE(code, lino, _end);

×

4453

      code = adjustSubTableForNextRow(pOperator, pSubTblsInfo);

×

4454

      QUERY_CHECK_CODE(code, lino, _end);

×

4456

      if (pBlock->info.rows >= capacity) {

×

4457

        break;

×

4461

    if (isTaskKilled(pTaskInfo)) {

×

4462

      return pTaskInfo->code;

×

4465

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pBlock, pTaskInfo);

×

4466

    if (finished || limitReached || pBlock->info.rows > 0) {

×

4471

  if (pBlock->info.rows > 0) {

×

4472

    *pResBlock = pBlock;

×

4475

_end:

×

4476

  if (code != TSDB_CODE_SUCCESS) {

×

4477

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

4478

    pTaskInfo->code = code;

×

4480

  return code;

×

4483

static int32_t startSubTablesTableMergeScan(SOperatorInfo* pOperator) {

×

4484

  int32_t              code = TSDB_CODE_SUCCESS;

×

4485

  int32_t              lino = 0;

×

4486

  STableMergeScanInfo* pInfo = pOperator->info;

×

4488

  code = initSubTablesMergeInfo(pInfo);

×

4489

  QUERY_CHECK_CODE(code, lino, _end);

×

4491

  code = initSubTableInputs(pOperator, pInfo);

×

4492

  QUERY_CHECK_CODE(code, lino, _end);

×

4494

  code = openSubTablesMergeSort(pInfo->pSubTablesMergeInfo);

×

4495

  QUERY_CHECK_CODE(code, lino, _end);

×

4497

_end:

×

4498

  if (code != TSDB_CODE_SUCCESS) {

×

4499

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

4501

  return code;

×

4504

static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {

×

4505

  STmsSubTablesMergeInfo* pSubTblsInfo = pInfo->pSubTablesMergeInfo;

×

4506

  if (pSubTblsInfo != NULL) {

×

4507

    tMergeTreeDestroy(&pSubTblsInfo->pTree);

×

4509

    for (int32_t i = 0; i < pSubTblsInfo->numSubTables; ++i) {

×

4510

      STmsSubTableInput* pInput = pSubTblsInfo->aInputs + i;

×

4511

      cleanupQueryTableDataCond(&pInput->tblCond);

×

4512

      blockDataDestroy(pInput->pReaderBlock);

×

4513

      blockDataDestroy(pInput->pPageBlock);

×

4514

      taosArrayDestroy(pInput->aBlockPages);

×

4515

      pInfo->base.readerAPI.tsdReaderClose(pInput->pReader);

×

4516

      pInput->pReader = NULL;

×

4519

    destroyDiskbasedBuf(pSubTblsInfo->pBlocksBuf);

×

4520

    taosMemoryFree(pSubTblsInfo->aInputs);

×

4522

    taosMemoryFree(pSubTblsInfo);

×

4523

    pInfo->pSubTablesMergeInfo = NULL;

×

4527

×

4529

void stopDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

×

4530

  STableMergeScanInfo* pInfo = pOperator->info;

×

4531

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

×

4532

  qDebug("%s table merge scan stop duration ", GET_TASKID(pTaskInfo));

×

4534

  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

×

4535

  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;

×

4536

  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;

×

4537

  pInfo->sortExecInfo.loops += sortExecInfo.loops;

×

4538

  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;

×

4539

  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

×

4541

  tsortDestroySortHandle(pInfo->pSortHandle);

×

4542

  pInfo->pSortHandle = NULL;

×

4543

×

4552

static int32_t createVTableMergeScanInfoFromBatchParam(SOperatorInfo* pOperator) {

×

4553

  int32_t                  code = TSDB_CODE_SUCCESS;

×

4554

  int32_t                  lino = 0;

×

4555

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

4556

  STableMergeScanInfo*     pInfo = pOperator->info;

×

4557

  SStorageAPI*             pAPI = &pTaskInfo->storageAPI;

×

4558

  STableListInfo*          pListInfo = pInfo->base.pTableListInfo;

×

4559

  SMetaReader              orgTable = {0};

×

4560

  SOperatorParam*          pGetParam = pOperator->pOperatorGetParam;

×

4561

  STableScanOperatorParam* pParam = (STableScanOperatorParam*)pGetParam->value;

×

4563

  if (pParam->paramType != DYN_TYPE_SCAN_PARAM) {

×

4564

    code = TSDB_CODE_INVALID_PARA;

×

4565

    QUERY_CHECK_CODE(code, lino, _return);

×

4567

  if (pInfo->pSubTablesMergeInfo) {

×

4568

    stopSubTablesTableMergeScan(pInfo);

×

4570

  if (pInfo->pSortHandle) {

×

4571

    stopDurationForGroupTableMergeScan(pOperator);

×

4573

  if (pInfo->base.dataReader) {

×

4574

    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);

×

4575

    pInfo->base.dataReader = NULL;

×

4578

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

×

4579

    if (pInfo->nextDurationBlocks[i]) {

×

4580

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

×

4581

      pInfo->nextDurationBlocks[i] = NULL;

×

4584

  pInfo->numNextDurationBlocks = 0;

×

4585

  pInfo->nextDurationBlocksIdx = 0;

×

4586

  pInfo->rtnNextDurationBlocks = false;

×

4587

  pInfo->bNewFilesetEvent = false;

×

4588

  pInfo->bNextDurationBlockEvent = false;

×

4589

  taosHashCleanup(pInfo->mSkipTables);

×

4590

  pInfo->mSkipTables = NULL;

×

4592

  if (pInfo->base.orgCond.colList && pInfo->base.orgCond.numOfCols > 0) {

×

4593

    cleanupQueryTableDataCond(&pInfo->base.cond);

×

4594

    code = dumpQueryTableCond(&pInfo->base.orgCond, &pInfo->base.cond);

×

4595

    QUERY_CHECK_CODE(code, lino, _return);

×

4598

  if (pParam->window.skey == INT64_MAX && pParam->window.ekey == INT64_MIN) {

×

4599

    if (pInfo->base.orgCond.colList && pInfo->base.orgCond.numOfCols > 0) {

×

4600

      pInfo->base.cond.twindows.skey = pInfo->base.orgCond.twindows.skey;

×

4601

      pInfo->base.cond.twindows.ekey = pInfo->base.orgCond.twindows.ekey;

×

4604

    pInfo->base.cond.twindows.skey = pParam->window.skey;

×

4605

    pInfo->base.cond.twindows.ekey = pParam->window.ekey;

×

4608

  pInfo->tableStartIndex = 0;

×

4609

  pInfo->tableEndIndex = 0;

×

4610

  pInfo->hasGroupId = false;

×

4611

  pInfo->groupId = pParam->groupid;

×

4612

  pInfo->bGroupProcessed = false;

×

4613

  pOperator->status = OP_NOT_OPENED;

×

4615

  pListInfo->oneTableForEachGroup = true;

×

4616

  pListInfo->numOfOuputGroups = 1;

×

4617

  taosHashClear(pListInfo->map);

×

4618

  taosArrayClear(pListInfo->pTableList);

×

4620

  int32_t tableIdx = 0;

×

4621

  if (pParam->pBatchTbInfo) {

×

4622

    int32_t numOfBatchTables = taosArrayGetSize(pParam->pBatchTbInfo);

×

4623

    for (int32_t i = 0; i < numOfBatchTables; ++i) {

×

4624

      SOrgTbInfo* pOrgTbInfo = taosArrayGet(pParam->pBatchTbInfo, i);

×

4625

      QUERY_CHECK_NULL(pOrgTbInfo, code, lino, _return, terrno);

×

4627

      const char* pTableName = strstr(pOrgTbInfo->tbName, ".");

×

4628

      pTableName = pTableName ? (pTableName + 1) : pOrgTbInfo->tbName;

×

4629

      pAPI->metaReaderFn.initReader(&orgTable, pInfo->base.readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

×

4630

      code = pAPI->metaReaderFn.getTableEntryByName(&orgTable, pTableName);

×

4631

      pAPI->metaReaderFn.readerReleaseLock(&orgTable);

×

4632

      QUERY_CHECK_CODE(code, lino, _return);

×

4634

      uint64_t uid = orgTable.me.uid;

×

4635

      code = taosHashPut(pListInfo->map, &uid, sizeof(uid), &tableIdx, sizeof(tableIdx));

×

4636

      if (code == TSDB_CODE_DUP_KEY) {

×

4637

        pAPI->metaReaderFn.clearReader(&orgTable);

×

4638

        continue;

×

4640

      QUERY_CHECK_CODE(code, lino, _return);

×

4642

      STableKeyInfo keyInfo = {.groupId = pParam->groupid, .uid = uid};

×

4643

      QUERY_CHECK_NULL(taosArrayPush(pListInfo->pTableList, &keyInfo), code, lino, _return, terrno);

×

4644

      ++tableIdx;

×

4646

      qDebug("%s table merge dynamic batch add table name:%s uid:%" PRIu64 " group:%" PRIu64, GET_TASKID(pTaskInfo),

×

4648

      pAPI->metaReaderFn.clearReader(&orgTable);

×

4652

  qDebug("%s table merge dynamic batch param consumed, group:%" PRIu64 ", tables:%d", GET_TASKID(pTaskInfo),

×

4655

_return:

×

4656

  if (code != TSDB_CODE_SUCCESS) {

×

4657

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

4659

  pAPI->metaReaderFn.clearReader(&orgTable);

×

4660

  return code;

×

4663

int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

×

4664

  int32_t              code = TSDB_CODE_SUCCESS;

×

4665

  int32_t              lino = 0;

×

4666

  int32_t              tableListSize = 0;

×

4667

  int64_t              st = taosGetTimestampUs();

×

4668

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

×

4669

  STableMergeScanInfo* pInfo = pOperator->info;

×

4671

  QRY_PARAM_CHECK(ppRes);

×

4673

  if (pOperator->status == OP_EXEC_DONE && !pOperator->pOperatorGetParam) {

×

4674

    *ppRes = NULL;

×

4675

    return TSDB_CODE_SUCCESS;

×

4678

  if (pOperator->pOperatorGetParam) {

×

4679

    code = createVTableMergeScanInfoFromBatchParam(pOperator);

×

4680

    QUERY_CHECK_CODE(code, lino, _end);

×

4682

    freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);

×

4683

    pOperator->pOperatorGetParam = NULL;

×

4684

    pOperator->dynamicTask = true;

×

4687

  code = pOperator->fpSet._openFn(pOperator);

×

4688

  QUERY_CHECK_CODE(code, lino, _end);

×

4690

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

×

4691

  QUERY_CHECK_CODE(code, lino, _end);

×

4693

  if (!pInfo->hasGroupId) {

×

4694

    pInfo->hasGroupId = true;

×

4696

    if (tableListSize == 0) {

×

4697

      setOperatorCompleted(pOperator);

×

4698

      (*ppRes) = NULL;

×

4699

      return code;

×

4702

    pInfo->tableStartIndex = 0;

×

4703

    STableKeyInfo* pTmpGpId = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

×

4704

    QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

×

4705

    pInfo->groupId = pTmpGpId->groupId;

×

4706

    code = startSubTablesTableMergeScan(pOperator);

×

4707

    QUERY_CHECK_CODE(code, lino, _end);

×

4710

  SSDataBlock* pBlock = NULL;

×

4711

  while (pInfo->tableStartIndex < tableListSize) {

×

4712

    if (isTaskKilled(pTaskInfo)) {

×

4713

      break;

×

4716

    code = getSubTablesSortedBlock(pOperator, pInfo->pResBlock, pOperator->resultInfo.capacity, &pBlock);

×

4717

    QUERY_CHECK_CODE(code, lino, _end);

×

4719

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

×

4720

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

×

4721

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

×

4723

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

×

4726

    if (pBlock != NULL) {

×

4727

      pBlock->info.id.groupId = pInfo->groupId;

×

4728

      pOperator->resultInfo.totalRows += pBlock->info.rows;

×

4729

      pInfo->bGroupProcessed = true;

×

4730

      break;

×

4733

      stopSubTablesTableMergeScan(pInfo);

×

4734

      if (pInfo->tableEndIndex >= tableListSize - 1) {

×

4735

        setOperatorCompleted(pOperator);

×

4736

        break;

×

4739

      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

×

4740

      STableKeyInfo* pTmpGpId = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

×

4741

      QUERY_CHECK_NULL(pTmpGpId, code, lino, _end, terrno);

×

4743

      pInfo->groupId = pTmpGpId->groupId;

×

4744

      code = startSubTablesTableMergeScan(pOperator);

×

4745

      QUERY_CHECK_CODE(code, lino, _end);

×

4746

      resetLimitInfoForNextGroup(&pInfo->limitInfo);

×

4750

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

×

4752

_end:

×

4753

  if (code != TSDB_CODE_SUCCESS) {

×

4754

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

4755

    pTaskInfo->code = code;

×

4756

    T_LONG_JMP(pTaskInfo->env, code);

×

4758

    (*ppRes) = pBlock;

×

4761

  return code;

×

4764

static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {

×

4765

  int32_t              code = TSDB_CODE_SUCCESS;

×

4766

  int32_t              lino = 0;

×

4767

  SOperatorInfo*       pOperator = (SOperatorInfo*)pTableMergeOpInfo;

×

4768

  STableMergeScanInfo* pInfo = pOperator->info;

×

4769

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

×

4771

  if (pInfo->mSkipTables == NULL) {

×

4772

    pInfo->mSkipTables = taosHashInit(pInfo->tableEndIndex - pInfo->tableStartIndex + 1,

×

4774

    QUERY_CHECK_NULL(pInfo->mSkipTables, code, lino, _end, terrno);

×

4776

  int bSkip = 1;

×

4777

  if (pInfo->mSkipTables != NULL) {

×

4778

    code = taosHashPut(pInfo->mSkipTables, &uid, sizeof(uid), &bSkip, sizeof(bSkip));

×

4779

    if (code == TSDB_CODE_DUP_KEY) {

×

4780

      code = TSDB_CODE_SUCCESS;

×

4782

    QUERY_CHECK_CODE(code, lino, _end);

×

4785

_end:

×

4786

  if (code != TSDB_CODE_SUCCESS) {

×

4787

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

4788

    pTaskInfo->code = code;

×

4789

    T_LONG_JMP(pTaskInfo->env, code);

×

4791

×

4793

static int32_t doGetBlockForTableMergeScan(SOperatorInfo* pOperator, bool* pFinished, bool* pSkipped) {

×

4794

  STableMergeScanInfo* pInfo = pOperator->info;

×

4795

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

×

4796

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

×

4797

  SSDataBlock*         pBlock = pInfo->pReaderBlock;

×

4798

  int32_t              code = 0;

×

4799

  bool                 hasNext = false;

×

4800

  STsdbReader*         reader = pInfo->base.dataReader;

×

4802

  code = pAPI->tsdReader.tsdNextDataBlock(reader, &hasNext);

×

4803

  if (code != 0) {

×

4804

    pAPI->tsdReader.tsdReaderReleaseDataBlock(reader);

×

4805

    qError("table merge scan fetch next data block error code: %d, %s", code, GET_TASKID(pTaskInfo));

×

4806

    pTaskInfo->code = code;

×

4807

    return code;

×

4810

  if (!hasNext || isTaskKilled(pTaskInfo)) {

×

4811

    if (isTaskKilled(pTaskInfo)) {

×

4812

      qInfo("table merge scan fetch next data block found task killed. %s", GET_TASKID(pTaskInfo));

×

4813

      pAPI->tsdReader.tsdReaderReleaseDataBlock(reader);

×

4815

    *pFinished = true;

×

4816

    return code;

×

4819

  uint32_t status = 0;

×

4820

  code = loadDataBlock(pOperator, &pInfo->base, pBlock, &status);

×

4822

  if (code != TSDB_CODE_SUCCESS) {

×

4823

    qInfo("table merge scan load datablock code %d, %s", code, GET_TASKID(pTaskInfo));

×

4824

    pTaskInfo->code = code;

×

4825

    return code;

×

4828

  if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {

×

4829

    *pFinished = true;

×

4830

    return code;

×

4834

  if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {

×

4835

    *pSkipped = true;

×

4836

    return code;

×

4839

  return code;

×

4842

static int32_t getBlockForTableMergeScan(void* param, SSDataBlock** ppBlock) {

×

4843

  STableMergeScanSortSourceParam* source = param;

×

4845

  SOperatorInfo*       pOperator = source->pOperator;

×

4846

  STableMergeScanInfo* pInfo = pOperator->info;

×

4847

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

×

4848

  SSDataBlock*         pBlock = NULL;

×

4849

  int64_t              st = taosGetTimestampUs();

×

4850

  int32_t              code = TSDB_CODE_SUCCESS;

×

4853

    if (pInfo->rtnNextDurationBlocks) {

×

4854

      qDebug("%s table merge scan return already fetched new duration blocks. index %d num of blocks %d",

×

4857

      if (pInfo->nextDurationBlocksIdx < pInfo->numNextDurationBlocks) {

×

4858

        pBlock = pInfo->nextDurationBlocks[pInfo->nextDurationBlocksIdx];

×

4859

        ++pInfo->nextDurationBlocksIdx;

×

4861

        for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

×

4862

          blockDataDestroy(pInfo->nextDurationBlocks[i]);

×

4863

          pInfo->nextDurationBlocks[i] = NULL;

×

4866

        pInfo->rtnNextDurationBlocks = false;

×

4867

        pInfo->nextDurationBlocksIdx = 0;

×

4868

        pInfo->numNextDurationBlocks = 0;

×

4869

        continue;

×

4872

      bool bFinished = false;

×

4873

      bool bSkipped = false;

×

4875

      code = doGetBlockForTableMergeScan(pOperator, &bFinished, &bSkipped);

×

4876

      if (code != 0) {

×

4877

        return code;

×

4880

      pBlock = pInfo->pReaderBlock;

×

4881

      qDebug("%s table merge scan fetch block. finished %d skipped %d next-duration-block %d new-fileset %d",

×

4883

      if (bFinished) {

×

4884

        pInfo->bNewFilesetEvent = false;

×

4885

        break;

×

4888

      if (pInfo->bNextDurationBlockEvent || pInfo->bNewFilesetEvent) {

×

4889

        if (!bSkipped) {

×

4890

          code = createOneDataBlock(pBlock, true, &pInfo->nextDurationBlocks[pInfo->numNextDurationBlocks]);

×

4891

          if (code) {

×

4892

            *ppBlock = NULL;

×

4893

            return code;

×

4896

          ++pInfo->numNextDurationBlocks;

×

4897

          if (pInfo->numNextDurationBlocks > 2) {

×

4898

            qError("%s table merge scan prefetch %d next duration blocks. end early.", GET_TASKID(pTaskInfo),

×

4900

            pInfo->bNewFilesetEvent = false;

×

4901

            break;

×

4905

        if (pInfo->bNewFilesetEvent) {

×

4906

          pInfo->rtnNextDurationBlocks = true;

×

4907

          *ppBlock = NULL;

×

4908

          return code;

×

4911

        if (pInfo->bNextDurationBlockEvent) {

×

4912

          pInfo->bNextDurationBlockEvent = false;

×

4913

          continue;

×

4916

      if (bSkipped) continue;

×

4919

    pBlock->info.id.groupId = tableListGetTableGroupId(pInfo->base.pTableListInfo, pBlock->info.id.uid);

×

4921

    pOperator->resultInfo.totalRows += pBlock->info.rows;

×

4922

    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

×

4923

    *ppBlock = pBlock;

×

4925

    return code;

×

4928

  *ppBlock = NULL;

×

4929

  return code;

×

4932

int32_t generateSortByTsPkInfo(SArray* colMatchInfo, int32_t order, SArray** ppSortArray) {

×

4933

  int32_t code = TSDB_CODE_SUCCESS;

×

4934

  int32_t lino = 0;

×

4935

  SArray* pSortInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));

×

4936

  QUERY_CHECK_NULL(pSortInfo, code, lino, _end, terrno);

×

4937

  SBlockOrderInfo biTs = {0};

×

4938

  SBlockOrderInfo biPk = {0};

×

4940

  int32_t tsTargetSlotId = 0;

×

4941

  int32_t pkTargetSlotId = -1;

×

4942

  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {

×

4943

    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);

×

4944

    QUERY_CHECK_NULL(colInfo, code, lino, _end, terrno);

×

4945

    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {

×

4946

      tsTargetSlotId = colInfo->dstSlotId;

×

4947

      biTs.order = order;

×

4948

      biTs.slotId = tsTargetSlotId;

×

4949

      biTs.nullFirst = (order == TSDB_ORDER_ASC);

×

4950

      biTs.compFn = getKeyComparFunc(TSDB_DATA_TYPE_TIMESTAMP, order);

×

4953

    if (colInfo->isPk) {

×

4954

      pkTargetSlotId = colInfo->dstSlotId;

×

4955

      biPk.order = order;

×

4956

      biPk.slotId = pkTargetSlotId;

×

4957

      biPk.nullFirst = (order == TSDB_ORDER_ASC);

×

4958

      biPk.compFn = getKeyComparFunc(colInfo->dataType.type, order);

×

4962

  void* tmp = taosArrayPush(pSortInfo, &biTs);

×

4963

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

4964

  if (pkTargetSlotId != -1) {

×

4965

    tmp = taosArrayPush(pSortInfo, &biPk);

×

4966

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

4969

  (*ppSortArray) = pSortInfo;

×

4971

_end:

×

4972

  if (code != TSDB_CODE_SUCCESS) {

×

4973

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

4975

  return code;

×

4978

void tableMergeScanTsdbNotifyCb(ETsdReaderNotifyType type, STsdReaderNotifyInfo* info, void* param) {

×

4979

  STableMergeScanInfo* pTmsInfo = param;

×

4980

  if (type == TSD_READER_NOTIFY_DURATION_START) {

×

4981

    pTmsInfo->bNewFilesetEvent = true;

×

4982

  } else if (type == TSD_READER_NOTIFY_NEXT_DURATION_BLOCK) {

×

4983

    pTmsInfo->bNextDurationBlockEvent = true;

×

4985

  qDebug("table merge scan receive notification. type %d, fileset %d", type, info->duration.filesetId);

×

4987

  return;

×

4990

int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {

×

4991

  STableMergeScanInfo* pInfo = pOperator->info;

×

4992

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

×

4994

  int32_t code = TSDB_CODE_SUCCESS;

×

4995

  int32_t lino = 0;

×

4996

  int32_t numOfTable = pInfo->tableEndIndex - pInfo->tableStartIndex + 1;

×

4998

  qDebug("%s table merge scan start duration ", GET_TASKID(pTaskInfo));

×

4999

  pInfo->bNewFilesetEvent = false;

×

5000

  pInfo->bNextDurationBlockEvent = false;

×

5002

  pInfo->sortBufSize = 2048 * pInfo->bufPageSize;

×

5003

  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;

×

5005

  pInfo->pSortHandle = NULL;

×

5006

  code = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,

×

5007

                               pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0, &pInfo->pSortHandle);

×

5008

  if (code) {

×

5009

    return code;

×

5012

  if (pInfo->bSortRowId && numOfTable != 1) {

×

5013

    int32_t memSize = 512 * 1024 * 1024;

×

5014

    code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);

×

5015

    QUERY_CHECK_CODE(code, lino, _end);

×

5018

  tsortSetMergeLimit(pInfo->pSortHandle, pInfo->mergeLimit);

×

5019

  tsortSetMergeLimitReachedFp(pInfo->pSortHandle, tableMergeScanDoSkipTable, pOperator);

×

5020

  tsortSetAbortCheckFn(pInfo->pSortHandle, isTaskKilled, pOperator->pTaskInfo);

×

5022

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getBlockForTableMergeScan, NULL, NULL);

×

5023

  QUERY_CHECK_CODE(code, lino, _end);

×

5025

  STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));

×

5026

  QUERY_CHECK_NULL(param, code, lino, _end, terrno);

×

5027

  param->pOperator = pOperator;

×

5029

  SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));

×

5030

  if (ps == NULL) {

×

5031

    taosMemoryFree(param);

×

5032

    QUERY_CHECK_NULL(ps, code, lino, _end, terrno);

×

5035

  ps->param = param;

×

5036

  ps->onlyRef = false;

×

5037

  code = tsortAddSource(pInfo->pSortHandle, ps);

×

5038

  QUERY_CHECK_CODE(code, lino, _end);

×

5040

  if (numOfTable == 1) {

×

5041

    tsortSetSingleTableMerge(pInfo->pSortHandle);

×

5043

    code = tsortOpen(pInfo->pSortHandle);

×

5044

    QUERY_CHECK_CODE(code, lino, _end);

×

5047

_end:

×

5048

  if (code != TSDB_CODE_SUCCESS) {

×

5049

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

5051

  return code;

×

5054

void startGroupTableMergeScan(SOperatorInfo* pOperator) {

×

5055

  int32_t              code = TSDB_CODE_SUCCESS;

×

5056

  int32_t              lino = 0;

×

5057

  STableMergeScanInfo* pInfo = pOperator->info;

×

5058

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

×

5059

  SReadHandle*         pHandle = &pInfo->base.readHandle;

×

5060

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

×

5061

  qDebug("%s table merge scan start group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

×

5064

    int32_t numOfTables = 0;

×

5065

    code = tableListGetSize(pInfo->base.pTableListInfo, &numOfTables);

×

5066

    QUERY_CHECK_CODE(code, lino, _end);

×

5068

    int32_t i = pInfo->tableStartIndex + 1;

×

5069

    for (; i < numOfTables; ++i) {

×

5070

      STableKeyInfo* tableKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, i);

×

5071

      QUERY_CHECK_NULL(tableKeyInfo, code, lino, _end, terrno);

×

5072

      if (tableKeyInfo->groupId != pInfo->groupId) {

×

5073

        break;

×

5076

    pInfo->tableEndIndex = i - 1;

×

5078

  pInfo->bGroupProcessed = false;

×

5079

  int32_t tableStartIdx = pInfo->tableStartIndex;

×

5080

  int32_t tableEndIdx = pInfo->tableEndIndex;

×

5082

  int32_t        numOfTable = tableEndIdx - tableStartIdx + 1;

×

5083

  STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);

×

5084

  code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock,

×

5085

                                       (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), &pInfo->mSkipTables);

×

5086

  QUERY_CHECK_CODE(code, lino, _end);

×

5087

  if (pInfo->filesetDelimited) {

×

5088

    pAPI->tsdReader.tsdSetFilesetDelimited(pInfo->base.dataReader);

×

5090

  pAPI->tsdReader.tsdSetSetNotifyCb(pInfo->base.dataReader, tableMergeScanTsdbNotifyCb, pInfo);

×

5092

  code = startDurationForGroupTableMergeScan(pOperator);

×

5093

  QUERY_CHECK_CODE(code, lino, _end);

×

5095

_end:

×

5096

  if (code != TSDB_CODE_SUCCESS) {

×

5097

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

5098

    pTaskInfo->code = code;

×

5099

    T_LONG_JMP(pTaskInfo->env, code);

×

5101

×

5103

void stopGroupTableMergeScan(SOperatorInfo* pOperator) {

×

5104

  STableMergeScanInfo* pInfo = pOperator->info;

×

5105

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

×

5106

  SStorageAPI*         pAPI = &pTaskInfo->storageAPI;

×

5108

  stopDurationForGroupTableMergeScan(pOperator);

×

5110

  if (pInfo->base.dataReader != NULL) {

×

5111

    pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);

×

5112

    pInfo->base.dataReader = NULL;

×

5114

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

×

5115

    if (pInfo->nextDurationBlocks[i]) {

×

5116

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

×

5117

      pInfo->nextDurationBlocks[i] = NULL;

×

5119

    pInfo->numNextDurationBlocks = 0;

×

5120

    pInfo->nextDurationBlocksIdx = 0;

×

5122

  resetLimitInfoForNextGroup(&pInfo->limitInfo);

×

5123

  taosHashCleanup(pInfo->mSkipTables);

×

5124

  pInfo->mSkipTables = NULL;

×

5125

  qDebug("%s table merge scan stop group %" PRIu64, GET_TASKID(pTaskInfo), pInfo->groupId);

×

5126

×

5130

SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,

×

5132

  STableMergeScanInfo* pInfo = pOperator->info;

×

5133

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

×

5134

  STupleHandle*        pTupleHandle = NULL;

×

5136

  blockDataCleanup(pResBlock);

×

5138

  while (1) {

×

5139

    while (1) {

×

5140

      pTupleHandle = NULL;

×

5141

      int32_t code = tsortNextTuple(pHandle, &pTupleHandle);

×

5142

      if (code != TSDB_CODE_SUCCESS) {

×

5143

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

5144

        T_LONG_JMP(pOperator->pTaskInfo->env, code);

×

5146

      if (pTupleHandle == NULL) {

×

5147

        break;

×

5150

      code = tsortAppendTupleToBlock(pInfo->pSortHandle, pResBlock, pTupleHandle);

×

5151

      if (code != TSDB_CODE_SUCCESS) {

×

5152

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

5153

        T_LONG_JMP(pOperator->pTaskInfo->env, terrno);

×

5156

      if (pResBlock->info.rows >= capacity) {

×

5157

        break;

×

5161

    if (tsortIsClosed(pHandle)) {

×

5162

      terrno = TSDB_CODE_TSC_QUERY_CANCELLED;

×

5163

      T_LONG_JMP(pOperator->pTaskInfo->env, terrno);

×

5166

    bool limitReached = applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo);

×

5167

    qDebug("%s get sorted row block, rows:%" PRId64 ", limit:%" PRId64, GET_TASKID(pTaskInfo), pResBlock->info.rows,

×

5169

    if (pTupleHandle == NULL || limitReached || pResBlock->info.rows > 0) {

×

5173

  return (pResBlock->info.rows > 0) ? pResBlock : NULL;

×

5176

int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

×

5177

  if (pOperator->status == OP_EXEC_DONE) {

×

5178

    (*ppRes) = NULL;

×

5179

    return TSDB_CODE_SUCCESS;

×

5182

  int32_t              code = TSDB_CODE_SUCCESS;

×

5183

  int32_t              lino = 0;

×

5184

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

×

5185

  STableMergeScanInfo* pInfo = pOperator->info;

×

5187

  code = pOperator->fpSet._openFn(pOperator);

×

5188

  QUERY_CHECK_CODE(code, lino, _end);

×

5190

  int64_t st = taosGetTimestampUs();

×

5192

  int32_t tableListSize = 0;

×

5193

  code = tableListGetSize(pInfo->base.pTableListInfo, &tableListSize);

×

5194

  QUERY_CHECK_CODE(code, lino, _end);

×

5196

  if (!pInfo->hasGroupId) {

×

5197

    pInfo->hasGroupId = true;

×

5199

    if (tableListSize == 0) {

×

5200

      setOperatorCompleted(pOperator);

×

5201

      (*ppRes) = NULL;

×

5202

      return code;

×

5204

    pInfo->tableStartIndex = 0;

×

5205

    STableKeyInfo* tmp = (STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

×

5206

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

5207

    pInfo->groupId = tmp->groupId;

×

5208

    startGroupTableMergeScan(pOperator);

×

5211

  SSDataBlock* pBlock = NULL;

×

5212

  while (pInfo->tableStartIndex < tableListSize) {

×

5213

    if (isTaskKilled(pTaskInfo)) {

×

5214

      goto _end;

×

5217

    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,

×

5219

    if (pBlock == NULL && !pInfo->bGroupProcessed && pInfo->needCountEmptyTable) {

×

5220

      STableKeyInfo* tbInfo = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

×

5221

      QUERY_CHECK_NULL(tbInfo, code, lino, _end, terrno);

×

5222

      pBlock = getOneRowResultBlock(pTaskInfo, &pInfo->base, pInfo->pResBlock, tbInfo);

×

5224

    if (pBlock != NULL) {

×

5225

      pBlock->info.id.groupId = pInfo->groupId;

×

5226

      pOperator->resultInfo.totalRows += pBlock->info.rows;

×

5227

      pInfo->bGroupProcessed = true;

×

5228

      break;

×

5230

      if (pInfo->bNewFilesetEvent) {

×

5231

        stopDurationForGroupTableMergeScan(pOperator);

×

5232

        code = startDurationForGroupTableMergeScan(pOperator);

×

5233

        QUERY_CHECK_CODE(code, lino, _end);

×

5236

        stopGroupTableMergeScan(pOperator);

×

5237

        if (pInfo->tableEndIndex >= tableListSize - 1) {

×

5238

          setOperatorCompleted(pOperator);

×

5239

          break;

×

5242

        pInfo->tableStartIndex = pInfo->tableEndIndex + 1;

×

5243

        STableKeyInfo* tmp = tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex);

×

5244

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

5245

        pInfo->groupId = tmp->groupId;

×

5246

        startGroupTableMergeScan(pOperator);

×

5247

        resetLimitInfoForNextGroup(&pInfo->limitInfo);

×

5252

  pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;

×

5254

_end:

×

5255

  if (code != TSDB_CODE_SUCCESS) {

×

5256

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

5257

    pTaskInfo->code = code;

×

5258

    T_LONG_JMP(pTaskInfo->env, code);

×

5260

    (*ppRes) = pBlock;

×

5263

  return code;

×

5266

static SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {

×

5267

  SSDataBlock* pRes = NULL;

×

5268

  int32_t      code = doTableMergeScanNext(pOperator, &pRes);

×

5269

  return pRes;

×

5272

void destroyTableMergeScanOperatorInfo(void* param) {

×

5273

  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;

×

5276

  if (pTableScanInfo->base.readerAPI.tsdReaderClose != NULL) {

×

5277

    pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);

×

5278

    pTableScanInfo->base.dataReader = NULL;

×

5281

  for (int32_t i = 0; i < pTableScanInfo->numNextDurationBlocks; ++i) {

×

5282

    if (pTableScanInfo->nextDurationBlocks[i] != NULL) {

×

5283

      blockDataDestroy(pTableScanInfo->nextDurationBlocks[i]);

×

5284

      pTableScanInfo->nextDurationBlocks[i] = NULL;

×

5288

  tsortDestroySortHandle(pTableScanInfo->pSortHandle);

×

5289

  pTableScanInfo->pSortHandle = NULL;

×

5290

  taosHashCleanup(pTableScanInfo->mSkipTables);

×

5291

  pTableScanInfo->mSkipTables = NULL;

×

5292

  blockDataDestroy(pTableScanInfo->pSortInputBlock);

×

5293

  pTableScanInfo->pSortInputBlock = NULL;

×

5296

  destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);

×

5298

  blockDataDestroy(pTableScanInfo->pResBlock);

×

5299

  pTableScanInfo->pResBlock = NULL;

×

5302

  blockDataDestroy(pTableScanInfo->pReaderBlock);

×

5303

  pTableScanInfo->pReaderBlock = NULL;

×

5304

  taosArrayDestroy(pTableScanInfo->pSortInfo);

×

5306

  stopSubTablesTableMergeScan(pTableScanInfo);

×

5308

  taosMemoryFreeClear(param);

×

5309

×

5311

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {

×

5312

  if (pOptr == NULL) {

×

5313

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_INVALID_PARA));

×

5314

    return TSDB_CODE_INVALID_PARA;

×

5317

  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));

×

5318

  if (!execInfo) {

×

5319

    return terrno;

×

5321

  STableMergeScanInfo* pInfo = pOptr->info;

×

5322

  execInfo->blockRecorder = pInfo->base.readRecorder;

×

5323

  execInfo->sortExecInfo = pInfo->sortExecInfo;

×

5325

  *pOptrExplain = execInfo;

×

5326

  *len = sizeof(STableMergeScanExecInfo);

×

5328

  return TSDB_CODE_SUCCESS;

×

5331

static int32_t resetTableMergeScanOperatorState(SOperatorInfo* pOper) {

×

5332

  int32_t         code = TSDB_CODE_SUCCESS;

×

5333

  STableMergeScanInfo*   pInfo = pOper->info;

×

5334

  pOper->status = OP_NOT_OPENED;

×

5336

  pInfo->tableEndIndex = 0;

×

5337

  pInfo->tableStartIndex = 0;

×

5338

  pInfo->hasGroupId = false;

×

5339

  if (pInfo->base.readerAPI.tsdReaderClose) {

×

5340

    pInfo->base.readerAPI.tsdReaderClose(pInfo->base.dataReader);

×

5342

  pInfo->base.dataReader = NULL;

×

5343

  pInfo->base.scanFlag = MAIN_SCAN;

×

5345

  pInfo->base.limitInfo = (SLimitInfo){0};

×

5346

  pInfo->base.limitInfo.limit.limit = -1;

×

5347

  pInfo->base.limitInfo.slimit.limit = -1;

×

5349

  tableListDestroy(pInfo->base.pTableListInfo);

×

5350

  pInfo->base.pTableListInfo = tableListCreate();

×

5351

  if (!pInfo->base.pTableListInfo) {

×

5352

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

5353

    return terrno;

×

5355

  SExecTaskInfo*         pTaskInfo = pOper->pTaskInfo;

×

5357

  STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pTaskInfo->pSubplan->pNode;

×

5358

  code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,

×

5360

                                  pTaskInfo->pSubplan->pTagCond, pTaskInfo->pSubplan->pTagIndexCond, pTaskInfo, NULL);

×

5361

  if (code) {

×

5362

    qError("%s failed to createScanTableListInfo, code:%s", __func__, tstrerror(code));

×

5363

    return code;

×

5366

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

×

5367

  cleanupQueryTableDataCond(&pInfo->base.cond);

×

5368

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, &pInfo->base.readHandle, false);

×

5369

  if (code) {

×

5370

    qError("%s failed to initQueryTableDataCond, code:%s", __func__, tstrerror(code));

×

5371

    return code;

×

5374

  tsortDestroySortHandle(pInfo->pSortHandle);

×

5375

  pInfo->pSortHandle = NULL;

×

5377

  taosHashCleanup(pInfo->mSkipTables);

×

5378

  pInfo->mSkipTables = NULL;

×

5379

  pOper->resultInfo.totalRows = 0;

×

5381

  pInfo->bGroupProcessed = false;

×

5382

  pInfo->bNewFilesetEvent = false;

×

5383

  pInfo->bNextDurationBlockEvent = false;

×

5384

  pInfo->rtnNextDurationBlocks = false;

×

5385

  pInfo->nextDurationBlocksIdx = 0;

×

5386

  pInfo->bSortRowId = false;

×

5388

  for (int32_t i = 0; i < pInfo->numNextDurationBlocks; ++i) {

×

5389

    if (pInfo->nextDurationBlocks[i] != NULL) {

×

5390

      blockDataDestroy(pInfo->nextDurationBlocks[i]);

×

5391

      pInfo->nextDurationBlocks[i] = NULL;

×

5394

  pInfo->numNextDurationBlocks = 0;

×

5396

  stopSubTablesTableMergeScan(pInfo);

×

5397

  return code;

×

5400

int32_t createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,

×

5403

  QRY_PARAM_CHECK(pOptrInfo);

×

5405

  int32_t              code = TSDB_CODE_SUCCESS;

×

5406

  int32_t              lino = 0;

×

5407

  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));

×

5408

  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

×

5409

  if (pInfo == NULL || pOperator == NULL) {

×

5410

    code = terrno;

×

5411

    goto _error;

×

5414

  pOperator->pPhyNode = pTableScanNode;

×

5415

  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

×

5417

  int32_t numOfCols = 0;

×

5418

  code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,

×

5420

  QUERY_CHECK_CODE(code, lino, _error);

×

5422

  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle, false);

×

5423

  QUERY_CHECK_CODE(code, lino, _error);

×

5425

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {

×

5426

    SExprSupp* pSup = &pInfo->base.pseudoSup;

×

5427

    code = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->pExprInfo, &pSup->numOfExprs);

×

5428

    QUERY_CHECK_CODE(code, lino, _error);

×

5430

    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset,

×

5432

    QUERY_CHECK_NULL(pSup->pCtx, code, lino, _error, terrno);

×

5435

  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};

×

5437

  if (tsMetaEntryCache && readHandle->streamRtInfo == NULL) {

×

5438

    pInfo->base.metaCache.pTableMetaEntryCache = taosLRUCacheInit(tsMetaEntryCacheSize, -1, .5);

×

5439

    QUERY_CHECK_NULL(pInfo->base.metaCache.pTableMetaEntryCache, code, lino, _error, terrno);

×

5442

  pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;

×

5443

  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;

×

5444

  pInfo->base.scanFlag = MAIN_SCAN;

×

5445

  pInfo->base.readHandle = *readHandle;

×

5447

  pInfo->readIdx = -1;

×

5449

  pInfo->base.limitInfo.limit.limit = -1;

×

5450

  pInfo->base.limitInfo.slimit.limit = -1;

×

5451

  pInfo->base.pTableListInfo = pTableListInfo;

×

5453

  pInfo->sample.sampleRatio = pTableScanNode->ratio;

×

5454

  pInfo->sample.seed = taosGetTimestampSec();

×

5456

  int32_t sortOrder = pInfo->base.cond.order;

×

5457

  if (pTableScanNode->scan.node.dynamicOp) {

×

5458

    memcpy(&pInfo->base.orgCond, &pInfo->base.cond, sizeof(SQueryTableDataCond));

×

5459

    sortOrder = pInfo->base.orgCond.order;

×

5460

    memset(&pInfo->base.cond, 0, sizeof(SQueryTableDataCond));

×

5463

  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,

×

5464

                            pTaskInfo->pStreamRuntimeInfo);

×

5465

  QUERY_CHECK_CODE(code, lino, _error);

×

5467

  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);

×

5469

  pInfo->mergeLimit = -1;

×

5470

  bool hasLimit = pInfo->limitInfo.limit.limit != -1 || pInfo->limitInfo.limit.offset != -1;

×

5471

  if (hasLimit) {

×

5472

    pInfo->mergeLimit = pInfo->limitInfo.limit.offset != -1

×

5473

                            ? pInfo->limitInfo.limit.limit + pInfo->limitInfo.limit.offset

×

5474

                            : pInfo->limitInfo.limit.limit;

×

5475

    pInfo->mSkipTables = NULL;

×

5478

  initResultSizeInfo(&pOperator->resultInfo, 1024);

×

5479

  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);

×

5480

  QUERY_CHECK_NULL(pInfo->pResBlock, code, lino, _error, terrno);

×

5481

  code = blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

×

5482

  QUERY_CHECK_CODE(code, lino, _error);

×

5483

  if (!hasLimit && blockDataGetRowSize(pInfo->pResBlock) >= 256 && !pTableScanNode->smallDataTsSort) {

×

5484

    pInfo->bSortRowId = true;

×

5486

    pInfo->bSortRowId = false;

×

5489

  code = prepareDataBlockBuf(pInfo->pResBlock, &pInfo->base.matchInfo);

×

5490

  QUERY_CHECK_CODE(code, lino, _error);

×

5492

  code = generateSortByTsPkInfo(pInfo->base.matchInfo.pList, sortOrder, &pInfo->pSortInfo);

×

5493

  QUERY_CHECK_CODE(code, lino, _error);

×

5495

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pReaderBlock);

×

5496

  QUERY_CHECK_CODE(code, lino, _error);

×

5498

  pInfo->needCountEmptyTable = tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable;

×

5500

  int32_t  rowSize = pInfo->pResBlock->info.rowSize;

×

5501

  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);

×

5503

  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);

×

5506

  code = createOneDataBlock(pInfo->pResBlock, false, &pInfo->pSortInputBlock);

×

5507

  QUERY_CHECK_CODE(code, lino, _error);

×

5509

  if (!tsExperimental) {

×

5510

    pInfo->filesetDelimited = false;

×

5512

    pInfo->filesetDelimited = pTableScanNode->filesetDelimited;

×

5515

  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,

×

5517

  pOperator->exprSupp.numOfExprs = numOfCols;

×

5519

  pOperator->fpSet = createOperatorFpSet(

×

5520

      optrDummyOpenFn, (pTableScanNode->paraTablesSort || pTableScanNode->scan.node.dynamicOp) ? doTableMergeScanParaSubTablesNext : doTableMergeScanNext, NULL,

×

5523

  setOperatorResetStateFn(pOperator, resetTableMergeScanOperatorState);

×

5524

  pOperator->cost.openCost = 0;

×

5526

  *pOptrInfo = pOperator;

×

5527

  return code;

×

5529

_error:

×

5530

  if (code != TSDB_CODE_SUCCESS) {

×

5531

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

5533

  pTaskInfo->code = code;

×

5534

  if (pInfo != NULL) {

×

5535

    pInfo->base.pTableListInfo = NULL;

×

5536

    destroyTableMergeScanOperatorInfo(pInfo);

×

5538

  if (pOperator != NULL) {

×

5539

    pOperator->info = NULL;

×

5540

    destroyOperator(pOperator);

×

5542

  return code;

×

5566

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {

×

5567

  if (scanCols != NULL) {

×

5568

    SNode* pNode = NULL;

×

5569

    FOREACH(pNode, scanCols) {

×

5570

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

×

5571

        return TSDB_CODE_QRY_SYS_ERROR;

×

5573

      STargetNode* targetNode = (STargetNode*)pNode;

×

5574

      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {

×

5575

        return TSDB_CODE_QRY_SYS_ERROR;

×

5577

      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);

×

5578

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

×

5579

        supp->dbNameSlotId = targetNode->slotId;

×

5580

      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

×

5581

        supp->stbNameSlotId = targetNode->slotId;

×

5585

  return TSDB_CODE_SUCCESS;

×

5588

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {

×

5589

  if (pseudoCols != NULL) {

×

5590

    SNode* pNode = NULL;

×

5591

    FOREACH(pNode, pseudoCols) {

×

5592

      if (nodeType(pNode) != QUERY_NODE_TARGET) {

×

5593

        return TSDB_CODE_QRY_SYS_ERROR;

×

5595

      STargetNode* targetNode = (STargetNode*)pNode;

×

5596

      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {

×

5597

        return TSDB_CODE_QRY_SYS_ERROR;

×

5599

      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);

×

5600

      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {

×

5601

        supp->tbCountSlotId = targetNode->slotId;

×

5605

  return TSDB_CODE_SUCCESS;

×

5608

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {

×

5609

  if (groupTags != NULL) {

×

5610

    SNode* pNode = NULL;

×

5611

    FOREACH(pNode, groupTags) {

×

5612

      if (nodeType(pNode) != QUERY_NODE_COLUMN) {

×

5613

        return TSDB_CODE_QRY_SYS_ERROR;

×

5615

      SColumnNode* colNode = (SColumnNode*)pNode;

×

5616

      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {

×

5617

        supp->groupByDbName = true;

×

5619

      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {

×

5620

        supp->groupByStbName = true;

×

5624

    tstrncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);

×

5625

    tstrncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);

×

5627

  return TSDB_CODE_SUCCESS;

×

5630

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,

×

5632

  int32_t code = 0;

×

5633

  code = tblCountScanGetInputs(groupTags, tableName, supp);

×

5634

  if (code != TSDB_CODE_SUCCESS) {

×

5635

    qError("%s get table count scan supp. get inputs error", GET_TASKID(taskInfo));

×

5636

    return code;

×

5639

  supp->dbNameSlotId = -1;

×

5640

  supp->stbNameSlotId = -1;

×

5641

  supp->tbCountSlotId = -1;

×

5643

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);

×

5644

  if (code != TSDB_CODE_SUCCESS) {

×

5645

    qError("%s get table count scan supp. get group tags slot id error", GET_TASKID(taskInfo));

×

5646

    return code;

×

5649

  code = tblCountScanGetCountSlotId(pseudoCols, supp);

×

5650

  if (code != TSDB_CODE_SUCCESS) {

×

5651

    qError("%s get table count scan supp. get count error", GET_TASKID(taskInfo));

×

5652

    return code;

×

5654

  return code;

×

5657

int32_t createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,

×

5659

  QRY_PARAM_CHECK(pOptrInfo);

×

5661

  int32_t                      code = TSDB_CODE_SUCCESS;

×

5662

  int32_t                      lino = 0;

×

5663

  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;

×

5664

  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));

×

5665

  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

×

5666

  if (!pInfo || !pOperator) {

×

5667

    code = terrno;

×

5668

    goto _error;

×

5671

  pInfo->readHandle = *readHandle;

×

5673

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;

×

5674

  initResultSizeInfo(&pOperator->resultInfo, 1);

×

5675

  pInfo->pRes = createDataBlockFromDescNode(pDescNode);

×

5676

  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

×

5678

  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

×

5679

  QUERY_CHECK_CODE(code, lino, _error);

×

5681

  code = getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,

×

5684

  QUERY_CHECK_CODE(code, lino, _error);

×

5686

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,

×

5688

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScanNext, NULL, destoryTableCountScanOperator,

×

5690

  *pOptrInfo = pOperator;

×

5691

  return code;

×

5693

_error:

×

5694

  if (pInfo != NULL) {

×

5695

    destoryTableCountScanOperator(pInfo);

×

5697

  if (pOperator != NULL) {

×

5698

    pOperator->info = NULL;

×

5699

    destroyOperator(pOperator);

×

5701

  pTaskInfo->code = code;

×

5702

  return code;

×

5705

int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,

×

5707

  int32_t code = TSDB_CODE_SUCCESS;

×

5708

  int32_t lino = 0;

×

5709

  if (pSupp->dbNameSlotId != -1) {

×

5710

    QUERY_CHECK_CONDITION((strlen(dbName) > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

5711

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);

×

5712

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

×

5714

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

5715

    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

×

5717

    varDataSetLen(varDbName, strlen(dbName));

×

5718

    code = colDataSetVal(colInfoData, 0, varDbName, false);

×

5719

    QUERY_CHECK_CODE(code, lino, _end);

×

5722

  if (pSupp->stbNameSlotId != -1) {

×

5723

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);

×

5724

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

×

5725

    if (strlen(stbName) != 0) {

×

5726

      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

5727

      tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);

×

5728

      varDataSetLen(varStbName, strlen(stbName));

×

5729

      code = colDataSetVal(colInfoData, 0, varStbName, false);

×

5730

      QUERY_CHECK_CODE(code, lino, _end);

×

5736

  if (pSupp->tbCountSlotId != -1) {

×

5737

    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);

×

5738

    QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);

×

5739

    code = colDataSetVal(colInfoData, 0, (char*)&count, false);

×

5740

    QUERY_CHECK_CODE(code, lino, _end);

×

5742

  pRes->info.rows = 1;

×

5744

_end:

×

5745

  if (code != TSDB_CODE_SUCCESS) {

×

5746

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

5748

  return code;

×

5751

static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {

×

5752

  STableCountScanSupp* pSupp = &pInfo->supp;

×

5753

  SSDataBlock*         pRes = pInfo->pRes;

×

5756

  getInfosDbMeta(NULL, &infodbTableNum);

×

5757

  infodbTableNum -= 1;

×

5759

  getPerfDbMeta(NULL, &perfdbTableNum);

×

5761

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

×

5762

    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);

×

5763

    return (pRes->info.rows > 0) ? pRes : NULL;

×

5765

    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);

×

5766

    return (pRes->info.rows > 0) ? pRes : NULL;

×

5770

static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,

×

5772

  int32_t        code = TSDB_CODE_SUCCESS;

×

5773

  int32_t        lino = 0;

×

5774

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

5775

  if (strlen(pSupp->stbNameFilter) != 0) {

×

5777

    code = fillTableCountScanDataBlock(pSupp, "", "", 0, pRes);

×

5778

    QUERY_CHECK_CODE(code, lino, _end);

×

5779

  } else if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {

×

5780

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

×

5781

    QUERY_CHECK_CODE(code, lino, _end);

×

5782

  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

×

5783

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

×

5784

    QUERY_CHECK_CODE(code, lino, _end);

×

5785

  } else if (strlen(pSupp->dbNameFilter) == 0) {

×

5786

    code = fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);

×

5787

    QUERY_CHECK_CODE(code, lino, _end);

×

5790

    code = fillTableCountScanDataBlock(pSupp, pSupp->dbNameFilter, pSupp->stbNameFilter, 0, pRes);

×

5791

    QUERY_CHECK_CODE(code, lino, _end);

×

5794

_end:

×

5795

  if (code != TSDB_CODE_SUCCESS) {

×

5796

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

5797

    pTaskInfo->code = code;

×

5798

    T_LONG_JMP(pTaskInfo->env, code);

×

5800

  setOperatorCompleted(pOperator);

×

5801

×

5803

static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

×

5806

  int32_t        code = TSDB_CODE_SUCCESS;

×

5807

  int32_t        lino = 0;

×

5808

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

5809

  if (pInfo->currGrpIdx == 0) {

×

5810

    uint64_t groupId = 0;

×

5811

    if (pSupp->groupByDbName) {

×

5812

      groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));

×

5814

      groupId = calcGroupId("", 0);

×

5817

    pRes->info.id.groupId = groupId;

×

5818

    code = fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);

×

5819

    QUERY_CHECK_CODE(code, lino, _end);

×

5820

  } else if (pInfo->currGrpIdx == 1) {

×

5821

    uint64_t groupId = 0;

×

5822

    if (pSupp->groupByDbName) {

×

5823

      groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));

×

5825

      groupId = calcGroupId("", 0);

×

5828

    pRes->info.id.groupId = groupId;

×

5829

    code = fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);

×

5830

    QUERY_CHECK_CODE(code, lino, _end);

×

5832

    setOperatorCompleted(pOperator);

×

5834

  pInfo->currGrpIdx++;

×

5836

_end:

×

5837

  if (code != TSDB_CODE_SUCCESS) {

×

5838

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

5839

    pTaskInfo->code = code;

×

5840

    T_LONG_JMP(pTaskInfo->env, code);

×

5842

×

5844

static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

×

5845

  int32_t                      code = TSDB_CODE_SUCCESS;

×

5846

  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;

×

5847

  STableCountScanOperatorInfo* pInfo = pOperator->info;

×

5848

  STableCountScanSupp*         pSupp = &pInfo->supp;

×

5849

  SSDataBlock*                 pRes = pInfo->pRes;

×

5851

  blockDataCleanup(pRes);

×

5852

  QRY_PARAM_CHECK(ppRes);

×

5854

  if (pOperator->status == OP_EXEC_DONE) {

×

5855

    return code;

×

5858

  if (pInfo->readHandle.mnd != NULL) {

×

5859

    (*ppRes) = buildSysDbTableCount(pOperator, pInfo);

×

5860

    return code;

×

5863

  code = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);

×

5864

  if (code != TSDB_CODE_SUCCESS) {

×

5865

    qError("%s failed since %s", __func__, tstrerror(code));

×

5866

    pTaskInfo->code = code;

×

5867

    T_LONG_JMP(pTaskInfo->env, code);

×

5869

  if (pRes->info.rows > 0) {

×

5870

    *ppRes = pRes;

×

5873

  return code;

×

5876

static int32_t buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

×

5878

  int32_t        code = TSDB_CODE_SUCCESS;

×

5879

  int32_t        lino = 0;

×

5880

  const char*    db = NULL;

×

5881

  int32_t        vgId = 0;

×

5882

  char           dbName[TSDB_DB_NAME_LEN] = {0};

×

5883

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

5884

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

5887

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);

×

5888

  SName sn = {0};

×

5890

  code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

×

5891

  QUERY_CHECK_CODE(code, lino, _end);

×

5893

  code = tNameGetDbName(&sn, dbName);

×

5894

  QUERY_CHECK_CODE(code, lino, _end);

×

5896

  if (pSupp->groupByDbName || pSupp->groupByStbName) {

×

5897

    code = buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);

×

5899

    code = buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);

×

5902

_end:

×

5903

  if (code != TSDB_CODE_SUCCESS) {

×

5904

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

5905

    pTaskInfo->code = code;

×

5908

  return code;

×

5911

static int32_t buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

×

5913

  int32_t        code = TSDB_CODE_SUCCESS;

×

5914

  int32_t        lino = 0;

×

5915

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

5916

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

5918

  if (pSupp->groupByStbName) {

×

5919

    if (pInfo->stbUidList == NULL) {

×

5920

      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));

×

5921

      QUERY_CHECK_NULL(pInfo->stbUidList, code, lino, _end, terrno);

×

5922

      code = pAPI->metaFn.storeGetTableList(pInfo->readHandle.vnode, TSDB_SUPER_TABLE, pInfo->stbUidList);

×

5923

      QUERY_CHECK_CODE(code, lino, _end);

×

5926

    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {

×

5927

      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);

×

5928

      code = buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid, pAPI);

×

5929

      QUERY_CHECK_CODE(code, lino, _end);

×

5931

      pInfo->currGrpIdx++;

×

5932

    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {

×

5933

      code = buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName, pAPI);

×

5934

      QUERY_CHECK_CODE(code, lino, _end);

×

5936

      pInfo->currGrpIdx++;

×

5938

      setOperatorCompleted(pOperator);

×

5941

    uint64_t groupId = calcGroupId(dbName, strlen(dbName));

×

5942

    pRes->info.id.groupId = groupId;

×

5944

    int64_t dbTableCount = 0;

×

5945

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &dbTableCount, NULL);

×

5946

    code = fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);

×

5947

    QUERY_CHECK_CODE(code, lino, _end);

×

5948

    setOperatorCompleted(pOperator);

×

5951

_end:

×

5952

  if (code != TSDB_CODE_SUCCESS) {

×

5953

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

5954

    pTaskInfo->code = code;

×

5956

  return code;

×

5959

static int32_t buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

×

5961

  int32_t        code = TSDB_CODE_SUCCESS;

×

5962

  int32_t        lino = 0;

×

5963

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

5964

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

5966

  if (strlen(pSupp->dbNameFilter) != 0 && strcmp(dbName, pSupp->dbNameFilter) != 0) {

×

5968

    code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, 0, pRes);

×

5969

    QUERY_CHECK_CODE(code, lino, _end);

×

5970

  } else if (strlen(pSupp->stbNameFilter) != 0) {

×

5971

    uint64_t uid = 0;

×

5972

    code = pAPI->metaFn.getTableUidByName(pInfo->readHandle.vnode, pSupp->stbNameFilter, &uid);

×

5973

    if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

×

5975

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, 0, pRes);

×

5976

      QUERY_CHECK_CODE(code, lino, _end);

×

5978

      QUERY_CHECK_CODE(code, lino, _end);

×

5980

      int64_t numOfChildTables = 0;

×

5981

      code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, uid, &numOfChildTables, NULL, NULL);

×

5982

      QUERY_CHECK_CODE(code, lino, _end);

×

5984

      code = fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, numOfChildTables, pRes);

×

5985

      QUERY_CHECK_CODE(code, lino, _end);

×

5988

    int64_t tbNumVnode = 0;

×

5989

    pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, &tbNumVnode, NULL);

×

5990

    code = fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);

×

5991

    QUERY_CHECK_CODE(code, lino, _end);

×

5994

_end:

×

5995

  if (code != TSDB_CODE_SUCCESS) {

×

5996

    pTaskInfo->code = code;

×

5997

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

6000

  setOperatorCompleted(pOperator);

×

6001

  return code;

×

6004

static int32_t buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

×

6006

  int32_t code = TSDB_CODE_SUCCESS;

×

6007

  int32_t lino = 0;

×

6008

  char    fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

×

6009

  if (pSupp->groupByDbName) {

×

6010

    snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");

×

6013

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

×

6014

  pRes->info.id.groupId = groupId;

×

6016

  int64_t numOfTables = 0;

×

6017

  pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);

×

6019

  if (numOfTables != 0) {

×

6020

    code = fillTableCountScanDataBlock(pSupp, dbName, "", numOfTables, pRes);

×

6023

_end:

×

6024

  if (code != TSDB_CODE_SUCCESS) {

×

6025

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

6027

  return code;

×

6030

static int32_t buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,

×

6032

  int32_t code = TSDB_CODE_SUCCESS;

×

6033

  int32_t lino = 0;

×

6034

  char    stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

6035

  code = pAPI->metaFn.getTableNameByUid(pInfo->readHandle.vnode, stbUid, stbName);

×

6036

  QUERY_CHECK_CODE(code, lino, _end);

×

6038

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};

×

6039

  if (pSupp->groupByDbName) {

×

6040

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, varDataVal(stbName));

×

6042

    (void)snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s", varDataVal(stbName));

×

6045

  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));

×

6046

  pRes->info.id.groupId = groupId;

×

6048

  int64_t ctbNum = 0;

×

6049

  code = pAPI->metaFn.getNumOfChildTables(pInfo->readHandle.vnode, stbUid, &ctbNum, NULL, NULL);

×

6050

  QUERY_CHECK_CODE(code, lino, _end);

×

6051

  code = fillTableCountScanDataBlock(pSupp, dbName, varDataVal(stbName), ctbNum, pRes);

×

6052

  QUERY_CHECK_CODE(code, lino, _end);

×

6054

_end:

×

6055

  if (code != TSDB_CODE_SUCCESS) {

×

6056

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

6058

  return code;

×

6061

static void destoryTableCountScanOperator(void* param) {

×

6062

  STableCountScanOperatorInfo* pTableCountScanInfo = param;

×

6063

  blockDataDestroy(pTableCountScanInfo->pRes);

×

6065

  taosArrayDestroy(pTableCountScanInfo->stbUidList);

×

6066

  taosMemoryFreeClear(param);

×

6067

×

taosdata / TDengine / #4986

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #4986

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