scanoperator.c 110.5 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"
17
#include "function.h"
18
#include "functionMgt.h"
H
Haojun Liao 已提交
19 20
#include "os.h"
#include "querynodes.h"
21
#include "systable.h"
H
Haojun Liao 已提交
22
#include "tname.h"
23
#include "ttime.h"
H
Haojun Liao 已提交
24 25 26 27 28 29 30 31 32

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

#include "executorimpl.h"
#include "query.h"
#include "tcompare.h"
#include "thash.h"
#include "ttypes.h"
33
#include "vnode.h"
H
Haojun Liao 已提交
34 35

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

38
static int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity);
L
Liu Jicong 已提交
39 40
static int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size,
                                     const char* dbName);
41

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

44
bool processBlockWithProbability(const SSampleExecInfo* pInfo) {
45 46 47 48 49 50 51 52 53 54 55 56
#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
}

57
static void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
H
Haojun Liao 已提交
58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85
  for (int32_t i = 0; i < numOfOutput; ++i) {
    SWITCH_ORDER(pCtx[i].order);
  }
}

static void setupQueryRangeForReverseScan(STableScanInfo* pTableScanInfo) {
#if 0
  int32_t numOfGroups = (int32_t)(GET_NUM_OF_TABLEGROUP(pRuntimeEnv));
  for(int32_t i = 0; i < numOfGroups; ++i) {
    SArray *group = GET_TABLEGROUP(pRuntimeEnv, i);
    SArray *tableKeyGroup = taosArrayGetP(pQueryAttr->tableGroupInfo.pGroupList, i);

    size_t t = taosArrayGetSize(group);
    for (int32_t j = 0; j < t; ++j) {
      STableQueryInfo *pCheckInfo = taosArrayGetP(group, j);
      updateTableQueryInfoForReverseScan(pCheckInfo);

      // update the last key in tableKeyInfo list, the tableKeyInfo is used to build the tsdbQueryHandle and decide
      // the start check timestamp of tsdbQueryHandle
//      STableKeyInfo *pTableKeyInfo = taosArrayGet(tableKeyGroup, j);
//      pTableKeyInfo->lastKey = pCheckInfo->lastKey;
//
//      assert(pCheckInfo->pTable == pTableKeyInfo->pTable);
    }
  }
#endif
}

86 87 88 89 90 91 92 93 94
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;
95
  // convert key to second
96 97 98 99 100 101 102
  key = convertTimePrecision(key, pInterval->precision, TSDB_TIME_PRECISION_MILLI) / 1000;

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

  struct tm tm;
103
  time_t    t = (time_t)key;
104 105 106 107 108
  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 已提交
109
  tw->skey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, pInterval->precision);
110 111 112 113

  mon = (int)(mon + interval);
  tm.tm_year = mon / 12;
  tm.tm_mon = mon % 12;
wafwerar's avatar
wafwerar 已提交
114
  tw->ekey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, pInterval->precision);
115 116 117 118

  tw->ekey -= 1;
}

119
static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockInfo, int32_t order) {
120 121 122 123 124 125 126
  STimeWindow w = {0};

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

127
  if (order == TSDB_ORDER_ASC) {
128
    w = getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.skey);
129 130 131 132 133 134
    assert(w.ekey >= pBlockInfo->window.skey);

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

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

      assert(w.ekey > pBlockInfo->window.ekey);
      if (w.skey <= pBlockInfo->window.ekey && w.skey > pBlockInfo->window.skey) {
        return true;
      }
    }
  } else {
147
    w = getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.ekey);
148 149 150 151 152 153
    assert(w.skey <= pBlockInfo->window.ekey);

    if (w.skey > pBlockInfo->window.skey) {
      return true;
    }

154
    while (1) {
155 156 157 158 159 160 161 162 163 164
      getNextTimeWindow(pInterval, &w, order);
      if (w.ekey < pBlockInfo->window.skey) {
        break;
      }

      assert(w.skey < pBlockInfo->window.skey);
      if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) {
        return true;
      }
    }
165 166 167 168 169
  }

  return false;
}

L
Liu Jicong 已提交
170 171
static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock,
                             uint32_t* status) {
172
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
173 174
  STableScanInfo* pInfo = pOperator->info;

175
  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;
H
Haojun Liao 已提交
176 177

  pCost->totalBlocks += 1;
178
  pCost->totalRows += pBlock->info.rows;
H
Haojun Liao 已提交
179

180
  *status = pInfo->dataBlockLoadFlag;
181 182
  if (pTableScanInfo->pFilterNode != NULL ||
      overlapWithTimeWindow(&pTableScanInfo->interval, &pBlock->info, pTableScanInfo->cond.order)) {
183 184 185 186
    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;
  }

  SDataBlockInfo* pBlockInfo = &pBlock->info;
187
  taosMemoryFreeClear(pBlock->pBlockAgg);
188 189

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {
190 191
    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
192 193 194
    pCost->filterOutBlocks += 1;
    return TSDB_CODE_SUCCESS;
  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {
195 196
    qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
197
    pCost->skipBlocks += 1;
198

199 200 201 202
    return TSDB_CODE_SUCCESS;
  } else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) {
    pCost->loadBlockStatis += 1;

L
Liu Jicong 已提交
203
    bool             allColumnsHaveAgg = true;
204
    SColumnDataAgg** pColAgg = NULL;
205

206
    int32_t code = tsdbRetrieveDatablockSMA(pTableScanInfo->dataReader, &pColAgg, &allColumnsHaveAgg);
207 208 209
    if (code != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, code);
    }
210

211
    if (allColumnsHaveAgg == true) {
212
      int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
213 214

      // todo create this buffer during creating operator
215 216 217 218
      if (pBlock->pBlockAgg == NULL) {
        pBlock->pBlockAgg = taosMemoryCalloc(numOfCols, POINTER_BYTES);
      }

219
      for (int32_t i = 0; i < taosArrayGetSize(pTableScanInfo->pColMatchInfo); ++i) {
220 221 222 223 224 225
        SColMatchInfo* pColMatchInfo = taosArrayGet(pTableScanInfo->pColMatchInfo, i);
        if (!pColMatchInfo->output) {
          continue;
        }
        pBlock->pBlockAgg[pColMatchInfo->targetSlotId] = pColAgg[i];
      }
H
Haojun Liao 已提交
226

227
      return TSDB_CODE_SUCCESS;
228
    } else {  // failed to load the block sma data, data block statistics does not exist, load data block instead
H
Haojun Liao 已提交
229
      *status = FUNC_DATA_REQUIRED_DATA_LOAD;
230
    }
H
Haojun Liao 已提交
231
  }
232

233
  ASSERT(*status == FUNC_DATA_REQUIRED_DATA_LOAD);
234

H
Haojun Liao 已提交
235 236 237 238 239 240 241 242 243 244
  // todo filter data block according to the block sma data firstly
#if 0
  if (!doFilterByBlockStatistics(pBlock->pBlockStatis, pTableScanInfo->pCtx, pBlockInfo->rows)) {
    pCost->filterOutBlocks += 1;
    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo), pBlockInfo->window.skey,
           pBlockInfo->window.ekey, pBlockInfo->rows);
    (*status) = FUNC_DATA_REQUIRED_FILTEROUT;
    return TSDB_CODE_SUCCESS;
  }
#endif
H
Haojun Liao 已提交
245

H
Haojun Liao 已提交
246 247
  pCost->totalCheckedRows += pBlock->info.rows;
  pCost->loadBlocks += 1;
248

H
Haojun Liao 已提交
249 250 251
  SArray* pCols = tsdbRetrieveDataBlock(pTableScanInfo->dataReader, NULL);
  if (pCols == NULL) {
    return terrno;
H
Haojun Liao 已提交
252 253
  }

254
  relocateColumnData(pBlock, pTableScanInfo->pColMatchInfo, pCols, true);
255 256

  // currently only the tbname pseudo column
257
  if (pTableScanInfo->pseudoSup.numOfExprs > 0) {
258
    SExprSupp* pSup = &pTableScanInfo->pseudoSup;
H
Haojun Liao 已提交
259

L
Liu Jicong 已提交
260 261
    int32_t code = addTagPseudoColumnData(&pTableScanInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pBlock,
                                          GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
262 263 264
    if (code != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, code);
    }
265 266
  }

267
  int64_t st = taosGetTimestampMs();
268
  doFilter(pTableScanInfo->pFilterNode, pBlock);
269

270 271 272
  int64_t et = taosGetTimestampMs();
  pTableScanInfo->readRecorder.filterTime += (et - st);

273 274
  if (pBlock->info.rows == 0) {
    pCost->filterOutBlocks += 1;
275 276
    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
277 278
  }

H
Haojun Liao 已提交
279 280 281
  return TSDB_CODE_SUCCESS;
}

282
static void prepareForDescendingScan(STableScanInfo* pTableScanInfo, SqlFunctionCtx* pCtx, int32_t numOfOutput) {
H
Haojun Liao 已提交
283 284 285
  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

  switchCtxOrder(pCtx, numOfOutput);
286
  //  setupQueryRangeForReverseScan(pTableScanInfo);
H
Haojun Liao 已提交
287

288
  pTableScanInfo->cond.order = TSDB_ORDER_DESC;
H
Haojun Liao 已提交
289 290
  STimeWindow* pTWindow = &pTableScanInfo->cond.twindows;
  TSWAP(pTWindow->skey, pTWindow->ekey);
H
Haojun Liao 已提交
291 292
}

H
Haojun Liao 已提交
293
int32_t addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr,
L
Liu Jicong 已提交
294
                               SSDataBlock* pBlock, const char* idStr) {
295
  // currently only the tbname pseudo column
296
  if (numOfPseudoExpr == 0) {
H
Haojun Liao 已提交
297
    return TSDB_CODE_SUCCESS;
298 299 300
  }

  SMetaReader mr = {0};
301
  metaReaderInit(&mr, pHandle->meta, 0);
H
Haojun Liao 已提交
302 303
  int32_t code = metaGetTableEntryByUid(&mr, pBlock->info.uid);
  if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
304
    qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.uid, tstrerror(terrno), idStr);
H
Haojun Liao 已提交
305 306 307
    metaReaderClear(&mr);
    return terrno;
  }
308

309 310
  for (int32_t j = 0; j < numOfPseudoExpr; ++j) {
    SExprInfo* pExpr = &pPseudoExpr[j];
311 312 313 314

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

    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotId);
D
dapan1121 已提交
315
    colInfoDataCleanup(pColInfoData, pBlock->info.rows);
316 317 318 319 320

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

    // this is to handle the tbname
    if (fmIsScanPseudoColumnFunc(functionId)) {
321
      setTbNameColData(pHandle->meta, pBlock, pColInfoData, functionId);
322
    } else {  // these are tags
wmmhello's avatar
wmmhello 已提交
323 324
      STagVal tagVal = {0};
      tagVal.cid = pExpr->base.pParam[0].pCol->colId;
325
      const char* p = metaGetTableTagVal(&mr.me, pColInfoData->info.type, &tagVal);
wmmhello's avatar
wmmhello 已提交
326

327 328 329 330
      char* data = NULL;
      if (pColInfoData->info.type != TSDB_DATA_TYPE_JSON && p != NULL) {
        data = tTagValToData((const STagVal*)p, false);
      } else {
wmmhello's avatar
wmmhello 已提交
331
        data = (char*)p;
wmmhello's avatar
wmmhello 已提交
332
      }
333

334
      for (int32_t i = 0; i < pBlock->info.rows; ++i) {
L
Liu Jicong 已提交
335 336
        colDataAppend(pColInfoData, i, data,
                      (data == NULL) || (pColInfoData->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data)));
337
      }
338

339 340
      if (data && (pColInfoData->info.type != TSDB_DATA_TYPE_JSON) && p != NULL &&
          IS_VAR_DATA_TYPE(((const STagVal*)p)->type)) {
wmmhello's avatar
wmmhello 已提交
341
        taosMemoryFree(data);
wmmhello's avatar
wmmhello 已提交
342
      }
343 344 345 346
    }
  }

  metaReaderClear(&mr);
H
Haojun Liao 已提交
347
  return TSDB_CODE_SUCCESS;
348 349
}

350 351 352 353
void setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId) {
  struct SScalarFuncExecFuncs fpSet = {0};
  fmGetScalarFuncExecFuncs(functionId, &fpSet);

354 355
  SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_BIGINT, sizeof(uint64_t), 1);
  colInfoDataEnsureCapacity(&infoData, 1);
356

357
  colDataAppendInt64(&infoData, 0, (int64_t*)&pBlock->info.uid);
358
  SScalarParam srcParam = {.numOfRows = pBlock->info.rows, .param = pMeta, .columnData = &infoData};
359 360 361

  SScalarParam param = {.columnData = pColInfoData};
  fpSet.process(&srcParam, 1, &param);
D
dapan1121 已提交
362
  colDataDestroy(&infoData);
363 364
}

365
static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
366
  STableScanInfo* pTableScanInfo = pOperator->info;
367
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
L
Liu Jicong 已提交
368
  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;
H
Haojun Liao 已提交
369

370 371
  int64_t st = taosGetTimestampUs();

372
  while (tsdbNextDataBlock(pTableScanInfo->dataReader)) {
373 374
    if (isTaskKilled(pTaskInfo)) {
      longjmp(pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
375
    }
H
Haojun Liao 已提交
376

377 378 379 380 381 382
    // process this data block based on the probabilities
    bool processThisBlock = processBlockWithProbability(&pTableScanInfo->sample);
    if (!processThisBlock) {
      continue;
    }

383 384 385 386 387 388 389 390
    blockDataCleanup(pBlock);

    SDataBlockInfo binfo = pBlock->info;
    tsdbRetrieveDataBlockInfo(pTableScanInfo->dataReader, &binfo);

    binfo.capacity = binfo.rows;
    blockDataEnsureCapacity(pBlock, binfo.rows);
    pBlock->info = binfo;
L
Liu Jicong 已提交
391
    ASSERT(binfo.uid != 0);
H
Haojun Liao 已提交
392

393 394 395 396 397 398
    uint32_t status = 0;
    int32_t  code = loadDataBlock(pOperator, pTableScanInfo, pBlock, &status);
    //    int32_t  code = loadDataBlockOnDemand(pOperator->pRuntimeEnv, pTableScanInfo, pBlock, &status);
    if (code != TSDB_CODE_SUCCESS) {
      longjmp(pOperator->pTaskInfo->env, code);
    }
399

400 401 402
    // 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;
403
    }
404

405
    uint64_t* groupId = taosHashGet(pTaskInfo->tableqinfoList.map, &pBlock->info.uid, sizeof(int64_t));
wmmhello's avatar
wmmhello 已提交
406 407 408 409
    if (groupId) {
      pBlock->info.groupId = *groupId;
    }

410 411 412 413
    pOperator->resultInfo.totalRows = pTableScanInfo->readRecorder.totalRows;
    pTableScanInfo->readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

    pOperator->cost.totalCost = pTableScanInfo->readRecorder.elapsedTime;
414 415

    // todo refactor
L
Liu Jicong 已提交
416 417 418 419 420
    /*pTableScanInfo->lastStatus.uid = pBlock->info.uid;*/
    /*pTableScanInfo->lastStatus.ts = pBlock->info.window.ekey;*/
    pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_DATA;
    pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.uid;
    pTaskInfo->streamInfo.lastStatus.ts = pBlock->info.window.ekey;
421

L
Liu Jicong 已提交
422
    ASSERT(pBlock->info.uid != 0);
423
    return pBlock;
H
Haojun Liao 已提交
424 425 426 427
  }
  return NULL;
}

wmmhello's avatar
wmmhello 已提交
428
static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
429 430 431 432
  STableScanInfo* pTableScanInfo = pOperator->info;
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

  // The read handle is not initialized yet, since no qualified tables exists
433
  if (pTableScanInfo->dataReader == NULL || pOperator->status == OP_EXEC_DONE) {
H
Haojun Liao 已提交
434 435 436
    return NULL;
  }

437 438
  // do the ascending order traverse in the first place.
  while (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
H
Haojun Liao 已提交
439 440 441 442
    SSDataBlock* p = doTableScanImpl(pOperator);
    if (p != NULL) {
      ASSERT(p->info.uid != 0);
      return p;
H
Haojun Liao 已提交
443 444
    }

445
    pTableScanInfo->scanTimes += 1;
446

447
    if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
448 449
      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
      pTableScanInfo->scanFlag = REPEAT_SCAN;
450
      qDebug("%s start to repeat ascending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
451

452
      // do prepare for the next round table scan operation
H
Haojun Liao 已提交
453
      tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
H
Haojun Liao 已提交
454
    }
455
  }
H
Haojun Liao 已提交
456

457
  int32_t total = pTableScanInfo->scanInfo.numOfAsc + pTableScanInfo->scanInfo.numOfDesc;
458
  if (pTableScanInfo->scanTimes < total) {
459
    if (pTableScanInfo->cond.order == TSDB_ORDER_ASC) {
H
Haojun Liao 已提交
460 461
      prepareForDescendingScan(pTableScanInfo, pOperator->exprSupp.pCtx, 0);
      tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
462
    }
H
Haojun Liao 已提交
463

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

466
    while (pTableScanInfo->scanTimes < total) {
H
Haojun Liao 已提交
467 468 469
      SSDataBlock* p = doTableScanImpl(pOperator);
      if (p != NULL) {
        return p;
470
      }
H
Haojun Liao 已提交
471

472
      pTableScanInfo->scanTimes += 1;
H
Haojun Liao 已提交
473

474
      if (pTableScanInfo->scanTimes < total) {
475 476
        setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
        pTableScanInfo->scanFlag = REPEAT_SCAN;
H
Haojun Liao 已提交
477

478 479
        qDebug("%s start to repeat descending order scan data blocks due to query func required",
               GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
480
        tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
481
      }
H
Haojun Liao 已提交
482 483 484
    }
  }

wmmhello's avatar
wmmhello 已提交
485 486 487 488 489 490 491
  return NULL;
}

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

L
Liu Jicong 已提交
492 493
  // if scan table by table
  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {
L
Liu Jicong 已提交
494
    if (pInfo->noTable) return NULL;
L
Liu Jicong 已提交
495
    while (1) {
L
Liu Jicong 已提交
496 497 498 499 500 501 502 503 504 505 506
      SSDataBlock* result = doTableScanGroup(pOperator);
      if (result) {
        return result;
      }
      // if no data, switch to next table and continue scan
      pInfo->currentTable++;
      if (pInfo->currentTable >= taosArrayGetSize(pTaskInfo->tableqinfoList.pTableList)) {
        return NULL;
      }
      STableKeyInfo* pTableInfo = taosArrayGet(pTaskInfo->tableqinfoList.pTableList, pInfo->currentTable);
      tsdbSetTableId(pInfo->dataReader, pTableInfo->uid);
H
Haojun Liao 已提交
507
      tsdbReaderReset(pInfo->dataReader, &pInfo->cond);
L
Liu Jicong 已提交
508 509 510 511
      pInfo->scanTimes = 0;
    }
  }

512
  if (pInfo->currentGroupId == -1) {
wmmhello's avatar
wmmhello 已提交
513
    pInfo->currentGroupId++;
wmmhello's avatar
wmmhello 已提交
514
    if (pInfo->currentGroupId >= taosArrayGetSize(pTaskInfo->tableqinfoList.pGroupList)) {
wmmhello's avatar
wmmhello 已提交
515
      setTaskStatus(pTaskInfo, TASK_COMPLETED);
wmmhello's avatar
wmmhello 已提交
516 517
      return NULL;
    }
H
Haojun Liao 已提交
518

519
    SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, pInfo->currentGroupId);
H
Haojun Liao 已提交
520 521 522

    tsdbReaderClose(pInfo->dataReader);

523 524
    int32_t code = tsdbReaderOpen(pInfo->readHandle.vnode, &pInfo->cond, tableList, (STsdbReader**)&pInfo->dataReader,
                                  GET_TASKID(pTaskInfo));
525 526 527 528
    if (code != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, code);
      return NULL;
    }
wmmhello's avatar
wmmhello 已提交
529 530 531
  }

  SSDataBlock* result = doTableScanGroup(pOperator);
532
  if (result) {
wmmhello's avatar
wmmhello 已提交
533 534 535 536 537
    return result;
  }

  pInfo->currentGroupId++;
  if (pInfo->currentGroupId >= taosArrayGetSize(pTaskInfo->tableqinfoList.pGroupList)) {
wmmhello's avatar
wmmhello 已提交
538
    setTaskStatus(pTaskInfo, TASK_COMPLETED);
wmmhello's avatar
wmmhello 已提交
539 540 541
    return NULL;
  }

542
  SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, pInfo->currentGroupId);
543
  //  tsdbSetTableList(pInfo->dataReader, tableList);
wmmhello's avatar
wmmhello 已提交
544

H
Haojun Liao 已提交
545
  tsdbReaderReset(pInfo->dataReader, &pInfo->cond);
wmmhello's avatar
wmmhello 已提交
546 547 548
  pInfo->scanTimes = 0;

  result = doTableScanGroup(pOperator);
549
  if (result) {
wmmhello's avatar
wmmhello 已提交
550 551 552
    return result;
  }

553 554
  setTaskStatus(pTaskInfo, TASK_COMPLETED);
  return NULL;
H
Haojun Liao 已提交
555 556
}

557 558
static int32_t getTableScannerExecInfo(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {
  SFileBlockLoadRecorder* pRecorder = taosMemoryCalloc(1, sizeof(SFileBlockLoadRecorder));
559
  STableScanInfo*         pTableScanInfo = pOptr->info;
560 561 562 563 564 565
  *pRecorder = pTableScanInfo->readRecorder;
  *pOptrExplain = pRecorder;
  *len = sizeof(SFileBlockLoadRecorder);
  return 0;
}

566 567
static void destroyTableScanOperatorInfo(void* param, int32_t numOfOutput) {
  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;
H
Haojun Liao 已提交
568
  blockDataDestroy(pTableScanInfo->pResBlock);
569
  cleanupQueryTableDataCond(&pTableScanInfo->cond);
H
Haojun Liao 已提交
570

H
refact  
Hongze Cheng 已提交
571
  tsdbReaderClose(pTableScanInfo->dataReader);
572 573 574 575

  if (pTableScanInfo->pColMatchInfo != NULL) {
    taosArrayDestroy(pTableScanInfo->pColMatchInfo);
  }
L
Liu Jicong 已提交
576

577
  cleanupExprSupp(&pTableScanInfo->pseudoSup);
D
dapan1121 已提交
578
  taosMemoryFreeClear(param);
579 580
}

wmmhello's avatar
wmmhello 已提交
581
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
582
                                           SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
583 584 585
  STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
586
    goto _error;
H
Haojun Liao 已提交
587 588
  }

589
  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;
590
  int32_t             numOfCols = 0;
591
  SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
L
Liu Jicong 已提交
592

593 594
  int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode);
  if (code != TSDB_CODE_SUCCESS) {
595
    goto _error;
596 597 598
  }

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {
599
    SExprSupp* pSup = &pInfo->pseudoSup;
600 601
    pSup->pExprInfo = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->numOfExprs);
    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset);
602 603
  }

604
  pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};
605
  //    pInfo->scanInfo = (SScanInfo){.numOfAsc = 0, .numOfDesc = 1}; // for debug purpose
606

607 608 609 610
  pInfo->readHandle = *readHandle;
  pInfo->interval = extractIntervalInfo(pTableScanNode);
  pInfo->sample.sampleRatio = pTableScanNode->ratio;
  pInfo->sample.seed = taosGetTimestampSec();
611

612
  pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired;
613 614 615 616
  pInfo->pResBlock = createResDataBlock(pDescNode);
  pInfo->pFilterNode = pTableScanNode->scan.node.pConditions;
  pInfo->scanFlag = MAIN_SCAN;
  pInfo->pColMatchInfo = pColList;
wmmhello's avatar
wmmhello 已提交
617
  pInfo->currentGroupId = -1;
618 619

  pOperator->name = "TableScanOperator";  // for debug purpose
L
Liu Jicong 已提交
620
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
621 622 623
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
624
  pOperator->exprSupp.numOfExprs = numOfCols;
625
  pOperator->pTaskInfo = pTaskInfo;
626

627 628
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, destroyTableScanOperatorInfo,
                                         NULL, NULL, getTableScannerExecInfo);
629 630 631

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

634
_error:
635 636 637 638 639
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);

  pTaskInfo->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
  return NULL;
H
Haojun Liao 已提交
640 641
}

642
SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
643
  STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
L
Liu Jicong 已提交
644
  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
H
Haojun Liao 已提交
645

L
Liu Jicong 已提交
646 647
  pInfo->dataReader = pReadHandle;
  //  pInfo->prevGroupId       = -1;
H
Haojun Liao 已提交
648

649
  pOperator->name = "TableSeqScanOperator";
H
Haojun Liao 已提交
650
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN;
651 652 653 654
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
  pOperator->pTaskInfo = pTaskInfo;
H
Haojun Liao 已提交
655

656
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScanImpl, NULL, NULL, NULL, NULL, NULL, NULL);
H
Haojun Liao 已提交
657 658 659
  return pOperator;
}

H
Haojun Liao 已提交
660 661
static int32_t doGetTableRowSize(void* pMeta, uint64_t uid, int32_t* rowLen, const char* idstr) {
  *rowLen = 0;
H
Haojun Liao 已提交
662

663
  SMetaReader mr = {0};
664
  metaReaderInit(&mr, pMeta, 0);
H
Haojun Liao 已提交
665 666
  int32_t code = metaGetTableEntryByUid(&mr, uid);
  if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
667
    qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", uid, tstrerror(terrno), idstr);
H
Haojun Liao 已提交
668 669 670 671
    metaReaderClear(&mr);
    return terrno;
  }

672 673
  if (mr.me.type == TSDB_SUPER_TABLE) {
    int32_t numOfCols = mr.me.stbEntry.schemaRow.nCols;
674
    for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
675
      (*rowLen) += mr.me.stbEntry.schemaRow.pSchema[i].bytes;
676 677 678
    }
  } else if (mr.me.type == TSDB_CHILD_TABLE) {
    uint64_t suid = mr.me.ctbEntry.suid;
H
Haojun Liao 已提交
679 680
    code = metaGetTableEntryByUid(&mr, suid);
    if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
681
      qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", suid, tstrerror(terrno), idstr);
H
Haojun Liao 已提交
682 683 684 685
      metaReaderClear(&mr);
      return terrno;
    }

686 687
    int32_t numOfCols = mr.me.stbEntry.schemaRow.nCols;

688
    for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
689
      (*rowLen) += mr.me.stbEntry.schemaRow.pSchema[i].bytes;
690 691 692
    }
  } else if (mr.me.type == TSDB_NORMAL_TABLE) {
    int32_t numOfCols = mr.me.ntbEntry.schemaRow.nCols;
693
    for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
694
      (*rowLen) += mr.me.ntbEntry.schemaRow.pSchema[i].bytes;
695 696 697 698
    }
  }

  metaReaderClear(&mr);
H
Haojun Liao 已提交
699
  return TSDB_CODE_SUCCESS;
700 701 702 703 704 705 706 707
}

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

  SBlockDistInfo* pBlockScanInfo = pOperator->info;
L
Liu Jicong 已提交
708
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
709 710

  STableBlockDistInfo blockDistInfo = {.minRows = INT_MAX, .maxRows = INT_MIN};
L
Liu Jicong 已提交
711 712
  int32_t code = doGetTableRowSize(pBlockScanInfo->readHandle.meta, pBlockScanInfo->uid, &blockDistInfo.rowSize,
                                   GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
713
  if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
714
    longjmp(pTaskInfo->env, code);
H
Haojun Liao 已提交
715
  }
716 717 718

  tsdbGetFileBlocksDistInfo(pBlockScanInfo->pHandle, &blockDistInfo);
  blockDistInfo.numOfInmemRows = (int32_t)tsdbGetNumOfRowsInMemTable(pBlockScanInfo->pHandle);
H
Haojun Liao 已提交
719

720
  SSDataBlock* pBlock = pBlockScanInfo->pResBlock;
H
Haojun Liao 已提交
721

722
  int32_t          slotId = pOperator->exprSupp.pExprInfo->base.resSchema.slotId;
723
  SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, slotId);
H
Haojun Liao 已提交
724

725
  int32_t len = tSerializeBlockDistInfo(NULL, 0, &blockDistInfo);
726
  char*   p = taosMemoryCalloc(1, len + VARSTR_HEADER_SIZE);
727 728 729
  tSerializeBlockDistInfo(varDataVal(p), len, &blockDistInfo);
  varDataSetLen(p, len);

730
  blockDataEnsureCapacity(pBlock, 1);
731 732
  colDataAppend(pColInfo, 0, p, false);
  taosMemoryFree(p);
H
Haojun Liao 已提交
733

734 735
  pBlock->info.rows = 1;

H
Haojun Liao 已提交
736 737 738 739
  pOperator->status = OP_EXEC_DONE;
  return pBlock;
}

740
static void destroyBlockDistScanOperatorInfo(void* param, int32_t numOfOutput) {
741
  SBlockDistInfo* pDistInfo = (SBlockDistInfo*)param;
742
  blockDataDestroy(pDistInfo->pResBlock);
L
Liu Jicong 已提交
743

D
dapan1121 已提交
744
  taosMemoryFreeClear(param);
745 746
}

747 748
SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SReadHandle* readHandle, uint64_t uid,
                                               SBlockDistScanPhysiNode* pBlockScanNode, SExecTaskInfo* pTaskInfo) {
749
  SBlockDistInfo* pInfo = taosMemoryCalloc(1, sizeof(SBlockDistInfo));
750
  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
H
Haojun Liao 已提交
751 752 753 754 755
  if (pInfo == NULL || pOperator == NULL) {
    pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
    goto _error;
  }

756
  pInfo->pHandle = dataReader;
757
  pInfo->readHandle = *readHandle;
758 759
  pInfo->uid = uid;
  pInfo->pResBlock = createResDataBlock(pBlockScanNode->node.pOutputDataBlockDesc);
760

761
  int32_t    numOfCols = 0;
762
  SExprInfo* pExprInfo = createExprInfo(pBlockScanNode->pScanPseudoCols, NULL, &numOfCols);
763
  int32_t    code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfCols);
764 765 766
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
H
Haojun Liao 已提交
767

768
  pOperator->name = "DataBlockDistScanOperator";
769
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN;
770 771 772
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
773 774 775 776
  pOperator->pTaskInfo = pTaskInfo;

  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doBlockInfoScan, NULL, NULL,
                                         destroyBlockDistScanOperatorInfo, NULL, NULL, NULL);
H
Haojun Liao 已提交
777 778
  return pOperator;

779
_error:
H
Haojun Liao 已提交
780 781 782 783 784
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  return NULL;
}

785
static void doClearBufferedBlocks(SStreamScanInfo* pInfo) {
H
Haojun Liao 已提交
786 787 788 789 790 791 792 793 794 795
  size_t total = taosArrayGetSize(pInfo->pBlockLists);

  pInfo->validBlockIndex = 0;
  for (int32_t i = 0; i < total; ++i) {
    SSDataBlock* p = taosArrayGetP(pInfo->pBlockLists, i);
    blockDataDestroy(p);
  }
  taosArrayClear(pInfo->pBlockLists);
}

796
static bool isSessionWindow(SStreamScanInfo* pInfo) {
L
Liu Jicong 已提交
797 798
  return pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||
         pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
5
54liuyao 已提交
799 800
}

801
static bool isStateWindow(SStreamScanInfo* pInfo) {
802
  return pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;
5
54liuyao 已提交
803
}
5
54liuyao 已提交
804

L
Liu Jicong 已提交
805 806
static bool isIntervalWindow(SStreamScanInfo* pInfo) {
  return pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL ||
5
54liuyao 已提交
807 808 809 810 811 812
         pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL ||
         pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL;
}

static bool isSignleIntervalWindow(SStreamScanInfo* pInfo) {
  return pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL;
L
Liu Jicong 已提交
813 814
}

815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835
static uint64_t getGroupId(SOperatorInfo* pOperator, uint64_t uid) {
  uint64_t* groupId = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &uid, sizeof(int64_t));
  if (groupId) {
    return *groupId;
  }
  return 0;
  /* Todo(liuyao) for partition by column
  recordNewGroupKeys(pTableScanInfo->pGroupCols, pTableScanInfo->pGroupColVals, pBlock, rowId);
  int32_t len = buildGroupKeys(pTableScanInfo->keyBuf, pTableScanInfo->pGroupColVals);
  uint64_t resId = 0;
  uint64_t* groupId = taosHashGet(pTableScanInfo->pGroupSet, pTableScanInfo->keyBuf, len);
  if (groupId) {
    return *groupId;
  } else if (len != 0) {
    resId = calcGroupId(pTableScanInfo->keyBuf, len);
    taosHashPut(pTableScanInfo->pGroupSet, pTableScanInfo->keyBuf, len, &resId, sizeof(uint64_t));
  }
  return resId;
  */
}

836
static void setGroupId(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t groupColIndex, int32_t rowIndex) {
837
  ASSERT(rowIndex < pBlock->info.rows);
838 839 840 841 842 843 844 845 846
  switch (pBlock->info.type) {
    case STREAM_DELETE_DATA:
    case STREAM_RETRIEVE: {
      SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, groupColIndex);
      uint64_t*        groupCol = (uint64_t*)pColInfo->pData;
      pInfo->groupId = groupCol[rowIndex];
    } break;
    default:
      break;
847 848 849
  }
}

L
Liu Jicong 已提交
850
void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin) {
H
Haojun Liao 已提交
851
  pTableScanInfo->cond.twindows = *pWin;
L
Liu Jicong 已提交
852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884
  pTableScanInfo->scanTimes = 0;
  pTableScanInfo->currentGroupId = -1;
}

static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t* pRowIndex) {
  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]};
  setGroupId(pInfo, pBlock, GROUPID_COLUMN_INDEX, *pRowIndex);
  (*pRowIndex)++;

  for (; *pRowIndex < pBlock->info.rows; (*pRowIndex)++) {
    if (win.skey == startData[*pRowIndex]) {
      win.ekey = TMAX(win.ekey, endData[*pRowIndex]);
      continue;
    }
    if (win.skey == endData[*pRowIndex]) {
      win.skey = TMIN(win.skey, startData[*pRowIndex]);
      continue;
    }
    ASSERT((win.skey > startData[*pRowIndex] && win.ekey < endData[*pRowIndex]) ||
           (isInTimeWindow(&win, startData[*pRowIndex], 0) || isInTimeWindow(&win, endData[*pRowIndex], 0)));
    break;
  }

  resetTableScanInfo(pInfo->pTableScanOp->info, &win);
885
  pInfo->pTableScanOp->status = OP_OPENED;
L
Liu Jicong 已提交
886 887 888
  return true;
}

889 890 891
static STimeWindow getSlidingWindow(TSKEY* tsCol, SInterval* pInterval, SDataBlockInfo* pDataBlockInfo,
                                    int32_t* pRowIndex) {
  SResultRowInfo dumyInfo;
5
54liuyao 已提交
892
  dumyInfo.cur.pageId = -1;
893
  STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCol[*pRowIndex], pInterval, TSDB_ORDER_ASC);
5
54liuyao 已提交
894 895 896
  STimeWindow endWin = win;
  STimeWindow preWin = win;
  while (1) {
897 898
    (*pRowIndex) += getNumOfRowsInTimeWindow(pDataBlockInfo, tsCol, *pRowIndex, endWin.ekey, binarySearchForKey, NULL,
                                             TSDB_ORDER_ASC);
5
54liuyao 已提交
899 900 901 902 903
    do {
      preWin = endWin;
      getNextTimeWindow(pInterval, &endWin, TSDB_ORDER_ASC);
    } while (tsCol[(*pRowIndex) - 1] >= endWin.skey);
    endWin = preWin;
904
    if (win.ekey == endWin.ekey || (*pRowIndex) == pDataBlockInfo->rows) {
5
54liuyao 已提交
905 906 907 908 909 910
      win.ekey = endWin.ekey;
      return win;
    }
    win.ekey = endWin.ekey;
  }
}
911
static bool prepareDataScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
912 913 914
  STimeWindow win = {
      .skey = INT64_MIN,
      .ekey = INT64_MAX,
915
  };
5
54liuyao 已提交
916
  bool needRead = false;
5
54liuyao 已提交
917 918
  if (!isStateWindow(pInfo) && (*pRowIndex) < pSDB->info.rows) {
    SColumnInfoData* pColDataInfo = taosArrayGet(pSDB->pDataBlock, tsColIndex);
919 920
    TSKEY*           tsCols = (TSKEY*)pColDataInfo->pData;
    SResultRowInfo   dumyInfo;
5
54liuyao 已提交
921
    dumyInfo.cur.pageId = -1;
5
54liuyao 已提交
922 923
    if (isSessionWindow(pInfo)) {
      SStreamAggSupporter* pAggSup = pInfo->sessionSup.pStreamAggSup;
924 925 926
      int64_t              gap = pInfo->sessionSup.gap;
      int32_t              winIndex = 0;
      SResultWindowInfo*   pCurWin =
927
          getSessionTimeWindow(pAggSup, tsCols[*pRowIndex], INT64_MIN, pSDB->info.groupId, gap, &winIndex);
5
54liuyao 已提交
928
      win = pCurWin->win;
L
Liu Jicong 已提交
929
      setGroupId(pInfo, pSDB, GROUPID_COLUMN_INDEX, *pRowIndex);
930
      (*pRowIndex) += updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, *pRowIndex, gap, NULL);
5
54liuyao 已提交
931
    } else {
932
      setGroupId(pInfo, pSDB, GROUPID_COLUMN_INDEX, *pRowIndex);
5
54liuyao 已提交
933 934 935 936 937
      pInfo->updateWin.skey = tsCols[*pRowIndex];
      win = getSlidingWindow(tsCols, &pInfo->interval, &pSDB->info, pRowIndex);
      pInfo->updateWin.ekey = tsCols[*pRowIndex - 1];
      // win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[*pRowIndex], &pInfo->interval, TSDB_ORDER_ASC);
      // (*pRowIndex) +=
938 939
      //     getNumOfRowsInTimeWindow(&pSDB->info, tsCols, *pRowIndex, win.ekey, binarySearchForKey, NULL,
      //     TSDB_ORDER_ASC);
5
54liuyao 已提交
940
    }
5
54liuyao 已提交
941 942 943 944 945
    needRead = true;
  } else if (isStateWindow(pInfo)) {
    SArray* pWins = pInfo->sessionSup.pStreamAggSup->pScanWindow;
    int32_t size = taosArrayGetSize(pWins);
    if (pInfo->scanWinIndex < size) {
946
      win = *(STimeWindow*)taosArrayGet(pWins, pInfo->scanWinIndex);
5
54liuyao 已提交
947 948 949 950 951 952 953 954
      pInfo->scanWinIndex++;
      needRead = true;
    } else {
      pInfo->scanWinIndex = 0;
      taosArrayClear(pWins);
    }
  }
  if (!needRead) {
5
54liuyao 已提交
955 956
    return false;
  }
L
Liu Jicong 已提交
957
  resetTableScanInfo(pInfo->pTableScanOp->info, &win);
5
54liuyao 已提交
958
  return true;
5
54liuyao 已提交
959 960
}

961
static void copyOneRow(SSDataBlock* dest, SSDataBlock* source, int32_t sourceRowId) {
962
  for (int32_t j = 0; j < taosArrayGetSize(source->pDataBlock); j++) {
963 964 965 966 967 968
    SColumnInfoData* pDestCol = (SColumnInfoData*)taosArrayGet(dest->pDataBlock, j);
    SColumnInfoData* pSourceCol = (SColumnInfoData*)taosArrayGet(source->pDataBlock, j);
    if (colDataIsNull_s(pSourceCol, sourceRowId)) {
      colDataAppendNULL(pDestCol, dest->info.rows);
    } else {
      colDataAppend(pDestCol, dest->info.rows, colDataGetData(pSourceCol, sourceRowId), false);
5
54liuyao 已提交
969 970
    }
  }
971
  dest->info.rows++;
5
54liuyao 已提交
972 973
}

L
Liu Jicong 已提交
974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993
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;
      return NULL;
    }

    if (pResult->info.groupId == pInfo->groupId) {
      return pResult;
    }
  }
}

994
static SSDataBlock* doDataScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
995 996
  while (1) {
    SSDataBlock* pResult = NULL;
997
    pResult = doTableScan(pInfo->pTableScanOp);
998
    if (pResult == NULL) {
5
54liuyao 已提交
999
      if (prepareDataScan(pInfo, pSDB, tsColIndex, pRowIndex)) {
1000
        // scan next window data
1001
        pResult = doTableScan(pInfo->pTableScanOp);
1002 1003 1004
      }
    }
    if (!pResult) {
5
54liuyao 已提交
1005
      pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
1006 1007
      return NULL;
    }
L
Liu Jicong 已提交
1008

1009
    if (pResult->info.groupId == pInfo->groupId) {
5
54liuyao 已提交
1010
      pResult->info.calWin = pInfo->updateWin;
1011
      return pResult;
5
54liuyao 已提交
1012 1013
    }
  }
1014

L
Liu Jicong 已提交
1015 1016 1017 1018
  /* Todo(liuyao) for partition by column
    SSDataBlock* pBlock = createOneDataBlock(pResult, true);
    blockDataCleanup(pResult);
    for (int32_t i = 0; i < pBlock->info.rows; i++) {
1019
      uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock->info.uid);
L
Liu Jicong 已提交
1020 1021 1022
      if (id == pInfo->groupId) {
        copyOneRow(pResult, pBlock, i);
      }
1023
    }
L
Liu Jicong 已提交
1024 1025
    return pResult;
  */
1026
}
L
Liu Jicong 已提交
1027 1028
static void generateIntervalTs(SStreamScanInfo* pInfo, SSDataBlock* pDelBlock, SOperatorInfo* pOperator,
                               SSDataBlock* pUpdateRes) {
1029 1030 1031 1032 1033 1034 1035
  if (pDelBlock->info.rows == 0) {
    return;
  }
  blockDataCleanup(pUpdateRes);
  blockDataEnsureCapacity(pUpdateRes, 64);
  ASSERT(taosArrayGetSize(pDelBlock->pDataBlock) >= 3);
  SColumnInfoData* pStartTsCol = taosArrayGet(pDelBlock->pDataBlock, START_TS_COLUMN_INDEX);
1036
  TSKEY*           startData = (TSKEY*)pStartTsCol->pData;
1037
  SColumnInfoData* pEndTsCol = taosArrayGet(pDelBlock->pDataBlock, END_TS_COLUMN_INDEX);
1038
  TSKEY*           endData = (TSKEY*)pEndTsCol->pData;
1039
  SColumnInfoData* pGpCol = taosArrayGet(pDelBlock->pDataBlock, UID_COLUMN_INDEX);
1040
  uint64_t*        uidCol = (uint64_t*)pGpCol->pData;
1041 1042

  SColumnInfoData* pDestTsCol = taosArrayGet(pUpdateRes->pDataBlock, START_TS_COLUMN_INDEX);
L
Liu Jicong 已提交
1043
  SColumnInfoData* pDestGpCol = taosArrayGet(pUpdateRes->pDataBlock, GROUPID_COLUMN_INDEX);
1044 1045 1046 1047
  for (int32_t i = pInfo->deleteDataIndex;
       i < pDelBlock->info.rows &&
       i < pDelBlock->info.capacity - (endData[i] - startData[i]) / pInfo->interval.interval - 1;
       i++) {
1048
    uint64_t groupId = getGroupId(pOperator, uidCol[i]);
1049
    for (TSKEY startTs = startData[i]; startTs <= endData[i];) {
1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063
      colDataAppend(pDestTsCol, pUpdateRes->info.rows, (const char*)&startTs, false);
      colDataAppend(pDestGpCol, pUpdateRes->info.rows, (const char*)&groupId, false);
      pUpdateRes->info.rows++;
      startTs = taosTimeAdd(startTs, pInfo->interval.interval, pInfo->interval.intervalUnit, pInfo->interval.precision);
    }
    pInfo->deleteDataIndex++;
  }

  if (pInfo->deleteDataIndex > 0 && pInfo->deleteDataIndex == pDelBlock->info.rows) {
    blockDataCleanup(pDelBlock);
    pInfo->deleteDataIndex = 0;
  }
}

L
Liu Jicong 已提交
1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100
static void generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pBlock, SOperatorInfo* pOperator,
                              SSDataBlock* pUpdateRes) {
  if (pBlock->info.rows == 0) {
    return;
  }
  blockDataCleanup(pUpdateRes);
  blockDataEnsureCapacity(pUpdateRes, pBlock->info.rows);
  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;
  SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);
  uint64_t*        uidCol = (uint64_t*)pGpCol->pData;

  SColumnInfoData* pDestStartCol = taosArrayGet(pUpdateRes->pDataBlock, START_TS_COLUMN_INDEX);
  SColumnInfoData* pDestEndCol = taosArrayGet(pUpdateRes->pDataBlock, END_TS_COLUMN_INDEX);
  SColumnInfoData* pDestGpCol = taosArrayGet(pUpdateRes->pDataBlock, GROUPID_COLUMN_INDEX);
  int32_t          dummy = 0;
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
    uint64_t groupId = getGroupId(pOperator, uidCol[i]);
    // gap must be 0.
    SResultWindowInfo* pStartWin =
        getCurSessionWindow(pInfo->sessionSup.pStreamAggSup, startData[i], endData[i], groupId, 0, &dummy);
    if (!pStartWin) {
      // window has been closed.
      continue;
    }
    SResultWindowInfo* pEndWin =
        getCurSessionWindow(pInfo->sessionSup.pStreamAggSup, endData[i], endData[i], groupId, 0, &dummy);
    ASSERT(pEndWin);
    colDataAppend(pDestStartCol, i, (const char*)&pStartWin->win.skey, false);
    colDataAppend(pDestEndCol, i, (const char*)&pEndWin->win.ekey, false);
    colDataAppend(pDestGpCol, i, (const char*)&groupId, false);
    pUpdateRes->info.rows++;
  }
}
1101
static void setUpdateData(SStreamScanInfo* pInfo, SSDataBlock* pBlock, SSDataBlock* pUpdateBlock) {
1102
  blockDataCleanup(pUpdateBlock);
5
54liuyao 已提交
1103
  int32_t size = taosArrayGetSize(pInfo->tsArray);
1104
  if (pInfo->tsArrayIndex < size) {
5
54liuyao 已提交
1105
    SColumnInfoData* pCol = (SColumnInfoData*)taosArrayGet(pUpdateBlock->pDataBlock, pInfo->primaryTsIndex);
5
54liuyao 已提交
1106
    ASSERT(pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
1107
    blockDataEnsureCapacity(pUpdateBlock, size);
1108 1109

    int32_t rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, pInfo->tsArrayIndex);
1110
    pInfo->groupId = getGroupId(pInfo->pTableScanOp, pBlock->info.uid);
1111
    int32_t i = 0;
L
Liu Jicong 已提交
1112
    for (; i < size; i++) {
1113
      rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, i + pInfo->tsArrayIndex);
1114
      uint64_t id = getGroupId(pInfo->pTableScanOp, pBlock->info.uid);
1115 1116
      if (pInfo->groupId != id) {
        break;
1117
      }
1118
      copyOneRow(pUpdateBlock, pBlock, rowId);
1119
    }
1120 1121 1122
    pUpdateBlock->info.rows = i;
    pInfo->tsArrayIndex += i;
    pUpdateBlock->info.groupId = pInfo->groupId;
5
54liuyao 已提交
1123
    pUpdateBlock->info.type = STREAM_CLEAR;
5
54liuyao 已提交
1124
    blockDataUpdateTsWindow(pUpdateBlock, 0);
1125 1126 1127 1128
  }
  // all rows have same group id
  ASSERT(pInfo->tsArrayIndex >= size);
  if (size > 0 && pInfo->tsArrayIndex == size) {
5
54liuyao 已提交
1129
    taosArrayClear(pInfo->tsArray);
5
54liuyao 已提交
1130
  }
1131 1132

  if (size == 0) {
L
Liu Jicong 已提交
1133
    generateIntervalTs(pInfo, pInfo->pDeleteDataRes, pInfo->pTableScanOp, pUpdateBlock);
1134
  }
5
54liuyao 已提交
1135 1136
}

1137
static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock* pBlock, bool out) {
1138 1139
  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
  ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP);
5
54liuyao 已提交
1140
  TSKEY* tsCol = (TSKEY*)pColDataInfo->pData;
1141
  for (int32_t rowId = 0; rowId < pBlock->info.rows; rowId++) {
5
54liuyao 已提交
1142 1143 1144 1145 1146
    SResultRowInfo dumyInfo;
    dumyInfo.cur.pageId = -1;
    STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCol[rowId], &pInfo->interval, TSDB_ORDER_ASC);
    // must check update info first.
    bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, tsCol[rowId]);
1147
    if ((update || (isSignleIntervalWindow(pInfo) && isCloseWindow(&win, &pInfo->twAggSup))) && out) {
1148 1149 1150
      taosArrayPush(pInfo->tsArray, &rowId);
    }
  }
1151 1152 1153 1154 1155 1156 1157
}

static void setBlockGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t uidColIndex) {
  ASSERT(taosArrayGetSize(pBlock->pDataBlock) >= 3);
  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, uidColIndex);
  uint64_t*        uidCol = (uint64_t*)pColDataInfo->pData;
  ASSERT(pBlock->info.rows > 0);
1158
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
1159
    uidCol[i] = getGroupId(pOperator, uidCol[i]);
5
54liuyao 已提交
1160 1161
  }
}
L
Liu Jicong 已提交
1162

L
Liu Jicong 已提交
1163 1164 1165 1166 1167
static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock) {
  SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
  SOperatorInfo*  pOperator = pInfo->pStreamScanOp;
  SExecTaskInfo*  pTaskInfo = pInfo->pStreamScanOp->pTaskInfo;

1168 1169
  blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows);

L
Liu Jicong 已提交
1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180
  pInfo->pRes->info.rows = pBlock->info.rows;
  pInfo->pRes->info.uid = pBlock->info.uid;
  pInfo->pRes->info.type = STREAM_NORMAL;

  uint64_t* groupIdPre = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &pBlock->info.uid, sizeof(int64_t));
  if (groupIdPre) {
    pInfo->pRes->info.groupId = *groupIdPre;
  } else {
    pInfo->pRes->info.groupId = 0;
  }

C
Cary Xu 已提交
1181 1182 1183 1184 1185 1186
  // for generating rollup SMA result, each time is an independent time serie.
  // TODO temporarily used, when the statement of "partition by tbname" is ready, remove this
  if (pInfo->assignBlockUid) {
    pInfo->pRes->info.groupId = pBlock->info.uid;
  }

L
Liu Jicong 已提交
1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197
  // todo extract method
  for (int32_t i = 0; i < taosArrayGetSize(pInfo->pColMatchInfo); ++i) {
    SColMatchInfo* pColMatchInfo = taosArrayGet(pInfo->pColMatchInfo, i);
    if (!pColMatchInfo->output) {
      continue;
    }

    bool colExists = false;
    for (int32_t j = 0; j < blockDataGetNumOfCols(pBlock); ++j) {
      SColumnInfoData* pResCol = bdGetColumnInfoData(pBlock, j);
      if (pResCol->info.colId == pColMatchInfo->colId) {
1198 1199
        SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId);
        colDataAssign(pDst, pResCol, pBlock->info.rows, &pInfo->pRes->info);
L
Liu Jicong 已提交
1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213
        colExists = true;
        break;
      }
    }

    // the required column does not exists in submit block, let's set it to be all null value
    if (!colExists) {
      SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId);
      colDataAppendNNULL(pDst, 0, pBlockInfo->rows);
    }
  }

  // currently only the tbname pseudo column
  if (pInfo->numOfPseudoExpr > 0) {
L
Liu Jicong 已提交
1214 1215
    int32_t code = addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, pInfo->pRes,
                                          GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
1216
    if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
1217
      blockDataFreeRes((SSDataBlock*)pBlock);
H
Haojun Liao 已提交
1218 1219
      longjmp(pTaskInfo->env, code);
    }
L
Liu Jicong 已提交
1220 1221 1222 1223
  }

  doFilter(pInfo->pCondition, pInfo->pRes);
  blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
L
Liu Jicong 已提交
1224
  blockDataFreeRes((SSDataBlock*)pBlock);
L
Liu Jicong 已提交
1225 1226
  return 0;
}
5
54liuyao 已提交
1227

1228
static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
1229
  // NOTE: this operator does never check if current status is done or not
1230 1231
  SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;
  SStreamScanInfo* pInfo = pOperator->info;
H
Haojun Liao 已提交
1232

L
Liu Jicong 已提交
1233
  qDebug("stream scan called");
L
Liu Jicong 已提交
1234 1235 1236 1237 1238 1239 1240 1241 1242
  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);
        if (setBlockIntoRes(pInfo, &ret.data) < 0) {
          ASSERT(0);
        }
L
Liu Jicong 已提交
1243
        // TODO clean data block
L
Liu Jicong 已提交
1244
        if (pInfo->pRes->info.rows > 0) {
L
Liu Jicong 已提交
1245
          qDebug("stream scan log return %d rows", pInfo->pRes->info.rows);
L
Liu Jicong 已提交
1246 1247 1248 1249 1250 1251 1252 1253
          return pInfo->pRes;
        }
      } else if (ret.fetchType == FETCH_TYPE__META) {
        ASSERT(0);
        pTaskInfo->streamInfo.lastStatus = ret.offset;
        pTaskInfo->streamInfo.metaBlk = ret.meta;
        return NULL;
      } else if (ret.fetchType == FETCH_TYPE__NONE) {
L
Liu Jicong 已提交
1254
        pTaskInfo->streamInfo.lastStatus = ret.offset;
L
Liu Jicong 已提交
1255
        ASSERT(pTaskInfo->streamInfo.lastStatus.version + 1 >= pTaskInfo->streamInfo.prepareStatus.version);
L
Liu Jicong 已提交
1256
        qDebug("stream scan log return null");
L
Liu Jicong 已提交
1257 1258 1259 1260 1261
        return NULL;
      } else {
        ASSERT(0);
      }
    }
L
Liu Jicong 已提交
1262 1263
  } else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
    SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
L
Liu Jicong 已提交
1264 1265 1266 1267 1268 1269
    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 已提交
1270 1271 1272 1273
  } else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_META) {
    // TODO scan meta
    ASSERT(0);
    return NULL;
H
Haojun Liao 已提交
1274 1275
  }

L
Liu Jicong 已提交
1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293
  if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__PREPARE) {
    STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
    memcpy(&pTSInfo->cond, &pTaskInfo->streamInfo.tableCond, sizeof(SQueryTableDataCond));
    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) {
      return pBlock;
    }
    // TODO fill in bloom filter
    pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__NONE;
    return NULL;
  }

5
54liuyao 已提交
1294
  size_t total = taosArrayGetSize(pInfo->pBlockLists);
1295
  // TODO: refactor
L
Liu Jicong 已提交
1296
  if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
H
Haojun Liao 已提交
1297
    if (pInfo->validBlockIndex >= total) {
L
Liu Jicong 已提交
1298
      /*doClearBufferedBlocks(pInfo);*/
L
Liu Jicong 已提交
1299
      /*pOperator->status = OP_EXEC_DONE;*/
H
Haojun Liao 已提交
1300 1301 1302
      return NULL;
    }

1303
    int32_t      current = pInfo->validBlockIndex++;
1304
    SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current);
1305
    // TODO move into scan
5
54liuyao 已提交
1306 1307
    pBlock->info.calWin.skey = INT64_MIN;
    pBlock->info.calWin.ekey = INT64_MAX;
1308
    blockDataUpdateTsWindow(pBlock, 0);
1309
    switch (pBlock->info.type) {
L
Liu Jicong 已提交
1310 1311 1312
      case STREAM_NORMAL:
      case STREAM_GET_ALL:
        return pBlock;
1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323
      case STREAM_RETRIEVE: {
        pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
        pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RETRIEVE;
        copyDataBlock(pInfo->pPullDataRes, pBlock);
        pInfo->pullDataResIndex = 0;
        prepareDataScan(pInfo, pInfo->pPullDataRes, START_TS_COLUMN_INDEX, &pInfo->pullDataResIndex);
        updateInfoAddCloseWindowSBF(pInfo->pUpdateInfo);
      } break;
      case STREAM_DELETE_DATA: {
        pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
        pInfo->updateResIndex = 0;
L
Liu Jicong 已提交
1324 1325 1326 1327 1328 1329 1330 1331 1332 1333
        if (isIntervalWindow(pInfo)) {
          copyDataBlock(pInfo->pDeleteDataRes, pBlock);
          generateIntervalTs(pInfo, pInfo->pDeleteDataRes, pInfo->pTableScanOp, pInfo->pUpdateRes);
          prepareDataScan(pInfo, pInfo->pUpdateRes, START_TS_COLUMN_INDEX, &pInfo->updateResIndex);
          pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
        } else {
          generateScanRange(pInfo, pBlock, pInfo->pTableScanOp, pInfo->pUpdateRes);
          prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
          pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
        }
1334 1335 1336 1337 1338
        pInfo->pUpdateRes->info.type = STREAM_DELETE_DATA;
        return pInfo->pUpdateRes;
      } break;
      default:
        break;
5
54liuyao 已提交
1339
    }
1340
    return pBlock;
L
Liu Jicong 已提交
1341
  } else if (pInfo->blockType == STREAM_INPUT__DATA_SUBMIT) {
L
Liu Jicong 已提交
1342
    qDebug("scan mode %d", pInfo->scanMode);
5
54liuyao 已提交
1343 1344 1345 1346 1347
    if (pInfo->scanMode == STREAM_SCAN_FROM_RES) {
      blockDataDestroy(pInfo->pUpdateRes);
      pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
      return pInfo->pRes;
    } else if (pInfo->scanMode == STREAM_SCAN_FROM_UPDATERES) {
L
Liu Jicong 已提交
1348 1349 1350 1351
      if (isStateWindow(pInfo)) {
        pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
      } else {
        pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
5
54liuyao 已提交
1352
        prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
5
54liuyao 已提交
1353
      }
5
54liuyao 已提交
1354
      return pInfo->pUpdateRes;
5
54liuyao 已提交
1355 1356 1357
    } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RETRIEVE) {
      SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pPullDataRes, 0, &pInfo->pullDataResIndex);
      if (pSDB != NULL) {
1358
        checkUpdateData(pInfo, true, pSDB, false);
L
Liu Jicong 已提交
1359
        pSDB->info.type = STREAM_PULL_DATA;
5
54liuyao 已提交
1360 1361 1362
        return pSDB;
      }
      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
1363 1364 1365 1366 1367 1368
    } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) {
      SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
      if (pSDB) {
        pSDB->info.type = STREAM_NORMAL;
        checkUpdateData(pInfo, true, pSDB, false);
        return pSDB;
5
54liuyao 已提交
1369
      }
1370 1371 1372 1373
      setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes);
      if (pInfo->pUpdateRes->info.rows > 0) {
        prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
        return pInfo->pUpdateRes;
5
54liuyao 已提交
1374
      }
1375
      pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
L
Liu Jicong 已提交
1376 1377 1378 1379 1380 1381 1382 1383
    } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE) {
      SSDataBlock* pSDB = doRangeScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
      if (pSDB) {
        pSDB->info.type = STREAM_NORMAL;
        checkUpdateData(pInfo, true, pSDB, false);
        return pSDB;
      }
      pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
1384 1385 1386 1387
    } else if (isStateWindow(pInfo)) {
      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
      pInfo->updateResIndex = pInfo->pUpdateRes->info.rows;
      if (prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex)) {
L
Liu Jicong 已提交
1388
        blockDataCleanup(pInfo->pUpdateRes);
1389 1390 1391 1392
        // return empty data blcok
        return pInfo->pUpdateRes;
      }
      pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
5
54liuyao 已提交
1393
    }
5
54liuyao 已提交
1394

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

1397
    int32_t totBlockNum = taosArrayGetSize(pInfo->pBlockLists);
1398

1399 1400 1401 1402 1403
    while (1) {
      if (pInfo->tqReader->pMsg == NULL) {
        if (pInfo->validBlockIndex >= totBlockNum) {
          return NULL;
        }
1404

1405 1406 1407 1408 1409 1410 1411 1412
        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 已提交
1413 1414
      }

1415 1416 1417 1418
      blockDataCleanup(pInfo->pRes);

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

1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431
        int32_t code = tqRetrieveDataBlock(&block, pInfo->tqReader);

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

        setBlockIntoRes(pInfo, &block);

        if (pBlockInfo->rows > 0) {
          break;
        }
      }
5
54liuyao 已提交
1432 1433
      if (pBlockInfo->rows > 0) {
        break;
J
jiacy-jcy 已提交
1434 1435
      } else {
        pInfo->tqReader->pMsg = NULL;
1436
        continue;
5
54liuyao 已提交
1437
      }
1438
      /*blockDataCleanup(pInfo->pRes);*/
H
Haojun Liao 已提交
1439 1440 1441 1442
    }

    // record the scan action.
    pInfo->numOfExec++;
1443
    pOperator->resultInfo.totalRows += pBlockInfo->rows;
H
Haojun Liao 已提交
1444

5
54liuyao 已提交
1445
    if (pBlockInfo->rows == 0) {
5
54liuyao 已提交
1446
      updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
L
Liu Jicong 已提交
1447
      /*pOperator->status = OP_EXEC_DONE;*/
5
54liuyao 已提交
1448
    } else if (pInfo->pUpdateInfo) {
1449
      pInfo->tsArrayIndex = 0;
1450 1451
      checkUpdateData(pInfo, true, pInfo->pRes, true);
      setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes);
5
54liuyao 已提交
1452
      pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlockInfo->window.ekey);
5
54liuyao 已提交
1453
      if (pInfo->pUpdateRes->info.rows > 0) {
5
54liuyao 已提交
1454
        if (pInfo->pUpdateRes->info.type == STREAM_CLEAR) {
5
54liuyao 已提交
1455 1456
          pInfo->updateResIndex = 0;
          pInfo->scanMode = STREAM_SCAN_FROM_UPDATERES;
5
54liuyao 已提交
1457
        } else if (pInfo->pUpdateRes->info.type == STREAM_INVERT) {
5
54liuyao 已提交
1458
          pInfo->scanMode = STREAM_SCAN_FROM_RES;
5
54liuyao 已提交
1459
          return pInfo->pUpdateRes;
5
54liuyao 已提交
1460
        }
5
54liuyao 已提交
1461
      }
1462
    }
1463

L
Liu Jicong 已提交
1464
    qDebug("scan rows: %d", pBlockInfo->rows);
5
54liuyao 已提交
1465
    return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes;
1466

L
Liu Jicong 已提交
1467 1468 1469
  } else {
    ASSERT(0);
    return NULL;
H
Haojun Liao 已提交
1470 1471 1472
  }
}

L
Liu Jicong 已提交
1473 1474 1475 1476 1477
static SSDataBlock* doRawScan(SOperatorInfo* pInfo) {
  //
  return NULL;
}

1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489
static SArray* extractTableIdList(const STableListInfo* pTableGroupInfo) {
  SArray* tableIdList = taosArrayInit(4, sizeof(uint64_t));

  // Transfer the Array of STableKeyInfo into uid list.
  for (int32_t i = 0; i < taosArrayGetSize(pTableGroupInfo->pTableList); ++i) {
    STableKeyInfo* pkeyInfo = taosArrayGet(pTableGroupInfo->pTableList, i);
    taosArrayPush(tableIdList, &pkeyInfo->uid);
  }

  return tableIdList;
}

L
Liu Jicong 已提交
1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502
// for subscribing db or stb (not including column),
// if this scan is used, meta data can be return
// and schemas are decided when scanning
SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,
                                         SExecTaskInfo* pTaskInfo, STimeWindowAggSupp* pTwSup) {
  // create operator
  // create tb reader
  // create meta reader
  // create tq reader

  return NULL;
}

1503 1504
static void destroyStreamScanOperatorInfo(void* param, int32_t numOfOutput) {
  SStreamScanInfo* pStreamScan = (SStreamScanInfo*)param;
L
Liu Jicong 已提交
1505
#if 1
1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516
  if (pStreamScan->pTableScanOp && pStreamScan->pTableScanOp->info) {
    STableScanInfo* pTableScanInfo = pStreamScan->pTableScanOp->info;
    destroyTableScanOperatorInfo(pTableScanInfo, 1);
  }
#endif
  if (pStreamScan->tqReader) {
    tqCloseReader(pStreamScan->tqReader);
  }
  if (pStreamScan->pColMatchInfo) {
    taosArrayDestroy(pStreamScan->pColMatchInfo);
  }
L
Liu Jicong 已提交
1517
  updateInfoDestroy(pStreamScan->pUpdateInfo);
1518 1519 1520 1521 1522
  blockDataDestroy(pStreamScan->pRes);
  blockDataDestroy(pStreamScan->pUpdateRes);
  blockDataDestroy(pStreamScan->pPullDataRes);
  blockDataDestroy(pStreamScan->pDeleteDataRes);
  taosArrayDestroy(pStreamScan->pBlockLists);
L
Liu Jicong 已提交
1523
  taosArrayDestroy(pStreamScan->tsArray);
1524 1525 1526
  taosMemoryFree(pStreamScan);
}

1527
SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond,
1528
                                            SExecTaskInfo* pTaskInfo) {
1529 1530
  SStreamScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamScanInfo));
  SOperatorInfo*   pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
1531

H
Haojun Liao 已提交
1532 1533
  if (pInfo == NULL || pOperator == NULL) {
    terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
1534
    goto _error;
H
Haojun Liao 已提交
1535 1536
  }

1537
  SScanPhysiNode*     pScanPhyNode = &pTableScanNode->scan;
1538
  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;
H
Haojun Liao 已提交
1539

1540 1541
  pInfo->pTagCond = pTagCond;

1542 1543 1544 1545 1546 1547
  pInfo->twAggSup = (STimeWindowAggSupp){
      .waterMark = pTableScanNode->watermark,
      .calTrigger = pTableScanNode->triggerType,
      .maxTs = INT64_MIN,
  };

1548
  int32_t numOfCols = 0;
1549
  pInfo->pColMatchInfo = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
1550 1551 1552

  int32_t numOfOutput = taosArrayGetSize(pInfo->pColMatchInfo);
  SArray* pColIds = taosArrayInit(numOfOutput, sizeof(int16_t));
1553
  for (int32_t i = 0; i < numOfOutput; ++i) {
1554 1555 1556
    SColMatchInfo* id = taosArrayGet(pInfo->pColMatchInfo, i);

    int16_t colId = id->colId;
1557
    taosArrayPush(pColIds, &colId);
1558
    if (id->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
5
54liuyao 已提交
1559 1560
      pInfo->primaryTsIndex = id->targetSlotId;
    }
H
Haojun Liao 已提交
1561 1562 1563 1564
  }

  pInfo->pBlockLists = taosArrayInit(4, POINTER_BYTES);
  if (pInfo->pBlockLists == NULL) {
1565 1566
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _error;
H
Haojun Liao 已提交
1567 1568
  }

1569
  pInfo->tsArray = taosArrayInit(4, sizeof(int32_t));
5
54liuyao 已提交
1570
  if (pInfo->tsArray == NULL) {
1571
    goto _error;
5
54liuyao 已提交
1572 1573
  }

5
54liuyao 已提交
1574
  if (pHandle->vnode) {
L
Liu Jicong 已提交
1575
    SOperatorInfo*  pTableScanOp = createTableScanOperatorInfo(pTableScanNode, pHandle, pTaskInfo);
L
Liu Jicong 已提交
1576
    STableScanInfo* pTSInfo = (STableScanInfo*)pTableScanOp->info;
1577
    if (pHandle->version > 0) {
L
Liu Jicong 已提交
1578
      pTSInfo->cond.endVersion = pHandle->version;
1579
    }
L
Liu Jicong 已提交
1580 1581

    SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, 0);
1582
    if (pHandle->initTableReader) {
L
Liu Jicong 已提交
1583 1584 1585
      pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;
      pTSInfo->dataReader = NULL;
      if (tsdbReaderOpen(pHandle->vnode, &pTSInfo->cond, tableList, &pTSInfo->dataReader, NULL) < 0) {
L
Liu Jicong 已提交
1586 1587
        ASSERT(0);
      }
L
Liu Jicong 已提交
1588 1589
    }

L
Liu Jicong 已提交
1590 1591 1592 1593
    if (pHandle->initTqReader) {
      ASSERT(pHandle->tqReader == NULL);
      pInfo->tqReader = tqOpenReader(pHandle->vnode);
      ASSERT(pInfo->tqReader);
1594
    } else {
L
Liu Jicong 已提交
1595 1596
      ASSERT(pHandle->tqReader);
      pInfo->tqReader = pHandle->tqReader;
1597 1598
    }

L
Liu Jicong 已提交
1599
    if (pTSInfo->interval.interval > 0) {
1600
      pInfo->pUpdateInfo = updateInfoInitP(&pTSInfo->interval, pInfo->twAggSup.waterMark);
L
Liu Jicong 已提交
1601 1602 1603
    } else {
      pInfo->pUpdateInfo = NULL;
    }
1604 1605

    pInfo->pTableScanOp = pTableScanOp;
L
Liu Jicong 已提交
1606
    pInfo->interval = pTSInfo->interval;
L
Liu Jicong 已提交
1607

L
Liu Jicong 已提交
1608 1609 1610
    pInfo->readHandle = *pHandle;
    pInfo->tableUid = pScanPhyNode->uid;

L
Liu Jicong 已提交
1611
    // set the extract column id to streamHandle
L
Liu Jicong 已提交
1612
    tqReaderSetColIdList(pInfo->tqReader, pColIds);
L
Liu Jicong 已提交
1613
    SArray* tableIdList = extractTableIdList(&pTaskInfo->tableqinfoList);
L
Liu Jicong 已提交
1614
    int32_t code = tqReaderSetTbUidList(pInfo->tqReader, tableIdList);
L
Liu Jicong 已提交
1615 1616 1617 1618 1619
    if (code != 0) {
      taosArrayDestroy(tableIdList);
      goto _error;
    }
    taosArrayDestroy(tableIdList);
L
Liu Jicong 已提交
1620
    memcpy(&pTaskInfo->streamInfo.tableCond, &pTSInfo->cond, sizeof(SQueryTableDataCond));
5
54liuyao 已提交
1621 1622
  }

1623 1624 1625 1626 1627
  // create the pseduo columns info
  if (pTableScanNode->scan.pScanPseudoCols != NULL) {
    pInfo->pPseudoExpr = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pInfo->numOfPseudoExpr);
  }

1628
  pInfo->pRes = createResDataBlock(pDescNode);
X
Xiaoyu Wang 已提交
1629
  pInfo->pUpdateRes = createResDataBlock(pDescNode);
1630 1631
  pInfo->pCondition = pScanPhyNode->node.pConditions;
  pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
1632 1633
  pInfo->sessionSup =
      (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1, .parentType = QUERY_NODE_PHYSICAL_PLAN};
1634
  pInfo->groupId = 0;
5
54liuyao 已提交
1635
  pInfo->pPullDataRes = createPullDataBlock();
1636
  pInfo->pStreamScanOp = pOperator;
1637 1638
  pInfo->deleteDataIndex = 0;
  pInfo->pDeleteDataRes = createPullDataBlock();
5
54liuyao 已提交
1639
  pInfo->updateWin = (STimeWindow){.skey = INT64_MAX, .ekey = INT64_MAX};
L
Liu Jicong 已提交
1640

1641
  pOperator->name = "StreamScanOperator";
L
Liu Jicong 已提交
1642
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
1643 1644 1645
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
1646
  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);
1647
  pOperator->pTaskInfo = pTaskInfo;
H
Haojun Liao 已提交
1648

1649 1650
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamScan, NULL, NULL, destroyStreamScanOperatorInfo,
                                         NULL, NULL, NULL);
1651

H
Haojun Liao 已提交
1652
  return pOperator;
1653

L
Liu Jicong 已提交
1654
_error:
1655 1656 1657
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  return NULL;
H
Haojun Liao 已提交
1658 1659 1660 1661 1662 1663 1664
}

static void destroySysScanOperator(void* param, int32_t numOfOutput) {
  SSysTableScanInfo* pInfo = (SSysTableScanInfo*)param;
  tsem_destroy(&pInfo->ready);
  blockDataDestroy(pInfo->pRes);

1665
  const char* name = tNameGetTableName(&pInfo->name);
1666
  if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0 ||
1667
      strncasecmp(name, TSDB_INS_TABLE_USER_TAGS, TSDB_TABLE_FNAME_LEN) == 0 || pInfo->pCur != NULL) {
H
Haojun Liao 已提交
1668
    metaCloseTbCursor(pInfo->pCur);
1669
    pInfo->pCur = NULL;
H
Haojun Liao 已提交
1670
  }
H
Haojun Liao 已提交
1671 1672

  taosArrayDestroy(pInfo->scanCols);
1673
  taosMemoryFreeClear(pInfo->pUser);
D
dapan1121 已提交
1674 1675

  taosMemoryFreeClear(param);
H
Haojun Liao 已提交
1676 1677
}

X
Xiaoyu Wang 已提交
1678
static int32_t getSysTableDbNameColId(const char* pTable) {
X
Xiaoyu Wang 已提交
1679 1680 1681
  // if (0 == strcmp(TSDB_INS_TABLE_USER_INDEXES, pTable)) {
  //   return 1;
  // }
X
Xiaoyu Wang 已提交
1682 1683 1684
  return TSDB_INS_USER_STABLES_DBNAME_COLID;
}

H
Haojun Liao 已提交
1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705
EDealRes getDBNameFromConditionWalker(SNode* pNode, void* pContext) {
  int32_t   code = TSDB_CODE_SUCCESS;
  ENodeType nType = nodeType(pNode);

  switch (nType) {
    case QUERY_NODE_OPERATOR: {
      SOperatorNode* node = (SOperatorNode*)pNode;
      if (OP_TYPE_EQUAL == node->opType) {
        *(int32_t*)pContext = 1;
        return DEAL_RES_CONTINUE;
      }

      *(int32_t*)pContext = 0;
      return DEAL_RES_IGNORE_CHILD;
    }
    case QUERY_NODE_COLUMN: {
      if (1 != *(int32_t*)pContext) {
        return DEAL_RES_CONTINUE;
      }

      SColumnNode* node = (SColumnNode*)pNode;
X
Xiaoyu Wang 已提交
1706
      if (getSysTableDbNameColId(node->tableName) == node->colId) {
H
Haojun Liao 已提交
1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722
        *(int32_t*)pContext = 2;
        return DEAL_RES_CONTINUE;
      }

      *(int32_t*)pContext = 0;
      return DEAL_RES_CONTINUE;
    }
    case QUERY_NODE_VALUE: {
      if (2 != *(int32_t*)pContext) {
        return DEAL_RES_CONTINUE;
      }

      SValueNode* node = (SValueNode*)pNode;
      char*       dbName = nodesGetValueFromNode(node);
      strncpy(pContext, varDataVal(dbName), varDataLen(dbName));
      *((char*)pContext + varDataLen(dbName)) = 0;
1723
      return DEAL_RES_END;  // stop walk
H
Haojun Liao 已提交
1724 1725 1726 1727 1728 1729 1730
    }
    default:
      break;
  }
  return DEAL_RES_CONTINUE;
}

1731
static void getDBNameFromCondition(SNode* pCondition, const char* dbName) {
H
Haojun Liao 已提交
1732 1733 1734
  if (NULL == pCondition) {
    return;
  }
L
Liu Jicong 已提交
1735
  nodesWalkExpr(pCondition, getDBNameFromConditionWalker, (char*)dbName);
H
Haojun Liao 已提交
1736 1737
}

D
dapan1121 已提交
1738
static int32_t loadSysTableCallback(void* param, SDataBuf* pMsg, int32_t code) {
H
Haojun Liao 已提交
1739 1740 1741 1742 1743 1744 1745
  SOperatorInfo*     operator=(SOperatorInfo*) param;
  SSysTableScanInfo* pScanResInfo = (SSysTableScanInfo*)operator->info;
  if (TSDB_CODE_SUCCESS == code) {
    pScanResInfo->pRsp = pMsg->pData;

    SRetrieveMetaTableRsp* pRsp = pScanResInfo->pRsp;
    pRsp->numOfRows = htonl(pRsp->numOfRows);
1746 1747 1748
    pRsp->useconds = htobe64(pRsp->useconds);
    pRsp->handle = htobe64(pRsp->handle);
    pRsp->compLen = htonl(pRsp->compLen);
H
Haojun Liao 已提交
1749 1750 1751 1752 1753
  } else {
    operator->pTaskInfo->code = code;
  }

  tsem_post(&pScanResInfo->ready);
wmmhello's avatar
wmmhello 已提交
1754
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1755 1756 1757 1758 1759 1760 1761
}

static SSDataBlock* doFilterResult(SSysTableScanInfo* pInfo) {
  if (pInfo->pCondition == NULL) {
    return pInfo->pRes->info.rows == 0 ? NULL : pInfo->pRes;
  }

1762 1763
  doFilter(pInfo->pCondition, pInfo->pRes);
#if 0
H
Haojun Liao 已提交
1764
  SFilterInfo* filter = NULL;
1765 1766

  int32_t code = filterInitFromNode(pInfo->pCondition, &filter, 0);
H
Haojun Liao 已提交
1767 1768 1769 1770 1771

  SFilterColumnParam param1 = {.numOfCols = pInfo->pRes->info.numOfCols, .pDataBlock = pInfo->pRes->pDataBlock};
  code = filterSetDataFromSlotId(filter, &param1);

  int8_t* rowRes = NULL;
L
Liu Jicong 已提交
1772
  bool    keep = filterExecute(filter, pInfo->pRes, &rowRes, NULL, param1.numOfCols);
D
dapan1121 已提交
1773
  filterFreeInfo(filter);
H
Haojun Liao 已提交
1774

1775
  SSDataBlock* px = createOneDataBlock(pInfo->pRes, false);
H
Haojun Liao 已提交
1776 1777 1778 1779 1780 1781 1782 1783
  blockDataEnsureCapacity(px, pInfo->pRes->info.rows);

  // TODO refactor
  int32_t numOfRow = 0;
  for (int32_t i = 0; i < pInfo->pRes->info.numOfCols; ++i) {
    SColumnInfoData* pDest = taosArrayGet(px->pDataBlock, i);
    SColumnInfoData* pSrc = taosArrayGet(pInfo->pRes->pDataBlock, i);

D
dapan1121 已提交
1784
    if (keep) {
1785
      colDataAssign(pDest, pSrc, pInfo->pRes->info.rows, &px->info);
D
dapan1121 已提交
1786 1787 1788 1789 1790 1791 1792
      numOfRow = pInfo->pRes->info.rows;
    } else if (NULL != rowRes) {
      numOfRow = 0;
      for (int32_t j = 0; j < pInfo->pRes->info.rows; ++j) {
        if (rowRes[j] == 0) {
          continue;
        }
1793

1794 1795 1796 1797 1798 1799
        if (colDataIsNull_s(pSrc, j)) {
          colDataAppendNULL(pDest, numOfRow);
        } else {
          colDataAppend(pDest, numOfRow, colDataGetData(pSrc, j), false);
        }

D
dapan1121 已提交
1800
        numOfRow += 1;
H
Haojun Liao 已提交
1801
      }
D
dapan1121 已提交
1802 1803
    } else {
      numOfRow = 0;
H
Haojun Liao 已提交
1804 1805 1806 1807 1808
    }
  }

  px->info.rows = numOfRow;
  pInfo->pRes = px;
1809
#endif
H
Haojun Liao 已提交
1810 1811 1812 1813

  return pInfo->pRes->info.rows == 0 ? NULL : pInfo->pRes;
}

1814
static SSDataBlock* buildInfoSchemaTableMetaBlock(char* tableName) {
L
Liu Jicong 已提交
1815 1816
  size_t               size = 0;
  const SSysTableMeta* pMeta = NULL;
1817 1818 1819
  getInfosDbMeta(&pMeta, &size);

  int32_t index = 0;
L
Liu Jicong 已提交
1820
  for (int32_t i = 0; i < size; ++i) {
1821
    if (strcmp(pMeta[i].name, tableName) == 0) {
1822 1823 1824 1825
      index = i;
      break;
    }
  }
1826

1827
  SSDataBlock* pBlock = createDataBlock();
L
Liu Jicong 已提交
1828
  for (int32_t i = 0; i < pMeta[index].colNum; ++i) {
L
Liu Jicong 已提交
1829 1830
    SColumnInfoData colInfoData =
        createColumnInfoData(pMeta[index].schema[i].type, pMeta[index].schema[i].bytes, i + 1);
1831
    blockDataAppendColInfo(pBlock, &colInfoData);
1832 1833
  }

1834 1835 1836
  return pBlock;
}

S
shenglian zhou 已提交
1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861
static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;
  SSysTableScanInfo* pInfo = pOperator->info;
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  if (pInfo->pCur == NULL) {
    pInfo->pCur = metaOpenTbCursor(pInfo->readHandle.meta);
  }

  blockDataCleanup(pInfo->pRes);
  int32_t numOfRows = 0;

  const char* db = NULL;
  int32_t     vgId = 0;
  vnodeGetInfo(pInfo->readHandle.vnode, &db, &vgId);

  SName sn = {0};
  char  dbname[TSDB_DB_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
  tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);

  tNameGetDbName(&sn, varDataVal(dbname));
  varDataSetLen(dbname, strlen(varDataVal(dbname)));

1862
  SSDataBlock* p = buildInfoSchemaTableMetaBlock(TSDB_INS_TABLE_USER_TAGS);
S
shenglian zhou 已提交
1863 1864 1865 1866
  blockDataEnsureCapacity(p, pOperator->resultInfo.capacity);

  int32_t ret = 0;
  while ((ret = metaTbCursorNext(pInfo->pCur)) == 0) {
1867 1868 1869
    if (pInfo->pCur->mr.me.type != TSDB_CHILD_TABLE) {
      continue;
    }
S
shenglian zhou 已提交
1870

1871 1872
    char tableName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_TO_VARSTR(tableName, pInfo->pCur->mr.me.name);
S
shenglian zhou 已提交
1873

1874 1875
    SMetaReader smr = {0};
    metaReaderInit(&smr, pInfo->readHandle.meta, 0);
S
shenglian zhou 已提交
1876

1877 1878 1879 1880 1881 1882 1883 1884 1885 1886
    uint64_t suid = pInfo->pCur->mr.me.ctbEntry.suid;
    int32_t  code = metaGetTableEntryByUid(&smr, suid);
    if (code != TSDB_CODE_SUCCESS) {
      qError("failed to get super table meta, uid:0x%" PRIx64 ", code:%s, %s", suid, tstrerror(terrno),
             GET_TASKID(pTaskInfo));
      metaReaderClear(&smr);
      metaCloseTbCursor(pInfo->pCur);
      pInfo->pCur = NULL;
      longjmp(pTaskInfo->env, terrno);
    }
S
shenglian zhou 已提交
1887

1888 1889
    char stableName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_TO_VARSTR(stableName, smr.me.name);
S
shenglian zhou 已提交
1890

1891 1892 1893
    int32_t numOfTags = smr.me.stbEntry.schemaTag.nCols;
    for (int32_t i = 0; i < numOfTags; ++i) {
      SColumnInfoData* pColInfoData = NULL;
S
shenglian zhou 已提交
1894

1895 1896 1897
      // table name
      pColInfoData = taosArrayGet(p->pDataBlock, 0);
      colDataAppend(pColInfoData, numOfRows, tableName, false);
S
shenglian zhou 已提交
1898

1899 1900 1901
      // database name
      pColInfoData = taosArrayGet(p->pDataBlock, 1);
      colDataAppend(pColInfoData, numOfRows, dbname, false);
S
shenglian zhou 已提交
1902 1903 1904

      // super table name
      pColInfoData = taosArrayGet(p->pDataBlock, 2);
1905
      colDataAppend(pColInfoData, numOfRows, stableName, false);
S
shenglian zhou 已提交
1906

1907 1908
      char tagName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
      STR_TO_VARSTR(tagName, smr.me.stbEntry.schemaTag.pSchema[i].name);
S
shenglian zhou 已提交
1909
      pColInfoData = taosArrayGet(p->pDataBlock, 3);
1910
      colDataAppend(pColInfoData, numOfRows, tagName, false);
S
shenglian zhou 已提交
1911

1912
      int8_t tagType = smr.me.stbEntry.schemaTag.pSchema[i].type;
S
shenglian zhou 已提交
1913
      pColInfoData = taosArrayGet(p->pDataBlock, 4);
1914 1915 1916 1917
      colDataAppend(pColInfoData, numOfRows, (char*)&tagType, false);

      STagVal tagVal = {0};
      tagVal.cid = smr.me.stbEntry.schemaTag.pSchema[i].colId;
S
shenglian zhou 已提交
1918 1919
      char*    tagData = NULL;
      uint32_t tagLen = 0;
1920

S
shenglian zhou 已提交
1921
      if (tagType == TSDB_DATA_TYPE_JSON) {
1922
        tagData = (char*)pInfo->pCur->mr.me.ctbEntry.pTags;
S
shenglian zhou 已提交
1923
      } else {
S
shenglian zhou 已提交
1924 1925 1926 1927 1928 1929 1930 1931 1932 1933
        bool exist = tTagGet((STag*)pInfo->pCur->mr.me.ctbEntry.pTags, &tagVal);
        if (exist) {
          if (IS_VAR_DATA_TYPE(tagType)) {
            tagData = (char*)tagVal.pData;
            tagLen = tagVal.nData;
          } else {
            tagData = (char*)&tagVal.i64;
            tagLen = tDataTypes[tagType].bytes;
          }
        }
S
shenglian zhou 已提交
1934 1935
      }

1936
      char* tagVarChar = NULL;
S
shenglian zhou 已提交
1937
      if (tagData != NULL) {
1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950
        if (tagType == TSDB_DATA_TYPE_JSON) {
          char* tagJson = parseTagDatatoJson(tagData);
          tagVarChar = taosMemoryMalloc(strlen(tagJson) + VARSTR_HEADER_SIZE);
          memcpy(varDataVal(tagVarChar), tagJson, strlen(tagJson));
          varDataSetLen(tagVarChar, strlen(tagJson));
          taosMemoryFree(tagJson);
        } else {
          int32_t bufSize = IS_VAR_DATA_TYPE(tagType) ? (tagLen + VARSTR_HEADER_SIZE)
                                                      : (3 + DBL_MANT_DIG - DBL_MIN_EXP + VARSTR_HEADER_SIZE);
          tagVarChar = taosMemoryMalloc(bufSize);
          int32_t len = -1;
          dataConverToStr(varDataVal(tagVarChar), tagType, tagData, tagLen, &len);
          varDataSetLen(tagVarChar, len);
S
shenglian zhou 已提交
1951
        }
1952
      }
S
shenglian zhou 已提交
1953 1954 1955 1956
      pColInfoData = taosArrayGet(p->pDataBlock, 5);
      colDataAppend(pColInfoData, numOfRows, tagVarChar,
                    (tagData == NULL) || (tagType == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(tagData)));
      taosMemoryFree(tagVarChar);
1957
      ++numOfRows;
S
shenglian zhou 已提交
1958
    }
1959
    metaReaderClear(&smr);
S
shenglian zhou 已提交
1960

1961
    if (numOfRows >= pOperator->resultInfo.capacity) {
S
shenglian zhou 已提交
1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984
      break;
    }
  }

  // todo temporarily free the cursor here, the true reason why the free is not valid needs to be found
  if (ret != 0) {
    metaCloseTbCursor(pInfo->pCur);
    pInfo->pCur = NULL;
    doSetOperatorCompleted(pOperator);
  }

  p->info.rows = numOfRows;
  pInfo->pRes->info.rows = numOfRows;

  relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
  doFilterResult(pInfo);

  blockDataDestroy(p);

  pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
  return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
}

1985
static SSDataBlock* sysTableScanUserTables(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
1986 1987
  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;
  SSysTableScanInfo* pInfo = pOperator->info;
1988 1989 1990
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }
H
Haojun Liao 已提交
1991

1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003
  // the retrieve is executed on the mnode, so return tables that belongs to the information schema database.
  if (pInfo->readHandle.mnd != NULL) {
    buildSysDbTableInfo(pInfo, pOperator->resultInfo.capacity);

    doFilterResult(pInfo);
    pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;

    doSetOperatorCompleted(pOperator);
    return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
  } else {
    if (pInfo->pCur == NULL) {
      pInfo->pCur = metaOpenTbCursor(pInfo->readHandle.meta);
2004 2005
    }

2006 2007
    blockDataCleanup(pInfo->pRes);
    int32_t numOfRows = 0;
2008

2009 2010 2011
    const char* db = NULL;
    int32_t     vgId = 0;
    vnodeGetInfo(pInfo->readHandle.vnode, &db, &vgId);
2012

2013 2014 2015
    SName sn = {0};
    char  dbname[TSDB_DB_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
    tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);
2016

2017 2018
    tNameGetDbName(&sn, varDataVal(dbname));
    varDataSetLen(dbname, strlen(varDataVal(dbname)));
2019

2020 2021
    SSDataBlock* p = buildInfoSchemaTableMetaBlock(TSDB_INS_TABLE_USER_TABLES);
    blockDataEnsureCapacity(p, pOperator->resultInfo.capacity);
2022

2023
    char n[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
2024

2025 2026 2027
    int32_t ret = 0;
    while ((ret = metaTbCursorNext(pInfo->pCur)) == 0) {
      STR_TO_VARSTR(n, pInfo->pCur->mr.me.name);
2028

2029 2030 2031
      // table name
      SColumnInfoData* pColInfoData = taosArrayGet(p->pDataBlock, 0);
      colDataAppend(pColInfoData, numOfRows, n, false);
2032

2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053
      // database name
      pColInfoData = taosArrayGet(p->pDataBlock, 1);
      colDataAppend(pColInfoData, numOfRows, dbname, false);

      // vgId
      pColInfoData = taosArrayGet(p->pDataBlock, 6);
      colDataAppend(pColInfoData, numOfRows, (char*)&vgId, false);

      int32_t tableType = pInfo->pCur->mr.me.type;
      if (tableType == TSDB_CHILD_TABLE) {
        // create time
        int64_t ts = pInfo->pCur->mr.me.ctbEntry.ctime;
        pColInfoData = taosArrayGet(p->pDataBlock, 2);
        colDataAppend(pColInfoData, numOfRows, (char*)&ts, false);

        SMetaReader mr = {0};
        metaReaderInit(&mr, pInfo->readHandle.meta, 0);

        uint64_t suid = pInfo->pCur->mr.me.ctbEntry.suid;
        int32_t  code = metaGetTableEntryByUid(&mr, suid);
        if (code != TSDB_CODE_SUCCESS) {
D
dapan1121 已提交
2054 2055
          qError("failed to get super table meta, cname:%s, suid:0x%" PRIx64 ", code:%s, %s", 
                 pInfo->pCur->mr.me.name, suid, tstrerror(terrno), GET_TASKID(pTaskInfo));
2056 2057 2058 2059 2060
          metaReaderClear(&mr);
          metaCloseTbCursor(pInfo->pCur);
          pInfo->pCur = NULL;
          longjmp(pTaskInfo->env, terrno);
        }
2061

2062 2063 2064
        // number of columns
        pColInfoData = taosArrayGet(p->pDataBlock, 3);
        colDataAppend(pColInfoData, numOfRows, (char*)&mr.me.stbEntry.schemaRow.nCols, false);
2065

2066 2067 2068
        // super table name
        STR_TO_VARSTR(n, mr.me.name);
        pColInfoData = taosArrayGet(p->pDataBlock, 4);
2069
        colDataAppend(pColInfoData, numOfRows, n, false);
2070
        metaReaderClear(&mr);
2071

2072 2073 2074 2075 2076 2077 2078 2079 2080 2081 2082
        // table comment
        pColInfoData = taosArrayGet(p->pDataBlock, 8);
        if (pInfo->pCur->mr.me.ctbEntry.commentLen > 0) {
          char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
          STR_TO_VARSTR(comment, pInfo->pCur->mr.me.ctbEntry.comment);
          colDataAppend(pColInfoData, numOfRows, comment, false);
        } else if (pInfo->pCur->mr.me.ctbEntry.commentLen == 0) {
          char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
          STR_TO_VARSTR(comment, "");
          colDataAppend(pColInfoData, numOfRows, comment, false);
        } else {
2083 2084
          colDataAppendNULL(pColInfoData, numOfRows);
        }
2085

2086 2087 2088 2089 2090 2091 2092 2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103 2104 2105 2106
        // uid
        pColInfoData = taosArrayGet(p->pDataBlock, 5);
        colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.uid, false);

        // ttl
        pColInfoData = taosArrayGet(p->pDataBlock, 7);
        colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ctbEntry.ttlDays, false);

        STR_TO_VARSTR(n, "CHILD_TABLE");
      } else if (tableType == TSDB_NORMAL_TABLE) {
        // create time
        pColInfoData = taosArrayGet(p->pDataBlock, 2);
        colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.ctime, false);

        // number of columns
        pColInfoData = taosArrayGet(p->pDataBlock, 3);
        colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.schemaRow.nCols, false);

        // super table name
        pColInfoData = taosArrayGet(p->pDataBlock, 4);
        colDataAppendNULL(pColInfoData, numOfRows);
2107

2108 2109 2110 2111 2112 2113 2114 2115 2116 2117 2118 2119
        // table comment
        pColInfoData = taosArrayGet(p->pDataBlock, 8);
        if (pInfo->pCur->mr.me.ntbEntry.commentLen > 0) {
          char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
          STR_TO_VARSTR(comment, pInfo->pCur->mr.me.ntbEntry.comment);
          colDataAppend(pColInfoData, numOfRows, comment, false);
        } else if (pInfo->pCur->mr.me.ntbEntry.commentLen == 0) {
          char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
          STR_TO_VARSTR(comment, "");
          colDataAppend(pColInfoData, numOfRows, comment, false);
        } else {
          colDataAppendNULL(pColInfoData, numOfRows);
2120
        }
2121 2122 2123 2124 2125 2126 2127 2128 2129 2130

        // uid
        pColInfoData = taosArrayGet(p->pDataBlock, 5);
        colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.uid, false);

        // ttl
        pColInfoData = taosArrayGet(p->pDataBlock, 7);
        colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.ttlDays, false);

        STR_TO_VARSTR(n, "NORMAL_TABLE");
H
Haojun Liao 已提交
2131 2132
      }

2133 2134 2135 2136 2137
      pColInfoData = taosArrayGet(p->pDataBlock, 9);
      colDataAppend(pColInfoData, numOfRows, n, false);

      if (++numOfRows >= pOperator->resultInfo.capacity) {
        break;
2138
      }
2139
    }
2140

2141 2142 2143 2144 2145 2146
    // todo temporarily free the cursor here, the true reason why the free is not valid needs to be found
    if (ret != 0) {
      metaCloseTbCursor(pInfo->pCur);
      pInfo->pCur = NULL;
      doSetOperatorCompleted(pOperator);
    }
H
Haojun Liao 已提交
2147

2148 2149
    p->info.rows = numOfRows;
    pInfo->pRes->info.rows = numOfRows;
H
Haojun Liao 已提交
2150

2151 2152
    relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
    doFilterResult(pInfo);
2153

2154 2155 2156 2157 2158 2159 2160
    blockDataDestroy(p);

    pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
    return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
  }
}

2161 2162 2163 2164 2165 2166 2167 2168 2169

static SSDataBlock* sysTableScanUserSTables(SOperatorInfo* pOperator) {
  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;
  SSysTableScanInfo* pInfo = pOperator->info;
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  pInfo->pRes->info.rows = 0;
D
dapan1121 已提交
2170
  pOperator->status = OP_EXEC_DONE;
2171 2172 2173 2174 2175

  pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
  return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
}

2176 2177 2178 2179 2180 2181
static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
  // build message and send to mnode to fetch the content of system tables.
  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;
  SSysTableScanInfo* pInfo = pOperator->info;

  const char* name = tNameGetTableName(&pInfo->name);
D
dapan1121 已提交
2182 2183 2184 2185 2186 2187
  if (pInfo->showRewrite) {
    char dbName[TSDB_DB_NAME_LEN] = {0};
    getDBNameFromCondition(pInfo->pCondition, dbName);
    sprintf(pInfo->req.db, "%d.%s", pInfo->accountId, dbName);
  }
  
2188 2189 2190 2191
  if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0) {
    return sysTableScanUserTables(pOperator);
  } else if (strncasecmp(name, TSDB_INS_TABLE_USER_TAGS, TSDB_TABLE_FNAME_LEN) == 0) {
    return sysTableScanUserTags(pOperator);
D
dapan1121 已提交
2192
  } else if (strncasecmp(name, TSDB_INS_TABLE_USER_STABLES, TSDB_TABLE_FNAME_LEN) == 0 && IS_SYS_DBNAME(pInfo->req.db)) {
2193
    return sysTableScanUserSTables(pOperator);
H
Haojun Liao 已提交
2194 2195 2196 2197 2198
  } else {  // load the meta from mnode of the given epset
    if (pOperator->status == OP_EXEC_DONE) {
      return NULL;
    }

2199 2200 2201
    while (1) {
      int64_t startTs = taosGetTimestampUs();
      strncpy(pInfo->req.tb, tNameGetTableName(&pInfo->name), tListLen(pInfo->req.tb));
2202
      strcpy(pInfo->req.user, pInfo->pUser);
H
Haojun Liao 已提交
2203

2204 2205 2206 2207 2208 2209 2210 2211 2212 2213 2214
      int32_t contLen = tSerializeSRetrieveTableReq(NULL, 0, &pInfo->req);
      char*   buf1 = taosMemoryCalloc(1, contLen);
      tSerializeSRetrieveTableReq(buf1, contLen, &pInfo->req);

      // send the fetch remote task result reques
      SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
      if (NULL == pMsgSendInfo) {
        qError("%s prepare message %d failed", GET_TASKID(pTaskInfo), (int32_t)sizeof(SMsgSendInfo));
        pTaskInfo->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
        return NULL;
      }
H
Haojun Liao 已提交
2215

L
Liu Jicong 已提交
2216 2217
      int32_t msgType = (strcasecmp(name, TSDB_INS_TABLE_DNODE_VARIABLES) == 0) ? TDMT_DND_SYSTABLE_RETRIEVE
                                                                                : TDMT_MND_SYSTABLE_RETRIEVE;
D
dapan1121 已提交
2218

2219 2220 2221
      pMsgSendInfo->param = pOperator;
      pMsgSendInfo->msgInfo.pData = buf1;
      pMsgSendInfo->msgInfo.len = contLen;
D
dapan1121 已提交
2222
      pMsgSendInfo->msgType = msgType;
2223
      pMsgSendInfo->fp = loadSysTableCallback;
D
dapan1121 已提交
2224
      pMsgSendInfo->requestId = pTaskInfo->id.queryId;
H
Haojun Liao 已提交
2225

2226
      int64_t transporterId = 0;
2227 2228
      int32_t code =
          asyncSendMsgToServer(pInfo->readHandle.pMsgCb->clientRpc, &pInfo->epSet, &transporterId, pMsgSendInfo);
2229
      tsem_wait(&pInfo->ready);
H
Haojun Liao 已提交
2230

2231 2232 2233 2234 2235
      if (pTaskInfo->code) {
        qDebug("%s load meta data from mnode failed, totalRows:%" PRIu64 ", code:%s", GET_TASKID(pTaskInfo),
               pInfo->loadInfo.totalRows, tstrerror(pTaskInfo->code));
        return NULL;
      }
H
Haojun Liao 已提交
2236

2237 2238
      SRetrieveMetaTableRsp* pRsp = pInfo->pRsp;
      pInfo->req.showId = pRsp->handle;
H
Haojun Liao 已提交
2239

2240 2241
      if (pRsp->numOfRows == 0 || pRsp->completed) {
        pOperator->status = OP_EXEC_DONE;
2242
        qDebug("%s load meta data from mnode completed, rowsOfSource:%d, totalRows:%" PRIu64, GET_TASKID(pTaskInfo),
2243
               pRsp->numOfRows, pInfo->loadInfo.totalRows);
H
Haojun Liao 已提交
2244

2245
        if (pRsp->numOfRows == 0) {
H
Haojun Liao 已提交
2246
          taosMemoryFree(pRsp);
2247 2248 2249
          return NULL;
        }
      }
H
Haojun Liao 已提交
2250

2251
      extractDataBlockFromFetchRsp(pInfo->pRes, &pInfo->loadInfo, pRsp->numOfRows, pRsp->data, pRsp->compLen,
L
Liu Jicong 已提交
2252
                                   pOperator->exprSupp.numOfExprs, startTs, NULL, pInfo->scanCols);
H
Haojun Liao 已提交
2253

2254 2255
      // todo log the filter info
      doFilterResult(pInfo);
H
Haojun Liao 已提交
2256
      taosMemoryFree(pRsp);
2257 2258
      if (pInfo->pRes->info.rows > 0) {
        return pInfo->pRes;
D
dapan1121 已提交
2259 2260
      } else if (pOperator->status == OP_EXEC_DONE) {
        return NULL;
2261
      }
2262
    }
H
Haojun Liao 已提交
2263 2264 2265
  }
}

2266
int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity) {
2267
  SSDataBlock* p = buildInfoSchemaTableMetaBlock(TSDB_INS_TABLE_USER_TABLES);
2268
  blockDataEnsureCapacity(p, capacity);
2269

L
Liu Jicong 已提交
2270
  size_t               size = 0;
2271 2272 2273 2274 2275 2276 2277 2278 2279
  const SSysTableMeta* pSysDbTableMeta = NULL;

  getInfosDbMeta(&pSysDbTableMeta, &size);
  p->info.rows = buildDbTableInfoBlock(p, pSysDbTableMeta, size, TSDB_INFORMATION_SCHEMA_DB);

  getPerfDbMeta(&pSysDbTableMeta, &size);
  p->info.rows = buildDbTableInfoBlock(p, pSysDbTableMeta, size, TSDB_PERFORMANCE_SCHEMA_DB);

  pInfo->pRes->info.rows = p->info.rows;
2280
  relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
2281 2282 2283
  blockDataDestroy(p);

  return pInfo->pRes->info.rows;
2284 2285
}

L
Liu Jicong 已提交
2286 2287 2288
int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size,
                              const char* dbName) {
  char    n[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
2289 2290
  int32_t numOfRows = p->info.rows;

L
Liu Jicong 已提交
2291
  for (int32_t i = 0; i < size; ++i) {
2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311
    const SSysTableMeta* pm = &pSysDbTableMeta[i];

    SColumnInfoData* pColInfoData = taosArrayGet(p->pDataBlock, 0);

    STR_TO_VARSTR(n, pm->name);
    colDataAppend(pColInfoData, numOfRows, n, false);

    // database name
    STR_TO_VARSTR(n, dbName);
    pColInfoData = taosArrayGet(p->pDataBlock, 1);
    colDataAppend(pColInfoData, numOfRows, n, false);

    // create time
    pColInfoData = taosArrayGet(p->pDataBlock, 2);
    colDataAppendNULL(pColInfoData, numOfRows);

    // number of columns
    pColInfoData = taosArrayGet(p->pDataBlock, 3);
    colDataAppend(pColInfoData, numOfRows, (char*)&pm->colNum, false);

L
Liu Jicong 已提交
2312
    for (int32_t j = 4; j <= 8; ++j) {
2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327
      pColInfoData = taosArrayGet(p->pDataBlock, j);
      colDataAppendNULL(pColInfoData, numOfRows);
    }

    STR_TO_VARSTR(n, "SYSTEM_TABLE");

    pColInfoData = taosArrayGet(p->pDataBlock, 9);
    colDataAppend(pColInfoData, numOfRows, n, false);

    numOfRows += 1;
  }

  return numOfRows;
}

2328
SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode* pScanPhyNode,
2329
                                              const char* pUser, SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
2330 2331 2332
  SSysTableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SSysTableScanInfo));
  SOperatorInfo*     pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
2333
    goto _error;
H
Haojun Liao 已提交
2334 2335
  }

2336 2337 2338
  SScanPhysiNode* pScanNode = &pScanPhyNode->scan;

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;
2339
  SSDataBlock*        pResBlock = createResDataBlock(pDescNode);
2340 2341

  int32_t num = 0;
2342
  SArray* colList = extractColMatchInfo(pScanNode->pScanCols, pDescNode, &num, COL_MATCH_FROM_COL_ID);
2343

2344 2345
  pInfo->accountId = pScanPhyNode->accountId;
  pInfo->pUser = taosMemoryStrDup((void*)pUser);
2346
  pInfo->showRewrite = pScanPhyNode->showRewrite;
2347 2348 2349
  pInfo->pRes = pResBlock;
  pInfo->pCondition = pScanNode->node.pConditions;
  pInfo->scanCols = colList;
2350

2351
  initResultSizeInfo(&pOperator->resultInfo, 4096);
H
Haojun Liao 已提交
2352

2353
  tNameAssign(&pInfo->name, &pScanNode->tableName);
2354
  const char* name = tNameGetTableName(&pInfo->name);
2355

2356 2357
  if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0 ||
      strncasecmp(name, TSDB_INS_TABLE_USER_TAGS, TSDB_TABLE_FNAME_LEN) == 0) {
L
Liu Jicong 已提交
2358
    pInfo->readHandle = *(SReadHandle*)readHandle;
2359
    blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
H
Haojun Liao 已提交
2360 2361
  } else {
    tsem_init(&pInfo->ready, 0, 0);
2362
    pInfo->epSet = pScanPhyNode->mgmtEpSet;
2363
    pInfo->readHandle = *(SReadHandle*)readHandle;
H
Haojun Liao 已提交
2364 2365
  }

2366
  pOperator->name = "SysTableScanOperator";
H
Haojun Liao 已提交
2367
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN;
2368 2369 2370
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
2371
  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pResBlock->pDataBlock);
2372
  pOperator->pTaskInfo = pTaskInfo;
2373

L
Liu Jicong 已提交
2374 2375
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doSysTableScan, NULL, NULL, destroySysScanOperator, NULL, NULL, NULL);
H
Haojun Liao 已提交
2376 2377

  return pOperator;
2378

2379
_error:
2380 2381 2382 2383
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
  return NULL;
H
Haojun Liao 已提交
2384
}
H
Haojun Liao 已提交
2385

2386
static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
2387 2388 2389 2390
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

2391 2392 2393
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

#if 0
H
Haojun Liao 已提交
2394 2395 2396 2397 2398 2399 2400 2401
  int32_t maxNumOfTables = (int32_t)pResultInfo->capacity;

  STagScanInfo *pInfo = pOperator->info;
  SSDataBlock  *pRes = pInfo->pRes;

  int32_t count = 0;
  SArray* pa = GET_TABLEGROUP(pRuntimeEnv, 0);

2402
  int32_t functionId = getExprFunctionId(&pOperator->exprSupp.pExprInfo[0]);
H
Haojun Liao 已提交
2403 2404 2405
  if (functionId == FUNCTION_TID_TAG) { // return the tags & table Id
    assert(pQueryAttr->numOfOutput == 1);

2406
    SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[0];
H
Haojun Liao 已提交
2407 2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423 2424 2425 2426 2427 2428 2429 2430 2431 2432 2433 2434 2435 2436 2437 2438 2439 2440 2441 2442 2443 2444 2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463 2464 2465
    int32_t rsize = pExprInfo->base.resSchema.bytes;

    count = 0;

    int16_t bytes = pExprInfo->base.resSchema.bytes;
    int16_t type  = pExprInfo->base.resSchema.type;

    for(int32_t i = 0; i < pQueryAttr->numOfTags; ++i) {
      if (pQueryAttr->tagColList[i].colId == pExprInfo->base.pColumns->info.colId) {
        bytes = pQueryAttr->tagColList[i].bytes;
        type = pQueryAttr->tagColList[i].type;
        break;
      }
    }

    SColumnInfoData* pColInfo = taosArrayGet(pRes->pDataBlock, 0);

    while(pInfo->curPos < pInfo->totalTables && count < maxNumOfTables) {
      int32_t i = pInfo->curPos++;
      STableQueryInfo *item = taosArrayGetP(pa, i);

      char *output = pColInfo->pData + count * rsize;
      varDataSetLen(output, rsize - VARSTR_HEADER_SIZE);

      output = varDataVal(output);
      STableId* id = TSDB_TABLEID(item->pTable);

      *(int16_t *)output = 0;
      output += sizeof(int16_t);

      *(int64_t *)output = id->uid;  // memory align problem, todo serialize
      output += sizeof(id->uid);

      *(int32_t *)output = id->tid;
      output += sizeof(id->tid);

      *(int32_t *)output = pQueryAttr->vgId;
      output += sizeof(pQueryAttr->vgId);

      char* data = NULL;
      if (pExprInfo->base.pColumns->info.colId == TSDB_TBNAME_COLUMN_INDEX) {
        data = tsdbGetTableName(item->pTable);
      } else {
        data = tsdbGetTableTagVal(item->pTable, pExprInfo->base.pColumns->info.colId, type, bytes);
      }

      doSetTagValueToResultBuf(output, data, type, bytes);
      count += 1;
    }

    //qDebug("QInfo:0x%"PRIx64" create (tableId, tag) info completed, rows:%d", GET_TASKID(pRuntimeEnv), count);
  } else if (functionId == FUNCTION_COUNT) {// handle the "count(tbname)" query
    SColumnInfoData* pColInfo = taosArrayGet(pRes->pDataBlock, 0);
    *(int64_t*)pColInfo->pData = pInfo->totalTables;
    count = 1;

    pOperator->status = OP_EXEC_DONE;
    //qDebug("QInfo:0x%"PRIx64" create count(tbname) query, res:%d rows:1", GET_TASKID(pRuntimeEnv), count);
  } else {  // return only the tags|table name etc.
2466
#endif
H
Haojun Liao 已提交
2467

2468
  STagScanInfo* pInfo = pOperator->info;
2469
  SExprInfo*    pExprInfo = &pOperator->exprSupp.pExprInfo[0];
2470
  SSDataBlock*  pRes = pInfo->pRes;
H
Haojun Liao 已提交
2471

wmmhello's avatar
wmmhello 已提交
2472 2473
  int32_t size = taosArrayGetSize(pInfo->pTableList->pTableList);
  if (size == 0) {
H
Haojun Liao 已提交
2474 2475 2476 2477
    setTaskStatus(pTaskInfo, TASK_COMPLETED);
    return NULL;
  }

2478 2479 2480
  char        str[512] = {0};
  int32_t     count = 0;
  SMetaReader mr = {0};
2481
  metaReaderInit(&mr, pInfo->readHandle.meta, 0);
H
Haojun Liao 已提交
2482

wmmhello's avatar
wmmhello 已提交
2483 2484
  while (pInfo->curPos < size && count < pOperator->resultInfo.capacity) {
    STableKeyInfo* item = taosArrayGet(pInfo->pTableList->pTableList, pInfo->curPos);
L
Liu Jicong 已提交
2485
    int32_t        code = metaGetTableEntryByUid(&mr, item->uid);
H
Haojun Liao 已提交
2486
    if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
2487 2488
      qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", item->uid, tstrerror(terrno),
             GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
2489 2490 2491
      metaReaderClear(&mr);
      longjmp(pTaskInfo->env, terrno);
    }
H
Haojun Liao 已提交
2492

2493
    for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {
2494 2495 2496 2497 2498 2499
      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);
2500
      } else {  // it is a tag value
wmmhello's avatar
wmmhello 已提交
2501 2502 2503 2504
        STagVal val = {0};
        val.cid = pExprInfo[j].base.pParam[0].pCol->colId;
        const char* p = metaGetTableTagVal(&mr.me, pDst->info.type, &val);

2505 2506 2507 2508
        char* data = NULL;
        if (pDst->info.type != TSDB_DATA_TYPE_JSON && p != NULL) {
          data = tTagValToData((const STagVal*)p, false);
        } else {
wmmhello's avatar
wmmhello 已提交
2509 2510
          data = (char*)p;
        }
L
Liu Jicong 已提交
2511 2512
        colDataAppend(pDst, count, data,
                      (data == NULL) || (pDst->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data)));
2513

2514 2515
        if (pDst->info.type != TSDB_DATA_TYPE_JSON && p != NULL && IS_VAR_DATA_TYPE(((const STagVal*)p)->type) &&
            data != NULL) {
wmmhello's avatar
wmmhello 已提交
2516
          taosMemoryFree(data);
wmmhello's avatar
wmmhello 已提交
2517
        }
H
Haojun Liao 已提交
2518 2519 2520
      }
    }

2521
    count += 1;
wmmhello's avatar
wmmhello 已提交
2522
    if (++pInfo->curPos >= size) {
2523
      doSetOperatorCompleted(pOperator);
H
Haojun Liao 已提交
2524 2525 2526
    }
  }

2527 2528
  metaReaderClear(&mr);

2529
  // qDebug("QInfo:0x%"PRIx64" create tag values results completed, rows:%d", GET_TASKID(pRuntimeEnv), count);
H
Haojun Liao 已提交
2530
  if (pOperator->status == OP_EXEC_DONE) {
2531
    setTaskStatus(pTaskInfo, TASK_COMPLETED);
H
Haojun Liao 已提交
2532 2533 2534
  }

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

2537
  return (pRes->info.rows == 0) ? NULL : pInfo->pRes;
H
Haojun Liao 已提交
2538 2539 2540 2541 2542
}

static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
  STagScanInfo* pInfo = (STagScanInfo*)param;
  pInfo->pRes = blockDataDestroy(pInfo->pRes);
L
Liu Jicong 已提交
2543

D
dapan1121 已提交
2544
  taosMemoryFreeClear(param);
H
Haojun Liao 已提交
2545 2546
}

2547 2548
SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode,
                                         STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo) {
2549
  STagScanInfo*  pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo));
H
Haojun Liao 已提交
2550 2551 2552 2553 2554
  SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

2555 2556
  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

2557
  int32_t    num = 0;
2558 2559
  int32_t    numOfExprs = 0;
  SExprInfo* pExprInfo = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &numOfExprs);
L
Liu Jicong 已提交
2560
  SArray*    colList = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID);
2561

2562 2563 2564 2565
  int32_t code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2566

2567 2568 2569 2570 2571
  pInfo->pTableList = pTableListInfo;
  pInfo->pColMatchInfo = colList;
  pInfo->pRes = createResDataBlock(pDescNode);
  pInfo->readHandle = *pReadHandle;
  pInfo->curPos = 0;
2572

2573
  pOperator->name = "TagScanOperator";
2574
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN;
2575

2576 2577 2578 2579
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
  pOperator->pTaskInfo = pTaskInfo;
2580

2581
  initResultSizeInfo(&pOperator->resultInfo, 4096);
2582 2583
  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

2584 2585
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doTagScan, NULL, NULL, destroyTagScanOperatorInfo, NULL, NULL, NULL);
H
Haojun Liao 已提交
2586 2587

  return pOperator;
2588

2589
_error:
H
Haojun Liao 已提交
2590 2591 2592 2593 2594
  taosMemoryFree(pInfo);
  taosMemoryFree(pOperator);
  terrno = TSDB_CODE_OUT_OF_MEMORY;
  return NULL;
}
2595 2596

typedef struct STableMergeScanInfo {
2597
  STableListInfo* tableListInfo;
S
slzhou 已提交
2598 2599 2600 2601
  int32_t         tableStartIndex;
  int32_t         tableEndIndex;
  bool            hasGroupId;
  uint64_t        groupId;
2602 2603 2604 2605 2606 2607
  SArray*         dataReaders;  // array of tsdbReaderT*
  SReadHandle     readHandle;
  int32_t         bufPageSize;
  uint32_t        sortBufSize;  // max buffer size for in-memory sort
  SArray*         pSortInfo;
  SSortHandle*    pSortHandle;
2608

2609 2610 2611
  SSDataBlock* pSortInputBlock;
  int64_t      startTs;  // sort start time
  SArray*      sortSourceParams;
2612 2613

  SFileBlockLoadRecorder readRecorder;
2614 2615 2616 2617 2618 2619 2620 2621 2622 2623 2624
  int64_t                numOfRows;
  SScanInfo              scanInfo;
  int32_t                scanTimes;
  SNode*                 pFilterNode;  // filter info, which is push down by optimizer
  SqlFunctionCtx*        pCtx;         // which belongs to the direct upstream operator operator query context
  SResultRowInfo*        pResultRowInfo;
  int32_t*               rowEntryInfoOffset;
  SExprInfo*             pExpr;
  SSDataBlock*           pResBlock;
  SArray*                pColMatchInfo;
  int32_t                numOfOutput;
2625 2626 2627 2628 2629 2630

  SExprInfo*      pPseudoExpr;
  int32_t         numOfPseudoExpr;
  SqlFunctionCtx* pPseudoCtx;

  SQueryTableDataCond cond;
2631 2632
  int32_t             scanFlag;  // table scan flag to denote if it is a repeat/reverse/main scan
  int32_t             dataBlockLoadFlag;
2633 2634 2635
  // if the upstream is an interval operator, the interval info is also kept here to get the time
  // window to check if current data block needs to be loaded.
  SInterval       interval;
2636 2637 2638
  SSampleExecInfo sample;  // sample execution info
} STableMergeScanInfo;

H
Haojun Liao 已提交
2639
int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
2640 2641
                                STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond,
                                const char* idStr) {
2642
  int32_t code = getTableList(pHandle->meta, pHandle->vnode, pScanNode, pTagCond, pTagIndexCond, pTableListInfo);
2643
  if (code != TSDB_CODE_SUCCESS) {
2644
    return code;
2645 2646 2647
  }

  if (taosArrayGetSize(pTableListInfo->pTableList) == 0) {
2648
    qDebug("no table qualified for query, %s" PRIx64, idStr);
2649 2650
    return TSDB_CODE_SUCCESS;
  }
2651

H
Haojun Liao 已提交
2652 2653
  pTableListInfo->needSortTableByGroupId = groupSort;
  code = generateGroupIdMap(pTableListInfo, pHandle, pGroupTags);
2654
  if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2655
    return code;
2656 2657 2658 2659 2660
  }

  return TSDB_CODE_SUCCESS;
}

S
slzhou 已提交
2661
int32_t createMultipleDataReaders(SQueryTableDataCond* pQueryCond, SReadHandle* pHandle, STableListInfo* pTableListInfo,
H
Haojun Liao 已提交
2662
                                  int32_t tableStartIdx, int32_t tableEndIdx, SArray* arrayReader, const char* idstr) {
S
slzhou 已提交
2663 2664 2665 2666
  for (int32_t i = tableStartIdx; i <= tableEndIdx; ++i) {
    SArray* subTableList = taosArrayInit(1, sizeof(STableKeyInfo));
    taosArrayPush(subTableList, taosArrayGet(pTableListInfo->pTableList, i));

H
Haojun Liao 已提交
2667 2668
    STsdbReader* pReader = NULL;
    tsdbReaderOpen(pHandle->vnode, pQueryCond, subTableList, &pReader, idstr);
S
slzhou 已提交
2669 2670 2671 2672 2673 2674 2675 2676
    taosArrayPush(arrayReader, &pReader);

    taosArrayDestroy(subTableList);
  }

  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
2677
// todo refactor
2678 2679
static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeScanInfo* pTableScanInfo,
                                         int32_t readerIdx, SSDataBlock* pBlock, uint32_t* status) {
L
Liu Jicong 已提交
2680
  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;
S
shenglian zhou 已提交
2681
  STableMergeScanInfo* pInfo = pOperator->info;
2682 2683 2684 2685 2686 2687 2688 2689 2690 2691 2692 2693 2694 2695 2696 2697 2698 2699 2700 2701 2702 2703 2704 2705 2706 2707

  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;

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

  *status = pInfo->dataBlockLoadFlag;
  if (pTableScanInfo->pFilterNode != NULL ||
      overlapWithTimeWindow(&pTableScanInfo->interval, &pBlock->info, pTableScanInfo->cond.order)) {
    (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;
  }

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

  if (*status == FUNC_DATA_REQUIRED_FILTEROUT) {
    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
    pCost->filterOutBlocks += 1;
    return TSDB_CODE_SUCCESS;
  } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {
    qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
    pCost->skipBlocks += 1;

    // clear all data in pBlock that are set when handing the previous block
2708
    for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
2709 2710 2711 2712 2713 2714 2715 2716 2717 2718
      SColumnInfoData* pcol = taosArrayGet(pBlock->pDataBlock, i);
      pcol->pData = NULL;
    }

    return TSDB_CODE_SUCCESS;
  } else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) {
    pCost->loadBlockStatis += 1;

    bool             allColumnsHaveAgg = true;
    SColumnDataAgg** pColAgg = NULL;
H
Hongze Cheng 已提交
2719
    STsdbReader*     reader = taosArrayGetP(pTableScanInfo->dataReaders, readerIdx);
2720
    tsdbRetrieveDatablockSMA(reader, &pColAgg, &allColumnsHaveAgg);
2721 2722

    if (allColumnsHaveAgg == true) {
2723
      int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
2724 2725 2726 2727 2728 2729 2730 2731 2732 2733 2734 2735 2736 2737 2738 2739 2740 2741 2742 2743 2744 2745 2746 2747 2748 2749 2750 2751 2752 2753 2754 2755 2756 2757 2758 2759

      // todo create this buffer during creating operator
      if (pBlock->pBlockAgg == NULL) {
        pBlock->pBlockAgg = taosMemoryCalloc(numOfCols, POINTER_BYTES);
      }

      for (int32_t i = 0; i < numOfCols; ++i) {
        SColMatchInfo* pColMatchInfo = taosArrayGet(pTableScanInfo->pColMatchInfo, i);
        if (!pColMatchInfo->output) {
          continue;
        }
        pBlock->pBlockAgg[pColMatchInfo->targetSlotId] = pColAgg[i];
      }

      return TSDB_CODE_SUCCESS;
    } else {  // failed to load the block sma data, data block statistics does not exist, load data block instead
      *status = FUNC_DATA_REQUIRED_DATA_LOAD;
    }
  }

  ASSERT(*status == FUNC_DATA_REQUIRED_DATA_LOAD);

  // todo filter data block according to the block sma data firstly
#if 0
  if (!doFilterByBlockStatistics(pBlock->pBlockStatis, pTableScanInfo->pCtx, pBlockInfo->rows)) {
    pCost->filterOutBlocks += 1;
    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo), pBlockInfo->window.skey,
           pBlockInfo->window.ekey, pBlockInfo->rows);
    (*status) = FUNC_DATA_REQUIRED_FILTEROUT;
    return TSDB_CODE_SUCCESS;
  }
#endif

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

H
Hongze Cheng 已提交
2760
  STsdbReader* reader = taosArrayGetP(pTableScanInfo->dataReaders, readerIdx);
2761 2762 2763 2764 2765
  SArray*      pCols = tsdbRetrieveDataBlock(reader, NULL);
  if (pCols == NULL) {
    return terrno;
  }

2766
  relocateColumnData(pBlock, pTableScanInfo->pColMatchInfo, pCols, true);
2767 2768 2769

  // currently only the tbname pseudo column
  if (pTableScanInfo->numOfPseudoExpr > 0) {
L
Liu Jicong 已提交
2770 2771
    int32_t code = addTagPseudoColumnData(&pTableScanInfo->readHandle, pTableScanInfo->pPseudoExpr,
                                          pTableScanInfo->numOfPseudoExpr, pBlock, GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
2772 2773 2774
    if (code != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, code);
    }
2775 2776 2777 2778 2779 2780 2781 2782 2783 2784 2785 2786 2787 2788 2789 2790 2791 2792 2793 2794
  }

  int64_t st = taosGetTimestampMs();
  doFilter(pTableScanInfo->pFilterNode, pBlock);

  int64_t et = taosGetTimestampMs();
  pTableScanInfo->readRecorder.filterTime += (et - st);

  if (pBlock->info.rows == 0) {
    pCost->filterOutBlocks += 1;
    qDebug("%s data block filter out, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
           pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
  }

  return TSDB_CODE_SUCCESS;
}

typedef struct STableMergeScanSortSourceParam {
  SOperatorInfo* pOperator;
  int32_t        readerIdx;
2795
  SSDataBlock*   inputBlock;
2796 2797 2798 2799 2800 2801
} STableMergeScanSortSourceParam;

static SSDataBlock* getTableDataBlock(void* param) {
  STableMergeScanSortSourceParam* source = param;
  SOperatorInfo*                  pOperator = source->pOperator;
  int32_t                         readerIdx = source->readerIdx;
2802
  SSDataBlock*                    pBlock = source->inputBlock;
2803 2804 2805 2806
  STableMergeScanInfo*            pTableScanInfo = pOperator->info;

  int64_t st = taosGetTimestampUs();

2807 2808
  blockDataCleanup(pBlock);

H
Hongze Cheng 已提交
2809
  STsdbReader* reader = taosArrayGetP(pTableScanInfo->dataReaders, readerIdx);
2810 2811 2812 2813 2814 2815 2816 2817 2818 2819 2820
  while (tsdbNextDataBlock(reader)) {
    if (isTaskKilled(pOperator->pTaskInfo)) {
      longjmp(pOperator->pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
    }

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

2821 2822 2823 2824
    blockDataCleanup(pBlock);
    SDataBlockInfo binfo = pBlock->info;
    tsdbRetrieveDataBlockInfo(reader, &binfo);

2825
    blockDataEnsureCapacity(pBlock, binfo.rows);
2826 2827 2828 2829
    pBlock->info.type = binfo.type;
    pBlock->info.uid = binfo.uid;
    pBlock->info.window = binfo.window;
    pBlock->info.rows = binfo.rows;
2830 2831 2832 2833 2834 2835 2836 2837 2838 2839 2840 2841 2842 2843 2844 2845 2846 2847 2848 2849 2850 2851 2852 2853 2854 2855

    uint32_t status = 0;
    int32_t  code = loadDataBlockFromOneTable(pOperator, pTableScanInfo, readerIdx, pBlock, &status);
    //    int32_t  code = loadDataBlockOnDemand(pOperator->pRuntimeEnv, pTableScanInfo, pBlock, &status);
    if (code != TSDB_CODE_SUCCESS) {
      longjmp(pOperator->pTaskInfo->env, code);
    }

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

    uint64_t* groupId = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &pBlock->info.uid, sizeof(int64_t));
    if (groupId) {
      pBlock->info.groupId = *groupId;
    }

    pOperator->resultInfo.totalRows = pTableScanInfo->readRecorder.totalRows;
    pTableScanInfo->readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;

    return pBlock;
  }
  return NULL;
}

2856 2857 2858 2859 2860 2861 2862 2863 2864
SArray* generateSortByTsInfo(SArray* colMatchInfo, int32_t order) {
  int32_t tsTargetSlotId = 0;
  for (int32_t i = 0; i < taosArrayGetSize(colMatchInfo); ++i) {
    SColMatchInfo* colInfo = taosArrayGet(colMatchInfo, i);
    if (colInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
      tsTargetSlotId = colInfo->targetSlotId;
    }
  }

2865 2866 2867
  SArray*         pList = taosArrayInit(1, sizeof(SBlockOrderInfo));
  SBlockOrderInfo bi = {0};
  bi.order = order;
2868
  bi.slotId = tsTargetSlotId;
2869 2870 2871 2872 2873 2874 2875
  bi.nullFirst = NULL_ORDER_FIRST;

  taosArrayPush(pList, &bi);

  return pList;
}

2876
int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {
2877 2878 2879
  STableMergeScanInfo* pInfo = pOperator->info;
  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

S
slzhou 已提交
2880 2881 2882 2883 2884 2885 2886 2887 2888 2889 2890
  {
    size_t  tableListSize = taosArrayGetSize(pInfo->tableListInfo->pTableList);
    int32_t i = pInfo->tableStartIndex + 1;
    for (; i < tableListSize; ++i) {
      STableKeyInfo* tableKeyInfo = taosArrayGet(pInfo->tableListInfo->pTableList, i);
      if (tableKeyInfo->groupId != pInfo->groupId) {
        break;
      }
    }
    pInfo->tableEndIndex = i - 1;
  }
2891

S
slzhou 已提交
2892 2893
  int32_t tableStartIdx = pInfo->tableStartIndex;
  int32_t tableEndIdx = pInfo->tableEndIndex;
2894

S
slzhou 已提交
2895 2896
  STableListInfo* tableListInfo = pInfo->tableListInfo;
  createMultipleDataReaders(&pInfo->cond, &pInfo->readHandle, tableListInfo, tableStartIdx, tableEndIdx,
2897
                            pInfo->dataReaders, GET_TASKID(pTaskInfo));
2898

2899 2900
  // 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 已提交
2901
  pInfo->sortBufSize = pInfo->bufPageSize * (tableEndIdx - tableStartIdx + 1 + 1);
2902
  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
L
Liu Jicong 已提交
2903 2904
  pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage,
                                             pInfo->pSortInputBlock, pTaskInfo->id.str);
2905 2906 2907 2908

  tsortSetFetchRawDataFp(pInfo->pSortHandle, getTableDataBlock, NULL, NULL);

  size_t numReaders = taosArrayGetSize(pInfo->dataReaders);
2909 2910 2911 2912 2913 2914 2915 2916
  for (int32_t i = 0; i < numReaders; ++i) {
    STableMergeScanSortSourceParam param = {0};
    param.readerIdx = i;
    param.pOperator = pOperator;
    param.inputBlock = createOneDataBlock(pInfo->pResBlock, false);
    taosArrayPush(pInfo->sortSourceParams, &param);
  }

2917 2918
  for (int32_t i = 0; i < numReaders; ++i) {
    SSortSource*                    ps = taosMemoryCalloc(1, sizeof(SSortSource));
2919
    STableMergeScanSortSourceParam* param = taosArrayGet(pInfo->sortSourceParams, i);
2920 2921 2922 2923 2924 2925 2926 2927 2928 2929
    ps->param = param;
    tsortAddSource(pInfo->pSortHandle, ps);
  }

  int32_t code = tsortOpen(pInfo->pSortHandle);

  if (code != TSDB_CODE_SUCCESS) {
    longjmp(pTaskInfo->env, terrno);
  }

2930 2931 2932 2933 2934 2935 2936 2937 2938 2939 2940
  return TSDB_CODE_SUCCESS;
}

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

  tsortDestroySortHandle(pInfo->pSortHandle);
  taosArrayClear(pInfo->sortSourceParams);

  for (int32_t i = 0; i < taosArrayGetSize(pInfo->dataReaders); ++i) {
H
Haojun Liao 已提交
2941 2942
    STsdbReader* reader = taosArrayGetP(pInfo->dataReaders, i);
    tsdbReaderClose(reader);
2943
  }
2944

2945 2946
  taosArrayDestroy(pInfo->dataReaders);
  pInfo->dataReaders = NULL;
2947 2948 2949
  return TSDB_CODE_SUCCESS;
}

2950
SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, int32_t capacity, SOperatorInfo* pOperator) {
2951 2952 2953 2954 2955 2956 2957 2958 2959 2960 2961
  STableMergeScanInfo* pInfo = pOperator->info;
  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

  SSDataBlock* p = tsortGetSortedDataBlock(pHandle);
  if (p == NULL) {
    return NULL;
  }

  blockDataEnsureCapacity(p, capacity);

  while (1) {
2962
    STupleHandle* pTupleHandle = tsortNextTuple(pHandle);
2963 2964 2965 2966
    if (pTupleHandle == NULL) {
      break;
    }

2967
    appendOneRowToDataBlock(p, pTupleHandle);
2968 2969 2970 2971 2972
    if (p->info.rows >= capacity) {
      break;
    }
  }

2973 2974
  qDebug("%s get sorted row blocks, rows:%d", GET_TASKID(pTaskInfo), p->info.rows);
  return (p->info.rows > 0) ? p : NULL;
2975 2976 2977 2978 2979 2980 2981 2982 2983 2984 2985 2986 2987 2988
}

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) {
    longjmp(pTaskInfo->env, code);
  }
S
slzhou 已提交
2989 2990 2991
  size_t tableListSize = taosArrayGetSize(pInfo->tableListInfo->pTableList);
  if (!pInfo->hasGroupId) {
    pInfo->hasGroupId = true;
2992

S
slzhou 已提交
2993
    if (tableListSize == 0) {
2994 2995 2996
      doSetOperatorCompleted(pOperator);
      return NULL;
    }
S
slzhou 已提交
2997 2998
    pInfo->tableStartIndex = 0;
    pInfo->groupId = ((STableKeyInfo*)taosArrayGet(pInfo->tableListInfo->pTableList, pInfo->tableStartIndex))->groupId;
2999 3000
    startGroupTableMergeScan(pOperator);
  }
S
slzhou 已提交
3001 3002 3003 3004 3005 3006 3007 3008 3009 3010 3011 3012 3013 3014 3015 3016 3017 3018
  SSDataBlock* pBlock = NULL;
  while (pInfo->tableStartIndex < tableListSize) {
    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator);
    if (pBlock != NULL) {
      pBlock->info.groupId = pInfo->groupId;
      pOperator->resultInfo.totalRows += pBlock->info.rows;
      return pBlock;
    } else {
      stopGroupTableMergeScan(pOperator);
      if (pInfo->tableEndIndex >= tableListSize - 1) {
        doSetOperatorCompleted(pOperator);
        break;
      }
      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;
      pInfo->groupId =
          ((STableKeyInfo*)taosArrayGet(pInfo->tableListInfo->pTableList, pInfo->tableStartIndex))->groupId;
      startGroupTableMergeScan(pOperator);
    }
wmmhello's avatar
wmmhello 已提交
3019 3020
  }

3021 3022 3023 3024 3025
  return pBlock;
}

void destroyTableMergeScanOperatorInfo(void* param, int32_t numOfOutput) {
  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;
3026
  cleanupQueryTableDataCond(&pTableScanInfo->cond);
3027 3028

  for (int32_t i = 0; i < taosArrayGetSize(pTableScanInfo->dataReaders); ++i) {
H
Hongze Cheng 已提交
3029
    STsdbReader* reader = taosArrayGetP(pTableScanInfo->dataReaders, i);
H
refact  
Hongze Cheng 已提交
3030
    tsdbReaderClose(reader);
3031 3032 3033 3034 3035 3036 3037 3038 3039 3040 3041
  }
  taosArrayDestroy(pTableScanInfo->dataReaders);

  if (pTableScanInfo->pColMatchInfo != NULL) {
    taosArrayDestroy(pTableScanInfo->pColMatchInfo);
  }

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

  taosArrayDestroy(pTableScanInfo->pSortInfo);
L
Liu Jicong 已提交
3042

D
dapan1121 已提交
3043
  taosMemoryFreeClear(param);
3044 3045
}

3046 3047
typedef struct STableMergeScanExecInfo {
  SFileBlockLoadRecorder blockRecorder;
L
Liu Jicong 已提交
3048
  SSortExecInfo          sortExecInfo;
3049 3050
} STableMergeScanExecInfo;

3051 3052
int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {
  ASSERT(pOptr != NULL);
3053 3054
  // TODO: merge these two info into one struct
  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));
L
Liu Jicong 已提交
3055
  STableMergeScanInfo*     pInfo = pOptr->info;
3056 3057 3058 3059 3060
  execInfo->blockRecorder = pInfo->readRecorder;
  execInfo->sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

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

3062 3063 3064
  return TSDB_CODE_SUCCESS;
}

S
slzhou 已提交
3065 3066 3067 3068 3069 3070
int32_t compareTableKeyInfoByGid(const void* p1, const void* p2) {
  const STableKeyInfo* info1 = p1;
  const STableKeyInfo* info2 = p2;
  return info1->groupId - info2->groupId;
}

3071
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo* pTableListInfo,
3072
                                                SReadHandle* readHandle, SExecTaskInfo* pTaskInfo) {
3073 3074 3075 3076 3077
  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));
  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }
3078
  if (pTableScanNode->pGroupTags) {
S
slzhou 已提交
3079 3080
    taosArraySort(pTableListInfo->pTableList, compareTableKeyInfoByGid);
  }
3081 3082 3083 3084

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

  int32_t numOfCols = 0;
L
Liu Jicong 已提交
3085
  SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
3086 3087 3088 3089 3090 3091 3092 3093

  int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {
    pInfo->pPseudoExpr = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pInfo->numOfPseudoExpr);
3094
    pInfo->pPseudoCtx = createSqlFunctionCtx(pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, &pInfo->rowEntryInfoOffset);
3095 3096 3097 3098
  }

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

L
Liu Jicong 已提交
3099 3100
  pInfo->readHandle = *readHandle;
  pInfo->interval = extractIntervalInfo(pTableScanNode);
3101
  pInfo->sample.sampleRatio = pTableScanNode->ratio;
L
Liu Jicong 已提交
3102 3103 3104
  pInfo->sample.seed = taosGetTimestampSec();
  pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired;
  pInfo->pFilterNode = pTableScanNode->scan.node.pConditions;
3105
  pInfo->tableListInfo = pTableListInfo;
L
Liu Jicong 已提交
3106 3107
  pInfo->scanFlag = MAIN_SCAN;
  pInfo->pColMatchInfo = pColList;
3108 3109

  pInfo->pResBlock = createResDataBlock(pDescNode);
3110 3111
  pInfo->dataReaders = taosArrayInit(64, POINTER_BYTES);
  pInfo->sortSourceParams = taosArrayInit(64, sizeof(STableMergeScanSortSourceParam));
3112

3113
  pInfo->pSortInfo = generateSortByTsInfo(pInfo->pColMatchInfo, pInfo->cond.order);
3114
  pInfo->pSortInputBlock = createOneDataBlock(pInfo->pResBlock, false);
3115

3116
  int32_t rowSize = pInfo->pResBlock->info.rowSize;
L
Liu Jicong 已提交
3117
  pInfo->bufPageSize = getProperSortPageSize(rowSize);
3118

L
Liu Jicong 已提交
3119
  pOperator->name = "TableMergeScanOperator";
3120
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN;
L
Liu Jicong 已提交
3121 3122 3123
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
L
Liu Jicong 已提交
3124
  pOperator->exprSupp.numOfExprs = numOfCols;
L
Liu Jicong 已提交
3125
  pOperator->pTaskInfo = pTaskInfo;
3126
  initResultSizeInfo(&pOperator->resultInfo, 1024);
3127 3128

  pOperator->fpSet =
3129 3130
      createOperatorFpSet(operatorDummyOpenFn, doTableMergeScan, NULL, NULL, destroyTableMergeScanOperatorInfo, NULL,
                          NULL, getTableMergeScanExplainExecInfo);
3131 3132 3133 3134 3135 3136 3137 3138 3139
  pOperator->cost.openCost = 0;
  return pOperator;

_error:
  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
  taosMemoryFree(pInfo);
  taosMemoryFree(pOperator);
  return NULL;
}