scanoperator.c 124.0 KB
Newer Older
H
Haojun Liao 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

16
#include "executorimpl.h"
H
Haojun Liao 已提交
17
#include "filter.h"
18
#include "function.h"
19
#include "functionMgt.h"
L
Liu Jicong 已提交
20
#include "os.h"
H
Haojun Liao 已提交
21
#include "querynodes.h"
22
#include "systable.h"
H
Haojun Liao 已提交
23
#include "tname.h"
24
#include "ttime.h"
H
Haojun Liao 已提交
25 26 27 28 29 30 31 32 33 34

#include "tdatablock.h"
#include "tmsg.h"

#include "query.h"
#include "tcompare.h"
#include "thash.h"
#include "ttypes.h"

#define SET_REVERSE_SCAN_FLAG(_info) ((_info)->scanFlag = REVERSE_SCAN)
35
#define SWITCH_ORDER(n)              (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC))
36

H
Haojun Liao 已提交
37 38 39 40 41 42 43 44 45 46 47 48
typedef struct STableMergeScanExecInfo {
  SFileBlockLoadRecorder blockRecorder;
  SSortExecInfo          sortExecInfo;
} STableMergeScanExecInfo;

typedef struct STableMergeScanSortSourceParam {
  SOperatorInfo* pOperator;
  int32_t        readerIdx;
  uint64_t       uid;
  SSDataBlock*   inputBlock;
} STableMergeScanSortSourceParam;

L
Liu Jicong 已提交
49
static bool processBlockWithProbability(const SSampleExecInfo* pInfo);
50

H
Haojun Liao 已提交
51
bool processBlockWithProbability(const SSampleExecInfo* pInfo) {
52 53 54 55 56 57 58 59 60 61 62 63
#if 0
  if (pInfo->sampleRatio == 1) {
    return true;
  }

  uint32_t val = taosRandR((uint32_t*) &pInfo->seed);
  return (val % ((uint32_t)(1/pInfo->sampleRatio))) == 0;
#else
  return true;
#endif
}

64
static void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
H
Haojun Liao 已提交
65 66 67 68 69
  for (int32_t i = 0; i < numOfOutput; ++i) {
    SWITCH_ORDER(pCtx[i].order);
  }
}

70 71 72 73 74 75 76 77 78
static void getNextTimeWindow(SInterval* pInterval, STimeWindow* tw, int32_t order) {
  int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);
  if (pInterval->intervalUnit != 'n' && pInterval->intervalUnit != 'y') {
    tw->skey += pInterval->sliding * factor;
    tw->ekey = tw->skey + pInterval->interval - 1;
    return;
  }

  int64_t key = tw->skey, interval = pInterval->interval;
79
  // convert key to second
80 81 82 83 84 85 86
  key = convertTimePrecision(key, pInterval->precision, TSDB_TIME_PRECISION_MILLI) / 1000;

  if (pInterval->intervalUnit == 'y') {
    interval *= 12;
  }

  struct tm tm;
87
  time_t    t = (time_t)key;
88 89 90 91 92
  taosLocalTime(&t, &tm);

  int mon = (int)(tm.tm_year * 12 + tm.tm_mon + interval * factor);
  tm.tm_year = mon / 12;
  tm.tm_mon = mon % 12;
wafwerar's avatar
wafwerar 已提交
93
  tw->skey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, pInterval->precision);
94 95 96 97

  mon = (int)(mon + interval);
  tm.tm_year = mon / 12;
  tm.tm_mon = mon % 12;
wafwerar's avatar
wafwerar 已提交
98
  tw->ekey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, pInterval->precision);
99 100 101 102

  tw->ekey -= 1;
}

103
static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockInfo, int32_t order) {
104 105 106 107 108 109 110
  STimeWindow w = {0};

  // 0 by default, which means it is not a interval operator of the upstream operator.
  if (pInterval->interval == 0) {
    return false;
  }

111
  if (order == TSDB_ORDER_ASC) {
112
    w = getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.skey);
113
    ASSERT(w.ekey >= pBlockInfo->window.skey);
114

115
    if (w.ekey < pBlockInfo->window.ekey) {
116 117 118
      return true;
    }

119 120
    while (1) {
      getNextTimeWindow(pInterval, &w, order);
121 122 123 124
      if (w.skey > pBlockInfo->window.ekey) {
        break;
      }

125
      ASSERT(w.ekey > pBlockInfo->window.ekey);
126
      if (TMAX(w.skey, pBlockInfo->window.skey) <= pBlockInfo->window.ekey) {
127 128 129 130
        return true;
      }
    }
  } else {
131
    w = getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.ekey);
132
    ASSERT(w.skey <= pBlockInfo->window.ekey);
133

134
    if (w.skey > pBlockInfo->window.skey) {
135 136 137
      return true;
    }

138
    while (1) {
139 140 141 142 143 144
      getNextTimeWindow(pInterval, &w, order);
      if (w.ekey < pBlockInfo->window.skey) {
        break;
      }

      assert(w.skey < pBlockInfo->window.skey);
145
      if (pBlockInfo->window.skey <= TMIN(w.ekey, pBlockInfo->window.ekey)) {
146 147 148
        return true;
      }
    }
149 150 151 152 153
  }

  return false;
}

154 155 156 157 158 159 160 161 162 163 164
// this function is for table scanner to extract temporary results of upstream aggregate results.
static SResultRow* getTableGroupOutputBuf(SOperatorInfo* pOperator, uint64_t groupId, SFilePage** pPage) {
  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
    return NULL;
  }

  int64_t buf[2] = {0};
  SET_RES_WINDOW_KEY((char*)buf, &groupId, sizeof(groupId), groupId);

  STableScanInfo* pTableScanInfo = pOperator->info;

S
slzhou 已提交
165 166
  SResultRowPosition* p1 = (SResultRowPosition*)tSimpleHashGet(pTableScanInfo->base.pdInfo.pAggSup->pResultRowHashTable,
                                                               buf, GET_RES_WINDOW_KEY_LEN(sizeof(groupId)));
167 168 169 170 171

  if (p1 == NULL) {
    return NULL;
  }

H
Haojun Liao 已提交
172
  *pPage = getBufPage(pTableScanInfo->base.pdInfo.pAggSup->pResultBuf, p1->pageId);
173 174 175 176 177 178
  return (SResultRow*)((char*)(*pPage) + p1->offset);
}

static int32_t doDynamicPruneDataBlock(SOperatorInfo* pOperator, SDataBlockInfo* pBlockInfo, uint32_t* status) {
  STableScanInfo* pTableScanInfo = pOperator->info;

H
Haojun Liao 已提交
179
  if (pTableScanInfo->base.pdInfo.pExprSup == NULL) {
180 181 182
    return TSDB_CODE_SUCCESS;
  }

H
Haojun Liao 已提交
183
  SExprSupp* pSup1 = pTableScanInfo->base.pdInfo.pExprSup;
184 185

  SFilePage*  pPage = NULL;
H
Haojun Liao 已提交
186
  SResultRow* pRow = getTableGroupOutputBuf(pOperator, pBlockInfo->id.groupId, &pPage);
187 188 189 190 191 192 193 194 195

  if (pRow == NULL) {
    return TSDB_CODE_SUCCESS;
  }

  bool notLoadBlock = true;
  for (int32_t i = 0; i < pSup1->numOfExprs; ++i) {
    int32_t functionId = pSup1->pCtx[i].functionId;

H
Haojun Liao 已提交
196
    SResultRowEntryInfo* pEntry = getResultEntryInfo(pRow, i, pTableScanInfo->base.pdInfo.pExprSup->rowEntryInfoOffset);
197 198 199 200 201 202 203 204 205

    int32_t reqStatus = fmFuncDynDataRequired(functionId, pEntry, &pBlockInfo->window);
    if (reqStatus != FUNC_DATA_REQUIRED_NOT_LOAD) {
      notLoadBlock = false;
      break;
    }
  }

  // release buffer pages
H
Haojun Liao 已提交
206
  releaseBufPage(pTableScanInfo->base.pdInfo.pAggSup->pResultBuf, pPage);
207 208 209 210 211 212 213 214

  if (notLoadBlock) {
    *status = FUNC_DATA_REQUIRED_NOT_LOAD;
  }

  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
215
static bool doFilterByBlockSMA(SFilterInfo* pFilterInfo, SColumnDataAgg** pColsAgg, int32_t numOfCols,
216
                               int32_t numOfRows) {
H
Haojun Liao 已提交
217
  if (pColsAgg == NULL || pFilterInfo == NULL) {
H
Haojun Liao 已提交
218 219 220
    return true;
  }

H
Haojun Liao 已提交
221
  bool keep = filterRangeExecute(pFilterInfo, pColsAgg, numOfCols, numOfRows);
H
Haojun Liao 已提交
222 223 224
  return keep;
}

H
Haojun Liao 已提交
225
static bool doLoadBlockSMA(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
226
  bool    allColumnsHaveAgg = true;
227
  int32_t code = tsdbRetrieveDatablockSMA(pTableScanInfo->dataReader, pBlock, &allColumnsHaveAgg);
H
Haojun Liao 已提交
228
  if (code != TSDB_CODE_SUCCESS) {
229
    T_LONG_JMP(pTaskInfo->env, code);
H
Haojun Liao 已提交
230 231 232 233 234 235
  }

  if (!allColumnsHaveAgg) {
    return false;
  }

236
#if 0
H
Haojun Liao 已提交
237 238 239 240 241 242 243
  //  if (allColumnsHaveAgg == true) {
  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

  // todo create this buffer during creating operator
  if (pBlock->pBlockAgg == NULL) {
    pBlock->pBlockAgg = taosMemoryCalloc(numOfCols, POINTER_BYTES);
    if (pBlock->pBlockAgg == NULL) {
244
      T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
H
Haojun Liao 已提交
245 246 247
    }
  }

248 249
  size_t num = taosArrayGetSize(pTableScanInfo->matchInfo.pList);
  for (int32_t i = 0; i < num; ++i) {
H
Haojun Liao 已提交
250 251
    SColMatchItem* pColMatchInfo = taosArrayGet(pTableScanInfo->matchInfo.pList, i);
    if (!pColMatchInfo->needOutput) {
H
Haojun Liao 已提交
252 253
      continue;
    }
H
Haojun Liao 已提交
254 255

    pBlock->pBlockAgg[pColMatchInfo->dstSlotId] = pColAgg[i];
H
Haojun Liao 已提交
256
  }
257
#endif
H
Haojun Liao 已提交
258 259 260 261

  return true;
}

H
Haojun Liao 已提交
262
static void doSetTagColumnData(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo,
263
                               int32_t rows) {
H
Haojun Liao 已提交
264 265 266
  if (pTableScanInfo->pseudoSup.numOfExprs > 0) {
    SExprSupp* pSup = &pTableScanInfo->pseudoSup;

267
    int32_t code = addTagPseudoColumnData(&pTableScanInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pBlock, rows,
268
                                          GET_TASKID(pTaskInfo), &pTableScanInfo->metaCache);
H
Haojun Liao 已提交
269
    // ignore the table not exists error, since this table may have been dropped during the scan procedure.
H
Haojun Liao 已提交
270
    if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_PAR_TABLE_NOT_EXIST) {
H
Haojun Liao 已提交
271 272
      T_LONG_JMP(pTaskInfo->env, code);
    }
H
Haojun Liao 已提交
273 274 275

    // reset the error code.
    terrno = 0;
H
Haojun Liao 已提交
276 277 278
  }
}

279 280
// todo handle the slimit info
void applyLimitOffset(SLimitInfo* pLimitInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo, SOperatorInfo* pOperator) {
281
  SLimit*     pLimit = &pLimitInfo->limit;
H
Haojun Liao 已提交
282
  const char* id = GET_TASKID(pTaskInfo);
283 284 285 286

  if (pLimit->offset > 0 && pLimitInfo->remainOffset > 0) {
    if (pLimitInfo->remainOffset >= pBlock->info.rows) {
      pLimitInfo->remainOffset -= pBlock->info.rows;
H
Haojun Liao 已提交
287
      blockDataEmpty(pBlock);
H
Haojun Liao 已提交
288
      qDebug("current block ignore due to offset, current:%" PRId64 ", %s", pLimitInfo->remainOffset, id);
289 290 291 292 293 294 295 296 297 298 299 300
    } else {
      blockDataTrimFirstNRows(pBlock, pLimitInfo->remainOffset);
      pLimitInfo->remainOffset = 0;
    }
  }

  if (pLimit->limit != -1 && pLimit->limit <= (pLimitInfo->numOfOutputRows + pBlock->info.rows)) {
    // limit the output rows
    int32_t overflowRows = pLimitInfo->numOfOutputRows + pBlock->info.rows - pLimit->limit;
    int32_t keep = pBlock->info.rows - overflowRows;

    blockDataKeepFirstNRows(pBlock, keep);
H
Haojun Liao 已提交
301
    qDebug("output limit %" PRId64 " has reached, %s", pLimit->limit, id);
302 303 304 305
    pOperator->status = OP_EXEC_DONE;
  }
}

H
Haojun Liao 已提交
306
static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableScanInfo, SSDataBlock* pBlock,
L
Liu Jicong 已提交
307
                             uint32_t* status) {
S
slzhou 已提交
308
  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;
309
  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;
H
Haojun Liao 已提交
310 311

  pCost->totalBlocks += 1;
312
  pCost->totalRows += pBlock->info.rows;
313

H
Haojun Liao 已提交
314
  bool loadSMA = false;
H
Haojun Liao 已提交
315
  *status = pTableScanInfo->dataBlockLoadFlag;
H
Haojun Liao 已提交
316
  if (pOperator->exprSupp.pFilterInfo != NULL ||
317
      overlapWithTimeWindow(&pTableScanInfo->pdInfo.interval, &pBlock->info, pTableScanInfo->cond.order)) {
318 319 320 321
    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;
  }

  SDataBlockInfo* pBlockInfo = &pBlock->info;
322
  taosMemoryFreeClear(pBlock->pBlockAgg);
323 324

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {
325 326
    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
327
    pCost->filterOutBlocks += 1;
328
    pCost->totalRows += pBlock->info.rows;
329 330
    return TSDB_CODE_SUCCESS;
  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {
331 332
    qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
333
    doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, 1);
334 335
    pCost->skipBlocks += 1;
    return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
336
  } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {
337
    pCost->loadBlockStatis += 1;
L
Liu Jicong 已提交
338
    loadSMA = true;  // mark the operation of load sma;
H
Haojun Liao 已提交
339
    bool success = doLoadBlockSMA(pTableScanInfo, pBlock, pTaskInfo);
L
Liu Jicong 已提交
340
    if (success) {  // failed to load the block sma data, data block statistics does not exist, load data block instead
341 342
      qDebug("%s data block SMA loaded, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
             pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
343
      doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, 1);
344 345
      return TSDB_CODE_SUCCESS;
    } else {
346
      qDebug("%s failed to load SMA, since not all columns have SMA", GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
347
      *status = FUNC_DATA_REQUIRED_DATA_LOAD;
348
    }
H
Haojun Liao 已提交
349
  }
350

H
Haojun Liao 已提交
351
  ASSERT(*status == FUNC_DATA_REQUIRED_DATA_LOAD);
352

H
Haojun Liao 已提交
353
  // try to filter data block according to sma info
H
Haojun Liao 已提交
354
  if (pOperator->exprSupp.pFilterInfo != NULL && (!loadSMA)) {
355 356 357
    bool success = doLoadBlockSMA(pTableScanInfo, pBlock, pTaskInfo);
    if (success) {
      size_t size = taosArrayGetSize(pBlock->pDataBlock);
H
Haojun Liao 已提交
358
      bool   keep = doFilterByBlockSMA(pOperator->exprSupp.pFilterInfo, pBlock->pBlockAgg, size, pBlockInfo->rows);
359 360 361 362 363 364 365 366
      if (!keep) {
        qDebug("%s data block filter out by block SMA, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
               pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
        pCost->filterOutBlocks += 1;
        (*status) = FUNC_DATA_REQUIRED_FILTEROUT;

        return TSDB_CODE_SUCCESS;
      }
367
    }
H
Haojun Liao 已提交
368
  }
369

370 371 372
  // free the sma info, since it should not be involved in later computing process.
  taosMemoryFreeClear(pBlock->pBlockAgg);

373
  // try to filter data block according to current results
374 375
  doDynamicPruneDataBlock(pOperator, pBlockInfo, status);
  if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {
376
    qDebug("%s data block skipped due to dynamic prune, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
377 378 379
           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
    pCost->skipBlocks += 1;

380
    *status = FUNC_DATA_REQUIRED_FILTEROUT;
381 382 383
    return TSDB_CODE_SUCCESS;
  }

H
Haojun Liao 已提交
384 385
  pCost->totalCheckedRows += pBlock->info.rows;
  pCost->loadBlocks += 1;
386

H
Haojun Liao 已提交
387 388
  SSDataBlock* p = tsdbRetrieveDataBlock(pTableScanInfo->dataReader, NULL);
  if (p == NULL) {
H
Haojun Liao 已提交
389
    return terrno;
H
Haojun Liao 已提交
390 391
  }

H
Haojun Liao 已提交
392
  ASSERT(p == pBlock);
393
  doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);
394

H
Haojun Liao 已提交
395 396
  // restore the previous value
  pCost->totalRows -= pBlock->info.rows;
397

H
Haojun Liao 已提交
398
  if (pOperator->exprSupp.pFilterInfo != NULL) {
399
    int64_t st = taosGetTimestampUs();
H
Haojun Liao 已提交
400
    doFilter(pBlock, pOperator->exprSupp.pFilterInfo, &pTableScanInfo->matchInfo);
401

402 403
    double el = (taosGetTimestampUs() - st) / 1000.0;
    pTableScanInfo->readRecorder.filterTime += el;
404

405 406 407 408 409 410 411
    if (pBlock->info.rows == 0) {
      pCost->filterOutBlocks += 1;
      qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d, elapsed time:%.2f ms",
             GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, el);
    } else {
      qDebug("%s data block filter applied, elapsed time:%.2f ms", GET_TASKID(pTaskInfo), el);
    }
412 413
  }

H
Haojun Liao 已提交
414
  applyLimitOffset(&pTableScanInfo->limitInfo, pBlock, pTaskInfo, pOperator);
415

H
Haojun Liao 已提交
416
  pCost->totalRows += pBlock->info.rows;
H
Haojun Liao 已提交
417
  pTableScanInfo->limitInfo.numOfOutputRows = pCost->totalRows;
H
Haojun Liao 已提交
418 419 420
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
421
static void prepareForDescendingScan(STableScanBase* pTableScanInfo, SqlFunctionCtx* pCtx, int32_t numOfOutput) {
H
Haojun Liao 已提交
422 423 424
  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

  switchCtxOrder(pCtx, numOfOutput);
425
  pTableScanInfo->cond.order = TSDB_ORDER_DESC;
H
Haojun Liao 已提交
426 427
  STimeWindow* pTWindow = &pTableScanInfo->cond.twindows;
  TSWAP(pTWindow->skey, pTWindow->ekey);
H
Haojun Liao 已提交
428 429
}

430 431
typedef struct STableCachedVal {
  const char* pName;
432
  STag*       pTags;
433 434
} STableCachedVal;

435 436 437 438 439 440 441 442 443 444 445
static void freeTableCachedVal(void* param) {
  if (param == NULL) {
    return;
  }

  STableCachedVal* pVal = param;
  taosMemoryFree((void*)pVal->pName);
  taosMemoryFree(pVal->pTags);
  taosMemoryFree(pVal);
}

H
Haojun Liao 已提交
446 447 448 449 450 451 452
static STableCachedVal* createTableCacheVal(const SMetaReader* pMetaReader) {
  STableCachedVal* pVal = taosMemoryMalloc(sizeof(STableCachedVal));
  pVal->pName = strdup(pMetaReader->me.name);
  pVal->pTags = NULL;

  // only child table has tag value
  if (pMetaReader->me.type == TSDB_CHILD_TABLE) {
453
    STag* pTag = (STag*)pMetaReader->me.ctbEntry.pTags;
H
Haojun Liao 已提交
454 455 456 457 458 459 460
    pVal->pTags = taosMemoryMalloc(pTag->len);
    memcpy(pVal->pTags, pTag, pTag->len);
  }

  return pVal;
}

461 462
// const void *key, size_t keyLen, void *value
static void freeCachedMetaItem(const void* key, size_t keyLen, void* value) { freeTableCachedVal(value); }
463

464 465
int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock,
                               int32_t rows, const char* idStr, STableMetaCacheInfo* pCache) {
466
  // currently only the tbname pseudo column
467
  if (numOfExpr <= 0) {
H
Haojun Liao 已提交
468
    return TSDB_CODE_SUCCESS;
469 470
  }

471 472
  int32_t code = 0;

473 474 475 476
  // backup the rows
  int32_t backupRows = pBlock->info.rows;
  pBlock->info.rows = rows;

477
  bool            freeReader = false;
478
  STableCachedVal val = {0};
479 480

  SMetaReader mr = {0};
481
  LRUHandle*  h = NULL;
482

483
  // 1. check if it is existed in meta cache
484
  if (pCache == NULL) {
485
    metaReaderInit(&mr, pHandle->meta, 0);
H
Haojun Liao 已提交
486
    code = metaGetTableEntryByUidCache(&mr, pBlock->info.id.uid);
487
    if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
488
      if (terrno == TSDB_CODE_PAR_TABLE_NOT_EXIST) {
S
slzhou 已提交
489 490
        qWarn("failed to get table meta, table may have been dropped, uid:0x%" PRIx64 ", code:%s, %s",
              pBlock->info.id.uid, tstrerror(terrno), idStr);
H
Haojun Liao 已提交
491
      } else {
S
slzhou 已提交
492 493
        qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid, tstrerror(terrno),
               idStr);
H
Haojun Liao 已提交
494
      }
495 496 497 498 499
      metaReaderClear(&mr);
      return terrno;
    }

    metaReaderReleaseLock(&mr);
500

501 502
    val.pName = mr.me.name;
    val.pTags = (STag*)mr.me.ctbEntry.pTags;
503 504

    freeReader = true;
505
  } else {
506 507
    pCache->metaFetch += 1;

H
Haojun Liao 已提交
508
    h = taosLRUCacheLookup(pCache->pTableMetaEntryCache, &pBlock->info.id.uid, sizeof(pBlock->info.id.uid));
509 510
    if (h == NULL) {
      metaReaderInit(&mr, pHandle->meta, 0);
H
Haojun Liao 已提交
511
      code = metaGetTableEntryByUidCache(&mr, pBlock->info.id.uid);
512
      if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
513
        if (terrno == TSDB_CODE_PAR_TABLE_NOT_EXIST) {
514
          qWarn("failed to get table meta, table may have been dropped, uid:0x%" PRIx64 ", code:%s, %s",
H
Haojun Liao 已提交
515
                pBlock->info.id.uid, tstrerror(terrno), idStr);
H
Haojun Liao 已提交
516
        } else {
H
Haojun Liao 已提交
517
          qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid, tstrerror(terrno),
518
                 idStr);
H
Haojun Liao 已提交
519
        }
520 521 522 523 524 525
        metaReaderClear(&mr);
        return terrno;
      }

      metaReaderReleaseLock(&mr);

H
Haojun Liao 已提交
526
      STableCachedVal* pVal = createTableCacheVal(&mr);
527

H
Haojun Liao 已提交
528
      val = *pVal;
529
      freeReader = true;
H
Haojun Liao 已提交
530

H
Haojun Liao 已提交
531
      int32_t ret = taosLRUCacheInsert(pCache->pTableMetaEntryCache, &pBlock->info.id.uid, sizeof(uint64_t), pVal,
532
                                       sizeof(STableCachedVal), freeCachedMetaItem, NULL, TAOS_LRU_PRIORITY_LOW);
533 534 535 536 537 538 539 540
      if (ret != TAOS_LRU_STATUS_OK) {
        qError("failed to put meta into lru cache, code:%d, %s", ret, idStr);
        freeTableCachedVal(pVal);
      }
    } else {
      pCache->cacheHit += 1;
      STableCachedVal* pVal = taosLRUCacheValue(pCache->pTableMetaEntryCache, h);
      val = *pVal;
H
Haojun Liao 已提交
541

H
Haojun Liao 已提交
542
      taosLRUCacheRelease(pCache->pTableMetaEntryCache, h, false);
543
    }
H
Haojun Liao 已提交
544

545 546
    qDebug("retrieve table meta from cache:%" PRIu64 ", hit:%" PRIu64 " miss:%" PRIu64 ", %s", pCache->metaFetch,
           pCache->cacheHit, (pCache->metaFetch - pCache->cacheHit), idStr);
H
Haojun Liao 已提交
547
  }
548

549 550
  for (int32_t j = 0; j < numOfExpr; ++j) {
    const SExprInfo* pExpr1 = &pExpr[j];
551
    int32_t          dstSlotId = pExpr1->base.resSchema.slotId;
552 553

    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotId);
D
dapan1121 已提交
554
    colInfoDataCleanup(pColInfoData, pBlock->info.rows);
555

556
    int32_t functionId = pExpr1->pExpr->_function.functionId;
557 558 559

    // this is to handle the tbname
    if (fmIsScanPseudoColumnFunc(functionId)) {
560
      setTbNameColData(pBlock, pColInfoData, functionId, val.pName);
561
    } else {  // these are tags
wmmhello's avatar
wmmhello 已提交
562
      STagVal tagVal = {0};
563 564
      tagVal.cid = pExpr1->base.pParam[0].pCol->colId;
      const char* p = metaGetTableTagVal(val.pTags, pColInfoData->info.type, &tagVal);
wmmhello's avatar
wmmhello 已提交
565

566 567 568 569
      char* data = NULL;
      if (pColInfoData->info.type != TSDB_DATA_TYPE_JSON && p != NULL) {
        data = tTagValToData((const STagVal*)p, false);
      } else {
wmmhello's avatar
wmmhello 已提交
570
        data = (char*)p;
wmmhello's avatar
wmmhello 已提交
571
      }
572

H
Haojun Liao 已提交
573 574 575
      bool isNullVal = (data == NULL) || (pColInfoData->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data));
      if (isNullVal) {
        colDataAppendNNULL(pColInfoData, 0, pBlock->info.rows);
H
Haojun Liao 已提交
576
      } else if (pColInfoData->info.type != TSDB_DATA_TYPE_JSON) {
H
Haojun Liao 已提交
577
        colDataAppendNItems(pColInfoData, 0, data, pBlock->info.rows);
H
Haojun Liao 已提交
578 579 580
        if (IS_VAR_DATA_TYPE(((const STagVal*)p)->type)) {
          taosMemoryFree(data);
        }
L
Liu Jicong 已提交
581
      } else {  // todo opt for json tag
H
Haojun Liao 已提交
582
        for (int32_t i = 0; i < pBlock->info.rows; ++i) {
H
Haojun Liao 已提交
583
          colDataAppend(pColInfoData, i, data, false);
H
Haojun Liao 已提交
584
        }
585 586 587 588
      }
    }
  }

589 590
  // restore the rows
  pBlock->info.rows = backupRows;
591 592 593 594
  if (freeReader) {
    metaReaderClear(&mr);
  }

H
Haojun Liao 已提交
595
  return TSDB_CODE_SUCCESS;
596 597
}

H
Haojun Liao 已提交
598
void setTbNameColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, const char* name) {
599 600 601
  struct SScalarFuncExecFuncs fpSet = {0};
  fmGetScalarFuncExecFuncs(functionId, &fpSet);

H
Haojun Liao 已提交
602
  size_t len = TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE;
603
  char   buf[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
H
Haojun Liao 已提交
604 605 606
  STR_TO_VARSTR(buf, name)

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

H
Haojun Liao 已提交
608 609
  colInfoDataEnsureCapacity(&infoData, 1, false);
  colDataAppend(&infoData, 0, buf, false);
610

H
Haojun Liao 已提交
611
  SScalarParam srcParam = {.numOfRows = pBlock->info.rows, .columnData = &infoData};
612
  SScalarParam param = {.columnData = pColInfoData};
H
Haojun Liao 已提交
613 614 615 616 617 618 619

  if (fpSet.process != NULL) {
    fpSet.process(&srcParam, 1, &param);
  } else {
    qError("failed to get the corresponding callback function, functionId:%d", functionId);
  }

D
dapan1121 已提交
620
  colDataDestroy(&infoData);
621 622
}

623
static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
624
  STableScanInfo* pTableScanInfo = pOperator->info;
625
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
L
Liu Jicong 已提交
626
  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;
H
Haojun Liao 已提交
627

628 629
  int64_t st = taosGetTimestampUs();

H
Haojun Liao 已提交
630
  while (tsdbNextDataBlock(pTableScanInfo->base.dataReader)) {
631
    if (isTaskKilled(pTaskInfo)) {
632
      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
633
    }
H
Haojun Liao 已提交
634

635 636 637 638 639 640
    // process this data block based on the probabilities
    bool processThisBlock = processBlockWithProbability(&pTableScanInfo->sample);
    if (!processThisBlock) {
      continue;
    }

H
Haojun Liao 已提交
641
    ASSERT(pBlock->info.id.uid != 0);
H
Haojun Liao 已提交
642
    pBlock->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid);
643

644
    uint32_t status = 0;
H
Haojun Liao 已提交
645
    int32_t  code = loadDataBlock(pOperator, &pTableScanInfo->base, pBlock, &status);
646 647
    //    int32_t  code = loadDataBlockOnDemand(pOperator->pRuntimeEnv, pTableScanInfo, pBlock, &status);
    if (code != TSDB_CODE_SUCCESS) {
648
      T_LONG_JMP(pOperator->pTaskInfo->env, code);
649
    }
650

651 652 653
    // current block is filter out according to filter condition, continue load the next block
    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {
      continue;
654
    }
655

H
Haojun Liao 已提交
656 657
    pOperator->resultInfo.totalRows = pTableScanInfo->base.readRecorder.totalRows;
    pTableScanInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;
658

H
Haojun Liao 已提交
659
    pOperator->cost.totalCost = pTableScanInfo->base.readRecorder.elapsedTime;
660 661

    // todo refactor
H
Haojun Liao 已提交
662
    /*pTableScanInfo->lastStatus.uid = pBlock->info.id.uid;*/
L
Liu Jicong 已提交
663 664
    /*pTableScanInfo->lastStatus.ts = pBlock->info.window.ekey;*/
    pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_DATA;
H
Haojun Liao 已提交
665
    pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.id.uid;
L
Liu Jicong 已提交
666
    pTaskInfo->streamInfo.lastStatus.ts = pBlock->info.window.ekey;
667

H
Haojun Liao 已提交
668
    ASSERT(pBlock->info.id.uid != 0);
669
    return pBlock;
H
Haojun Liao 已提交
670 671 672 673
  }
  return NULL;
}

H
Haojun Liao 已提交
674
static SSDataBlock* doGroupedTableScan(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
675 676 677 678
  STableScanInfo* pTableScanInfo = pOperator->info;
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

  // The read handle is not initialized yet, since no qualified tables exists
H
Haojun Liao 已提交
679
  if (pTableScanInfo->base.dataReader == NULL || pOperator->status == OP_EXEC_DONE) {
H
Haojun Liao 已提交
680 681 682
    return NULL;
  }

683 684
  // do the ascending order traverse in the first place.
  while (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
H
Haojun Liao 已提交
685 686 687
    SSDataBlock* p = doTableScanImpl(pOperator);
    if (p != NULL) {
      return p;
H
Haojun Liao 已提交
688 689
    }

690
    pTableScanInfo->scanTimes += 1;
691

692
    if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
693
      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
H
Haojun Liao 已提交
694
      pTableScanInfo->base.scanFlag = REPEAT_SCAN;
695
      qDebug("start to repeat ascending order scan data blocks due to query func required, %s", GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
696

697
      // do prepare for the next round table scan operation
H
Haojun Liao 已提交
698
      tsdbReaderReset(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond);
H
Haojun Liao 已提交
699
    }
700
  }
H
Haojun Liao 已提交
701

702
  int32_t total = pTableScanInfo->scanInfo.numOfAsc + pTableScanInfo->scanInfo.numOfDesc;
703
  if (pTableScanInfo->scanTimes < total) {
H
Haojun Liao 已提交
704 705 706
    if (pTableScanInfo->base.cond.order == TSDB_ORDER_ASC) {
      prepareForDescendingScan(&pTableScanInfo->base, pOperator->exprSupp.pCtx, 0);
      tsdbReaderReset(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond);
707
      qDebug("%s start to descending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
708
    }
H
Haojun Liao 已提交
709

710
    while (pTableScanInfo->scanTimes < total) {
H
Haojun Liao 已提交
711 712 713
      SSDataBlock* p = doTableScanImpl(pOperator);
      if (p != NULL) {
        return p;
714
      }
H
Haojun Liao 已提交
715

716
      pTableScanInfo->scanTimes += 1;
H
Haojun Liao 已提交
717

718
      if (pTableScanInfo->scanTimes < total) {
719
        setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
H
Haojun Liao 已提交
720
        pTableScanInfo->base.scanFlag = REPEAT_SCAN;
H
Haojun Liao 已提交
721

722
        qDebug("%s start to repeat descending order scan data blocks", GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
723
        tsdbReaderReset(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond);
724
      }
H
Haojun Liao 已提交
725 726 727
    }
  }

wmmhello's avatar
wmmhello 已提交
728 729 730 731 732 733 734
  return NULL;
}

static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
  STableScanInfo* pInfo = pOperator->info;
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

735
  // scan table one by one sequentially
L
Liu Jicong 已提交
736
  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {
H
Haojun Liao 已提交
737
    int32_t numOfTables = tableListGetSize(pTaskInfo->pTableInfoList);
H
Haojun Liao 已提交
738

L
Liu Jicong 已提交
739
    while (1) {
H
Haojun Liao 已提交
740
      SSDataBlock* result = doGroupedTableScan(pOperator);
L
Liu Jicong 已提交
741 742 743
      if (result) {
        return result;
      }
H
Haojun Liao 已提交
744

L
Liu Jicong 已提交
745 746
      // if no data, switch to next table and continue scan
      pInfo->currentTable++;
H
Haojun Liao 已提交
747
      if (pInfo->currentTable >= numOfTables) {
L
Liu Jicong 已提交
748 749
        return NULL;
      }
H
Haojun Liao 已提交
750

H
Haojun Liao 已提交
751
      STableKeyInfo* pTableInfo = tableListGetInfo(pTaskInfo->pTableInfoList, pInfo->currentTable);
H
Haojun Liao 已提交
752
      tsdbSetTableList(pInfo->base.dataReader, pTableInfo, 1);
L
Liu Jicong 已提交
753 754
      qDebug("set uid:%" PRIu64 " into scanner, total tables:%d, index:%d %s", pTableInfo->uid, numOfTables,
             pInfo->currentTable, pTaskInfo->id.str);
H
Haojun Liao 已提交
755

H
Haojun Liao 已提交
756
      tsdbReaderReset(pInfo->base.dataReader, &pInfo->base.cond);
L
Liu Jicong 已提交
757 758
      pInfo->scanTimes = 0;
    }
759 760
  } else {  // scan table group by group sequentially
    if (pInfo->currentGroupId == -1) {
H
Haojun Liao 已提交
761
      if ((++pInfo->currentGroupId) >= tableListGetOutputGroups(pTaskInfo->pTableInfoList)) {
H
Haojun Liao 已提交
762
        setOperatorCompleted(pOperator);
763 764
        return NULL;
      }
765

5
54liuyao 已提交
766
      int32_t        num = 0;
767
      STableKeyInfo* pList = NULL;
H
Haojun Liao 已提交
768
      tableListGetGroupList(pTaskInfo->pTableInfoList, pInfo->currentGroupId, &pList, &num);
H
Haojun Liao 已提交
769
      ASSERT(pInfo->base.dataReader == NULL);
770

L
Liu Jicong 已提交
771 772
      int32_t code = tsdbReaderOpen(pInfo->base.readHandle.vnode, &pInfo->base.cond, pList, num, pInfo->pResBlock,
                                    (STsdbReader**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo));
773 774 775
      if (code != TSDB_CODE_SUCCESS) {
        T_LONG_JMP(pTaskInfo->env, code);
      }
wmmhello's avatar
wmmhello 已提交
776
    }
H
Haojun Liao 已提交
777

H
Haojun Liao 已提交
778
    SSDataBlock* result = doGroupedTableScan(pOperator);
779
    if (result != NULL) {
H
Haojun Liao 已提交
780
      ASSERT(result->info.id.uid != 0);
781 782
      return result;
    }
H
Haojun Liao 已提交
783

H
Haojun Liao 已提交
784
    if ((++pInfo->currentGroupId) >= tableListGetOutputGroups(pTaskInfo->pTableInfoList)) {
H
Haojun Liao 已提交
785
      setOperatorCompleted(pOperator);
786 787
      return NULL;
    }
wmmhello's avatar
wmmhello 已提交
788

789 790
    // reset value for the next group data output
    pOperator->status = OP_OPENED;
H
Haojun Liao 已提交
791 792
    pInfo->base.limitInfo.numOfOutputRows = 0;
    pInfo->base.limitInfo.remainOffset = pInfo->base.limitInfo.limit.offset;
wmmhello's avatar
wmmhello 已提交
793

5
54liuyao 已提交
794
    int32_t        num = 0;
795
    STableKeyInfo* pList = NULL;
H
Haojun Liao 已提交
796
    tableListGetGroupList(pTaskInfo->pTableInfoList, pInfo->currentGroupId, &pList, &num);
wmmhello's avatar
wmmhello 已提交
797

H
Haojun Liao 已提交
798 799
    tsdbSetTableList(pInfo->base.dataReader, pList, num);
    tsdbReaderReset(pInfo->base.dataReader, &pInfo->base.cond);
800
    pInfo->scanTimes = 0;
wmmhello's avatar
wmmhello 已提交
801

H
Haojun Liao 已提交
802
    result = doGroupedTableScan(pOperator);
803 804 805
    if (result != NULL) {
      return result;
    }
806

H
Haojun Liao 已提交
807
    setOperatorCompleted(pOperator);
808 809
    return NULL;
  }
H
Haojun Liao 已提交
810 811
}

812 813
static int32_t getTableScannerExecInfo(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {
  SFileBlockLoadRecorder* pRecorder = taosMemoryCalloc(1, sizeof(SFileBlockLoadRecorder));
814
  STableScanInfo*         pTableScanInfo = pOptr->info;
H
Haojun Liao 已提交
815
  *pRecorder = pTableScanInfo->base.readRecorder;
816 817 818 819 820
  *pOptrExplain = pRecorder;
  *len = sizeof(SFileBlockLoadRecorder);
  return 0;
}

821
static void destroyTableScanOperatorInfo(void* param) {
822
  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;
H
Haojun Liao 已提交
823
  blockDataDestroy(pTableScanInfo->pResBlock);
H
Haojun Liao 已提交
824
  cleanupQueryTableDataCond(&pTableScanInfo->base.cond);
H
Haojun Liao 已提交
825

H
Haojun Liao 已提交
826 827
  tsdbReaderClose(pTableScanInfo->base.dataReader);
  pTableScanInfo->base.dataReader = NULL;
828

H
Haojun Liao 已提交
829 830
  if (pTableScanInfo->base.matchInfo.pList != NULL) {
    taosArrayDestroy(pTableScanInfo->base.matchInfo.pList);
831
  }
L
Liu Jicong 已提交
832

H
Haojun Liao 已提交
833 834
  taosLRUCacheCleanup(pTableScanInfo->base.metaCache.pTableMetaEntryCache);
  cleanupExprSupp(&pTableScanInfo->base.pseudoSup);
D
dapan1121 已提交
835
  taosMemoryFreeClear(param);
836 837
}

838
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
839
                                           SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
840 841 842
  STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
843
    goto _error;
H
Haojun Liao 已提交
844 845
  }

846
  SScanPhysiNode*     pScanNode = &pTableScanNode->scan;
H
Haojun Liao 已提交
847
  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;
848 849

  int32_t numOfCols = 0;
850
  int32_t code =
H
Haojun Liao 已提交
851
      extractColMatchInfo(pScanNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->base.matchInfo);
852 853 854 855
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
856
  initLimitInfo(pScanNode->node.pLimit, pScanNode->node.pSlimit, &pInfo->base.limitInfo);
H
Haojun Liao 已提交
857
  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode);
858
  if (code != TSDB_CODE_SUCCESS) {
859
    goto _error;
860 861
  }

H
Haojun Liao 已提交
862
  if (pScanNode->pScanPseudoCols != NULL) {
H
Haojun Liao 已提交
863
    SExprSupp* pSup = &pInfo->base.pseudoSup;
H
Haojun Liao 已提交
864
    pSup->pExprInfo = createExprInfo(pScanNode->pScanPseudoCols, NULL, &pSup->numOfExprs);
865
    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset);
866 867
  }

868
  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};
H
Haojun Liao 已提交
869 870

  pInfo->base.scanFlag = MAIN_SCAN;
H
Haojun Liao 已提交
871 872
  pInfo->base.pdInfo.interval = extractIntervalInfo(pTableScanNode);
  pInfo->base.readHandle = *readHandle;
H
Haojun Liao 已提交
873 874
  pInfo->base.dataBlockLoadFlag = pTableScanNode->dataRequired;

875 876
  pInfo->sample.sampleRatio = pTableScanNode->ratio;
  pInfo->sample.seed = taosGetTimestampSec();
877

H
Haojun Liao 已提交
878
  initResultSizeInfo(&pOperator->resultInfo, 4096);
H
Haojun Liao 已提交
879
  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);
H
Haojun Liao 已提交
880
  blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);
H
Haojun Liao 已提交
881

H
Haojun Liao 已提交
882 883 884
  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
H
Haojun Liao 已提交
885 886
  }

wmmhello's avatar
wmmhello 已提交
887
  pInfo->currentGroupId = -1;
888
  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;
889
  pInfo->hasGroupByTag = pTableScanNode->pGroupTags ? true : false;
890

L
Liu Jicong 已提交
891 892
  setOperatorInfo(pOperator, "TableScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, false, OP_NOT_OPENED, pInfo,
                  pTaskInfo);
893
  pOperator->exprSupp.numOfExprs = numOfCols;
894

H
Haojun Liao 已提交
895 896
  pInfo->base.metaCache.pTableMetaEntryCache = taosLRUCacheInit(1024 * 128, -1, .5);
  if (pInfo->base.metaCache.pTableMetaEntryCache == NULL) {
897 898 899
    code = terrno;
    goto _error;
  }
900

H
Haojun Liao 已提交
901
  taosLRUCacheSetStrictCapacity(pInfo->base.metaCache.pTableMetaEntryCache, false);
902 903
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableScan, NULL, destroyTableScanOperatorInfo,
                                         optrDefaultBufFn, getTableScannerExecInfo);
904 905 906

  // for non-blocking operator, the open cost is always 0
  pOperator->cost.openCost = 0;
H
Haojun Liao 已提交
907
  return pOperator;
908

909
_error:
910 911 912
  if (pInfo != NULL) {
    destroyTableScanOperatorInfo(pInfo);
  }
913

914 915
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
916
  return NULL;
H
Haojun Liao 已提交
917 918
}

919
SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
920
  STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
L
Liu Jicong 已提交
921
  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
H
Haojun Liao 已提交
922

H
Haojun Liao 已提交
923
  pInfo->base.dataReader = pReadHandle;
L
Liu Jicong 已提交
924
  //  pInfo->prevGroupId       = -1;
H
Haojun Liao 已提交
925

L
Liu Jicong 已提交
926 927
  setOperatorInfo(pOperator, "TableSeqScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN, false, OP_NOT_OPENED,
                  pInfo, pTaskInfo);
928
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableScanImpl, NULL, NULL, optrDefaultBufFn, NULL);
H
Haojun Liao 已提交
929 930 931
  return pOperator;
}

932
static FORCE_INLINE void doClearBufferedBlocks(SStreamScanInfo* pInfo) {
L
Liu Jicong 已提交
933 934
  taosArrayClear(pInfo->pBlockLists);
  pInfo->validBlockIndex = 0;
H
Haojun Liao 已提交
935 936
}

937
static bool isSessionWindow(SStreamScanInfo* pInfo) {
H
Haojun Liao 已提交
938
  return pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
5
54liuyao 已提交
939 940
}

941
static bool isStateWindow(SStreamScanInfo* pInfo) {
942
  return pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;
5
54liuyao 已提交
943
}
5
54liuyao 已提交
944

L
Liu Jicong 已提交
945
static bool isIntervalWindow(SStreamScanInfo* pInfo) {
946 947 948
  return pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL ||
         pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL ||
         pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL;
5
54liuyao 已提交
949 950 951
}

static bool isSignleIntervalWindow(SStreamScanInfo* pInfo) {
952
  return pInfo->windowSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL;
L
Liu Jicong 已提交
953 954
}

955 956 957 958
static bool isSlidingWindow(SStreamScanInfo* pInfo) {
  return isIntervalWindow(pInfo) && pInfo->interval.interval != pInfo->interval.sliding;
}

959
static void setGroupId(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t groupColIndex, int32_t rowIndex) {
960 961
  SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, groupColIndex);
  uint64_t*        groupCol = (uint64_t*)pColInfo->pData;
962
  ASSERT(rowIndex < pBlock->info.rows);
963
  pInfo->groupId = groupCol[rowIndex];
964 965
}

L
Liu Jicong 已提交
966
void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin) {
H
Haojun Liao 已提交
967
  pTableScanInfo->base.cond.twindows = *pWin;
L
Liu Jicong 已提交
968 969
  pTableScanInfo->scanTimes = 0;
  pTableScanInfo->currentGroupId = -1;
H
Haojun Liao 已提交
970 971
  tsdbReaderClose(pTableScanInfo->base.dataReader);
  pTableScanInfo->base.dataReader = NULL;
972 973
}

L
Liu Jicong 已提交
974 975
static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbUid, TSKEY startTs, TSKEY endTs,
                                       int64_t maxVersion) {
976
  STableKeyInfo tblInfo = {.uid = tbUid, .groupId = 0};
977

978
  STableScanInfo*     pTableScanInfo = pTableScanOp->info;
H
Haojun Liao 已提交
979
  SQueryTableDataCond cond = pTableScanInfo->base.cond;
980 981 982 983 984 985 986 987 988

  cond.startVersion = -1;
  cond.endVersion = maxVersion;
  cond.twindows = (STimeWindow){.skey = startTs, .ekey = endTs};

  SExecTaskInfo* pTaskInfo = pTableScanOp->pTaskInfo;

  SSDataBlock* pBlock = pTableScanInfo->pResBlock;
  STsdbReader* pReader = NULL;
L
Liu Jicong 已提交
989 990
  int32_t      code = tsdbReaderOpen(pTableScanInfo->base.readHandle.vnode, &cond, &tblInfo, 1, pBlock,
                                     (STsdbReader**)&pReader, GET_TASKID(pTaskInfo));
991 992
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
dengyihao's avatar
dengyihao 已提交
993
    T_LONG_JMP(pTaskInfo->env, code);
994 995 996
    return NULL;
  }

H
Haojun Liao 已提交
997
  if (tsdbNextDataBlock(pReader)) {
L
Liu Jicong 已提交
998
    /*SSDataBlock* p = */ tsdbRetrieveDataBlock(pReader, NULL);
H
Haojun Liao 已提交
999
    doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, pBlock->info.rows);
H
Haojun Liao 已提交
1000
    pBlock->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid);
1001 1002 1003 1004
  }

  tsdbReaderClose(pReader);
  qDebug("retrieve prev rows:%d, skey:%" PRId64 ", ekey:%" PRId64 " uid:%" PRIu64 ", max ver:%" PRId64
5
54liuyao 已提交
1005 1006
         ", suid:%" PRIu64,
         pBlock->info.rows, startTs, endTs, tbUid, maxVersion, cond.suid);
1007 1008

  return pBlock->info.rows > 0 ? pBlock : NULL;
1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019
}

static uint64_t getGroupIdByCol(SStreamScanInfo* pInfo, uint64_t uid, TSKEY ts, int64_t maxVersion) {
  SSDataBlock* pPreRes = readPreVersionData(pInfo->pTableScanOp, uid, ts, ts, maxVersion);
  if (!pPreRes || pPreRes->info.rows == 0) {
    return 0;
  }
  ASSERT(pPreRes->info.rows == 1);
  return calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pPreRes, 0);
}

5
54liuyao 已提交
1020
static uint64_t getGroupIdByUid(SStreamScanInfo* pInfo, uint64_t uid) {
H
Haojun Liao 已提交
1021
  return getTableGroupId(pInfo->pTableScanOp->pTaskInfo->pTableInfoList, uid);
1022 1023
}

5
54liuyao 已提交
1024 1025 1026 1027 1028 1029 1030 1031
static uint64_t getGroupIdByData(SStreamScanInfo* pInfo, uint64_t uid, TSKEY ts, int64_t maxVersion) {
  if (pInfo->partitionSup.needCalc) {
    return getGroupIdByCol(pInfo, uid, ts, maxVersion);
  }

  return getGroupIdByUid(pInfo, uid);
}

L
Liu Jicong 已提交
1032
static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t* pRowIndex) {
5
54liuyao 已提交
1033 1034 1035
  if (pBlock->info.rows == 0) {
    return false;
  }
L
Liu Jicong 已提交
1036 1037 1038 1039 1040 1041 1042 1043 1044 1045
  if ((*pRowIndex) == pBlock->info.rows) {
    return false;
  }

  ASSERT(taosArrayGetSize(pBlock->pDataBlock) >= 3);
  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
  TSKEY*           startData = (TSKEY*)pStartTsCol->pData;
  SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
  TSKEY*           endData = (TSKEY*)pEndTsCol->pData;
  STimeWindow      win = {.skey = startData[*pRowIndex], .ekey = endData[*pRowIndex]};
1046 1047 1048
  SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
  uint64_t*        gpData = (uint64_t*)pGpCol->pData;
  uint64_t         groupId = gpData[*pRowIndex];
1049 1050 1051 1052 1053 1054

  SColumnInfoData* pCalStartTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
  TSKEY*           calStartData = (TSKEY*)pCalStartTsCol->pData;
  SColumnInfoData* pCalEndTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
  TSKEY*           calEndData = (TSKEY*)pCalEndTsCol->pData;

L
Liu Jicong 已提交
1055
  setGroupId(pInfo, pBlock, GROUPID_COLUMN_INDEX, *pRowIndex);
1056 1057 1058 1059
  if (isSlidingWindow(pInfo)) {
    pInfo->updateWin.skey = calStartData[*pRowIndex];
    pInfo->updateWin.ekey = calEndData[*pRowIndex];
  }
L
Liu Jicong 已提交
1060 1061 1062
  (*pRowIndex)++;

  for (; *pRowIndex < pBlock->info.rows; (*pRowIndex)++) {
1063
    if (win.skey == startData[*pRowIndex] && groupId == gpData[*pRowIndex]) {
L
Liu Jicong 已提交
1064 1065 1066
      win.ekey = TMAX(win.ekey, endData[*pRowIndex]);
      continue;
    }
1067
    if (win.skey == endData[*pRowIndex] && groupId == gpData[*pRowIndex]) {
L
Liu Jicong 已提交
1068 1069 1070
      win.skey = TMIN(win.skey, startData[*pRowIndex]);
      continue;
    }
1071 1072
    ASSERT(!(win.skey > startData[*pRowIndex] && win.ekey < endData[*pRowIndex]) ||
           !(isInTimeWindow(&win, startData[*pRowIndex], 0) || isInTimeWindow(&win, endData[*pRowIndex], 0)));
L
Liu Jicong 已提交
1073 1074 1075 1076
    break;
  }

  resetTableScanInfo(pInfo->pTableScanOp->info, &win);
1077
  pInfo->pTableScanOp->status = OP_OPENED;
L
Liu Jicong 已提交
1078 1079 1080
  return true;
}

5
54liuyao 已提交
1081
static STimeWindow getSlidingWindow(TSKEY* startTsCol, TSKEY* endTsCol, uint64_t* gpIdCol, SInterval* pInterval,
1082
                                    SDataBlockInfo* pDataBlockInfo, int32_t* pRowIndex, bool hasGroup) {
H
Haojun Liao 已提交
1083
  SResultRowInfo dumyInfo = {0};
5
54liuyao 已提交
1084
  dumyInfo.cur.pageId = -1;
1085
  STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, startTsCol[*pRowIndex], pInterval, TSDB_ORDER_ASC);
5
54liuyao 已提交
1086 1087
  STimeWindow endWin = win;
  STimeWindow preWin = win;
5
54liuyao 已提交
1088
  uint64_t    groupId = gpIdCol[*pRowIndex];
H
Haojun Liao 已提交
1089

5
54liuyao 已提交
1090
  while (1) {
1091 1092 1093
    if (hasGroup) {
      (*pRowIndex) += 1;
    } else {
5
54liuyao 已提交
1094
      while ((groupId == gpIdCol[(*pRowIndex)] && startTsCol[*pRowIndex] <= endWin.ekey)) {
5
54liuyao 已提交
1095 1096 1097 1098 1099
        (*pRowIndex) += 1;
        if ((*pRowIndex) == pDataBlockInfo->rows) {
          break;
        }
      }
1100
    }
5
54liuyao 已提交
1101

5
54liuyao 已提交
1102 1103 1104
    do {
      preWin = endWin;
      getNextTimeWindow(pInterval, &endWin, TSDB_ORDER_ASC);
1105
    } while (endTsCol[(*pRowIndex) - 1] >= endWin.skey);
5
54liuyao 已提交
1106
    endWin = preWin;
5
54liuyao 已提交
1107
    if (win.ekey == endWin.ekey || (*pRowIndex) == pDataBlockInfo->rows || groupId != gpIdCol[*pRowIndex]) {
5
54liuyao 已提交
1108 1109 1110 1111 1112 1113
      win.ekey = endWin.ekey;
      return win;
    }
    win.ekey = endWin.ekey;
  }
}
5
54liuyao 已提交
1114

L
Liu Jicong 已提交
1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125
static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
  while (1) {
    SSDataBlock* pResult = NULL;
    pResult = doTableScan(pInfo->pTableScanOp);
    if (!pResult && prepareRangeScan(pInfo, pSDB, pRowIndex)) {
      // scan next window data
      pResult = doTableScan(pInfo->pTableScanOp);
    }
    if (!pResult) {
      blockDataCleanup(pSDB);
      *pRowIndex = 0;
5
54liuyao 已提交
1126
      pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
H
Hongze Cheng 已提交
1127
      STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;
H
Haojun Liao 已提交
1128 1129
      tsdbReaderClose(pTableScanInfo->base.dataReader);
      pTableScanInfo->base.dataReader = NULL;
1130 1131
      return NULL;
    }
L
Liu Jicong 已提交
1132

H
Haojun Liao 已提交
1133
    doFilter(pResult, pInfo->pTableScanOp->exprSupp.pFilterInfo, NULL);
1134 1135 1136 1137
    if (pResult->info.rows == 0) {
      continue;
    }

1138 1139 1140 1141 1142 1143 1144 1145
    if (pInfo->partitionSup.needCalc) {
      SSDataBlock* tmpBlock = createOneDataBlock(pResult, true);
      blockDataCleanup(pResult);
      for (int32_t i = 0; i < tmpBlock->info.rows; i++) {
        if (calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, tmpBlock, i) == pInfo->groupId) {
          for (int32_t j = 0; j < pInfo->pTableScanOp->exprSupp.numOfExprs; j++) {
            SColumnInfoData* pSrcCol = taosArrayGet(tmpBlock->pDataBlock, j);
            SColumnInfoData* pDestCol = taosArrayGet(pResult->pDataBlock, j);
L
Liu Jicong 已提交
1146 1147
            bool             isNull = colDataIsNull(pSrcCol, tmpBlock->info.rows, i, NULL);
            char*            pSrcData = colDataGetData(pSrcCol, i);
1148 1149 1150 1151 1152
            colDataAppend(pDestCol, pResult->info.rows, pSrcData, isNull);
          }
          pResult->info.rows++;
        }
      }
H
Haojun Liao 已提交
1153 1154 1155

      blockDataDestroy(tmpBlock);

1156 1157 1158 1159
      if (pResult->info.rows > 0) {
        pResult->info.calWin = pInfo->updateWin;
        return pResult;
      }
H
Haojun Liao 已提交
1160
    } else if (pResult->info.id.groupId == pInfo->groupId) {
5
54liuyao 已提交
1161
      pResult->info.calWin = pInfo->updateWin;
1162
      return pResult;
5
54liuyao 已提交
1163 1164
    }
  }
1165
}
1166

1167
static int32_t generateSessionScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) {
5
54liuyao 已提交
1168
  blockDataCleanup(pDestBlock);
1169 1170
  if (pSrcBlock->info.rows == 0) {
    return TSDB_CODE_SUCCESS;
1171
  }
1172
  int32_t code = blockDataEnsureCapacity(pDestBlock, pSrcBlock->info.rows);
1173
  if (code != TSDB_CODE_SUCCESS) {
1174
    return code;
L
Liu Jicong 已提交
1175
  }
1176 1177
  ASSERT(taosArrayGetSize(pSrcBlock->pDataBlock) >= 3);
  SColumnInfoData* pStartTsCol = taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX);
L
Liu Jicong 已提交
1178
  TSKEY*           startData = (TSKEY*)pStartTsCol->pData;
1179
  SColumnInfoData* pEndTsCol = taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX);
L
Liu Jicong 已提交
1180
  TSKEY*           endData = (TSKEY*)pEndTsCol->pData;
1181 1182
  SColumnInfoData* pUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX);
  uint64_t*        uidCol = (uint64_t*)pUidCol->pData;
L
Liu Jicong 已提交
1183

1184 1185
  SColumnInfoData* pDestStartCol = taosArrayGet(pDestBlock->pDataBlock, START_TS_COLUMN_INDEX);
  SColumnInfoData* pDestEndCol = taosArrayGet(pDestBlock->pDataBlock, END_TS_COLUMN_INDEX);
5
54liuyao 已提交
1186
  SColumnInfoData* pDestUidCol = taosArrayGet(pDestBlock->pDataBlock, UID_COLUMN_INDEX);
1187
  SColumnInfoData* pDestGpCol = taosArrayGet(pDestBlock->pDataBlock, GROUPID_COLUMN_INDEX);
5
54liuyao 已提交
1188 1189
  SColumnInfoData* pDestCalStartTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
  SColumnInfoData* pDestCalEndTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
L
Liu Jicong 已提交
1190
  int64_t          version = pSrcBlock->info.version - 1;
1191
  for (int32_t i = 0; i < pSrcBlock->info.rows; i++) {
1192
    uint64_t groupId = getGroupIdByData(pInfo, uidCol[i], startData[i], version);
L
Liu Jicong 已提交
1193
    // gap must be 0.
5
54liuyao 已提交
1194
    SSessionKey startWin = {0};
1195
    getCurSessionWindow(pInfo->windowSup.pStreamAggSup, startData[i], startData[i], groupId, &startWin);
5
54liuyao 已提交
1196
    if (IS_INVALID_SESSION_WIN_KEY(startWin)) {
L
Liu Jicong 已提交
1197 1198 1199
      // window has been closed.
      continue;
    }
5
54liuyao 已提交
1200 1201 1202 1203 1204 1205
    SSessionKey endWin = {0};
    getCurSessionWindow(pInfo->windowSup.pStreamAggSup, endData[i], endData[i], groupId, &endWin);
    ASSERT(!IS_INVALID_SESSION_WIN_KEY(endWin));
    colDataAppend(pDestStartCol, i, (const char*)&startWin.win.skey, false);
    colDataAppend(pDestEndCol, i, (const char*)&endWin.win.ekey, false);

5
54liuyao 已提交
1206
    colDataAppendNULL(pDestUidCol, i);
L
Liu Jicong 已提交
1207
    colDataAppend(pDestGpCol, i, (const char*)&groupId, false);
5
54liuyao 已提交
1208 1209
    colDataAppendNULL(pDestCalStartTsCol, i);
    colDataAppendNULL(pDestCalEndTsCol, i);
1210
    pDestBlock->info.rows++;
L
Liu Jicong 已提交
1211
  }
1212
  return TSDB_CODE_SUCCESS;
L
Liu Jicong 已提交
1213
}
1214 1215 1216 1217 1218 1219

static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) {
  blockDataCleanup(pDestBlock);
  int32_t rows = pSrcBlock->info.rows;
  if (rows == 0) {
    return TSDB_CODE_SUCCESS;
1220
  }
1221

1222 1223
  SColumnInfoData* pSrcStartTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX);
  SColumnInfoData* pSrcEndTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX);
1224 1225
  SColumnInfoData* pSrcUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX);
  SColumnInfoData* pSrcGpCol = taosArrayGet(pSrcBlock->pDataBlock, GROUPID_COLUMN_INDEX);
5
54liuyao 已提交
1226

L
Liu Jicong 已提交
1227
  uint64_t* srcUidData = (uint64_t*)pSrcUidCol->pData;
1228
  ASSERT(pSrcStartTsCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
5
54liuyao 已提交
1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264
  TSKEY*  srcStartTsCol = (TSKEY*)pSrcStartTsCol->pData;
  TSKEY*  srcEndTsCol = (TSKEY*)pSrcEndTsCol->pData;
  int64_t version = pSrcBlock->info.version - 1;

  if (pInfo->partitionSup.needCalc && srcStartTsCol[0] != srcEndTsCol[0]) {
    uint64_t     srcUid = srcUidData[0];
    TSKEY        startTs = srcStartTsCol[0];
    TSKEY        endTs = srcEndTsCol[0];
    SSDataBlock* pPreRes = readPreVersionData(pInfo->pTableScanOp, srcUid, startTs, endTs, version);
    printDataBlock(pPreRes, "pre res");
    blockDataCleanup(pSrcBlock);
    int32_t code = blockDataEnsureCapacity(pSrcBlock, pPreRes->info.rows);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

    SColumnInfoData* pTsCol = (SColumnInfoData*)taosArrayGet(pPreRes->pDataBlock, pInfo->primaryTsIndex);
    rows = pPreRes->info.rows;

    for (int32_t i = 0; i < rows; i++) {
      uint64_t groupId = calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pPreRes, i);
      appendOneRowToStreamSpecialBlock(pSrcBlock, ((TSKEY*)pTsCol->pData) + i, ((TSKEY*)pTsCol->pData) + i, &srcUid,
                                       &groupId, NULL);
    }
    printDataBlock(pSrcBlock, "new delete");
  }
  uint64_t* srcGp = (uint64_t*)pSrcGpCol->pData;
  srcStartTsCol = (TSKEY*)pSrcStartTsCol->pData;
  srcEndTsCol = (TSKEY*)pSrcEndTsCol->pData;
  srcUidData = (uint64_t*)pSrcUidCol->pData;

  int32_t code = blockDataEnsureCapacity(pDestBlock, rows);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

1265 1266
  SColumnInfoData* pStartTsCol = taosArrayGet(pDestBlock->pDataBlock, START_TS_COLUMN_INDEX);
  SColumnInfoData* pEndTsCol = taosArrayGet(pDestBlock->pDataBlock, END_TS_COLUMN_INDEX);
1267
  SColumnInfoData* pDeUidCol = taosArrayGet(pDestBlock->pDataBlock, UID_COLUMN_INDEX);
1268 1269 1270
  SColumnInfoData* pGpCol = taosArrayGet(pDestBlock->pDataBlock, GROUPID_COLUMN_INDEX);
  SColumnInfoData* pCalStartTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
  SColumnInfoData* pCalEndTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
1271
  for (int32_t i = 0; i < rows;) {
1272
    uint64_t srcUid = srcUidData[i];
5
54liuyao 已提交
1273 1274 1275 1276 1277
    uint64_t groupId = srcGp[i];
    if (groupId == 0) {
      groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], version);
    }
    TSKEY calStartTs = srcStartTsCol[i];
1278
    colDataAppend(pCalStartTsCol, pDestBlock->info.rows, (const char*)(&calStartTs), false);
5
54liuyao 已提交
1279
    STimeWindow win = getSlidingWindow(srcStartTsCol, srcEndTsCol, srcGp, &pInfo->interval, &pSrcBlock->info, &i,
1280 1281
                                       pInfo->partitionSup.needCalc);
    TSKEY       calEndTs = srcStartTsCol[i - 1];
1282 1283
    colDataAppend(pCalEndTsCol, pDestBlock->info.rows, (const char*)(&calEndTs), false);
    colDataAppend(pDeUidCol, pDestBlock->info.rows, (const char*)(&srcUid), false);
1284 1285 1286 1287
    colDataAppend(pStartTsCol, pDestBlock->info.rows, (const char*)(&win.skey), false);
    colDataAppend(pEndTsCol, pDestBlock->info.rows, (const char*)(&win.ekey), false);
    colDataAppend(pGpCol, pDestBlock->info.rows, (const char*)(&groupId), false);
    pDestBlock->info.rows++;
5
54liuyao 已提交
1288
  }
1289 1290
  return TSDB_CODE_SUCCESS;
}
1291

1292
static int32_t generateDeleteResultBlock(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) {
5
54liuyao 已提交
1293 1294 1295
  blockDataCleanup(pDestBlock);
  int32_t rows = pSrcBlock->info.rows;
  if (rows == 0) {
1296 1297
    return TSDB_CODE_SUCCESS;
  }
5
54liuyao 已提交
1298
  int32_t code = blockDataEnsureCapacity(pDestBlock, rows);
1299 1300 1301 1302
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

5
54liuyao 已提交
1303 1304 1305 1306 1307 1308 1309 1310 1311 1312
  SColumnInfoData* pSrcStartTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX);
  SColumnInfoData* pSrcEndTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX);
  SColumnInfoData* pSrcUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX);
  uint64_t*        srcUidData = (uint64_t*)pSrcUidCol->pData;
  SColumnInfoData* pSrcGpCol = taosArrayGet(pSrcBlock->pDataBlock, GROUPID_COLUMN_INDEX);
  uint64_t*        srcGp = (uint64_t*)pSrcGpCol->pData;
  ASSERT(pSrcStartTsCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
  TSKEY*  srcStartTsCol = (TSKEY*)pSrcStartTsCol->pData;
  TSKEY*  srcEndTsCol = (TSKEY*)pSrcEndTsCol->pData;
  int64_t version = pSrcBlock->info.version - 1;
1313
  for (int32_t i = 0; i < pSrcBlock->info.rows; i++) {
5
54liuyao 已提交
1314 1315
    uint64_t srcUid = srcUidData[i];
    uint64_t groupId = srcGp[i];
L
Liu Jicong 已提交
1316
    char*    tbname[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN] = {0};
5
54liuyao 已提交
1317 1318 1319
    if (groupId == 0) {
      groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], version);
    }
L
Liu Jicong 已提交
1320
    if (pInfo->tbnameCalSup.pExprInfo) {
1321 1322 1323
      void* parTbname = NULL;
      streamStateGetParName(pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState, groupId, &parTbname);

L
Liu Jicong 已提交
1324 1325
      memcpy(varDataVal(tbname), parTbname, TSDB_TABLE_NAME_LEN);
      varDataSetLen(tbname, strlen(varDataVal(tbname)));
L
Liu Jicong 已提交
1326
      tdbFree(parTbname);
L
Liu Jicong 已提交
1327 1328 1329
    }
    appendOneRowToStreamSpecialBlock(pDestBlock, srcStartTsCol + i, srcEndTsCol + i, srcUidData + i, &groupId,
                                     tbname[0] == 0 ? NULL : tbname);
1330 1331 1332 1333
  }
  return TSDB_CODE_SUCCESS;
}

1334 1335 1336 1337
static int32_t generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) {
  int32_t code = TSDB_CODE_SUCCESS;
  if (isIntervalWindow(pInfo)) {
    code = generateIntervalScanRange(pInfo, pSrcBlock, pDestBlock);
1338
  } else if (isSessionWindow(pInfo) || isStateWindow(pInfo)) {
1339
    code = generateSessionScanRange(pInfo, pSrcBlock, pDestBlock);
5
54liuyao 已提交
1340 1341
  } else {
    code = generateDeleteResultBlock(pInfo, pSrcBlock, pDestBlock);
1342
  }
1343
  pDestBlock->info.type = STREAM_CLEAR;
1344
  pDestBlock->info.version = pSrcBlock->info.version;
1345
  pDestBlock->info.dataLoad = 1;
1346 1347 1348 1349
  blockDataUpdateTsWindow(pDestBlock, 0);
  return code;
}

L
Liu Jicong 已提交
1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379
#if 0
void calBlockTag(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
  SExprSupp*    pTagCalSup = &pInfo->tagCalSup;
  SStreamState* pState = pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState;
  if (pTagCalSup == NULL || pTagCalSup->numOfExprs == 0) return;
  if (pBlock == NULL || pBlock->info.rows == 0) return;

  void*   tag = NULL;
  int32_t tagLen = 0;
  if (streamStateGetParTag(pState, pBlock->info.id.groupId, &tag, &tagLen) == 0) {
    pBlock->info.tagLen = tagLen;
    void* pTag = taosMemoryRealloc(pBlock->info.pTag, tagLen);
    if (pTag == NULL) {
      tdbFree(tag);
      taosMemoryFree(pBlock->info.pTag);
      pBlock->info.pTag = NULL;
      pBlock->info.tagLen = 0;
      return;
    }
    pBlock->info.pTag = pTag;
    memcpy(pBlock->info.pTag, tag, tagLen);
    tdbFree(tag);
    return;
  } else {
    pBlock->info.pTag = NULL;
  }
  tdbFree(tag);
}
#endif

L
Liu Jicong 已提交
1380
void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
1381 1382
  SExprSupp*    pTbNameCalSup = &pInfo->tbnameCalSup;
  SStreamState* pState = pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState;
L
Liu Jicong 已提交
1383 1384
  if (pTbNameCalSup == NULL || pTbNameCalSup->numOfExprs == 0) return;
  if (pBlock == NULL || pBlock->info.rows == 0) return;
1385 1386

  void* tbname = NULL;
L
Liu Jicong 已提交
1387
  if (streamStateGetParName(pState, pBlock->info.id.groupId, &tbname) == 0) {
1388
    memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
L
Liu Jicong 已提交
1389 1390 1391 1392
    tdbFree(tbname);
    return;
  } else {
    pBlock->info.parTbName[0] = 0;
L
Liu Jicong 已提交
1393
  }
1394
  tdbFree(tbname);
L
Liu Jicong 已提交
1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413

  SSDataBlock* pSrcBlock = blockCopyOneRow(pBlock, 0);
  ASSERT(pSrcBlock->info.rows == 1);

  SSDataBlock* pResBlock = createDataBlock();
  pResBlock->info.rowSize = VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN;
  SColumnInfoData data = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, TSDB_TABLE_NAME_LEN, 0);
  taosArrayPush(pResBlock->pDataBlock, &data);
  blockDataEnsureCapacity(pResBlock, 1);

  projectApplyFunctions(pTbNameCalSup->pExprInfo, pResBlock, pSrcBlock, pTbNameCalSup->pCtx, 1, NULL);
  ASSERT(pResBlock->info.rows == 1);
  ASSERT(taosArrayGetSize(pResBlock->pDataBlock) == 1);
  SColumnInfoData* pCol = taosArrayGet(pResBlock->pDataBlock, 0);
  ASSERT(pCol->info.type == TSDB_DATA_TYPE_VARCHAR);

  void* pData = colDataGetData(pCol, 0);
  // TODO check tbname validation
  if (pData != (void*)-1 && pData != NULL) {
1414
    memset(pBlock->info.parTbName, 0, TSDB_TABLE_NAME_LEN);
L
Liu Jicong 已提交
1415
    int32_t len = TMIN(varDataLen(pData), TSDB_TABLE_NAME_LEN - 1);
1416 1417
    memcpy(pBlock->info.parTbName, varDataVal(pData), len);
    /*pBlock->info.parTbName[len + 1] = 0;*/
L
Liu Jicong 已提交
1418 1419 1420 1421
  } else {
    pBlock->info.parTbName[0] = 0;
  }

H
Haojun Liao 已提交
1422 1423
  if (pBlock->info.id.groupId && pBlock->info.parTbName[0]) {
    streamStatePutParName(pState, pBlock->info.id.groupId, pBlock->info.parTbName);
L
Liu Jicong 已提交
1424 1425
  }

L
Liu Jicong 已提交
1426 1427 1428 1429
  blockDataDestroy(pSrcBlock);
  blockDataDestroy(pResBlock);
}

1430 1431
void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid,
                                      uint64_t* pGp, void* pTbName) {
1432 1433
  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
  SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
1434 1435
  SColumnInfoData* pUidCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);
  SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
1436 1437
  SColumnInfoData* pCalStartCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
  SColumnInfoData* pCalEndCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
1438
  SColumnInfoData* pTableCol = taosArrayGet(pBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);
1439 1440
  colDataAppend(pStartTsCol, pBlock->info.rows, (const char*)pStartTs, false);
  colDataAppend(pEndTsCol, pBlock->info.rows, (const char*)pEndTs, false);
1441 1442
  colDataAppend(pUidCol, pBlock->info.rows, (const char*)pUid, false);
  colDataAppend(pGpCol, pBlock->info.rows, (const char*)pGp, false);
1443 1444
  colDataAppend(pCalStartCol, pBlock->info.rows, (const char*)pStartTs, false);
  colDataAppend(pCalEndCol, pBlock->info.rows, (const char*)pEndTs, false);
1445
  colDataAppend(pTableCol, pBlock->info.rows, (const char*)pTbName, pTbName == NULL);
1446
  pBlock->info.rows++;
5
54liuyao 已提交
1447 1448
}

1449
static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock* pBlock, bool out) {
1450 1451
  if (out) {
    blockDataCleanup(pInfo->pUpdateDataRes);
5
54liuyao 已提交
1452
    blockDataEnsureCapacity(pInfo->pUpdateDataRes, pBlock->info.rows * 2);
1453
  }
1454 1455
  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
  ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP);
5
54liuyao 已提交
1456
  TSKEY* tsCol = (TSKEY*)pColDataInfo->pData;
H
Haojun Liao 已提交
1457
  bool   tableInserted = updateInfoIsTableInserted(pInfo->pUpdateInfo, pBlock->info.id.uid);
1458
  for (int32_t rowId = 0; rowId < pBlock->info.rows; rowId++) {
5
54liuyao 已提交
1459 1460
    SResultRowInfo dumyInfo;
    dumyInfo.cur.pageId = -1;
L
Liu Jicong 已提交
1461
    bool        isClosed = false;
5
54liuyao 已提交
1462
    STimeWindow win = {.skey = INT64_MIN, .ekey = INT64_MAX};
L
Liu Jicong 已提交
1463
    if (tableInserted && isOverdue(tsCol[rowId], &pInfo->twAggSup)) {
5
54liuyao 已提交
1464 1465 1466
      win = getActiveTimeWindow(NULL, &dumyInfo, tsCol[rowId], &pInfo->interval, TSDB_ORDER_ASC);
      isClosed = isCloseWindow(&win, &pInfo->twAggSup);
    }
5
54liuyao 已提交
1467
    // must check update info first.
H
Haojun Liao 已提交
1468
    bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.id.uid, tsCol[rowId]);
L
Liu Jicong 已提交
1469
    bool closedWin = isClosed && isSignleIntervalWindow(pInfo) &&
H
Haojun Liao 已提交
1470
                     isDeletedStreamWindow(&win, pBlock->info.id.groupId,
1471
                                           pInfo->pTableScanOp->pTaskInfo->streamInfo.pState, &pInfo->twAggSup);
L
Liu Jicong 已提交
1472
    if ((update || closedWin) && out) {
L
Liu Jicong 已提交
1473
      qDebug("stream update check not pass, update %d, closedWin %d", update, closedWin);
5
54liuyao 已提交
1474
      uint64_t gpId = 0;
H
Haojun Liao 已提交
1475
      appendOneRowToStreamSpecialBlock(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.id.uid, &gpId,
1476
                                       NULL);
5
54liuyao 已提交
1477 1478
      if (closedWin && pInfo->partitionSup.needCalc) {
        gpId = calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pBlock, rowId);
S
slzhou 已提交
1479 1480
        appendOneRowToStreamSpecialBlock(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.id.uid,
                                         &gpId, NULL);
5
54liuyao 已提交
1481
      }
1482 1483
    }
  }
1484 1485
  if (out && pInfo->pUpdateDataRes->info.rows > 0) {
    pInfo->pUpdateDataRes->info.version = pBlock->info.version;
1486
    pInfo->pUpdateDataRes->info.dataLoad = 1;
1487
    blockDataUpdateTsWindow(pInfo->pUpdateDataRes, 0);
1488
    pInfo->pUpdateDataRes->info.type = pInfo->partitionSup.needCalc ? STREAM_DELETE_DATA : STREAM_CLEAR;
5
54liuyao 已提交
1489 1490
  }
}
L
Liu Jicong 已提交
1491

1492
static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock, bool filter) {
L
Liu Jicong 已提交
1493 1494
  SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
  SOperatorInfo*  pOperator = pInfo->pStreamScanOp;
L
Liu Jicong 已提交
1495
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
L
Liu Jicong 已提交
1496

1497 1498
  blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows);

L
Liu Jicong 已提交
1499
  pInfo->pRes->info.rows = pBlock->info.rows;
H
Haojun Liao 已提交
1500
  pInfo->pRes->info.id.uid = pBlock->info.id.uid;
L
Liu Jicong 已提交
1501
  pInfo->pRes->info.type = STREAM_NORMAL;
1502
  pInfo->pRes->info.version = pBlock->info.version;
L
Liu Jicong 已提交
1503

H
Haojun Liao 已提交
1504
  pInfo->pRes->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid);
L
Liu Jicong 已提交
1505 1506

  // todo extract method
H
Haojun Liao 已提交
1507 1508 1509
  for (int32_t i = 0; i < taosArrayGetSize(pInfo->matchInfo.pList); ++i) {
    SColMatchItem* pColMatchInfo = taosArrayGet(pInfo->matchInfo.pList, i);
    if (!pColMatchInfo->needOutput) {
L
Liu Jicong 已提交
1510 1511 1512 1513 1514 1515 1516
      continue;
    }

    bool colExists = false;
    for (int32_t j = 0; j < blockDataGetNumOfCols(pBlock); ++j) {
      SColumnInfoData* pResCol = bdGetColumnInfoData(pBlock, j);
      if (pResCol->info.colId == pColMatchInfo->colId) {
H
Haojun Liao 已提交
1517
        SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pColMatchInfo->dstSlotId);
1518
        colDataAssign(pDst, pResCol, pBlock->info.rows, &pInfo->pRes->info);
L
Liu Jicong 已提交
1519 1520 1521 1522 1523 1524 1525
        colExists = true;
        break;
      }
    }

    // the required column does not exists in submit block, let's set it to be all null value
    if (!colExists) {
H
Haojun Liao 已提交
1526
      SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pColMatchInfo->dstSlotId);
L
Liu Jicong 已提交
1527 1528 1529 1530 1531 1532
      colDataAppendNNULL(pDst, 0, pBlockInfo->rows);
    }
  }

  // currently only the tbname pseudo column
  if (pInfo->numOfPseudoExpr > 0) {
L
Liu Jicong 已提交
1533
    int32_t code = addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, pInfo->pRes,
1534
                                          pInfo->pRes->info.rows, GET_TASKID(pTaskInfo), NULL);
K
kailixu 已提交
1535 1536
    // ignore the table not exists error, since this table may have been dropped during the scan procedure.
    if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_PAR_TABLE_NOT_EXIST) {
L
Liu Jicong 已提交
1537
      blockDataFreeRes((SSDataBlock*)pBlock);
1538
      T_LONG_JMP(pTaskInfo->env, code);
H
Haojun Liao 已提交
1539
    }
K
kailixu 已提交
1540 1541 1542

    // reset the error code.
    terrno = 0;
L
Liu Jicong 已提交
1543 1544
  }

1545
  if (filter) {
H
Haojun Liao 已提交
1546
    doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
1547
  }
1548

1549
  pInfo->pRes->info.dataLoad = 1;
L
Liu Jicong 已提交
1550
  blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
L
Liu Jicong 已提交
1551
  blockDataFreeRes((SSDataBlock*)pBlock);
L
Liu Jicong 已提交
1552

L
Liu Jicong 已提交
1553
  calBlockTbName(pInfo, pInfo->pRes);
L
Liu Jicong 已提交
1554 1555
  return 0;
}
5
54liuyao 已提交
1556

L
Liu Jicong 已提交
1557
static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
1558 1559
  SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;
  SStreamScanInfo* pInfo = pOperator->info;
H
Haojun Liao 已提交
1560

L
Liu Jicong 已提交
1561
  qDebug("queue scan called");
L
Liu Jicong 已提交
1562 1563 1564 1565 1566 1567 1568 1569

  if (pTaskInfo->streamInfo.pReq != NULL) {
    if (pInfo->tqReader->pMsg == NULL) {
      pInfo->tqReader->pMsg = pTaskInfo->streamInfo.pReq;
      const SSubmitReq* pSubmit = pInfo->tqReader->pMsg;
      if (tqReaderSetDataMsg(pInfo->tqReader, pSubmit, 0) < 0) {
        qError("submit msg messed up when initing stream submit block %p", pSubmit);
        pInfo->tqReader->pMsg = NULL;
L
Liu Jicong 已提交
1570
        pTaskInfo->streamInfo.pReq = NULL;
L
Liu Jicong 已提交
1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586
        ASSERT(0);
      }
    }

    blockDataCleanup(pInfo->pRes);
    SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;

    while (tqNextDataBlock(pInfo->tqReader)) {
      SSDataBlock block = {0};

      int32_t code = tqRetrieveDataBlock(&block, pInfo->tqReader);

      if (code != TSDB_CODE_SUCCESS || block.info.rows == 0) {
        continue;
      }

1587
      setBlockIntoRes(pInfo, &block, true);
L
Liu Jicong 已提交
1588 1589 1590 1591 1592 1593 1594 1595

      if (pBlockInfo->rows > 0) {
        return pInfo->pRes;
      }
    }

    pInfo->tqReader->pMsg = NULL;
    pTaskInfo->streamInfo.pReq = NULL;
L
Liu Jicong 已提交
1596
    return NULL;
L
Liu Jicong 已提交
1597 1598
  }

L
Liu Jicong 已提交
1599 1600 1601
  if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
    SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
    if (pResult && pResult->info.rows > 0) {
L
Liu Jicong 已提交
1602
      qDebug("queue scan tsdb return %d rows", pResult->info.rows);
1603
      pTaskInfo->streamInfo.returned = 1;
L
Liu Jicong 已提交
1604 1605
      return pResult;
    } else {
1606 1607
      if (!pTaskInfo->streamInfo.returned) {
        STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
H
Haojun Liao 已提交
1608 1609
        tsdbReaderClose(pTSInfo->base.dataReader);
        pTSInfo->base.dataReader = NULL;
1610
        tqOffsetResetToLog(&pTaskInfo->streamInfo.prepareStatus, pTaskInfo->streamInfo.snapshotVer);
1611
        qDebug("queue scan tsdb over, switch to wal ver %" PRId64 "", pTaskInfo->streamInfo.snapshotVer + 1);
1612
        if (tqSeekVer(pInfo->tqReader, pTaskInfo->streamInfo.snapshotVer + 1) < 0) {
1613
          tqOffsetResetToLog(&pTaskInfo->streamInfo.lastStatus, pTaskInfo->streamInfo.snapshotVer);
1614 1615 1616 1617
          return NULL;
        }
        ASSERT(pInfo->tqReader->pWalReader->curVersion == pTaskInfo->streamInfo.snapshotVer + 1);
      } else {
L
Liu Jicong 已提交
1618 1619
        return NULL;
      }
1620 1621 1622
    }
  }

L
Liu Jicong 已提交
1623 1624 1625 1626 1627 1628
  if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__LOG) {
    while (1) {
      SFetchRet ret = {0};
      tqNextBlock(pInfo->tqReader, &ret);
      if (ret.fetchType == FETCH_TYPE__DATA) {
        blockDataCleanup(pInfo->pRes);
1629
        if (setBlockIntoRes(pInfo, &ret.data, true) < 0) {
L
Liu Jicong 已提交
1630 1631 1632
          ASSERT(0);
        }
        if (pInfo->pRes->info.rows > 0) {
L
Liu Jicong 已提交
1633
          pOperator->status = OP_EXEC_RECV;
L
Liu Jicong 已提交
1634
          qDebug("queue scan log return %d rows", pInfo->pRes->info.rows);
L
Liu Jicong 已提交
1635 1636 1637 1638
          return pInfo->pRes;
        }
      } else if (ret.fetchType == FETCH_TYPE__META) {
        ASSERT(0);
L
Liu Jicong 已提交
1639 1640 1641
        //        pTaskInfo->streamInfo.lastStatus = ret.offset;
        //        pTaskInfo->streamInfo.metaBlk = ret.meta;
        //        return NULL;
L
Liu Jicong 已提交
1642 1643
      } else if (ret.fetchType == FETCH_TYPE__NONE ||
                 (ret.fetchType == FETCH_TYPE__SEP && pOperator->status == OP_EXEC_RECV)) {
L
Liu Jicong 已提交
1644
        pTaskInfo->streamInfo.lastStatus = ret.offset;
1645 1646 1647 1648
        ASSERT(pTaskInfo->streamInfo.lastStatus.version >= pTaskInfo->streamInfo.prepareStatus.version);
        ASSERT(pTaskInfo->streamInfo.lastStatus.version + 1 == pInfo->tqReader->pWalReader->curVersion);
        char formatBuf[80];
        tFormatOffset(formatBuf, 80, &ret.offset);
L
Liu Jicong 已提交
1649
        qDebug("queue scan log return null, offset %s", formatBuf);
L
Liu Jicong 已提交
1650
        pOperator->status = OP_OPENED;
L
Liu Jicong 已提交
1651 1652 1653
        return NULL;
      }
    }
L
Liu Jicong 已提交
1654
#if 0
1655
    } else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
L
Liu Jicong 已提交
1656
    SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
L
Liu Jicong 已提交
1657 1658 1659 1660 1661 1662
    if (pResult && pResult->info.rows > 0) {
      qDebug("stream scan tsdb return %d rows", pResult->info.rows);
      return pResult;
    }
    qDebug("stream scan tsdb return null");
    return NULL;
L
Liu Jicong 已提交
1663
#endif
L
Liu Jicong 已提交
1664 1665 1666
  } else {
    ASSERT(0);
    return NULL;
H
Haojun Liao 已提交
1667
  }
L
Liu Jicong 已提交
1668 1669
}

L
Liu Jicong 已提交
1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697
static int32_t filterDelBlockByUid(SSDataBlock* pDst, const SSDataBlock* pSrc, SStreamScanInfo* pInfo) {
  STqReader* pReader = pInfo->tqReader;
  int32_t    rows = pSrc->info.rows;
  blockDataEnsureCapacity(pDst, rows);

  SColumnInfoData* pSrcStartCol = taosArrayGet(pSrc->pDataBlock, START_TS_COLUMN_INDEX);
  uint64_t*        startCol = (uint64_t*)pSrcStartCol->pData;
  SColumnInfoData* pSrcEndCol = taosArrayGet(pSrc->pDataBlock, END_TS_COLUMN_INDEX);
  uint64_t*        endCol = (uint64_t*)pSrcEndCol->pData;
  SColumnInfoData* pSrcUidCol = taosArrayGet(pSrc->pDataBlock, UID_COLUMN_INDEX);
  uint64_t*        uidCol = (uint64_t*)pSrcUidCol->pData;

  SColumnInfoData* pDstStartCol = taosArrayGet(pDst->pDataBlock, START_TS_COLUMN_INDEX);
  SColumnInfoData* pDstEndCol = taosArrayGet(pDst->pDataBlock, END_TS_COLUMN_INDEX);
  SColumnInfoData* pDstUidCol = taosArrayGet(pDst->pDataBlock, UID_COLUMN_INDEX);
  int32_t          j = 0;
  for (int32_t i = 0; i < rows; i++) {
    if (taosHashGet(pReader->tbIdHash, &uidCol[i], sizeof(uint64_t))) {
      colDataAppend(pDstStartCol, j, (const char*)&startCol[i], false);
      colDataAppend(pDstEndCol, j, (const char*)&endCol[i], false);
      colDataAppend(pDstUidCol, j, (const char*)&uidCol[i], false);

      colDataAppendNULL(taosArrayGet(pDst->pDataBlock, GROUPID_COLUMN_INDEX), j);
      colDataAppendNULL(taosArrayGet(pDst->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX), j);
      colDataAppendNULL(taosArrayGet(pDst->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX), j);
      j++;
    }
  }
L
Liu Jicong 已提交
1698
  uint32_t cap = pDst->info.capacity;
L
Liu Jicong 已提交
1699 1700
  pDst->info = pSrc->info;
  pDst->info.rows = j;
L
Liu Jicong 已提交
1701
  pDst->info.capacity = cap;
L
Liu Jicong 已提交
1702 1703 1704 1705

  return 0;
}

5
54liuyao 已提交
1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722
// for partition by tag
static void setBlockGroupIdByUid(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
  TSKEY*           startTsCol = (TSKEY*)pStartTsCol->pData;
  SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
  uint64_t*        gpCol = (uint64_t*)pGpCol->pData;
  SColumnInfoData* pUidCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);
  uint64_t*        uidCol = (uint64_t*)pUidCol->pData;
  int32_t          rows = pBlock->info.rows;
  if (!pInfo->partitionSup.needCalc) {
    for (int32_t i = 0; i < rows; i++) {
      uint64_t groupId = getGroupIdByUid(pInfo, uidCol[i]);
      colDataAppend(pGpCol, i, (const char*)&groupId, false);
    }
  }
}

L
Liu Jicong 已提交
1723 1724 1725 1726 1727
static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
  // NOTE: this operator does never check if current status is done or not
  SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;
  SStreamScanInfo* pInfo = pOperator->info;

L
Liu Jicong 已提交
1728
  qDebug("stream scan called");
L
Liu Jicong 已提交
1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760 1761
#if 0
  SStreamState* pState = pTaskInfo->streamInfo.pState;
  if (pState) {
    printf(">>>>>>>> stream write backend\n");
    SWinKey key = {
        .ts = 1,
        .groupId = 2,
    };
    char tmp[100] = "abcdefg1";
    if (streamStatePut(pState, &key, &tmp, strlen(tmp) + 1) < 0) {
      ASSERT(0);
    }

    key.ts = 2;
    char tmp2[100] = "abcdefg2";
    if (streamStatePut(pState, &key, &tmp2, strlen(tmp2) + 1) < 0) {
      ASSERT(0);
    }

    key.groupId = 5;
    key.ts = 1;
    char tmp3[100] = "abcdefg3";
    if (streamStatePut(pState, &key, &tmp3, strlen(tmp3) + 1) < 0) {
      ASSERT(0);
    }

    char*   val2 = NULL;
    int32_t sz;
    if (streamStateGet(pState, &key, (void**)&val2, &sz) < 0) {
      ASSERT(0);
    }
    printf("stream read %s %d\n", val2, sz);
    streamFreeVal(val2);
H
Haojun Liao 已提交
1762
  }
L
Liu Jicong 已提交
1763
#endif
H
Haojun Liao 已提交
1764

1765 1766
  if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__PREPARE1 ||
      pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__PREPARE2) {
L
Liu Jicong 已提交
1767
    STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
H
Haojun Liao 已提交
1768
    memcpy(&pTSInfo->base.cond, &pTaskInfo->streamInfo.tableCond, sizeof(SQueryTableDataCond));
1769
    if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__PREPARE1) {
H
Haojun Liao 已提交
1770 1771 1772 1773
      pTSInfo->base.cond.startVersion = 0;
      pTSInfo->base.cond.endVersion = pTaskInfo->streamInfo.fillHistoryVer1;
      qDebug("stream recover step 1, from %" PRId64 " to %" PRId64, pTSInfo->base.cond.startVersion,
             pTSInfo->base.cond.endVersion);
1774
    } else {
H
Haojun Liao 已提交
1775 1776 1777 1778
      pTSInfo->base.cond.startVersion = pTaskInfo->streamInfo.fillHistoryVer1 + 1;
      pTSInfo->base.cond.endVersion = pTaskInfo->streamInfo.fillHistoryVer2;
      qDebug("stream recover step 2, from %" PRId64 " to %" PRId64, pTSInfo->base.cond.startVersion,
             pTSInfo->base.cond.endVersion);
1779
    }
L
Liu Jicong 已提交
1780 1781

    /*resetTableScanInfo(pTSInfo, pWin);*/
H
Haojun Liao 已提交
1782 1783
    tsdbReaderClose(pTSInfo->base.dataReader);
    pTSInfo->base.dataReader = NULL;
L
Liu Jicong 已提交
1784

L
Liu Jicong 已提交
1785 1786 1787 1788 1789 1790 1791 1792
    pTSInfo->scanTimes = 0;
    pTSInfo->currentGroupId = -1;
    pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__SCAN;
  }

  if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__SCAN) {
    SSDataBlock* pBlock = doTableScan(pInfo->pTableScanOp);
    if (pBlock != NULL) {
L
Liu Jicong 已提交
1793
      calBlockTbName(pInfo, pBlock);
1794
      if (pInfo->pUpdateInfo) {
L
Liu Jicong 已提交
1795 1796
        TSKEY maxTs = updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex);
        pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
1797
      }
1798
      qDebug("stream recover scan get block, rows %d", pBlock->info.rows);
L
Liu Jicong 已提交
1799
      printDataBlock(pBlock, "scan recover");
L
Liu Jicong 已提交
1800 1801 1802
      return pBlock;
    }
    pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__NONE;
L
Liu Jicong 已提交
1803
    STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
H
Haojun Liao 已提交
1804 1805
    tsdbReaderClose(pTSInfo->base.dataReader);
    pTSInfo->base.dataReader = NULL;
1806

H
Haojun Liao 已提交
1807 1808
    pTSInfo->base.cond.startVersion = -1;
    pTSInfo->base.cond.endVersion = -1;
L
Liu Jicong 已提交
1809

L
Liu Jicong 已提交
1810 1811 1812
    return NULL;
  }

5
54liuyao 已提交
1813
  size_t total = taosArrayGetSize(pInfo->pBlockLists);
5
54liuyao 已提交
1814
// TODO: refactor
L
Liu Jicong 已提交
1815
FETCH_NEXT_BLOCK:
L
Liu Jicong 已提交
1816
  if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
H
Haojun Liao 已提交
1817
    if (pInfo->validBlockIndex >= total) {
L
Liu Jicong 已提交
1818
      doClearBufferedBlocks(pInfo);
L
Liu Jicong 已提交
1819
      /*pOperator->status = OP_EXEC_DONE;*/
H
Haojun Liao 已提交
1820 1821 1822
      return NULL;
    }

1823
    int32_t      current = pInfo->validBlockIndex++;
1824
    SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current);
H
Haojun Liao 已提交
1825 1826
    if (pBlock->info.id.groupId && pBlock->info.parTbName[0]) {
      streamStatePutParName(pTaskInfo->streamInfo.pState, pBlock->info.id.groupId, pBlock->info.parTbName);
1827
    }
1828
    // TODO move into scan
5
54liuyao 已提交
1829 1830
    pBlock->info.calWin.skey = INT64_MIN;
    pBlock->info.calWin.ekey = INT64_MAX;
1831
    pBlock->info.dataLoad = 1;
1832
    blockDataUpdateTsWindow(pBlock, 0);
1833
    switch (pBlock->info.type) {
L
Liu Jicong 已提交
1834 1835 1836
      case STREAM_NORMAL:
      case STREAM_GET_ALL:
        return pBlock;
1837 1838 1839
      case STREAM_RETRIEVE: {
        pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
        pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RETRIEVE;
1840 1841
        copyDataBlock(pInfo->pUpdateRes, pBlock);
        prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
1842 1843 1844
        updateInfoAddCloseWindowSBF(pInfo->pUpdateInfo);
      } break;
      case STREAM_DELETE_DATA: {
1845
        printDataBlock(pBlock, "stream scan delete recv");
L
Liu Jicong 已提交
1846
        SSDataBlock* pDelBlock = NULL;
L
Liu Jicong 已提交
1847
        if (pInfo->tqReader) {
L
Liu Jicong 已提交
1848
          pDelBlock = createSpecialDataBlock(STREAM_DELETE_DATA);
L
Liu Jicong 已提交
1849
          filterDelBlockByUid(pDelBlock, pBlock, pInfo);
L
Liu Jicong 已提交
1850 1851
        } else {
          pDelBlock = pBlock;
L
Liu Jicong 已提交
1852
        }
5
54liuyao 已提交
1853 1854
        setBlockGroupIdByUid(pInfo, pDelBlock);
        printDataBlock(pDelBlock, "stream scan delete recv filtered");
5
54liuyao 已提交
1855 1856 1857 1858 1859 1860
        if (pDelBlock->info.rows == 0) {
          if (pInfo->tqReader) {
            blockDataDestroy(pDelBlock);
          }
          goto FETCH_NEXT_BLOCK;
        }
1861
        if (!isIntervalWindow(pInfo) && !isSessionWindow(pInfo) && !isStateWindow(pInfo)) {
L
Liu Jicong 已提交
1862
          generateDeleteResultBlock(pInfo, pDelBlock, pInfo->pDeleteDataRes);
1863
          pInfo->pDeleteDataRes->info.type = STREAM_DELETE_RESULT;
L
Liu Jicong 已提交
1864
          printDataBlock(pDelBlock, "stream scan delete result");
H
Haojun Liao 已提交
1865 1866
          blockDataDestroy(pDelBlock);

L
Liu Jicong 已提交
1867 1868 1869 1870 1871
          if (pInfo->pDeleteDataRes->info.rows > 0) {
            return pInfo->pDeleteDataRes;
          } else {
            goto FETCH_NEXT_BLOCK;
          }
1872 1873 1874
        } else {
          pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
          pInfo->updateResIndex = 0;
L
Liu Jicong 已提交
1875
          generateScanRange(pInfo, pDelBlock, pInfo->pUpdateRes);
1876 1877 1878
          prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
          copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes);
          pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA;
L
Liu Jicong 已提交
1879 1880 1881 1882
          printDataBlock(pDelBlock, "stream scan delete data");
          if (pInfo->tqReader) {
            blockDataDestroy(pDelBlock);
          }
L
Liu Jicong 已提交
1883
          if (pInfo->pDeleteDataRes->info.rows > 0) {
5
54liuyao 已提交
1884
            pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
L
Liu Jicong 已提交
1885 1886 1887 1888
            return pInfo->pDeleteDataRes;
          } else {
            goto FETCH_NEXT_BLOCK;
          }
1889
        }
1890 1891 1892
      } break;
      default:
        break;
5
54liuyao 已提交
1893
    }
1894
    // printDataBlock(pBlock, "stream scan recv");
1895
    return pBlock;
L
Liu Jicong 已提交
1896
  } else if (pInfo->blockType == STREAM_INPUT__DATA_SUBMIT) {
L
Liu Jicong 已提交
1897
    qDebug("scan mode %d", pInfo->scanMode);
5
54liuyao 已提交
1898 1899 1900 1901 1902 1903
    switch (pInfo->scanMode) {
      case STREAM_SCAN_FROM_RES: {
        blockDataDestroy(pInfo->pUpdateRes);
        pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
        return pInfo->pRes;
      } break;
1904
      case STREAM_SCAN_FROM_DELETE_DATA: {
1905 1906 1907 1908 1909 1910 1911
        generateScanRange(pInfo, pInfo->pUpdateDataRes, pInfo->pUpdateRes);
        prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
        pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
        copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes);
        pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA;
        return pInfo->pDeleteDataRes;
      } break;
5
54liuyao 已提交
1912 1913 1914 1915 1916 1917 1918 1919 1920 1921
      case STREAM_SCAN_FROM_UPDATERES: {
        generateScanRange(pInfo, pInfo->pUpdateDataRes, pInfo->pUpdateRes);
        prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
        pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
        return pInfo->pUpdateRes;
      } break;
      case STREAM_SCAN_FROM_DATAREADER_RANGE:
      case STREAM_SCAN_FROM_DATAREADER_RETRIEVE: {
        SSDataBlock* pSDB = doRangeScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
        if (pSDB) {
1922
          STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;
H
Haojun Liao 已提交
1923 1924
          uint64_t        version = getReaderMaxVersion(pTableScanInfo->base.dataReader);
          updateInfoSetScanRange(pInfo->pUpdateInfo, &pTableScanInfo->base.cond.twindows, pInfo->groupId, version);
5
54liuyao 已提交
1925 1926
          pSDB->info.type = pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE ? STREAM_NORMAL : STREAM_PULL_DATA;
          checkUpdateData(pInfo, true, pSDB, false);
1927
          // printDataBlock(pSDB, "stream scan update");
L
Liu Jicong 已提交
1928
          calBlockTbName(pInfo, pSDB);
5
54liuyao 已提交
1929 1930
          return pSDB;
        }
1931
        blockDataCleanup(pInfo->pUpdateDataRes);
5
54liuyao 已提交
1932 1933 1934 1935
        pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
      } break;
      default:
        break;
1936
    }
1937

1938
    SStreamAggSupporter* pSup = pInfo->windowSup.pStreamAggSup;
5
54liuyao 已提交
1939
    if (isStateWindow(pInfo) && pSup->pScanBlock->info.rows > 0) {
1940 1941
      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
      pInfo->updateResIndex = 0;
5
54liuyao 已提交
1942 1943
      copyDataBlock(pInfo->pUpdateRes, pSup->pScanBlock);
      blockDataCleanup(pSup->pScanBlock);
1944 1945
      prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
      return pInfo->pUpdateRes;
5
54liuyao 已提交
1946
    }
5
54liuyao 已提交
1947

H
Haojun Liao 已提交
1948 1949
    SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;

1950
    int32_t totBlockNum = taosArrayGetSize(pInfo->pBlockLists);
1951

L
Liu Jicong 已提交
1952
  NEXT_SUBMIT_BLK:
1953 1954 1955
    while (1) {
      if (pInfo->tqReader->pMsg == NULL) {
        if (pInfo->validBlockIndex >= totBlockNum) {
5
54liuyao 已提交
1956
          updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
L
Liu Jicong 已提交
1957
          doClearBufferedBlocks(pInfo);
L
Liu Jicong 已提交
1958
          qDebug("stream scan return empty, consume block %d", totBlockNum);
1959 1960
          return NULL;
        }
1961

1962 1963 1964 1965 1966 1967 1968 1969
        int32_t     current = pInfo->validBlockIndex++;
        SSubmitReq* pSubmit = taosArrayGetP(pInfo->pBlockLists, current);
        if (tqReaderSetDataMsg(pInfo->tqReader, pSubmit, 0) < 0) {
          qError("submit msg messed up when initing stream submit block %p, current %d, total %d", pSubmit, current,
                 totBlockNum);
          pInfo->tqReader->pMsg = NULL;
          continue;
        }
H
Haojun Liao 已提交
1970 1971
      }

1972 1973 1974 1975
      blockDataCleanup(pInfo->pRes);

      while (tqNextDataBlock(pInfo->tqReader)) {
        SSDataBlock block = {0};
1976

1977 1978 1979 1980 1981 1982
        int32_t code = tqRetrieveDataBlock(&block, pInfo->tqReader);

        if (code != TSDB_CODE_SUCCESS || block.info.rows == 0) {
          continue;
        }

1983
        setBlockIntoRes(pInfo, &block, false);
1984

H
Haojun Liao 已提交
1985
        if (updateInfoIgnore(pInfo->pUpdateInfo, &pInfo->pRes->info.window, pInfo->pRes->info.id.groupId,
L
Liu Jicong 已提交
1986
                             pInfo->pRes->info.version)) {
1987 1988 1989 1990 1991
          printDataBlock(pInfo->pRes, "stream scan ignore");
          blockDataCleanup(pInfo->pRes);
          continue;
        }

1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007
        if (pInfo->pUpdateInfo) {
          checkUpdateData(pInfo, true, pInfo->pRes, true);
          pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlockInfo->window.ekey);
          if (pInfo->pUpdateDataRes->info.rows > 0) {
            pInfo->updateResIndex = 0;
            if (pInfo->pUpdateDataRes->info.type == STREAM_CLEAR) {
              pInfo->scanMode = STREAM_SCAN_FROM_UPDATERES;
            } else if (pInfo->pUpdateDataRes->info.type == STREAM_INVERT) {
              pInfo->scanMode = STREAM_SCAN_FROM_RES;
              return pInfo->pUpdateDataRes;
            } else if (pInfo->pUpdateDataRes->info.type == STREAM_DELETE_DATA) {
              pInfo->scanMode = STREAM_SCAN_FROM_DELETE_DATA;
            }
          }
        }

H
Haojun Liao 已提交
2008
        doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
2009
        pInfo->pRes->info.dataLoad = 1;
2010 2011 2012
        blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);

        if (pBlockInfo->rows > 0 || pInfo->pUpdateDataRes->info.rows > 0) {
2013 2014 2015
          break;
        }
      }
2016
      if (pBlockInfo->rows > 0 || pInfo->pUpdateDataRes->info.rows > 0) {
5
54liuyao 已提交
2017
        break;
J
jiacy-jcy 已提交
2018 2019
      } else {
        pInfo->tqReader->pMsg = NULL;
2020
        continue;
5
54liuyao 已提交
2021
      }
2022
      /*blockDataCleanup(pInfo->pRes);*/
H
Haojun Liao 已提交
2023 2024 2025 2026
    }

    // record the scan action.
    pInfo->numOfExec++;
2027
    pOperator->resultInfo.totalRows += pBlockInfo->rows;
2028
    // printDataBlock(pInfo->pRes, "stream scan");
H
Haojun Liao 已提交
2029

L
Liu Jicong 已提交
2030
    qDebug("scan rows: %d", pBlockInfo->rows);
L
Liu Jicong 已提交
2031 2032 2033
    if (pBlockInfo->rows > 0) {
      return pInfo->pRes;
    }
2034 2035 2036 2037 2038 2039

    if (pInfo->pUpdateDataRes->info.rows > 0) {
      goto FETCH_NEXT_BLOCK;
    }

    goto NEXT_SUBMIT_BLK;
L
Liu Jicong 已提交
2040 2041 2042
  } else {
    ASSERT(0);
    return NULL;
H
Haojun Liao 已提交
2043 2044 2045
  }
}

H
Haojun Liao 已提交
2046
static SArray* extractTableIdList(const STableListInfo* pTableListInfo) {
2047 2048 2049
  SArray* tableIdList = taosArrayInit(4, sizeof(uint64_t));

  // Transfer the Array of STableKeyInfo into uid list.
H
Haojun Liao 已提交
2050 2051 2052
  size_t size = tableListGetSize(pTableListInfo);
  for (int32_t i = 0; i < size; ++i) {
    STableKeyInfo* pkeyInfo = tableListGetInfo(pTableListInfo, i);
2053 2054 2055 2056 2057 2058
    taosArrayPush(tableIdList, &pkeyInfo->uid);
  }

  return tableIdList;
}

2059
static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
L
Liu Jicong 已提交
2060 2061
  // NOTE: this operator does never check if current status is done or not
  SExecTaskInfo*      pTaskInfo = pOperator->pTaskInfo;
2062
  SStreamRawScanInfo* pInfo = pOperator->info;
L
Liu Jicong 已提交
2063
  pTaskInfo->streamInfo.metaRsp.metaRspLen = 0;  // use metaRspLen !=0 to judge if data is meta
wmmhello's avatar
wmmhello 已提交
2064
  pTaskInfo->streamInfo.metaRsp.metaRsp = NULL;
2065

wmmhello's avatar
wmmhello 已提交
2066
  qDebug("tmqsnap doRawScan called");
L
Liu Jicong 已提交
2067
  if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
2068
    if (pInfo->dataReader && tsdbNextDataBlock(pInfo->dataReader)) {
wmmhello's avatar
wmmhello 已提交
2069
      if (isTaskKilled(pTaskInfo)) {
2070
        longjmp(pTaskInfo->env, pTaskInfo->code);
wmmhello's avatar
wmmhello 已提交
2071
      }
2072

H
Haojun Liao 已提交
2073 2074
      SSDataBlock* pBlock = tsdbRetrieveDataBlock(pInfo->dataReader, NULL);
      if (pBlock == NULL) {
wmmhello's avatar
wmmhello 已提交
2075
        longjmp(pTaskInfo->env, terrno);
wmmhello's avatar
wmmhello 已提交
2076 2077
      }

H
Haojun Liao 已提交
2078
      qDebug("tmqsnap doRawScan get data uid:%" PRId64 "", pBlock->info.id.uid);
wmmhello's avatar
wmmhello 已提交
2079
      pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_DATA;
H
Haojun Liao 已提交
2080
      pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.id.uid;
wmmhello's avatar
wmmhello 已提交
2081 2082 2083
      pTaskInfo->streamInfo.lastStatus.ts = pBlock->info.window.ekey;
      return pBlock;
    }
wmmhello's avatar
wmmhello 已提交
2084 2085

    SMetaTableInfo mtInfo = getUidfromSnapShot(pInfo->sContext);
L
Liu Jicong 已提交
2086
    if (mtInfo.uid == 0) {  // read snapshot done, change to get data from wal
wmmhello's avatar
wmmhello 已提交
2087 2088
      qDebug("tmqsnap read snapshot done, change to get data from wal");
      pTaskInfo->streamInfo.prepareStatus.uid = mtInfo.uid;
wmmhello's avatar
wmmhello 已提交
2089 2090
      pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__LOG;
      pTaskInfo->streamInfo.lastStatus.version = pInfo->sContext->snapVersion;
L
Liu Jicong 已提交
2091
    } else {
wmmhello's avatar
wmmhello 已提交
2092 2093
      pTaskInfo->streamInfo.prepareStatus.uid = mtInfo.uid;
      pTaskInfo->streamInfo.prepareStatus.ts = INT64_MIN;
2094
      qDebug("tmqsnap change get data uid:%" PRId64 "", mtInfo.uid);
wmmhello's avatar
wmmhello 已提交
2095 2096
      qStreamPrepareScan(pTaskInfo, &pTaskInfo->streamInfo.prepareStatus, pInfo->sContext->subType);
    }
2097
    tDeleteSSchemaWrapper(mtInfo.schema);
wmmhello's avatar
wmmhello 已提交
2098
    qDebug("tmqsnap stream scan tsdb return null");
wmmhello's avatar
wmmhello 已提交
2099
    return NULL;
L
Liu Jicong 已提交
2100 2101 2102 2103 2104 2105 2106
  } else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_META) {
    SSnapContext* sContext = pInfo->sContext;
    void*         data = NULL;
    int32_t       dataLen = 0;
    int16_t       type = 0;
    int64_t       uid = 0;
    if (getMetafromSnapShot(sContext, &data, &dataLen, &type, &uid) < 0) {
wmmhello's avatar
wmmhello 已提交
2107
      qError("tmqsnap getMetafromSnapShot error");
wmmhello's avatar
wmmhello 已提交
2108
      taosMemoryFreeClear(data);
2109 2110 2111
      return NULL;
    }

L
Liu Jicong 已提交
2112
    if (!sContext->queryMetaOrData) {  // change to get data next poll request
wmmhello's avatar
wmmhello 已提交
2113 2114 2115 2116
      pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_META;
      pTaskInfo->streamInfo.lastStatus.uid = uid;
      pTaskInfo->streamInfo.metaRsp.rspOffset.type = TMQ_OFFSET__SNAPSHOT_DATA;
      pTaskInfo->streamInfo.metaRsp.rspOffset.uid = 0;
wmmhello's avatar
wmmhello 已提交
2117
      pTaskInfo->streamInfo.metaRsp.rspOffset.ts = INT64_MIN;
L
Liu Jicong 已提交
2118
    } else {
wmmhello's avatar
wmmhello 已提交
2119 2120 2121 2122 2123 2124 2125
      pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_META;
      pTaskInfo->streamInfo.lastStatus.uid = uid;
      pTaskInfo->streamInfo.metaRsp.rspOffset = pTaskInfo->streamInfo.lastStatus;
      pTaskInfo->streamInfo.metaRsp.resMsgType = type;
      pTaskInfo->streamInfo.metaRsp.metaRspLen = dataLen;
      pTaskInfo->streamInfo.metaRsp.metaRsp = data;
    }
2126

wmmhello's avatar
wmmhello 已提交
2127
    return NULL;
2128
  }
L
Liu Jicong 已提交
2129 2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152 2153 2154 2155 2156 2157 2158 2159 2160 2161 2162 2163 2164 2165 2166
  //  else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__LOG) {
  //    int64_t fetchVer = pTaskInfo->streamInfo.prepareStatus.version + 1;
  //
  //    while(1){
  //      if (tqFetchLog(pInfo->tqReader->pWalReader, pInfo->sContext->withMeta, &fetchVer, &pInfo->pCkHead) < 0) {
  //        qDebug("tmqsnap tmq poll: consumer log end. offset %" PRId64, fetchVer);
  //        pTaskInfo->streamInfo.lastStatus.version = fetchVer;
  //        pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__LOG;
  //        return NULL;
  //      }
  //      SWalCont* pHead = &pInfo->pCkHead->head;
  //      qDebug("tmqsnap tmq poll: consumer log offset %" PRId64 " msgType %d", fetchVer, pHead->msgType);
  //
  //      if (pHead->msgType == TDMT_VND_SUBMIT) {
  //        SSubmitReq* pCont = (SSubmitReq*)&pHead->body;
  //        tqReaderSetDataMsg(pInfo->tqReader, pCont, 0);
  //        SSDataBlock* block = tqLogScanExec(pInfo->sContext->subType, pInfo->tqReader, pInfo->pFilterOutTbUid,
  //        &pInfo->pRes); if(block){
  //          pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__LOG;
  //          pTaskInfo->streamInfo.lastStatus.version = fetchVer;
  //          qDebug("tmqsnap fetch data msg, ver:%" PRId64 ", type:%d", pHead->version, pHead->msgType);
  //          return block;
  //        }else{
  //          fetchVer++;
  //        }
  //      } else{
  //        ASSERT(pInfo->sContext->withMeta);
  //        ASSERT(IS_META_MSG(pHead->msgType));
  //        qDebug("tmqsnap fetch meta msg, ver:%" PRId64 ", type:%d", pHead->version, pHead->msgType);
  //        pTaskInfo->streamInfo.metaRsp.rspOffset.version = fetchVer;
  //        pTaskInfo->streamInfo.metaRsp.rspOffset.type = TMQ_OFFSET__LOG;
  //        pTaskInfo->streamInfo.metaRsp.resMsgType = pHead->msgType;
  //        pTaskInfo->streamInfo.metaRsp.metaRspLen = pHead->bodyLen;
  //        pTaskInfo->streamInfo.metaRsp.metaRsp = taosMemoryMalloc(pHead->bodyLen);
  //        memcpy(pTaskInfo->streamInfo.metaRsp.metaRsp, pHead->body, pHead->bodyLen);
  //        return NULL;
  //      }
  //    }
2167 2168 2169
  return NULL;
}

wmmhello's avatar
wmmhello 已提交
2170
static void destroyRawScanOperatorInfo(void* param) {
wmmhello's avatar
wmmhello 已提交
2171 2172 2173 2174 2175 2176
  SStreamRawScanInfo* pRawScan = (SStreamRawScanInfo*)param;
  tsdbReaderClose(pRawScan->dataReader);
  destroySnapContext(pRawScan->sContext);
  taosMemoryFree(pRawScan);
}

L
Liu Jicong 已提交
2177 2178 2179
// for subscribing db or stb (not including column),
// if this scan is used, meta data can be return
// and schemas are decided when scanning
2180
SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pTaskInfo) {
L
Liu Jicong 已提交
2181 2182 2183 2184 2185
  // create operator
  // create tb reader
  // create meta reader
  // create tq reader

H
Haojun Liao 已提交
2186 2187
  int32_t code = TSDB_CODE_SUCCESS;

2188
  SStreamRawScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamRawScanInfo));
L
Liu Jicong 已提交
2189
  SOperatorInfo*      pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
2190
  if (pInfo == NULL || pOperator == NULL) {
H
Haojun Liao 已提交
2191 2192
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _end;
2193 2194
  }

wmmhello's avatar
wmmhello 已提交
2195 2196
  pInfo->vnode = pHandle->vnode;

2197
  pInfo->sContext = pHandle->sContext;
L
Liu Jicong 已提交
2198 2199
  setOperatorInfo(pOperator, "RawScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, false, OP_NOT_OPENED, pInfo,
                  pTaskInfo);
2200

2201
  pOperator->fpSet = createOperatorFpSet(NULL, doRawScan, NULL, destroyRawScanOperatorInfo, optrDefaultBufFn, NULL);
2202
  return pOperator;
H
Haojun Liao 已提交
2203

L
Liu Jicong 已提交
2204
_end:
H
Haojun Liao 已提交
2205 2206 2207 2208
  taosMemoryFree(pInfo);
  taosMemoryFree(pOperator);
  pTaskInfo->code = code;
  return NULL;
L
Liu Jicong 已提交
2209 2210
}

2211
static void destroyStreamScanOperatorInfo(void* param) {
2212 2213
  SStreamScanInfo* pStreamScan = (SStreamScanInfo*)param;
  if (pStreamScan->pTableScanOp && pStreamScan->pTableScanOp->info) {
5
54liuyao 已提交
2214
    destroyOperatorInfo(pStreamScan->pTableScanOp);
2215 2216 2217 2218
  }
  if (pStreamScan->tqReader) {
    tqCloseReader(pStreamScan->tqReader);
  }
H
Haojun Liao 已提交
2219 2220
  if (pStreamScan->matchInfo.pList) {
    taosArrayDestroy(pStreamScan->matchInfo.pList);
2221
  }
C
Cary Xu 已提交
2222 2223
  if (pStreamScan->pPseudoExpr) {
    destroyExprInfo(pStreamScan->pPseudoExpr, pStreamScan->numOfPseudoExpr);
L
Liu Jicong 已提交
2224
    taosMemoryFree(pStreamScan->pPseudoExpr);
C
Cary Xu 已提交
2225
  }
C
Cary Xu 已提交
2226

L
Liu Jicong 已提交
2227 2228
  cleanupExprSupp(&pStreamScan->tbnameCalSup);

L
Liu Jicong 已提交
2229
  updateInfoDestroy(pStreamScan->pUpdateInfo);
2230 2231 2232 2233
  blockDataDestroy(pStreamScan->pRes);
  blockDataDestroy(pStreamScan->pUpdateRes);
  blockDataDestroy(pStreamScan->pPullDataRes);
  blockDataDestroy(pStreamScan->pDeleteDataRes);
5
54liuyao 已提交
2234
  blockDataDestroy(pStreamScan->pUpdateDataRes);
2235 2236 2237 2238
  taosArrayDestroy(pStreamScan->pBlockLists);
  taosMemoryFree(pStreamScan);
}

2239
SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond,
2240
                                            SExecTaskInfo* pTaskInfo) {
2241 2242
  SStreamScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamScanInfo));
  SOperatorInfo*   pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
2243

H
Haojun Liao 已提交
2244
  if (pInfo == NULL || pOperator == NULL) {
S
Shengliang Guan 已提交
2245
    terrno = TSDB_CODE_OUT_OF_MEMORY;
2246
    goto _error;
H
Haojun Liao 已提交
2247 2248
  }

2249
  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;
2250
  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;
H
Haojun Liao 已提交
2251

2252
  pInfo->pTagCond = pTagCond;
2253
  pInfo->pGroupTags = pTableScanNode->pGroupTags;
2254

2255
  int32_t numOfCols = 0;
2256 2257
  int32_t code =
      extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);
H
Haojun Liao 已提交
2258 2259 2260
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2261

H
Haojun Liao 已提交
2262
  int32_t numOfOutput = taosArrayGetSize(pInfo->matchInfo.pList);
2263
  SArray* pColIds = taosArrayInit(numOfOutput, sizeof(int16_t));
2264
  for (int32_t i = 0; i < numOfOutput; ++i) {
H
Haojun Liao 已提交
2265
    SColMatchItem* id = taosArrayGet(pInfo->matchInfo.pList, i);
2266 2267

    int16_t colId = id->colId;
2268
    taosArrayPush(pColIds, &colId);
2269
    if (id->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
H
Haojun Liao 已提交
2270
      pInfo->primaryTsIndex = id->dstSlotId;
5
54liuyao 已提交
2271
    }
H
Haojun Liao 已提交
2272 2273
  }

L
Liu Jicong 已提交
2274 2275 2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286
  if (pTableScanNode->pSubtable != NULL) {
    SExprInfo* pSubTableExpr = taosMemoryCalloc(1, sizeof(SExprInfo));
    if (pSubTableExpr == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto _error;
    }
    pInfo->tbnameCalSup.pExprInfo = pSubTableExpr;
    createExprFromOneNode(pSubTableExpr, pTableScanNode->pSubtable, 0);
    if (initExprSupp(&pInfo->tbnameCalSup, pSubTableExpr, 1) != 0) {
      goto _error;
    }
  }

2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299
  if (pTableScanNode->pTags != NULL) {
    int32_t    numOfTags;
    SExprInfo* pTagExpr = createExprInfo(pTableScanNode->pTags, NULL, &numOfTags);
    if (pTagExpr == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto _error;
    }
    if (initExprSupp(&pInfo->tagCalSup, pTagExpr, numOfTags) != 0) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto _error;
    }
  }

H
Haojun Liao 已提交
2300 2301
  pInfo->pBlockLists = taosArrayInit(4, POINTER_BYTES);
  if (pInfo->pBlockLists == NULL) {
2302 2303
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _error;
H
Haojun Liao 已提交
2304 2305
  }

5
54liuyao 已提交
2306
  if (pHandle->vnode) {
L
Liu Jicong 已提交
2307
    SOperatorInfo*  pTableScanOp = createTableScanOperatorInfo(pTableScanNode, pHandle, pTaskInfo);
L
Liu Jicong 已提交
2308
    STableScanInfo* pTSInfo = (STableScanInfo*)pTableScanOp->info;
2309
    if (pHandle->version > 0) {
H
Haojun Liao 已提交
2310
      pTSInfo->base.cond.endVersion = pHandle->version;
2311
    }
L
Liu Jicong 已提交
2312

2313
    STableKeyInfo* pList = NULL;
5
54liuyao 已提交
2314
    int32_t        num = 0;
H
Haojun Liao 已提交
2315
    tableListGetGroupList(pTaskInfo->pTableInfoList, 0, &pList, &num);
2316

2317
    if (pHandle->initTableReader) {
L
Liu Jicong 已提交
2318
      pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;
H
Haojun Liao 已提交
2319
      pTSInfo->base.dataReader = NULL;
L
Liu Jicong 已提交
2320 2321
      code = tsdbReaderOpen(pHandle->vnode, &pTSInfo->base.cond, pList, num, pTSInfo->pResBlock,
                            &pTSInfo->base.dataReader, NULL);
dengyihao's avatar
dengyihao 已提交
2322 2323
      if (code != 0) {
        terrno = code;
H
Haojun Liao 已提交
2324
        destroyTableScanOperatorInfo(pTableScanOp);
2325
        goto _error;
L
Liu Jicong 已提交
2326
      }
L
Liu Jicong 已提交
2327 2328
    }

L
Liu Jicong 已提交
2329 2330 2331 2332
    if (pHandle->initTqReader) {
      ASSERT(pHandle->tqReader == NULL);
      pInfo->tqReader = tqOpenReader(pHandle->vnode);
      ASSERT(pInfo->tqReader);
2333
    } else {
L
Liu Jicong 已提交
2334 2335
      ASSERT(pHandle->tqReader);
      pInfo->tqReader = pHandle->tqReader;
2336 2337
    }

2338
    pInfo->pUpdateInfo = NULL;
2339
    pInfo->pTableScanOp = pTableScanOp;
2340 2341 2342
    if (pInfo->pTableScanOp->pTaskInfo->streamInfo.pState) {
      streamStateSetNumber(pInfo->pTableScanOp->pTaskInfo->streamInfo.pState, -1);
    }
L
Liu Jicong 已提交
2343

L
Liu Jicong 已提交
2344 2345
    pInfo->readHandle = *pHandle;
    pInfo->tableUid = pScanPhyNode->uid;
L
Liu Jicong 已提交
2346
    pTaskInfo->streamInfo.snapshotVer = pHandle->version;
L
Liu Jicong 已提交
2347

L
Liu Jicong 已提交
2348
    // set the extract column id to streamHandle
L
Liu Jicong 已提交
2349
    tqReaderSetColIdList(pInfo->tqReader, pColIds);
H
Haojun Liao 已提交
2350
    SArray* tableIdList = extractTableIdList(pTaskInfo->pTableInfoList);
2351
    code = tqReaderSetTbUidList(pInfo->tqReader, tableIdList);
L
Liu Jicong 已提交
2352 2353 2354 2355 2356
    if (code != 0) {
      taosArrayDestroy(tableIdList);
      goto _error;
    }
    taosArrayDestroy(tableIdList);
H
Haojun Liao 已提交
2357
    memcpy(&pTaskInfo->streamInfo.tableCond, &pTSInfo->base.cond, sizeof(SQueryTableDataCond));
L
Liu Jicong 已提交
2358 2359
  } else {
    taosArrayDestroy(pColIds);
5
54liuyao 已提交
2360 2361
  }

2362 2363 2364 2365 2366
  // create the pseduo columns info
  if (pTableScanNode->scan.pScanPseudoCols != NULL) {
    pInfo->pPseudoExpr = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pInfo->numOfPseudoExpr);
  }

H
Haojun Liao 已提交
2367 2368 2369 2370 2371
  code = filterInitFromNode((SNode*)pScanPhyNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
2372
  pInfo->pRes = createDataBlockFromDescNode(pDescNode);
2373
  pInfo->pUpdateRes = createSpecialDataBlock(STREAM_CLEAR);
2374
  pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
L
Liu Jicong 已提交
2375
  pInfo->windowSup = (SWindowSupporter){.pStreamAggSup = NULL, .gap = -1, .parentType = QUERY_NODE_PHYSICAL_PLAN};
2376
  pInfo->groupId = 0;
2377
  pInfo->pPullDataRes = createSpecialDataBlock(STREAM_RETRIEVE);
2378
  pInfo->pStreamScanOp = pOperator;
2379
  pInfo->deleteDataIndex = 0;
2380
  pInfo->pDeleteDataRes = createSpecialDataBlock(STREAM_DELETE_DATA);
5
54liuyao 已提交
2381
  pInfo->updateWin = (STimeWindow){.skey = INT64_MAX, .ekey = INT64_MAX};
2382
  pInfo->pUpdateDataRes = createSpecialDataBlock(STREAM_CLEAR);
X
Xiaoyu Wang 已提交
2383
  pInfo->assignBlockUid = pTableScanNode->assignBlockUid;
2384
  pInfo->partitionSup.needCalc = false;
L
Liu Jicong 已提交
2385

L
Liu Jicong 已提交
2386 2387
  setOperatorInfo(pOperator, "StreamScanOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, false, OP_NOT_OPENED, pInfo,
                  pTaskInfo);
2388
  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);
H
Haojun Liao 已提交
2389

L
Liu Jicong 已提交
2390
  __optr_fn_t nextFn = pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM ? doStreamScan : doQueueScan;
L
Liu Jicong 已提交
2391 2392
  pOperator->fpSet =
      createOperatorFpSet(optrDummyOpenFn, nextFn, NULL, destroyStreamScanOperatorInfo, optrDefaultBufFn, NULL);
2393

H
Haojun Liao 已提交
2394
  return pOperator;
2395

L
Liu Jicong 已提交
2396
_error:
H
Haojun Liao 已提交
2397 2398 2399 2400 2401 2402 2403 2404
  if (pColIds != NULL) {
    taosArrayDestroy(pColIds);
  }

  if (pInfo != NULL) {
    destroyStreamScanOperatorInfo(pInfo);
  }

2405 2406
  taosMemoryFreeClear(pOperator);
  return NULL;
H
Haojun Liao 已提交
2407 2408
}

2409
static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
2410 2411 2412 2413
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

2414 2415 2416
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

  STagScanInfo* pInfo = pOperator->info;
2417
  SExprInfo*    pExprInfo = &pOperator->exprSupp.pExprInfo[0];
2418
  SSDataBlock*  pRes = pInfo->pRes;
2419
  blockDataCleanup(pRes);
H
Haojun Liao 已提交
2420

H
Haojun Liao 已提交
2421
  int32_t size = tableListGetSize(pTaskInfo->pTableInfoList);
wmmhello's avatar
wmmhello 已提交
2422
  if (size == 0) {
H
Haojun Liao 已提交
2423 2424 2425 2426
    setTaskStatus(pTaskInfo, TASK_COMPLETED);
    return NULL;
  }

2427 2428 2429
  char        str[512] = {0};
  int32_t     count = 0;
  SMetaReader mr = {0};
2430
  metaReaderInit(&mr, pInfo->readHandle.meta, 0);
H
Haojun Liao 已提交
2431

wmmhello's avatar
wmmhello 已提交
2432
  while (pInfo->curPos < size && count < pOperator->resultInfo.capacity) {
H
Haojun Liao 已提交
2433
    STableKeyInfo* item = tableListGetInfo(pTaskInfo->pTableInfoList, pInfo->curPos);
L
Liu Jicong 已提交
2434
    int32_t        code = metaGetTableEntryByUid(&mr, item->uid);
2435
    tDecoderClear(&mr.coder);
H
Haojun Liao 已提交
2436
    if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
2437 2438
      qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", item->uid, tstrerror(terrno),
             GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
2439
      metaReaderClear(&mr);
2440
      T_LONG_JMP(pTaskInfo->env, terrno);
H
Haojun Liao 已提交
2441
    }
H
Haojun Liao 已提交
2442

2443
    for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {
2444 2445 2446 2447 2448 2449
      SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pExprInfo[j].base.resSchema.slotId);

      // refactor later
      if (fmIsScanPseudoColumnFunc(pExprInfo[j].pExpr->_function.functionId)) {
        STR_TO_VARSTR(str, mr.me.name);
        colDataAppend(pDst, count, str, false);
2450
      } else {  // it is a tag value
wmmhello's avatar
wmmhello 已提交
2451 2452
        STagVal val = {0};
        val.cid = pExprInfo[j].base.pParam[0].pCol->colId;
2453
        const char* p = metaGetTableTagVal(mr.me.ctbEntry.pTags, pDst->info.type, &val);
wmmhello's avatar
wmmhello 已提交
2454

2455 2456 2457 2458
        char* data = NULL;
        if (pDst->info.type != TSDB_DATA_TYPE_JSON && p != NULL) {
          data = tTagValToData((const STagVal*)p, false);
        } else {
wmmhello's avatar
wmmhello 已提交
2459 2460
          data = (char*)p;
        }
L
Liu Jicong 已提交
2461 2462
        colDataAppend(pDst, count, data,
                      (data == NULL) || (pDst->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data)));
2463

2464 2465
        if (pDst->info.type != TSDB_DATA_TYPE_JSON && p != NULL && IS_VAR_DATA_TYPE(((const STagVal*)p)->type) &&
            data != NULL) {
wmmhello's avatar
wmmhello 已提交
2466
          taosMemoryFree(data);
wmmhello's avatar
wmmhello 已提交
2467
        }
H
Haojun Liao 已提交
2468 2469 2470
      }
    }

2471
    count += 1;
wmmhello's avatar
wmmhello 已提交
2472
    if (++pInfo->curPos >= size) {
H
Haojun Liao 已提交
2473
      setOperatorCompleted(pOperator);
H
Haojun Liao 已提交
2474 2475 2476
    }
  }

2477 2478
  metaReaderClear(&mr);

2479
  // qDebug("QInfo:0x%"PRIx64" create tag values results completed, rows:%d", GET_TASKID(pRuntimeEnv), count);
H
Haojun Liao 已提交
2480
  if (pOperator->status == OP_EXEC_DONE) {
2481
    setTaskStatus(pTaskInfo, TASK_COMPLETED);
H
Haojun Liao 已提交
2482 2483 2484
  }

  pRes->info.rows = count;
wmmhello's avatar
wmmhello 已提交
2485
  pOperator->resultInfo.totalRows += count;
2486

2487
  return (pRes->info.rows == 0) ? NULL : pInfo->pRes;
H
Haojun Liao 已提交
2488 2489
}

2490
static void destroyTagScanOperatorInfo(void* param) {
H
Haojun Liao 已提交
2491 2492
  STagScanInfo* pInfo = (STagScanInfo*)param;
  pInfo->pRes = blockDataDestroy(pInfo->pRes);
H
Haojun Liao 已提交
2493
  taosArrayDestroy(pInfo->matchInfo.pList);
D
dapan1121 已提交
2494
  taosMemoryFreeClear(param);
H
Haojun Liao 已提交
2495 2496
}

S
slzhou 已提交
2497 2498
SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode,
                                         SExecTaskInfo* pTaskInfo) {
2499
  STagScanInfo*  pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo));
H
Haojun Liao 已提交
2500 2501 2502 2503 2504
  SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

2505 2506 2507 2508
  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

  int32_t    numOfExprs = 0;
  SExprInfo* pExprInfo = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &numOfExprs);
2509
  int32_t    code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs);
2510 2511 2512
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2513

H
Haojun Liao 已提交
2514 2515
  int32_t num = 0;
  code = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);
2516 2517 2518
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2519

H
Haojun Liao 已提交
2520
  pInfo->pRes = createDataBlockFromDescNode(pDescNode);
2521 2522
  pInfo->readHandle = *pReadHandle;
  pInfo->curPos = 0;
2523

L
Liu Jicong 已提交
2524 2525
  setOperatorInfo(pOperator, "TagScanOperator", QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN, false, OP_NOT_OPENED, pInfo,
                  pTaskInfo);
2526
  initResultSizeInfo(&pOperator->resultInfo, 4096);
2527 2528
  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

L
Liu Jicong 已提交
2529 2530
  pOperator->fpSet =
      createOperatorFpSet(optrDummyOpenFn, doTagScan, NULL, destroyTagScanOperatorInfo, optrDefaultBufFn, NULL);
H
Haojun Liao 已提交
2531 2532

  return pOperator;
2533

2534
_error:
H
Haojun Liao 已提交
2535 2536 2537 2538 2539
  taosMemoryFree(pInfo);
  taosMemoryFree(pOperator);
  terrno = TSDB_CODE_OUT_OF_MEMORY;
  return NULL;
}
2540

dengyihao's avatar
dengyihao 已提交
2541
static SSDataBlock* getTableDataBlockImpl(void* param) {
dengyihao's avatar
opt mem  
dengyihao 已提交
2542 2543 2544 2545 2546 2547 2548
  STableMergeScanSortSourceParam* source = param;
  SOperatorInfo*                  pOperator = source->pOperator;
  STableMergeScanInfo*            pInfo = pOperator->info;
  SExecTaskInfo*                  pTaskInfo = pOperator->pTaskInfo;
  int32_t                         readIdx = source->readerIdx;
  SSDataBlock*                    pBlock = source->inputBlock;

H
Haojun Liao 已提交
2549
  SQueryTableDataCond* pQueryCond = taosArrayGet(pInfo->queryConds, readIdx);
dengyihao's avatar
opt mem  
dengyihao 已提交
2550

L
Liu Jicong 已提交
2551 2552
  int64_t      st = taosGetTimestampUs();
  void*        p = tableListGetInfo(pTaskInfo->pTableInfoList, readIdx + pInfo->tableStartIndex);
H
Haojun Liao 已提交
2553
  SReadHandle* pHandle = &pInfo->base.readHandle;
dengyihao's avatar
dengyihao 已提交
2554

L
Liu Jicong 已提交
2555 2556
  int32_t code =
      tsdbReaderOpen(pHandle->vnode, pQueryCond, p, 1, pBlock, &pInfo->base.dataReader, GET_TASKID(pTaskInfo));
dengyihao's avatar
dengyihao 已提交
2557
  if (code != 0) {
H
Haojun Liao 已提交
2558
    T_LONG_JMP(pTaskInfo->env, code);
dengyihao's avatar
dengyihao 已提交
2559
  }
dengyihao's avatar
opt mem  
dengyihao 已提交
2560

H
Haojun Liao 已提交
2561
  STsdbReader* reader = pInfo->base.dataReader;
dengyihao's avatar
opt mem  
dengyihao 已提交
2562
  while (tsdbNextDataBlock(reader)) {
H
Haojun Liao 已提交
2563
    if (isTaskKilled(pTaskInfo)) {
2564
      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
dengyihao's avatar
opt mem  
dengyihao 已提交
2565 2566 2567
    }

    // process this data block based on the probabilities
H
Haojun Liao 已提交
2568
    bool processThisBlock = processBlockWithProbability(&pInfo->sample);
dengyihao's avatar
opt mem  
dengyihao 已提交
2569 2570 2571 2572
    if (!processThisBlock) {
      continue;
    }

H
Haojun Liao 已提交
2573
    if (pQueryCond->order == TSDB_ORDER_ASC) {
dengyihao's avatar
opt mem  
dengyihao 已提交
2574 2575 2576 2577
      pQueryCond->twindows.skey = pBlock->info.window.ekey + 1;
    } else {
      pQueryCond->twindows.ekey = pBlock->info.window.skey - 1;
    }
dengyihao's avatar
opt mem  
dengyihao 已提交
2578 2579

    uint32_t status = 0;
H
Haojun Liao 已提交
2580
    loadDataBlock(pOperator, &pInfo->base, pBlock, &status);
S
slzhou 已提交
2581
    //    code = loadDataBlockFromOneTable(pOperator, pTableScanInfo, pBlock, &status);
dengyihao's avatar
opt mem  
dengyihao 已提交
2582
    if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2583
      T_LONG_JMP(pTaskInfo->env, code);
dengyihao's avatar
opt mem  
dengyihao 已提交
2584 2585 2586 2587 2588 2589 2590
    }

    // current block is filter out according to filter condition, continue load the next block
    if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) {
      continue;
    }

H
Haojun Liao 已提交
2591
    pBlock->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid);
dengyihao's avatar
opt mem  
dengyihao 已提交
2592

H
Haojun Liao 已提交
2593
    pOperator->resultInfo.totalRows += pBlock->info.rows;
H
Haojun Liao 已提交
2594
    pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;
dengyihao's avatar
opt mem  
dengyihao 已提交
2595

H
Haojun Liao 已提交
2596 2597
    tsdbReaderClose(pInfo->base.dataReader);
    pInfo->base.dataReader = NULL;
dengyihao's avatar
opt mem  
dengyihao 已提交
2598 2599
    return pBlock;
  }
H
Haojun Liao 已提交
2600

H
Haojun Liao 已提交
2601 2602
  tsdbReaderClose(pInfo->base.dataReader);
  pInfo->base.dataReader = NULL;
dengyihao's avatar
opt mem  
dengyihao 已提交
2603 2604 2605
  return NULL;
}

2606 2607 2608
SArray* generateSortByTsInfo(SArray* colMatchInfo, int32_t order) {
  int32_t tsTargetSlotId = 0;
  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {
H
Haojun Liao 已提交
2609
    SColMatchItem* colInfo = taosArrayGet(colMatchInfo, i);
2610
    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
H
Haojun Liao 已提交
2611
      tsTargetSlotId = colInfo->dstSlotId;
2612 2613 2614
    }
  }

2615 2616 2617
  SArray*         pList = taosArrayInit(1, sizeof(SBlockOrderInfo));
  SBlockOrderInfo bi = {0};
  bi.order = order;
2618
  bi.slotId = tsTargetSlotId;
2619 2620 2621 2622 2623 2624 2625
  bi.nullFirst = NULL_ORDER_FIRST;

  taosArrayPush(pList, &bi);

  return pList;
}

H
Haojun Liao 已提交
2626
int32_t dumpQueryTableCond(const SQueryTableDataCond* src, SQueryTableDataCond* dst) {
dengyihao's avatar
opt mem  
dengyihao 已提交
2627 2628 2629 2630 2631 2632 2633
  memcpy((void*)dst, (void*)src, sizeof(SQueryTableDataCond));
  dst->colList = taosMemoryCalloc(src->numOfCols, sizeof(SColumnInfo));
  for (int i = 0; i < src->numOfCols; i++) {
    dst->colList[i] = src->colList[i];
  }
  return 0;
}
H
Haojun Liao 已提交
2634

2635
int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {
2636 2637 2638
  STableMergeScanInfo* pInfo = pOperator->info;
  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

S
slzhou 已提交
2639
  {
H
Haojun Liao 已提交
2640
    size_t  numOfTables = tableListGetSize(pTaskInfo->pTableInfoList);
S
slzhou 已提交
2641
    int32_t i = pInfo->tableStartIndex + 1;
H
Haojun Liao 已提交
2642
    for (; i < numOfTables; ++i) {
H
Haojun Liao 已提交
2643
      STableKeyInfo* tableKeyInfo = tableListGetInfo(pTaskInfo->pTableInfoList, i);
S
slzhou 已提交
2644 2645 2646 2647 2648 2649
      if (tableKeyInfo->groupId != pInfo->groupId) {
        break;
      }
    }
    pInfo->tableEndIndex = i - 1;
  }
2650

S
slzhou 已提交
2651 2652
  int32_t tableStartIdx = pInfo->tableStartIndex;
  int32_t tableEndIdx = pInfo->tableEndIndex;
2653

H
Haojun Liao 已提交
2654
  pInfo->base.dataReader = NULL;
2655

2656 2657
  // todo the total available buffer should be determined by total capacity of buffer of this task.
  // the additional one is reserved for merge result
S
slzhou 已提交
2658
  pInfo->sortBufSize = pInfo->bufPageSize * (tableEndIdx - tableStartIdx + 1 + 1);
2659
  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
L
Liu Jicong 已提交
2660 2661
  pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage,
                                             pInfo->pSortInputBlock, pTaskInfo->id.str);
2662

dengyihao's avatar
dengyihao 已提交
2663
  tsortSetFetchRawDataFp(pInfo->pSortHandle, getTableDataBlockImpl, NULL, NULL);
dengyihao's avatar
opt mem  
dengyihao 已提交
2664 2665 2666 2667 2668 2669

  // one table has one data block
  int32_t numOfTable = tableEndIdx - tableStartIdx + 1;
  pInfo->queryConds = taosArrayInit(numOfTable, sizeof(SQueryTableDataCond));

  for (int32_t i = 0; i < numOfTable; ++i) {
2670 2671 2672 2673
    STableMergeScanSortSourceParam param = {0};
    param.readerIdx = i;
    param.pOperator = pOperator;
    param.inputBlock = createOneDataBlock(pInfo->pResBlock, false);
H
Haojun Liao 已提交
2674 2675
    blockDataEnsureCapacity(param.inputBlock, pOperator->resultInfo.capacity);

2676
    taosArrayPush(pInfo->sortSourceParams, &param);
dengyihao's avatar
opt mem  
dengyihao 已提交
2677 2678

    SQueryTableDataCond cond;
H
Haojun Liao 已提交
2679
    dumpQueryTableCond(&pInfo->base.cond, &cond);
dengyihao's avatar
opt mem  
dengyihao 已提交
2680
    taosArrayPush(pInfo->queryConds, &cond);
2681 2682
  }

dengyihao's avatar
opt mem  
dengyihao 已提交
2683
  for (int32_t i = 0; i < numOfTable; ++i) {
2684
    SSortSource*                    ps = taosMemoryCalloc(1, sizeof(SSortSource));
2685
    STableMergeScanSortSourceParam* param = taosArrayGet(pInfo->sortSourceParams, i);
2686
    ps->param = param;
2687
    ps->onlyRef = true;
2688 2689 2690 2691 2692 2693
    tsortAddSource(pInfo->pSortHandle, ps);
  }

  int32_t code = tsortOpen(pInfo->pSortHandle);

  if (code != TSDB_CODE_SUCCESS) {
2694
    T_LONG_JMP(pTaskInfo->env, terrno);
2695 2696
  }

2697 2698 2699 2700 2701 2702 2703
  return TSDB_CODE_SUCCESS;
}

int32_t stopGroupTableMergeScan(SOperatorInfo* pOperator) {
  STableMergeScanInfo* pInfo = pOperator->info;
  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

dengyihao's avatar
dengyihao 已提交
2704
  int32_t numOfTable = taosArrayGetSize(pInfo->queryConds);
2705

2706 2707 2708 2709 2710 2711 2712
  SSortExecInfo sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);
  pInfo->sortExecInfo.sortMethod = sortExecInfo.sortMethod;
  pInfo->sortExecInfo.sortBuffer = sortExecInfo.sortBuffer;
  pInfo->sortExecInfo.loops += sortExecInfo.loops;
  pInfo->sortExecInfo.readBytes += sortExecInfo.readBytes;
  pInfo->sortExecInfo.writeBytes += sortExecInfo.writeBytes;

dengyihao's avatar
dengyihao 已提交
2713
  for (int32_t i = 0; i < numOfTable; ++i) {
2714 2715 2716
    STableMergeScanSortSourceParam* param = taosArrayGet(pInfo->sortSourceParams, i);
    blockDataDestroy(param->inputBlock);
  }
2717 2718
  taosArrayClear(pInfo->sortSourceParams);

2719
  tsortDestroySortHandle(pInfo->pSortHandle);
dengyihao's avatar
dengyihao 已提交
2720
  pInfo->pSortHandle = NULL;
2721

dengyihao's avatar
opt mem  
dengyihao 已提交
2722 2723 2724
  for (int32_t i = 0; i < taosArrayGetSize(pInfo->queryConds); i++) {
    SQueryTableDataCond* cond = taosArrayGet(pInfo->queryConds, i);
    taosMemoryFree(cond->colList);
2725
  }
dengyihao's avatar
opt mem  
dengyihao 已提交
2726 2727 2728
  taosArrayDestroy(pInfo->queryConds);
  pInfo->queryConds = NULL;

2729 2730 2731
  return TSDB_CODE_SUCCESS;
}

L
Liu Jicong 已提交
2732 2733
SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock* pResBlock, int32_t capacity,
                                              SOperatorInfo* pOperator) {
2734 2735 2736
  STableMergeScanInfo* pInfo = pOperator->info;
  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

2737
  blockDataCleanup(pResBlock);
2738 2739

  while (1) {
2740
    STupleHandle* pTupleHandle = tsortNextTuple(pHandle);
2741 2742 2743 2744
    if (pTupleHandle == NULL) {
      break;
    }

2745 2746
    appendOneRowToDataBlock(pResBlock, pTupleHandle);
    if (pResBlock->info.rows >= capacity) {
2747 2748 2749 2750
      break;
    }
  }

2751
  qDebug("%s get sorted row blocks, rows:%d", GET_TASKID(pTaskInfo), pResBlock->info.rows);
2752 2753 2754
  applyLimitOffset(&pInfo->limitInfo, pResBlock, pTaskInfo, pOperator);
  pInfo->limitInfo.numOfOutputRows += pResBlock->info.rows;

2755
  return (pResBlock->info.rows > 0) ? pResBlock : NULL;
2756 2757 2758 2759 2760 2761 2762 2763 2764 2765 2766 2767
}

SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;
  STableMergeScanInfo* pInfo = pOperator->info;

  int32_t code = pOperator->fpSet._openFn(pOperator);
  if (code != TSDB_CODE_SUCCESS) {
2768
    T_LONG_JMP(pTaskInfo->env, code);
2769
  }
2770

H
Haojun Liao 已提交
2771
  size_t tableListSize = tableListGetSize(pTaskInfo->pTableInfoList);
S
slzhou 已提交
2772 2773
  if (!pInfo->hasGroupId) {
    pInfo->hasGroupId = true;
2774

S
slzhou 已提交
2775
    if (tableListSize == 0) {
H
Haojun Liao 已提交
2776
      setOperatorCompleted(pOperator);
2777 2778
      return NULL;
    }
S
slzhou 已提交
2779
    pInfo->tableStartIndex = 0;
H
Haojun Liao 已提交
2780
    pInfo->groupId = ((STableKeyInfo*)tableListGetInfo(pTaskInfo->pTableInfoList, pInfo->tableStartIndex))->groupId;
2781 2782
    startGroupTableMergeScan(pOperator);
  }
2783

S
slzhou 已提交
2784 2785
  SSDataBlock* pBlock = NULL;
  while (pInfo->tableStartIndex < tableListSize) {
L
Liu Jicong 已提交
2786 2787
    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pInfo->pResBlock, pOperator->resultInfo.capacity,
                                              pOperator);
S
slzhou 已提交
2788
    if (pBlock != NULL) {
H
Haojun Liao 已提交
2789
      pBlock->info.id.groupId = pInfo->groupId;
S
slzhou 已提交
2790 2791 2792 2793 2794
      pOperator->resultInfo.totalRows += pBlock->info.rows;
      return pBlock;
    } else {
      stopGroupTableMergeScan(pOperator);
      if (pInfo->tableEndIndex >= tableListSize - 1) {
H
Haojun Liao 已提交
2795
        setOperatorCompleted(pOperator);
S
slzhou 已提交
2796 2797 2798
        break;
      }
      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;
H
Haojun Liao 已提交
2799
      pInfo->groupId = tableListGetInfo(pTaskInfo->pTableInfoList, pInfo->tableStartIndex)->groupId;
S
slzhou 已提交
2800 2801
      startGroupTableMergeScan(pOperator);
    }
wmmhello's avatar
wmmhello 已提交
2802 2803
  }

2804 2805 2806
  return pBlock;
}

2807
void destroyTableMergeScanOperatorInfo(void* param) {
2808
  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;
H
Haojun Liao 已提交
2809
  cleanupQueryTableDataCond(&pTableScanInfo->base.cond);
2810

dengyihao's avatar
dengyihao 已提交
2811 2812 2813
  int32_t numOfTable = taosArrayGetSize(pTableScanInfo->queryConds);

  for (int32_t i = 0; i < numOfTable; i++) {
H
Haojun Liao 已提交
2814 2815
    STableMergeScanSortSourceParam* p = taosArrayGet(pTableScanInfo->sortSourceParams, i);
    blockDataDestroy(p->inputBlock);
2816
  }
H
Haojun Liao 已提交
2817

2818
  taosArrayDestroy(pTableScanInfo->sortSourceParams);
dengyihao's avatar
dengyihao 已提交
2819 2820
  tsortDestroySortHandle(pTableScanInfo->pSortHandle);
  pTableScanInfo->pSortHandle = NULL;
2821

H
Haojun Liao 已提交
2822 2823
  tsdbReaderClose(pTableScanInfo->base.dataReader);
  pTableScanInfo->base.dataReader = NULL;
dengyihao's avatar
opt mem  
dengyihao 已提交
2824

dengyihao's avatar
opt mem  
dengyihao 已提交
2825 2826 2827
  for (int i = 0; i < taosArrayGetSize(pTableScanInfo->queryConds); i++) {
    SQueryTableDataCond* pCond = taosArrayGet(pTableScanInfo->queryConds, i);
    taosMemoryFree(pCond->colList);
2828
  }
dengyihao's avatar
opt mem  
dengyihao 已提交
2829
  taosArrayDestroy(pTableScanInfo->queryConds);
2830

H
Haojun Liao 已提交
2831 2832
  if (pTableScanInfo->base.matchInfo.pList != NULL) {
    taosArrayDestroy(pTableScanInfo->base.matchInfo.pList);
2833 2834 2835 2836 2837 2838
  }

  pTableScanInfo->pResBlock = blockDataDestroy(pTableScanInfo->pResBlock);
  pTableScanInfo->pSortInputBlock = blockDataDestroy(pTableScanInfo->pSortInputBlock);

  taosArrayDestroy(pTableScanInfo->pSortInfo);
H
Haojun Liao 已提交
2839
  cleanupExprSupp(&pTableScanInfo->base.pseudoSup);
L
Liu Jicong 已提交
2840

H
Haojun Liao 已提交
2841 2842 2843 2844
  tsdbReaderClose(pTableScanInfo->base.dataReader);
  pTableScanInfo->base.dataReader = NULL;
  taosLRUCacheCleanup(pTableScanInfo->base.metaCache.pTableMetaEntryCache);

D
dapan1121 已提交
2845
  taosMemoryFreeClear(param);
2846 2847 2848 2849
}

int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {
  ASSERT(pOptr != NULL);
2850 2851
  // TODO: merge these two info into one struct
  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));
L
Liu Jicong 已提交
2852
  STableMergeScanInfo*     pInfo = pOptr->info;
H
Haojun Liao 已提交
2853
  execInfo->blockRecorder = pInfo->base.readRecorder;
2854
  execInfo->sortExecInfo = pInfo->sortExecInfo;
2855 2856 2857

  *pOptrExplain = execInfo;
  *len = sizeof(STableMergeScanExecInfo);
L
Liu Jicong 已提交
2858

2859 2860 2861
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
2862 2863
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
                                                SExecTaskInfo* pTaskInfo) {
2864 2865 2866 2867 2868
  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));
  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }
2869

2870 2871 2872
  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;

  int32_t numOfCols = 0;
2873
  int32_t code = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID,
H
Haojun Liao 已提交
2874
                                     &pInfo->base.matchInfo);
H
Haojun Liao 已提交
2875 2876 2877
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2878

H
Haojun Liao 已提交
2879
  code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode);
2880
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2881
    taosArrayDestroy(pInfo->base.matchInfo.pList);
2882 2883 2884 2885
    goto _error;
  }

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {
H
Haojun Liao 已提交
2886
    SExprSupp* pSup = &pInfo->base.pseudoSup;
2887 2888
    pSup->pExprInfo = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->numOfExprs);
    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset);
2889 2890 2891 2892
  }

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

H
Haojun Liao 已提交
2893 2894 2895 2896 2897 2898
  pInfo->base.metaCache.pTableMetaEntryCache = taosLRUCacheInit(1024 * 128, -1, .5);
  if (pInfo->base.metaCache.pTableMetaEntryCache == NULL) {
    code = terrno;
    goto _error;
  }

H
Haojun Liao 已提交
2899 2900
  pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;
  pInfo->base.scanFlag = MAIN_SCAN;
H
Haojun Liao 已提交
2901
  pInfo->base.readHandle = *readHandle;
2902 2903 2904

  pInfo->base.limitInfo.limit.limit = -1;
  pInfo->base.limitInfo.slimit.limit = -1;
H
Haojun Liao 已提交
2905

2906
  pInfo->sample.sampleRatio = pTableScanNode->ratio;
L
Liu Jicong 已提交
2907
  pInfo->sample.seed = taosGetTimestampSec();
H
Haojun Liao 已提交
2908 2909 2910 2911 2912 2913

  code = filterInitFromNode((SNode*)pTableScanNode->scan.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
2914
  initResultSizeInfo(&pOperator->resultInfo, 1024);
H
Haojun Liao 已提交
2915
  pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);
H
Haojun Liao 已提交
2916 2917
  blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);

2918
  pInfo->sortSourceParams = taosArrayInit(64, sizeof(STableMergeScanSortSourceParam));
2919

H
Haojun Liao 已提交
2920
  pInfo->pSortInfo = generateSortByTsInfo(pInfo->base.matchInfo.pList, pInfo->base.cond.order);
2921
  pInfo->pSortInputBlock = createOneDataBlock(pInfo->pResBlock, false);
2922
  initLimitInfo(pTableScanNode->scan.node.pLimit, pTableScanNode->scan.node.pSlimit, &pInfo->limitInfo);
2923

dengyihao's avatar
dengyihao 已提交
2924
  int32_t  rowSize = pInfo->pResBlock->info.rowSize;
A
Alex Duan 已提交
2925 2926
  uint32_t nCols = taosArrayGetSize(pInfo->pResBlock->pDataBlock);
  pInfo->bufPageSize = getProperSortPageSize(rowSize, nCols);
2927

L
Liu Jicong 已提交
2928 2929
  setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,
                  pInfo, pTaskInfo);
L
Liu Jicong 已提交
2930
  pOperator->exprSupp.numOfExprs = numOfCols;
2931

2932 2933
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableMergeScan, NULL, destroyTableMergeScanOperatorInfo,
                                         optrDefaultBufFn, getTableMergeScanExplainExecInfo);
2934 2935 2936 2937 2938 2939 2940 2941 2942
  pOperator->cost.openCost = 0;
  return pOperator;

_error:
  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
  taosMemoryFree(pInfo);
  taosMemoryFree(pOperator);
  return NULL;
}
S
shenglian zhou 已提交
2943 2944 2945 2946

// ====================================================================================================================
// TableCountScanOperator
static SSDataBlock* doTableCountScan(SOperatorInfo* pOperator);
S
slzhou 已提交
2947
static void         destoryTableCountScanOperator(void* param);
S
slzhou 已提交
2948 2949 2950 2951 2952 2953
static void         buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,
                                                   SSDataBlock* pRes, char* dbName, tb_uid_t stbUid);
static void         buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,
                                                   SSDataBlock* pRes, char* dbName);
static void         buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
                                              STableCountScanSupp* pSupp, SSDataBlock* pRes, char* dbName);
L
Liu Jicong 已提交
2954 2955
static void         buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
                                                STableCountScanSupp* pSupp, SSDataBlock* pRes, int32_t vgId, char* dbName);
S
slzhou 已提交
2956 2957 2958 2959 2960 2961 2962
static SSDataBlock* buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
                                           STableCountScanSupp* pSupp, SSDataBlock* pRes);
static void         buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
                                                STableCountScanSupp* pSupp, SSDataBlock* pRes, size_t infodbTableNum,
                                                size_t perfdbTableNum);
static void         buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,
                                               size_t infodbTableNum, size_t perfdbTableNum);
S
slzhou 已提交
2963 2964 2965 2966 2967 2968 2969 2970 2971 2972 2973 2974 2975 2976 2977 2978 2979 2980 2981 2982 2983 2984 2985 2986 2987 2988 2989 2990 2991 2992 2993 2994 2995 2996 2997 2998 2999 3000 3001 3002 3003 3004 3005 3006 3007 3008 3009 3010 3011 3012 3013 3014 3015 3016 3017 3018 3019 3020 3021 3022 3023
static const char*  GROUP_TAG_DB_NAME = "db_name";
static const char*  GROUP_TAG_STABLE_NAME = "stable_name";

int32_t tblCountScanGetGroupTagsSlotId(const SNodeList* scanCols, STableCountScanSupp* supp) {
  if (scanCols != NULL) {
    SNode* pNode = NULL;
    FOREACH(pNode, scanCols) {
      if (nodeType(pNode) != QUERY_NODE_TARGET) {
        return TSDB_CODE_QRY_SYS_ERROR;
      }
      STargetNode* targetNode = (STargetNode*)pNode;
      if (nodeType(targetNode->pExpr) != QUERY_NODE_COLUMN) {
        return TSDB_CODE_QRY_SYS_ERROR;
      }
      SColumnNode* colNode = (SColumnNode*)(targetNode->pExpr);
      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {
        supp->dbNameSlotId = targetNode->slotId;
      } else if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {
        supp->stbNameSlotId = targetNode->slotId;
      }
    }
  }
  return TSDB_CODE_SUCCESS;
}

int32_t tblCountScanGetCountSlotId(const SNodeList* pseudoCols, STableCountScanSupp* supp) {
  if (pseudoCols != NULL) {
    SNode* pNode = NULL;
    FOREACH(pNode, pseudoCols) {
      if (nodeType(pNode) != QUERY_NODE_TARGET) {
        return TSDB_CODE_QRY_SYS_ERROR;
      }
      STargetNode* targetNode = (STargetNode*)pNode;
      if (nodeType(targetNode->pExpr) != QUERY_NODE_FUNCTION) {
        return TSDB_CODE_QRY_SYS_ERROR;
      }
      SFunctionNode* funcNode = (SFunctionNode*)(targetNode->pExpr);
      if (funcNode->funcType == FUNCTION_TYPE_TABLE_COUNT) {
        supp->tbCountSlotId = targetNode->slotId;
      }
    }
  }
  return TSDB_CODE_SUCCESS;
}

int32_t tblCountScanGetInputs(SNodeList* groupTags, SName* tableName, STableCountScanSupp* supp) {
  if (groupTags != NULL) {
    SNode* pNode = NULL;
    FOREACH(pNode, groupTags) {
      if (nodeType(pNode) != QUERY_NODE_COLUMN) {
        return TSDB_CODE_QRY_SYS_ERROR;
      }
      SColumnNode* colNode = (SColumnNode*)pNode;
      if (strcmp(colNode->colName, GROUP_TAG_DB_NAME) == 0) {
        supp->groupByDbName = true;
      }
      if (strcmp(colNode->colName, GROUP_TAG_STABLE_NAME) == 0) {
        supp->groupByStbName = true;
      }
    }
  } else {
S
slzhou 已提交
3024 3025
    strncpy(supp->dbNameFilter, tNameGetDbNameP(tableName), TSDB_DB_NAME_LEN);
    strncpy(supp->stbNameFilter, tNameGetTableName(tableName), TSDB_TABLE_NAME_LEN);
S
slzhou 已提交
3026 3027 3028 3029 3030 3031 3032 3033 3034 3035 3036 3037 3038 3039 3040 3041 3042 3043 3044 3045 3046 3047 3048 3049 3050 3051 3052 3053
  }
  return TSDB_CODE_SUCCESS;
}

int32_t getTableCountScanSupp(SNodeList* groupTags, SName* tableName, SNodeList* scanCols, SNodeList* pseudoCols,
                              STableCountScanSupp* supp, SExecTaskInfo* taskInfo) {
  int32_t code = 0;
  code = tblCountScanGetInputs(groupTags, tableName, supp);
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s get table count scan supp. get inputs error", GET_TASKID(taskInfo));
    return code;
  }
  supp->dbNameSlotId = -1;
  supp->stbNameSlotId = -1;
  supp->tbCountSlotId = -1;

  code = tblCountScanGetGroupTagsSlotId(scanCols, supp);
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s get table count scan supp. get group tags slot id error", GET_TASKID(taskInfo));
    return code;
  }
  code = tblCountScanGetCountSlotId(pseudoCols, supp);
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s get table count scan supp. get count error", GET_TASKID(taskInfo));
    return code;
  }
  return code;
}
S
shenglian zhou 已提交
3054

S
slzhou 已提交
3055
SOperatorInfo* createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pTblCountScanNode,
S
shenglian zhou 已提交
3056 3057 3058
                                                SExecTaskInfo* pTaskInfo) {
  int32_t code = TSDB_CODE_SUCCESS;

S
slzhou 已提交
3059
  SScanPhysiNode*              pScanNode = &pTblCountScanNode->scan;
S
slzhou 已提交
3060
  STableCountScanOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanOperatorInfo));
S
slzhou 已提交
3061
  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
S
shenglian zhou 已提交
3062 3063 3064 3065 3066 3067 3068 3069 3070

  if (!pInfo || !pOperator) {
    goto _error;
  }

  pInfo->readHandle = *readHandle;

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;
  initResultSizeInfo(&pOperator->resultInfo, 1);
3071
  pInfo->pRes = createDataBlockFromDescNode(pDescNode);
S
shenglian zhou 已提交
3072 3073
  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

S
slzhou 已提交
3074 3075 3076
  getTableCountScanSupp(pTblCountScanNode->pGroupTags, &pTblCountScanNode->scan.tableName,
                        pTblCountScanNode->scan.pScanCols, pTblCountScanNode->scan.pScanPseudoCols, &pInfo->supp,
                        pTaskInfo);
S
shenglian zhou 已提交
3077 3078 3079

  setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED,
                  pInfo, pTaskInfo);
L
Liu Jicong 已提交
3080 3081
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScan, NULL, destoryTableCountScanOperator,
                                         optrDefaultBufFn, NULL);
S
shenglian zhou 已提交
3082 3083 3084 3085 3086 3087 3088 3089 3090 3091 3092
  return pOperator;

_error:
  if (pInfo != NULL) {
    destoryTableCountScanOperator(pInfo);
  }
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}

S
slzhou 已提交
3093 3094 3095
void fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, char* stbName, int64_t count,
                                 SSDataBlock* pRes) {
  if (pSupp->dbNameSlotId != -1) {
3096
    ASSERT(strlen(dbName));
S
slzhou 已提交
3097
    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->dbNameSlotId);
H
Haojun Liao 已提交
3098 3099 3100 3101

    char varDbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
    tstrncpy(varDataVal(varDbName), dbName, TSDB_DB_NAME_LEN);

S
slzhou 已提交
3102 3103 3104 3105 3106 3107
    varDataSetLen(varDbName, strlen(dbName));
    colDataAppend(colInfoData, 0, varDbName, false);
  }

  if (pSupp->stbNameSlotId != -1) {
    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->stbNameSlotId);
3108
    if (strlen(stbName) != 0) {
S
slzhou 已提交
3109
      char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
H
Haojun Liao 已提交
3110
      strncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);
3111 3112 3113 3114 3115
      varDataSetLen(varStbName, strlen(stbName));
      colDataAppend(colInfoData, 0, varStbName, false);
    } else {
      colDataAppendNULL(colInfoData, 0);
    }
S
slzhou 已提交
3116 3117 3118
  }

  if (pSupp->tbCountSlotId != -1) {
S
slzhou 已提交
3119
    SColumnInfoData* colInfoData = taosArrayGet(pRes->pDataBlock, pSupp->tbCountSlotId);
S
slzhou 已提交
3120 3121 3122 3123 3124
    colDataAppend(colInfoData, 0, (char*)&count, false);
  }
  pRes->info.rows = 1;
}

S
slzhou 已提交
3125
static SSDataBlock* buildSysDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo) {
S
slzhou 已提交
3126 3127 3128
  STableCountScanSupp* pSupp = &pInfo->supp;
  SSDataBlock*         pRes = pInfo->pRes;

S
slzhou 已提交
3129
  size_t infodbTableNum;
S
slzhou 已提交
3130
  getInfosDbMeta(NULL, &infodbTableNum);
S
slzhou 已提交
3131
  size_t perfdbTableNum;
S
slzhou 已提交
3132 3133 3134
  getPerfDbMeta(NULL, &perfdbTableNum);

  if (pSupp->groupByDbName) {
S
slzhou 已提交
3135
    buildSysDbGroupedTableCount(pOperator, pInfo, pSupp, pRes, infodbTableNum, perfdbTableNum);
S
slzhou 已提交
3136 3137
    return (pRes->info.rows > 0) ? pRes : NULL;
  } else {
S
slzhou 已提交
3138
    buildSysDbFilterTableCount(pOperator, pSupp, pRes, infodbTableNum, perfdbTableNum);
S
slzhou 已提交
3139 3140 3141 3142
    return (pRes->info.rows > 0) ? pRes : NULL;
  }
}

S
slzhou 已提交
3143 3144 3145 3146 3147 3148 3149 3150 3151 3152 3153 3154 3155 3156 3157 3158 3159 3160 3161 3162 3163 3164 3165 3166 3167 3168 3169 3170 3171
static void buildSysDbFilterTableCount(SOperatorInfo* pOperator, STableCountScanSupp* pSupp, SSDataBlock* pRes,
                                       size_t infodbTableNum, size_t perfdbTableNum) {
  if (strcmp(pSupp->dbNameFilter, TSDB_INFORMATION_SCHEMA_DB) == 0) {
    fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);
  } else if (strcmp(pSupp->dbNameFilter, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {
    fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);
  } else if (strlen(pSupp->dbNameFilter) == 0) {
    fillTableCountScanDataBlock(pSupp, "", "", infodbTableNum + perfdbTableNum, pRes);
  }
  setOperatorCompleted(pOperator);
}

static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
                                        STableCountScanSupp* pSupp, SSDataBlock* pRes, size_t infodbTableNum,
                                        size_t perfdbTableNum) {
  if (pInfo->currGrpIdx == 0) {
    uint64_t groupId = calcGroupId(TSDB_INFORMATION_SCHEMA_DB, strlen(TSDB_INFORMATION_SCHEMA_DB));
    pRes->info.id.groupId = groupId;
    fillTableCountScanDataBlock(pSupp, TSDB_INFORMATION_SCHEMA_DB, "", infodbTableNum, pRes);
  } else if (pInfo->currGrpIdx == 1) {
    uint64_t groupId = calcGroupId(TSDB_PERFORMANCE_SCHEMA_DB, strlen(TSDB_PERFORMANCE_SCHEMA_DB));
    pRes->info.id.groupId = groupId;
    fillTableCountScanDataBlock(pSupp, TSDB_PERFORMANCE_SCHEMA_DB, "", perfdbTableNum, pRes);
  } else {
    setOperatorCompleted(pOperator);
  }
  pInfo->currGrpIdx++;
}

S
shenglian zhou 已提交
3172
static SSDataBlock* doTableCountScan(SOperatorInfo* pOperator) {
S
slzhou 已提交
3173 3174 3175 3176
  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;
  STableCountScanOperatorInfo* pInfo = pOperator->info;
  STableCountScanSupp*         pSupp = &pInfo->supp;
  SSDataBlock*                 pRes = pInfo->pRes;
S
slzhou 已提交
3177
  blockDataCleanup(pRes);
3178

S
slzhou 已提交
3179 3180 3181
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }
S
slzhou 已提交
3182
  if (pInfo->readHandle.mnd != NULL) {
S
slzhou 已提交
3183
    return buildSysDbTableCount(pOperator, pInfo);
S
slzhou 已提交
3184
  }
S
slzhou 已提交
3185

S
slzhou 已提交
3186 3187 3188 3189 3190
  return buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);
}

static SSDataBlock* buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
                                           STableCountScanSupp* pSupp, SSDataBlock* pRes) {
S
slzhou 已提交
3191 3192
  const char* db = NULL;
  int32_t     vgId = 0;
S
slzhou 已提交
3193
  char        dbName[TSDB_DB_NAME_LEN] = {0};
S
slzhou 已提交
3194

S
slzhou 已提交
3195 3196 3197 3198 3199 3200
  // get dbname
  vnodeGetInfo(pInfo->readHandle.vnode, &db, &vgId);
  SName sn = {0};
  tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);
  tNameGetDbName(&sn, dbName);

S
slzhou 已提交
3201
  if (pSupp->groupByDbName) {
S
slzhou 已提交
3202 3203 3204 3205 3206 3207 3208 3209 3210 3211 3212 3213 3214 3215
    buildVnodeGroupedTableCount(pOperator, pInfo, pSupp, pRes, vgId, dbName);
  } else {
    buildVnodeFilteredTbCount(pOperator, pInfo, pSupp, pRes, dbName);
  }
  return pRes->info.rows > 0 ? pRes : NULL;
}

static void buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
                                        STableCountScanSupp* pSupp, SSDataBlock* pRes, int32_t vgId, char* dbName) {
  if (pSupp->groupByStbName) {
    if (pInfo->stbUidList == NULL) {
      pInfo->stbUidList = taosArrayInit(16, sizeof(tb_uid_t));
      if (vnodeGetStbIdList(pInfo->readHandle.vnode, 0, pInfo->stbUidList) < 0) {
        qError("vgId:%d, failed to get stb id list error: %s", vgId, terrstr());
S
slzhou 已提交
3216
      }
S
slzhou 已提交
3217 3218 3219 3220 3221 3222 3223 3224 3225 3226
    }
    if (pInfo->currGrpIdx < taosArrayGetSize(pInfo->stbUidList)) {
      tb_uid_t stbUid = *(tb_uid_t*)taosArrayGet(pInfo->stbUidList, pInfo->currGrpIdx);
      buildVnodeGroupedStbTableCount(pInfo, pSupp, pRes, dbName, stbUid);

      pInfo->currGrpIdx++;
    } else if (pInfo->currGrpIdx == taosArrayGetSize(pInfo->stbUidList)) {
      buildVnodeGroupedNtbTableCount(pInfo, pSupp, pRes, dbName);

      pInfo->currGrpIdx++;
S
slzhou 已提交
3227
    } else {
S
slzhou 已提交
3228
      setOperatorCompleted(pOperator);
S
slzhou 已提交
3229 3230
    }
  } else {
S
slzhou 已提交
3231 3232 3233 3234 3235 3236 3237 3238 3239 3240 3241 3242 3243 3244 3245 3246 3247
    uint64_t groupId = calcGroupId(dbName, strlen(dbName));
    pRes->info.id.groupId = groupId;
    int64_t dbTableCount = metaGetTbNum(pInfo->readHandle.meta);
    fillTableCountScanDataBlock(pSupp, dbName, "", dbTableCount, pRes);
    setOperatorCompleted(pOperator);
  }
}

static void buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,
                                      STableCountScanSupp* pSupp, SSDataBlock* pRes, char* dbName) {
  if (strlen(pSupp->dbNameFilter) != 0) {
    if (strlen(pSupp->stbNameFilter) != 0) {
      tb_uid_t      uid = metaGetTableEntryUidByName(pInfo->readHandle.meta, pSupp->stbNameFilter);
      SMetaStbStats stats = {0};
      metaGetStbStats(pInfo->readHandle.meta, uid, &stats);
      int64_t ctbNum = stats.ctbNum;
      fillTableCountScanDataBlock(pSupp, dbName, pSupp->stbNameFilter, ctbNum, pRes);
S
slzhou 已提交
3248 3249 3250
    } else {
      int64_t tbNumVnode = metaGetTbNum(pInfo->readHandle.meta);
      fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);
S
slzhou 已提交
3251
    }
S
slzhou 已提交
3252 3253 3254
  } else {
    int64_t tbNumVnode = metaGetTbNum(pInfo->readHandle.meta);
    fillTableCountScanDataBlock(pSupp, dbName, "", tbNumVnode, pRes);
S
slzhou 已提交
3255
  }
S
slzhou 已提交
3256 3257 3258 3259 3260 3261 3262 3263 3264 3265 3266 3267 3268 3269 3270 3271 3272 3273 3274 3275 3276 3277 3278 3279 3280 3281 3282 3283
  setOperatorCompleted(pOperator);
}

static void buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,
                                           SSDataBlock* pRes, char* dbName) {
  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};
  snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, "");
  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));
  pRes->info.id.groupId = groupId;
  int64_t ntbNum = metaGetNtbNum(pInfo->readHandle.meta);
  fillTableCountScanDataBlock(pSupp, dbName, "", ntbNum, pRes);
}

static void buildVnodeGroupedStbTableCount(STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp,
                                           SSDataBlock* pRes, char* dbName, tb_uid_t stbUid) {
  char stbName[TSDB_TABLE_NAME_LEN] = {0};
  metaGetTableSzNameByUid(pInfo->readHandle.meta, stbUid, stbName);

  char fullStbName[TSDB_TABLE_FNAME_LEN] = {0};
  snprintf(fullStbName, TSDB_TABLE_FNAME_LEN, "%s.%s", dbName, stbName);
  uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));
  pRes->info.id.groupId = groupId;

  SMetaStbStats stats = {0};
  metaGetStbStats(pInfo->readHandle.meta, stbUid, &stats);
  int64_t ctbNum = stats.ctbNum;

  fillTableCountScanDataBlock(pSupp, dbName, stbName, ctbNum, pRes);
S
shenglian zhou 已提交
3284 3285 3286
}

static void destoryTableCountScanOperator(void* param) {
S
slzhou 已提交
3287
  STableCountScanOperatorInfo* pTableCountScanInfo = param;
S
shenglian zhou 已提交
3288 3289
  blockDataDestroy(pTableCountScanInfo->pRes);

S
slzhou 已提交
3290
  taosArrayDestroy(pTableCountScanInfo->stbUidList);
S
shenglian zhou 已提交
3291 3292
  taosMemoryFreeClear(param);
}