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

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

#include "executorimpl.h"
#include "query.h"
#include "tcompare.h"
#include "thash.h"
#include "ttypes.h"
36
#include "vnode.h"
H
Haojun Liao 已提交
37

wmmhello's avatar
wmmhello 已提交
38 39
#include "executorInt.h"

H
Haojun Liao 已提交
40
#define SET_REVERSE_SCAN_FLAG(_info) ((_info)->scanFlag = REVERSE_SCAN)
41
#define SWITCH_ORDER(n)              (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC))
42

43
static int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity);
L
Liu Jicong 已提交
44 45
static int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size,
                                     const char* dbName);
46

47 48 49
static void addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr,
                                   SSDataBlock* pBlock);
static bool processBlockWithProbability(const SSampleExecInfo* pInfo);
50

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

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

64
static void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
H
Haojun Liao 已提交
65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92
  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
}

93 94 95 96 97 98 99 100 101
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;
102
  // convert key to second
103 104 105 106 107 108 109
  key = convertTimePrecision(key, pInterval->precision, TSDB_TIME_PRECISION_MILLI) / 1000;

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

  struct tm tm;
110
  time_t    t = (time_t)key;
111 112 113 114 115
  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 已提交
116
  tw->skey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, pInterval->precision);
117 118 119 120

  mon = (int)(mon + interval);
  tm.tm_year = mon / 12;
  tm.tm_mon = mon % 12;
wafwerar's avatar
wafwerar 已提交
121
  tw->ekey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, pInterval->precision);
122 123 124 125

  tw->ekey -= 1;
}

126
static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockInfo, int32_t order) {
127 128 129 130 131 132 133
  STimeWindow w = {0};

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

134
  if (order == TSDB_ORDER_ASC) {
135
    getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.skey, &w);
136 137 138 139 140 141
    assert(w.ekey >= pBlockInfo->window.skey);

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

142 143
    while (1) {
      getNextTimeWindow(pInterval, &w, order);
144 145 146 147 148 149 150 151 152 153
      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 {
154 155 156 157 158 159 160
    getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.ekey, &w);
    assert(w.skey <= pBlockInfo->window.ekey);

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

161
    while (1) {
162 163 164 165 166 167 168 169 170 171
      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;
      }
    }
172 173 174 175 176
  }

  return false;
}

L
Liu Jicong 已提交
177 178
static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock,
                             uint32_t* status) {
179
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
180 181
  STableScanInfo* pInfo = pOperator->info;

182
  SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;
H
Haojun Liao 已提交
183 184

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

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

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

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

    // clear all data in pBlock that are set when handing the previous block
207
    for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
208 209 210 211
      SColumnInfoData* pcol = taosArrayGet(pBlock->pDataBlock, i);
      pcol->pData = NULL;
    }

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

L
Liu Jicong 已提交
216
    bool             allColumnsHaveAgg = true;
217
    SColumnDataAgg** pColAgg = NULL;
218
    tsdbRetrieveDataBlockStatisInfo(pTableScanInfo->dataReader, &pColAgg, &allColumnsHaveAgg);
219

220
    if (allColumnsHaveAgg == true) {
221
      int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
222 223

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

228
      for (int32_t i = 0; i < taosArrayGetSize(pTableScanInfo->pColMatchInfo); ++i) {
229 230 231 232 233 234
        SColMatchInfo* pColMatchInfo = taosArrayGet(pTableScanInfo->pColMatchInfo, i);
        if (!pColMatchInfo->output) {
          continue;
        }
        pBlock->pBlockAgg[pColMatchInfo->targetSlotId] = pColAgg[i];
      }
H
Haojun Liao 已提交
235

236
      return TSDB_CODE_SUCCESS;
237
    } else {  // failed to load the block sma data, data block statistics does not exist, load data block instead
H
Haojun Liao 已提交
238
      *status = FUNC_DATA_REQUIRED_DATA_LOAD;
239
    }
H
Haojun Liao 已提交
240
  }
241

242
  ASSERT(*status == FUNC_DATA_REQUIRED_DATA_LOAD);
243

H
Haojun Liao 已提交
244 245 246 247 248 249 250 251 252 253
  // 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 已提交
254

H
Haojun Liao 已提交
255 256
  pCost->totalCheckedRows += pBlock->info.rows;
  pCost->loadBlocks += 1;
257

H
Haojun Liao 已提交
258 259 260
  SArray* pCols = tsdbRetrieveDataBlock(pTableScanInfo->dataReader, NULL);
  if (pCols == NULL) {
    return terrno;
H
Haojun Liao 已提交
261 262
  }

263
  relocateColumnData(pBlock, pTableScanInfo->pColMatchInfo, pCols, true);
264 265

  // currently only the tbname pseudo column
266
  if (pTableScanInfo->pseudoSup.numOfExprs > 0) {
267
    SExprSupp* pSup = &pTableScanInfo->pseudoSup;
268
    addTagPseudoColumnData(&pTableScanInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pBlock);
269 270
  }

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

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

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

H
Haojun Liao 已提交
283 284 285
  return TSDB_CODE_SUCCESS;
}

286
static void prepareForDescendingScan(STableScanInfo* pTableScanInfo, SqlFunctionCtx* pCtx, int32_t numOfOutput) {
H
Haojun Liao 已提交
287 288 289
  SET_REVERSE_SCAN_FLAG(pTableScanInfo);

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

292
  pTableScanInfo->cond.order = TSDB_ORDER_DESC;
293 294 295 296
  for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
    STimeWindow* pTWindow = &pTableScanInfo->cond.twindows[i];
    TSWAP(pTWindow->skey, pTWindow->ekey);
  }
297 298 299

  SQueryTableDataCond* pCond = &pTableScanInfo->cond;
  taosqsort(pCond->twindows, pCond->numOfTWindows, sizeof(STimeWindow), pCond, compareTimeWindow);
H
Haojun Liao 已提交
300 301
}

302 303
void addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr,
                            SSDataBlock* pBlock) {
304
  // currently only the tbname pseudo column
305
  if (numOfPseudoExpr == 0) {
306 307 308 309
    return;
  }

  SMetaReader mr = {0};
310
  metaReaderInit(&mr, pHandle->meta, 0);
311 312
  metaGetTableEntryByUid(&mr, pBlock->info.uid);

313 314
  for (int32_t j = 0; j < numOfPseudoExpr; ++j) {
    SExprInfo* pExpr = &pPseudoExpr[j];
315 316 317 318

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

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

320
    colInfoDataEnsureCapacity(pColInfoData, pBlock->info.rows);
D
dapan1121 已提交
321
    colInfoDataCleanup(pColInfoData, pBlock->info.rows);
322 323 324 325 326

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

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

333 334 335 336
      char* data = NULL;
      if (pColInfoData->info.type != TSDB_DATA_TYPE_JSON && p != NULL) {
        data = tTagValToData((const STagVal*)p, false);
      } else {
wmmhello's avatar
wmmhello 已提交
337
        data = (char*)p;
wmmhello's avatar
wmmhello 已提交
338
      }
339

340
      for (int32_t i = 0; i < pBlock->info.rows; ++i) {
L
Liu Jicong 已提交
341 342
        colDataAppend(pColInfoData, i, data,
                      (data == NULL) || (pColInfoData->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data)));
343
      }
344

345 346
      if (data && (pColInfoData->info.type != TSDB_DATA_TYPE_JSON) && p != NULL &&
          IS_VAR_DATA_TYPE(((const STagVal*)p)->type)) {
wmmhello's avatar
wmmhello 已提交
347
        taosMemoryFree(data);
wmmhello's avatar
wmmhello 已提交
348
      }
349 350 351 352 353 354
    }
  }

  metaReaderClear(&mr);
}

355 356 357 358
void setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId) {
  struct SScalarFuncExecFuncs fpSet = {0};
  fmGetScalarFuncExecFuncs(functionId, &fpSet);

359 360
  SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_BIGINT, sizeof(uint64_t), 1);
  colInfoDataEnsureCapacity(&infoData, 1);
361

362
  colDataAppendInt64(&infoData, 0, (int64_t*)&pBlock->info.uid);
363
  SScalarParam srcParam = {.numOfRows = pBlock->info.rows, .param = pMeta, .columnData = &infoData};
364 365 366 367 368

  SScalarParam param = {.columnData = pColInfoData};
  fpSet.process(&srcParam, 1, &param);
}

369
static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
370
  STableScanInfo* pTableScanInfo = pOperator->info;
371
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
L
Liu Jicong 已提交
372
  SSDataBlock*    pBlock = pTableScanInfo->pResBlock;
H
Haojun Liao 已提交
373

374 375
  int64_t st = taosGetTimestampUs();

376
  while (tsdbNextDataBlock(pTableScanInfo->dataReader)) {
377 378
    if (isTaskKilled(pTaskInfo)) {
      longjmp(pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
379
    }
H
Haojun Liao 已提交
380

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

387 388 389 390 391 392 393 394
    blockDataCleanup(pBlock);

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

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

397 398 399 400 401 402
    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);
    }
403

404 405 406
    // 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;
407
    }
408

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

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

    pOperator->cost.totalCost = pTableScanInfo->readRecorder.elapsedTime;
418 419

    // todo refactor
L
Liu Jicong 已提交
420 421
    pTableScanInfo->lastStatus.uid = pBlock->info.uid;
    pTableScanInfo->lastStatus.ts = pBlock->info.window.ekey;
422

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

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

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

438 439
  // do the ascending order traverse in the first place.
  while (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
440 441 442
    while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
      SSDataBlock* p = doTableScanImpl(pOperator);
      if (p != NULL) {
L
Liu Jicong 已提交
443
        ASSERT(p->info.uid != 0);
444 445 446 447 448 449
        return p;
      }
      pTableScanInfo->curTWinIdx += 1;
      if (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
        tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, pTableScanInfo->curTWinIdx);
      }
H
Haojun Liao 已提交
450 451
    }

452
    pTableScanInfo->scanTimes += 1;
453

454
    if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
455 456
      setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
      pTableScanInfo->scanFlag = REPEAT_SCAN;
457 458 459
      qDebug("%s start to repeat ascending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
      for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
        STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
460
        qDebug("%s qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
461
      }
462 463 464
      // do prepare for the next round table scan operation
      tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
      pTableScanInfo->curTWinIdx = 0;
H
Haojun Liao 已提交
465
    }
466
  }
H
Haojun Liao 已提交
467

468
  int32_t total = pTableScanInfo->scanInfo.numOfAsc + pTableScanInfo->scanInfo.numOfDesc;
469
  if (pTableScanInfo->scanTimes < total) {
470
    if (pTableScanInfo->cond.order == TSDB_ORDER_ASC) {
471
      prepareForDescendingScan(pTableScanInfo, pTableScanInfo->pCtx, 0);
472 473
      tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
      pTableScanInfo->curTWinIdx = 0;
474
    }
H
Haojun Liao 已提交
475

476 477 478
    qDebug("%s start to descending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
    for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
      STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
479
      qDebug("%s qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
480
    }
481

482
    while (pTableScanInfo->scanTimes < total) {
483 484 485 486 487 488 489 490 491
      while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
        SSDataBlock* p = doTableScanImpl(pOperator);
        if (p != NULL) {
          return p;
        }
        pTableScanInfo->curTWinIdx += 1;
        if (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
          tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, pTableScanInfo->curTWinIdx);
        }
492
      }
H
Haojun Liao 已提交
493

494
      pTableScanInfo->scanTimes += 1;
H
Haojun Liao 已提交
495

496
      if (pTableScanInfo->scanTimes < total) {
497 498
        setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
        pTableScanInfo->scanFlag = REPEAT_SCAN;
H
Haojun Liao 已提交
499

500 501
        qDebug("%s start to repeat descending order scan data blocks due to query func required",
               GET_TASKID(pTaskInfo));
502 503
        for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
          STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
504
          qDebug("%s qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
505
        }
506
        tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
507
        pTableScanInfo->curTWinIdx = 0;
508
      }
H
Haojun Liao 已提交
509 510 511
    }
  }

wmmhello's avatar
wmmhello 已提交
512 513 514 515 516 517 518
  return NULL;
}

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

L
Liu Jicong 已提交
519 520 521
  // if scan table by table
  if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {
    // check status
L
Liu Jicong 已提交
522
    while (1) {
L
Liu Jicong 已提交
523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540
      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);
      /*pTableInfo->uid */
      tsdbSetTableId(pInfo->dataReader, pTableInfo->uid);
      tsdbResetReadHandle(pInfo->dataReader, &pInfo->cond, 0);
      pInfo->scanTimes = 0;
      pInfo->curTWinIdx = 0;
    }
  }

541
  if (pInfo->currentGroupId == -1) {
wmmhello's avatar
wmmhello 已提交
542
    pInfo->currentGroupId++;
wmmhello's avatar
wmmhello 已提交
543
    if (pInfo->currentGroupId >= taosArrayGetSize(pTaskInfo->tableqinfoList.pGroupList)) {
wmmhello's avatar
wmmhello 已提交
544
      setTaskStatus(pTaskInfo, TASK_COMPLETED);
wmmhello's avatar
wmmhello 已提交
545 546
      return NULL;
    }
547
    SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, pInfo->currentGroupId);
wmmhello's avatar
wmmhello 已提交
548
    tsdbCleanupReadHandle(pInfo->dataReader);
549 550
    tsdbReaderT* pReader =
        tsdbReaderOpen(pInfo->readHandle.vnode, &pInfo->cond, tableList, pInfo->queryId, pInfo->taskId);
wmmhello's avatar
wmmhello 已提交
551 552 553 554
    pInfo->dataReader = pReader;
  }

  SSDataBlock* result = doTableScanGroup(pOperator);
555
  if (result) {
wmmhello's avatar
wmmhello 已提交
556 557 558 559 560
    return result;
  }

  pInfo->currentGroupId++;
  if (pInfo->currentGroupId >= taosArrayGetSize(pTaskInfo->tableqinfoList.pGroupList)) {
wmmhello's avatar
wmmhello 已提交
561
    setTaskStatus(pTaskInfo, TASK_COMPLETED);
wmmhello's avatar
wmmhello 已提交
562 563 564
    return NULL;
  }

565
  SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, pInfo->currentGroupId);
wmmhello's avatar
wmmhello 已提交
566 567 568 569 570 571 572
  tsdbSetTableList(pInfo->dataReader, tableList);

  tsdbResetReadHandle(pInfo->dataReader, &pInfo->cond, 0);
  pInfo->curTWinIdx = 0;
  pInfo->scanTimes = 0;

  result = doTableScanGroup(pOperator);
573
  if (result) {
wmmhello's avatar
wmmhello 已提交
574 575 576
    return result;
  }

577 578
  setTaskStatus(pTaskInfo, TASK_COMPLETED);
  return NULL;
H
Haojun Liao 已提交
579 580
}

581 582
static int32_t getTableScannerExecInfo(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {
  SFileBlockLoadRecorder* pRecorder = taosMemoryCalloc(1, sizeof(SFileBlockLoadRecorder));
583
  STableScanInfo*         pTableScanInfo = pOptr->info;
584 585 586 587 588 589
  *pRecorder = pTableScanInfo->readRecorder;
  *pOptrExplain = pRecorder;
  *len = sizeof(SFileBlockLoadRecorder);
  return 0;
}

590 591
static void destroyTableScanOperatorInfo(void* param, int32_t numOfOutput) {
  STableScanInfo* pTableScanInfo = (STableScanInfo*)param;
H
Haojun Liao 已提交
592
  blockDataDestroy(pTableScanInfo->pResBlock);
593
  cleanupQueryTableDataCond(&pTableScanInfo->cond);
H
Haojun Liao 已提交
594

595 596 597 598 599 600 601
  tsdbCleanupReadHandle(pTableScanInfo->dataReader);

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

wmmhello's avatar
wmmhello 已提交
602 603
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
                                           SExecTaskInfo* pTaskInfo, uint64_t queryId, uint64_t taskId) {
H
Haojun Liao 已提交
604 605 606
  STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
607
    goto _error;
H
Haojun Liao 已提交
608 609
  }

L
Liu Jicong 已提交
610
  // taosSsleep(20);
D
dapan1121 已提交
611

612
  SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;
613
  int32_t             numOfCols = 0;
614
  SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
L
Liu Jicong 已提交
615

616 617
  int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode);
  if (code != TSDB_CODE_SUCCESS) {
618
    goto _error;
619 620 621
  }

  if (pTableScanNode->scan.pScanPseudoCols != NULL) {
622
    SExprSupp* pSup = &pInfo->pseudoSup;
623 624
    pSup->pExprInfo = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->numOfExprs);
    pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset);
625 626
  }

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

630 631 632 633
  pInfo->readHandle = *readHandle;
  pInfo->interval = extractIntervalInfo(pTableScanNode);
  pInfo->sample.sampleRatio = pTableScanNode->ratio;
  pInfo->sample.seed = taosGetTimestampSec();
634

635
  pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired;
636 637 638 639 640
  pInfo->pResBlock = createResDataBlock(pDescNode);
  pInfo->pFilterNode = pTableScanNode->scan.node.pConditions;
  pInfo->scanFlag = MAIN_SCAN;
  pInfo->pColMatchInfo = pColList;
  pInfo->curTWinIdx = 0;
wmmhello's avatar
wmmhello 已提交
641 642 643
  pInfo->queryId = queryId;
  pInfo->taskId = taskId;
  pInfo->currentGroupId = -1;
644 645

  pOperator->name = "TableScanOperator";  // for debug purpose
L
Liu Jicong 已提交
646
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
647 648 649
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
650
  pOperator->exprSupp.numOfExprs = numOfCols;
651
  pOperator->pTaskInfo = pTaskInfo;
652

653 654
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, destroyTableScanOperatorInfo,
                                         NULL, NULL, getTableScannerExecInfo);
655 656 657

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

660
_error:
661 662 663 664 665
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);

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

668
SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
669
  STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
L
Liu Jicong 已提交
670
  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
H
Haojun Liao 已提交
671

L
Liu Jicong 已提交
672 673
  pInfo->dataReader = pReadHandle;
  //  pInfo->prevGroupId       = -1;
H
Haojun Liao 已提交
674

675
  pOperator->name = "TableSeqScanOperator";
H
Haojun Liao 已提交
676
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN;
677 678 679 680
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
  pOperator->pTaskInfo = pTaskInfo;
H
Haojun Liao 已提交
681

682
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScanImpl, NULL, NULL, NULL, NULL, NULL, NULL);
H
Haojun Liao 已提交
683 684 685
  return pOperator;
}

686 687
static int32_t doGetTableRowSize(void* pMeta, uint64_t uid) {
  int32_t rowLen = 0;
H
Haojun Liao 已提交
688

689
  SMetaReader mr = {0};
690 691
  metaReaderInit(&mr, pMeta, 0);
  metaGetTableEntryByUid(&mr, uid);
692 693
  if (mr.me.type == TSDB_SUPER_TABLE) {
    int32_t numOfCols = mr.me.stbEntry.schemaRow.nCols;
694
    for (int32_t i = 0; i < numOfCols; ++i) {
695 696 697 698 699 700 701
      rowLen += mr.me.stbEntry.schemaRow.pSchema[i].bytes;
    }
  } else if (mr.me.type == TSDB_CHILD_TABLE) {
    uint64_t suid = mr.me.ctbEntry.suid;
    metaGetTableEntryByUid(&mr, suid);
    int32_t numOfCols = mr.me.stbEntry.schemaRow.nCols;

702
    for (int32_t i = 0; i < numOfCols; ++i) {
703 704 705 706
      rowLen += mr.me.stbEntry.schemaRow.pSchema[i].bytes;
    }
  } else if (mr.me.type == TSDB_NORMAL_TABLE) {
    int32_t numOfCols = mr.me.ntbEntry.schemaRow.nCols;
707
    for (int32_t i = 0; i < numOfCols; ++i) {
708 709 710 711 712
      rowLen += mr.me.ntbEntry.schemaRow.pSchema[i].bytes;
    }
  }

  metaReaderClear(&mr);
713 714 715 716 717 718 719 720 721 722 723 724
  return rowLen;
}

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

  SBlockDistInfo* pBlockScanInfo = pOperator->info;

  STableBlockDistInfo blockDistInfo = {.minRows = INT_MAX, .maxRows = INT_MIN};
  blockDistInfo.rowSize = doGetTableRowSize(pBlockScanInfo->readHandle.meta, pBlockScanInfo->uid);
725 726 727

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

729
  SSDataBlock* pBlock = pBlockScanInfo->pResBlock;
H
Haojun Liao 已提交
730

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

734
  int32_t len = tSerializeBlockDistInfo(NULL, 0, &blockDistInfo);
735
  char*   p = taosMemoryCalloc(1, len + VARSTR_HEADER_SIZE);
736 737 738
  tSerializeBlockDistInfo(varDataVal(p), len, &blockDistInfo);
  varDataSetLen(p, len);

739
  blockDataEnsureCapacity(pBlock, 1);
740 741
  colDataAppend(pColInfo, 0, p, false);
  taosMemoryFree(p);
H
Haojun Liao 已提交
742

743 744
  pBlock->info.rows = 1;

H
Haojun Liao 已提交
745 746 747 748
  pOperator->status = OP_EXEC_DONE;
  return pBlock;
}

749
static void destroyBlockDistScanOperatorInfo(void* param, int32_t numOfOutput) {
750
  SBlockDistInfo* pDistInfo = (SBlockDistInfo*)param;
751 752 753
  blockDataDestroy(pDistInfo->pResBlock);
}

754 755
SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SReadHandle* readHandle, uint64_t uid,
                                               SBlockDistScanPhysiNode* pBlockScanNode, SExecTaskInfo* pTaskInfo) {
756
  SBlockDistInfo* pInfo = taosMemoryCalloc(1, sizeof(SBlockDistInfo));
757
  SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
H
Haojun Liao 已提交
758 759 760 761 762
  if (pInfo == NULL || pOperator == NULL) {
    pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
    goto _error;
  }

763
  pInfo->pHandle = dataReader;
764
  pInfo->readHandle = *readHandle;
765 766
  pInfo->uid = uid;
  pInfo->pResBlock = createResDataBlock(pBlockScanNode->node.pOutputDataBlockDesc);
767

768
  int32_t    numOfCols = 0;
769
  SExprInfo* pExprInfo = createExprInfo(pBlockScanNode->pScanPseudoCols, NULL, &numOfCols);
770
  int32_t    code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfCols);
771 772 773
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
H
Haojun Liao 已提交
774

775
  pOperator->name = "DataBlockDistScanOperator";
776
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN;
777 778 779
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
780 781 782 783
  pOperator->pTaskInfo = pTaskInfo;

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

786
_error:
H
Haojun Liao 已提交
787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  return NULL;
}

static void doClearBufferedBlocks(SStreamBlockScanInfo* pInfo) {
  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);
}

H
Haojun Liao 已提交
803
static bool isSessionWindow(SStreamBlockScanInfo* pInfo) {
804
  return pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
5
54liuyao 已提交
805 806
}

H
Haojun Liao 已提交
807
static bool isStateWindow(SStreamBlockScanInfo* pInfo) {
808
  return pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;
5
54liuyao 已提交
809
}
5
54liuyao 已提交
810

5
54liuyao 已提交
811
static bool prepareDataScan(SStreamBlockScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
812 813 814
  STimeWindow win = {
      .skey = INT64_MIN,
      .ekey = INT64_MAX,
815
  };
5
54liuyao 已提交
816
  bool needRead = false;
5
54liuyao 已提交
817 818
  if (!isStateWindow(pInfo) && (*pRowIndex) < pSDB->info.rows) {
    SColumnInfoData* pColDataInfo = taosArrayGet(pSDB->pDataBlock, tsColIndex);
819 820
    TSKEY*           tsCols = (TSKEY*)pColDataInfo->pData;
    SResultRowInfo   dumyInfo;
5
54liuyao 已提交
821
    dumyInfo.cur.pageId = -1;
5
54liuyao 已提交
822 823
    if (isSessionWindow(pInfo)) {
      SStreamAggSupporter* pAggSup = pInfo->sessionSup.pStreamAggSup;
824 825 826
      int64_t              gap = pInfo->sessionSup.gap;
      int32_t              winIndex = 0;
      SResultWindowInfo*   pCurWin =
5
54liuyao 已提交
827
          getSessionTimeWindow(pAggSup, tsCols[(*pRowIndex)], INT64_MIN, pSDB->info.groupId, gap, &winIndex);
5
54liuyao 已提交
828
      win = pCurWin->win;
829
      (*pRowIndex) += updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, (*pRowIndex), gap, NULL);
5
54liuyao 已提交
830
    } else {
831 832 833 834
      win =
          getActiveTimeWindow(NULL, &dumyInfo, tsCols[(*pRowIndex)], &pInfo->interval, pInfo->interval.precision, NULL);
      (*pRowIndex) += getNumOfRowsInTimeWindow(&pSDB->info, tsCols, (*pRowIndex), win.ekey, binarySearchForKey, NULL,
                                               TSDB_ORDER_ASC);
5
54liuyao 已提交
835
    }
5
54liuyao 已提交
836 837 838 839 840
    needRead = true;
  } else if (isStateWindow(pInfo)) {
    SArray* pWins = pInfo->sessionSup.pStreamAggSup->pScanWindow;
    int32_t size = taosArrayGetSize(pWins);
    if (pInfo->scanWinIndex < size) {
841
      win = *(STimeWindow*)taosArrayGet(pWins, pInfo->scanWinIndex);
5
54liuyao 已提交
842 843 844 845 846 847 848 849
      pInfo->scanWinIndex++;
      needRead = true;
    } else {
      pInfo->scanWinIndex = 0;
      taosArrayClear(pWins);
    }
  }
  if (!needRead) {
5
54liuyao 已提交
850 851
    return false;
  }
L
Liu Jicong 已提交
852
  STableScanInfo* pTableScanInfo = pInfo->pSnapshotReadOp->info;
5
54liuyao 已提交
853 854
  pTableScanInfo->cond.twindows[0] = win;
  pTableScanInfo->curTWinIdx = 0;
855
  //  tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
5
54liuyao 已提交
856 857 858
  // if (!pTableScanInfo->dataReader) {
  //   return false;
  // }
5
54liuyao 已提交
859
  pTableScanInfo->scanTimes = 0;
wmmhello's avatar
wmmhello 已提交
860
  pTableScanInfo->currentGroupId = -1;
5
54liuyao 已提交
861
  return true;
5
54liuyao 已提交
862 863
}

864
static void copyOneRow(SSDataBlock* dest, SSDataBlock* source, int32_t sourceRowId) {
865
  for (int32_t j = 0; j < taosArrayGetSize(source->pDataBlock); j++) {
866 867 868 869 870 871
    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 已提交
872 873
    }
  }
874
  dest->info.rows++;
5
54liuyao 已提交
875 876
}

877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897
static uint64_t getGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t rowId) {
  uint64_t* groupId = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &pBlock->info.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;
  */
}

5
54liuyao 已提交
898
static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
899 900
  while (1) {
    SSDataBlock* pResult = NULL;
L
Liu Jicong 已提交
901
    pResult = doTableScan(pInfo->pSnapshotReadOp);
902
    if (pResult == NULL) {
5
54liuyao 已提交
903
      if (prepareDataScan(pInfo, pSDB, tsColIndex, pRowIndex)) {
904
        // scan next window data
L
Liu Jicong 已提交
905
        pResult = doTableScan(pInfo->pSnapshotReadOp);
906 907 908 909 910
      }
    }
    if (!pResult) {
      return NULL;
    }
L
Liu Jicong 已提交
911

912 913
    if (pResult->info.groupId == pInfo->groupId) {
      return pResult;
5
54liuyao 已提交
914 915
    }
  }
916

L
Liu Jicong 已提交
917 918 919 920 921 922 923 924
  /* Todo(liuyao) for partition by column
    SSDataBlock* pBlock = createOneDataBlock(pResult, true);
    blockDataCleanup(pResult);
    for (int32_t i = 0; i < pBlock->info.rows; i++) {
      uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock, i);
      if (id == pInfo->groupId) {
        copyOneRow(pResult, pBlock, i);
      }
925
    }
L
Liu Jicong 已提交
926 927
    return pResult;
  */
928 929 930 931
}

static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDataBlock* pUpdateBlock) {
  blockDataCleanup(pUpdateBlock);
5
54liuyao 已提交
932
  int32_t size = taosArrayGetSize(pInfo->tsArray);
933
  if (pInfo->tsArrayIndex < size) {
5
54liuyao 已提交
934
    SColumnInfoData* pCol = (SColumnInfoData*)taosArrayGet(pUpdateBlock->pDataBlock, pInfo->primaryTsIndex);
5
54liuyao 已提交
935
    ASSERT(pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
936
    blockDataEnsureCapacity(pUpdateBlock, size);
937 938

    int32_t rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, pInfo->tsArrayIndex);
L
Liu Jicong 已提交
939
    pInfo->groupId = getGroupId(pInfo->pSnapshotReadOp, pBlock, rowId);
940
    int32_t i = 0;
L
Liu Jicong 已提交
941
    for (; i < size; i++) {
942
      rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, i + pInfo->tsArrayIndex);
L
Liu Jicong 已提交
943
      uint64_t id = getGroupId(pInfo->pSnapshotReadOp, pBlock, rowId);
944 945
      if (pInfo->groupId != id) {
        break;
946
      }
947
      copyOneRow(pUpdateBlock, pBlock, rowId);
948
    }
949 950 951
    pUpdateBlock->info.rows = i;
    pInfo->tsArrayIndex += i;
    pUpdateBlock->info.groupId = pInfo->groupId;
5
54liuyao 已提交
952
    pUpdateBlock->info.type = STREAM_CLEAR;
5
54liuyao 已提交
953
    blockDataUpdateTsWindow(pUpdateBlock, 0);
954 955 956 957
  }
  // all rows have same group id
  ASSERT(pInfo->tsArrayIndex >= size);
  if (size > 0 && pInfo->tsArrayIndex == size) {
5
54liuyao 已提交
958
    taosArrayClear(pInfo->tsArray);
5
54liuyao 已提交
959 960 961
  }
}

962 963 964 965 966 967 968 969 970 971 972
static void getUpdateDataBlock(SStreamBlockScanInfo* pInfo, bool invertible, SSDataBlock* pBlock,
                               SSDataBlock* pUpdateBlock) {
  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
  ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP);
  TSKEY* ts = (TSKEY*)pColDataInfo->pData;
  for (int32_t rowId = 0; rowId < pBlock->info.rows; rowId++) {
    if (updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, ts[rowId])) {
      taosArrayPush(pInfo->tsArray, &rowId);
    }
  }
  if (!pUpdateBlock) {
5
54liuyao 已提交
973
    taosArrayClear(pInfo->tsArray);
974
    return;
5
54liuyao 已提交
975
  }
976 977 978 979 980 981
  setUpdateData(pInfo, pBlock, pUpdateBlock);
  // Todo(liuyao) get from tsdb
  //  SSDataBlock* p = createOneDataBlock(pBlock, true);
  //  p->info.type = STREAM_INVERT;
  //  taosArrayClear(pInfo->tsArray);
  //  return p;
5
54liuyao 已提交
982 983
}

984
static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
985 986 987 988
  // NOTE: this operator does never check if current status is done or not
  SExecTaskInfo*        pTaskInfo = pOperator->pTaskInfo;
  SStreamBlockScanInfo* pInfo = pOperator->info;

989
  pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
990
  if (pTaskInfo->code != TSDB_CODE_SUCCESS || pOperator->status == OP_EXEC_DONE) {
H
Haojun Liao 已提交
991 992 993
    return NULL;
  }

5
54liuyao 已提交
994
  size_t total = taosArrayGetSize(pInfo->pBlockLists);
995
  // TODO: refactor
L
Liu Jicong 已提交
996
  if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
H
Haojun Liao 已提交
997
    if (pInfo->validBlockIndex >= total) {
L
Liu Jicong 已提交
998
      /*doClearBufferedBlocks(pInfo);*/
999
      pOperator->status = OP_EXEC_DONE;
H
Haojun Liao 已提交
1000 1001 1002
      return NULL;
    }

1003
    int32_t      current = pInfo->validBlockIndex++;
1004 1005
    SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current);
    blockDataUpdateTsWindow(pBlock, 0);
5
54liuyao 已提交
1006
    if (pBlock->info.type == STREAM_RETRIEVE) {
L
Liu Jicong 已提交
1007
      pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
5
54liuyao 已提交
1008 1009 1010 1011 1012 1013
      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RETRIEVE;
      copyDataBlock(pInfo->pPullDataRes, pBlock);
      pInfo->pullDataResIndex = 0;
      prepareDataScan(pInfo, pInfo->pPullDataRes, 0, &pInfo->pullDataResIndex);
      updateInfoAddCloseWindowSBF(pInfo->pUpdateInfo);
    }
1014
    return pBlock;
L
Liu Jicong 已提交
1015
  } else if (pInfo->blockType == STREAM_INPUT__DATA_SUBMIT) {
5
54liuyao 已提交
1016 1017 1018 1019 1020 1021
    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) {
      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
5
54liuyao 已提交
1022
      if (!isStateWindow(pInfo)) {
5
54liuyao 已提交
1023
        prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
5
54liuyao 已提交
1024
      }
5
54liuyao 已提交
1025
      return pInfo->pUpdateRes;
5
54liuyao 已提交
1026 1027 1028 1029
    } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RETRIEVE) {
      SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pPullDataRes, 0, &pInfo->pullDataResIndex);
      if (pSDB != NULL) {
        getUpdateDataBlock(pInfo, true, pSDB, NULL);
L
Liu Jicong 已提交
1030
        pSDB->info.type = STREAM_PULL_DATA;
5
54liuyao 已提交
1031 1032 1033
        return pSDB;
      }
      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
5
54liuyao 已提交
1034
    } else {
1035
      if (isStateWindow(pInfo) && taosArrayGetSize(pInfo->sessionSup.pStreamAggSup->pScanWindow) > 0) {
5
54liuyao 已提交
1036 1037
        pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
        pInfo->updateResIndex = pInfo->pUpdateRes->info.rows;
5
54liuyao 已提交
1038
        prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
5
54liuyao 已提交
1039 1040
      }
      if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) {
5
54liuyao 已提交
1041
        SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
5
54liuyao 已提交
1042
        if (pSDB == NULL) {
1043 1044 1045
          setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes);
          if (pInfo->pUpdateRes->info.rows > 0) {
            if (!isStateWindow(pInfo)) {
5
54liuyao 已提交
1046 1047 1048
              // Todo(liuyao) mybe can delete this.
              bool test = prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
              ASSERT(test == false);
1049 1050 1051 1052 1053
            }
            return pInfo->pUpdateRes;
          } else {
            pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
          }
5
54liuyao 已提交
1054
        } else {
5
54liuyao 已提交
1055
          pSDB->info.type = STREAM_NORMAL;
5
54liuyao 已提交
1056 1057 1058
          getUpdateDataBlock(pInfo, true, pSDB, NULL);
          return pSDB;
        }
5
54liuyao 已提交
1059
      }
5
54liuyao 已提交
1060
    }
5
54liuyao 已提交
1061

H
Haojun Liao 已提交
1062 1063 1064
    SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
    blockDataCleanup(pInfo->pRes);

1065
    while (tqNextDataBlock(pInfo->streamBlockReader)) {
1066
      SSDataBlock block = {0};
1067

1068
      // todo refactor
1069 1070 1071 1072 1073
      int32_t code = tqRetrieveDataBlock(&block, pInfo->streamBlockReader);

      uint64_t groupId = block.info.groupId;
      uint64_t uid = block.info.uid;
      int32_t  numOfRows = block.info.rows;
H
Haojun Liao 已提交
1074

1075 1076 1077
      if (code != TSDB_CODE_SUCCESS || numOfRows == 0) {
        pTaskInfo->code = code;
        return NULL;
H
Haojun Liao 已提交
1078 1079
      }

1080 1081
      pInfo->pRes->info.groupId = groupId;
      pInfo->pRes->info.rows = numOfRows;
1082
      pInfo->pRes->info.uid = uid;
5
54liuyao 已提交
1083
      pInfo->pRes->info.type = STREAM_NORMAL;
1084
      pInfo->pRes->info.capacity = numOfRows;
H
Haojun Liao 已提交
1085

1086 1087 1088 1089
      // 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 = uid;
1090 1091
      } else {
        pInfo->pRes->info.groupId = groupId;
1092 1093
      }

1094 1095 1096 1097 1098
      uint64_t* groupIdPre = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &uid, sizeof(int64_t));
      if (groupIdPre) {
        pInfo->pRes->info.groupId = *groupIdPre;
      }

1099
      // todo extract method
1100
      for (int32_t i = 0; i < taosArrayGetSize(pInfo->pColMatchInfo); ++i) {
1101
        SColMatchInfo* pColMatchInfo = taosArrayGet(pInfo->pColMatchInfo, i);
H
Haojun Liao 已提交
1102 1103 1104 1105
        if (!pColMatchInfo->output) {
          continue;
        }

1106
        bool colExists = false;
1107 1108
        for (int32_t j = 0; j < blockDataGetNumOfCols(&block); ++j) {
          SColumnInfoData* pResCol = bdGetColumnInfoData(&block, j);
1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120
          if (pResCol->info.colId == pColMatchInfo->colId) {
            taosArraySet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId, pResCol);
            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);
        }
H
Haojun Liao 已提交
1121 1122
      }

H
Haojun Liao 已提交
1123
      taosArrayDestroy(block.pDataBlock);
H
Haojun Liao 已提交
1124 1125
      if (pInfo->pRes->pDataBlock == NULL) {
        // TODO add log
5
54liuyao 已提交
1126
        updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
1127
        pOperator->status = OP_EXEC_DONE;
H
Haojun Liao 已提交
1128 1129 1130
        pTaskInfo->code = terrno;
        return NULL;
      }
5
54liuyao 已提交
1131

1132 1133 1134 1135 1136
      // currently only the tbname pseudo column
      if (pInfo->numOfPseudoExpr > 0) {
        addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, pInfo->pRes);
      }

1137
      doFilter(pInfo->pCondition, pInfo->pRes);
1138
      blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
5
54liuyao 已提交
1139 1140 1141
      if (pBlockInfo->rows > 0) {
        break;
      }
H
Haojun Liao 已提交
1142 1143 1144 1145
    }

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

5
54liuyao 已提交
1148
    if (pBlockInfo->rows == 0) {
5
54liuyao 已提交
1149
      updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
1150
      pOperator->status = OP_EXEC_DONE;
5
54liuyao 已提交
1151
    } else if (pInfo->pUpdateInfo) {
1152
      pInfo->tsArrayIndex = 0;
5
54liuyao 已提交
1153 1154
      getUpdateDataBlock(pInfo, true, pInfo->pRes, pInfo->pUpdateRes);
      if (pInfo->pUpdateRes->info.rows > 0) {
5
54liuyao 已提交
1155
        if (pInfo->pUpdateRes->info.type == STREAM_CLEAR) {
5
54liuyao 已提交
1156 1157
          pInfo->updateResIndex = 0;
          pInfo->scanMode = STREAM_SCAN_FROM_UPDATERES;
5
54liuyao 已提交
1158
        } else if (pInfo->pUpdateRes->info.type == STREAM_INVERT) {
5
54liuyao 已提交
1159
          pInfo->scanMode = STREAM_SCAN_FROM_RES;
5
54liuyao 已提交
1160
          return pInfo->pUpdateRes;
5
54liuyao 已提交
1161
        }
5
54liuyao 已提交
1162
      }
1163 1164
    }

5
54liuyao 已提交
1165
    return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes;
L
Liu Jicong 已提交
1166

L
Liu Jicong 已提交
1167 1168 1169
  } else if (pInfo->blockType == STREAM_INPUT__DATA_SCAN) {
    // check reader last status
    // if not match, reset status
L
Liu Jicong 已提交
1170
    SSDataBlock* pResult = doTableScan(pInfo->pSnapshotReadOp);
L
Liu Jicong 已提交
1171 1172
    return pResult && pResult->info.rows > 0 ? pResult : NULL;

L
Liu Jicong 已提交
1173 1174 1175
  } else {
    ASSERT(0);
    return NULL;
H
Haojun Liao 已提交
1176 1177 1178
  }
}

1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190
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;
}

1191 1192 1193
SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode,
                                            SExecTaskInfo* pTaskInfo, STimeWindowAggSupp* pTwSup, uint64_t queryId,
                                            uint64_t taskId) {
H
Haojun Liao 已提交
1194 1195
  SStreamBlockScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamBlockScanInfo));
  SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
1196

H
Haojun Liao 已提交
1197 1198
  if (pInfo == NULL || pOperator == NULL) {
    terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
1199
    goto _error;
H
Haojun Liao 已提交
1200 1201
  }

1202 1203 1204
  SScanPhysiNode* pScanPhyNode = &pTableScanNode->scan;

  SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;
H
Haojun Liao 已提交
1205

1206
  int32_t numOfCols = 0;
1207
  pInfo->pColMatchInfo = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
1208 1209 1210

  int32_t numOfOutput = taosArrayGetSize(pInfo->pColMatchInfo);
  SArray* pColIds = taosArrayInit(numOfOutput, sizeof(int16_t));
1211
  for (int32_t i = 0; i < numOfOutput; ++i) {
1212 1213 1214
    SColMatchInfo* id = taosArrayGet(pInfo->pColMatchInfo, i);

    int16_t colId = id->colId;
1215
    taosArrayPush(pColIds, &colId);
1216
    if (id->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
5
54liuyao 已提交
1217 1218
      pInfo->primaryTsIndex = id->targetSlotId;
    }
H
Haojun Liao 已提交
1219 1220 1221 1222
  }

  pInfo->pBlockLists = taosArrayInit(4, POINTER_BYTES);
  if (pInfo->pBlockLists == NULL) {
1223 1224
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _error;
H
Haojun Liao 已提交
1225 1226
  }

1227
  pInfo->tsArray = taosArrayInit(4, sizeof(int32_t));
5
54liuyao 已提交
1228
  if (pInfo->tsArray == NULL) {
1229
    goto _error;
5
54liuyao 已提交
1230 1231
  }

L
Liu Jicong 已提交
1232
  if (pHandle) {
wmmhello's avatar
wmmhello 已提交
1233
    SOperatorInfo*  pTableScanDummy = createTableScanOperatorInfo(pTableScanNode, pHandle, pTaskInfo, queryId, taskId);
L
Liu Jicong 已提交
1234
    STableScanInfo* pSTInfo = (STableScanInfo*)pTableScanDummy->info;
L
Liu Jicong 已提交
1235 1236 1237 1238 1239 1240 1241

    SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, 0);
    if (pHandle->tqReader) {
      pSTInfo->scanMode = TABLE_SCAN__TABLE_ORDER;
      pSTInfo->dataReader = tsdbReaderOpen(pHandle->vnode, &pSTInfo->cond, tableList, 0, 0);
    }

L
Liu Jicong 已提交
1242 1243 1244 1245 1246
    if (pSTInfo->interval.interval > 0) {
      pInfo->pUpdateInfo = updateInfoInitP(&pSTInfo->interval, pTwSup->waterMark);
    } else {
      pInfo->pUpdateInfo = NULL;
    }
L
Liu Jicong 已提交
1247
    pInfo->pSnapshotReadOp = pTableScanDummy;
L
Liu Jicong 已提交
1248 1249
    pInfo->interval = pSTInfo->interval;

L
Liu Jicong 已提交
1250 1251 1252 1253 1254
    pInfo->readHandle = *pHandle;
    ASSERT(pHandle->reader);
    pInfo->streamBlockReader = pHandle->reader;
    pInfo->tableUid = pScanPhyNode->uid;

L
Liu Jicong 已提交
1255
    // set the extract column id to streamHandle
L
Liu Jicong 已提交
1256
    tqReadHandleSetColIdList((SStreamReader*)pHandle->reader, pColIds);
L
Liu Jicong 已提交
1257 1258 1259 1260 1261 1262 1263
    SArray* tableIdList = extractTableIdList(&pTaskInfo->tableqinfoList);
    int32_t code = tqReadHandleSetTbUidList(pHandle->reader, tableIdList);
    if (code != 0) {
      taosArrayDestroy(tableIdList);
      goto _error;
    }
    taosArrayDestroy(tableIdList);
5
54liuyao 已提交
1264 1265
  }

1266 1267 1268 1269 1270
  // create the pseduo columns info
  if (pTableScanNode->scan.pScanPseudoCols != NULL) {
    pInfo->pPseudoExpr = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pInfo->numOfPseudoExpr);
  }

1271
  pInfo->pRes = createResDataBlock(pDescNode);
X
Xiaoyu Wang 已提交
1272
  pInfo->pUpdateRes = createResDataBlock(pDescNode);
1273 1274 1275
  pInfo->pCondition = pScanPhyNode->node.pConditions;
  pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
  pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1};
1276
  pInfo->groupId = 0;
5
54liuyao 已提交
1277
  pInfo->pPullDataRes = createPullDataBlock();
L
Liu Jicong 已提交
1278

1279
  pOperator->name = "StreamBlockScanOperator";
L
Liu Jicong 已提交
1280
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
1281 1282 1283
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
1284
  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);
1285
  pOperator->pTaskInfo = pTaskInfo;
H
Haojun Liao 已提交
1286

1287 1288
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doStreamBlockScan, NULL, NULL, operatorDummyCloseFn, NULL, NULL, NULL);
1289

H
Haojun Liao 已提交
1290
  return pOperator;
1291

L
Liu Jicong 已提交
1292
_error:
1293 1294 1295
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  return NULL;
H
Haojun Liao 已提交
1296 1297 1298 1299 1300 1301 1302
}

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

1303
  const char* name = tNameGetTableName(&pInfo->name);
1304
  if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0 || pInfo->pCur != NULL) {
H
Haojun Liao 已提交
1305
    metaCloseTbCursor(pInfo->pCur);
1306
    pInfo->pCur = NULL;
H
Haojun Liao 已提交
1307
  }
H
Haojun Liao 已提交
1308 1309

  taosArrayDestroy(pInfo->scanCols);
H
Haojun Liao 已提交
1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349
}

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;
      if (TSDB_INS_USER_STABLES_DBNAME_COLID == node->colId) {
        *(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;
1350
      return DEAL_RES_END;  // stop walk
H
Haojun Liao 已提交
1351 1352 1353 1354 1355 1356 1357
    }
    default:
      break;
  }
  return DEAL_RES_CONTINUE;
}

1358
static void getDBNameFromCondition(SNode* pCondition, const char* dbName) {
H
Haojun Liao 已提交
1359 1360 1361
  if (NULL == pCondition) {
    return;
  }
L
Liu Jicong 已提交
1362
  nodesWalkExpr(pCondition, getDBNameFromConditionWalker, (char*)dbName);
H
Haojun Liao 已提交
1363 1364
}

D
dapan1121 已提交
1365
static int32_t loadSysTableCallback(void* param, SDataBuf* pMsg, int32_t code) {
H
Haojun Liao 已提交
1366 1367 1368 1369 1370 1371 1372
  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);
1373 1374 1375
    pRsp->useconds = htobe64(pRsp->useconds);
    pRsp->handle = htobe64(pRsp->handle);
    pRsp->compLen = htonl(pRsp->compLen);
H
Haojun Liao 已提交
1376 1377 1378 1379 1380
  } else {
    operator->pTaskInfo->code = code;
  }

  tsem_post(&pScanResInfo->ready);
wmmhello's avatar
wmmhello 已提交
1381
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1382 1383 1384 1385 1386 1387 1388
}

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

1389 1390
  doFilter(pInfo->pCondition, pInfo->pRes);
#if 0
H
Haojun Liao 已提交
1391
  SFilterInfo* filter = NULL;
1392 1393

  int32_t code = filterInitFromNode(pInfo->pCondition, &filter, 0);
H
Haojun Liao 已提交
1394 1395 1396 1397 1398

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

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

1402
  SSDataBlock* px = createOneDataBlock(pInfo->pRes, false);
H
Haojun Liao 已提交
1403 1404 1405 1406 1407 1408 1409 1410
  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 已提交
1411
    if (keep) {
1412
      colDataAssign(pDest, pSrc, pInfo->pRes->info.rows, &px->info);
D
dapan1121 已提交
1413 1414 1415 1416 1417 1418 1419
      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;
        }
1420

1421 1422 1423 1424 1425 1426
        if (colDataIsNull_s(pSrc, j)) {
          colDataAppendNULL(pDest, numOfRow);
        } else {
          colDataAppend(pDest, numOfRow, colDataGetData(pSrc, j), false);
        }

D
dapan1121 已提交
1427
        numOfRow += 1;
H
Haojun Liao 已提交
1428
      }
D
dapan1121 已提交
1429 1430
    } else {
      numOfRow = 0;
H
Haojun Liao 已提交
1431 1432 1433 1434 1435
    }
  }

  px->info.rows = numOfRow;
  pInfo->pRes = px;
1436
#endif
H
Haojun Liao 已提交
1437 1438 1439 1440

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

1441
static SSDataBlock* buildSysTableMetaBlock() {
L
Liu Jicong 已提交
1442 1443
  size_t               size = 0;
  const SSysTableMeta* pMeta = NULL;
1444 1445 1446
  getInfosDbMeta(&pMeta, &size);

  int32_t index = 0;
L
Liu Jicong 已提交
1447 1448
  for (int32_t i = 0; i < size; ++i) {
    if (strcmp(pMeta[i].name, TSDB_INS_TABLE_USER_TABLES) == 0) {
1449 1450 1451 1452
      index = i;
      break;
    }
  }
1453

1454
  SSDataBlock* pBlock = createDataBlock();
L
Liu Jicong 已提交
1455
  for (int32_t i = 0; i < pMeta[index].colNum; ++i) {
L
Liu Jicong 已提交
1456 1457
    SColumnInfoData colInfoData =
        createColumnInfoData(pMeta[index].schema[i].type, pMeta[index].schema[i].bytes, i + 1);
1458
    blockDataAppendColInfo(pBlock, &colInfoData);
1459 1460
  }

1461 1462 1463
  return pBlock;
}

1464
static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
1465 1466 1467 1468 1469
  // build message and send to mnode to fetch the content of system tables.
  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;
  SSysTableScanInfo* pInfo = pOperator->info;

  // retrieve local table list info from vnode
1470 1471
  const char* name = tNameGetTableName(&pInfo->name);
  if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0) {
1472 1473 1474 1475
    if (pOperator->status == OP_EXEC_DONE) {
      return NULL;
    }

1476 1477
    // the retrieve is executed on the mnode, so return tables that belongs to the information schema database.
    if (pInfo->readHandle.mnd != NULL) {
1478
      buildSysDbTableInfo(pInfo, pOperator->resultInfo.capacity);
1479

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

1483
      doSetOperatorCompleted(pOperator);
1484 1485 1486 1487 1488
      return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
    } else {
      if (pInfo->pCur == NULL) {
        pInfo->pCur = metaOpenTbCursor(pInfo->readHandle.meta);
      }
1489

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

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

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

1501 1502
      tNameGetDbName(&sn, varDataVal(dbname));
      varDataSetLen(dbname, strlen(varDataVal(dbname)));
1503

1504
      SSDataBlock* p = buildSysTableMetaBlock();
1505
      blockDataEnsureCapacity(p, pOperator->resultInfo.capacity);
1506

1507
      char n[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
1508 1509 1510

      int32_t ret = 0;
      while ((ret = metaTbCursorNext(pInfo->pCur)) == 0) {
1511
        STR_TO_VARSTR(n, pInfo->pCur->mr.me.name);
1512

1513 1514 1515
        // table name
        SColumnInfoData* pColInfoData = taosArrayGet(p->pDataBlock, 0);
        colDataAppend(pColInfoData, numOfRows, n, false);
1516

1517 1518 1519
        // database name
        pColInfoData = taosArrayGet(p->pDataBlock, 1);
        colDataAppend(pColInfoData, numOfRows, dbname, false);
1520

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

1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537
        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);
          metaGetTableEntryByUid(&mr, pInfo->pCur->mr.me.ctbEntry.suid);

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

          // super table name
wmmhello's avatar
wmmhello 已提交
1541
          STR_TO_VARSTR(n, mr.me.name);
1542
          pColInfoData = taosArrayGet(p->pDataBlock, 4);
wmmhello's avatar
wmmhello 已提交
1543
          colDataAppend(pColInfoData, numOfRows, n, false);
1544 1545
          metaReaderClear(&mr);

wmmhello's avatar
wmmhello 已提交
1546 1547
          // table comment
          pColInfoData = taosArrayGet(p->pDataBlock, 8);
L
Liu Jicong 已提交
1548
          if (pInfo->pCur->mr.me.ctbEntry.commentLen > 0) {
wmmhello's avatar
wmmhello 已提交
1549 1550 1551
            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);
L
Liu Jicong 已提交
1552
          } else if (pInfo->pCur->mr.me.ctbEntry.commentLen == 0) {
wmmhello's avatar
wmmhello 已提交
1553 1554 1555
            char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
            STR_TO_VARSTR(comment, "");
            colDataAppend(pColInfoData, numOfRows, comment, false);
L
Liu Jicong 已提交
1556
          } else {
wmmhello's avatar
wmmhello 已提交
1557 1558 1559
            colDataAppendNULL(pColInfoData, numOfRows);
          }

1560 1561 1562 1563 1564 1565 1566 1567
          // 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);

wmmhello's avatar
wmmhello 已提交
1568
          STR_TO_VARSTR(n, "CHILD_TABLE");
1569 1570 1571 1572 1573 1574 1575
        } 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);
1576
          colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.schemaRow.nCols, false);
1577 1578 1579 1580 1581

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

wmmhello's avatar
wmmhello 已提交
1582 1583
          // table comment
          pColInfoData = taosArrayGet(p->pDataBlock, 8);
L
Liu Jicong 已提交
1584
          if (pInfo->pCur->mr.me.ntbEntry.commentLen > 0) {
wmmhello's avatar
wmmhello 已提交
1585 1586 1587
            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);
L
Liu Jicong 已提交
1588
          } else if (pInfo->pCur->mr.me.ntbEntry.commentLen == 0) {
wmmhello's avatar
wmmhello 已提交
1589 1590 1591
            char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
            STR_TO_VARSTR(comment, "");
            colDataAppend(pColInfoData, numOfRows, comment, false);
L
Liu Jicong 已提交
1592
          } else {
wmmhello's avatar
wmmhello 已提交
1593 1594 1595
            colDataAppendNULL(pColInfoData, numOfRows);
          }

1596 1597 1598 1599 1600 1601 1602 1603
          // 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);

wmmhello's avatar
wmmhello 已提交
1604
          STR_TO_VARSTR(n, "NORMAL_TABLE");
1605
        }
1606

1607
        pColInfoData = taosArrayGet(p->pDataBlock, 9);
wmmhello's avatar
wmmhello 已提交
1608
        colDataAppend(pColInfoData, numOfRows, n, false);
1609

1610
        if (++numOfRows >= pOperator->resultInfo.capacity) {
1611 1612
          break;
        }
H
Haojun Liao 已提交
1613 1614
      }

1615 1616 1617 1618 1619 1620 1621
      // 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);
      }

1622 1623
      p->info.rows = numOfRows;
      pInfo->pRes->info.rows = numOfRows;
H
Haojun Liao 已提交
1624

1625
      relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
1626
      doFilterResult(pInfo);
H
Haojun Liao 已提交
1627

1628 1629
      blockDataDestroy(p);

1630 1631 1632
      pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
      return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
    }
H
Haojun Liao 已提交
1633 1634 1635 1636 1637
  } else {  // load the meta from mnode of the given epset
    if (pOperator->status == OP_EXEC_DONE) {
      return NULL;
    }

1638 1639 1640
    while (1) {
      int64_t startTs = taosGetTimestampUs();
      strncpy(pInfo->req.tb, tNameGetTableName(&pInfo->name), tListLen(pInfo->req.tb));
1641
      strcpy(pInfo->req.user, pInfo->pUser);
H
Haojun Liao 已提交
1642

1643 1644 1645 1646 1647
      if (pInfo->showRewrite) {
        char dbName[TSDB_DB_NAME_LEN] = {0};
        getDBNameFromCondition(pInfo->pCondition, dbName);
        sprintf(pInfo->req.db, "%d.%s", pInfo->accountId, dbName);
      }
H
Haojun Liao 已提交
1648

1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659
      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 已提交
1660

L
Liu Jicong 已提交
1661 1662
      int32_t msgType = (strcasecmp(name, TSDB_INS_TABLE_DNODE_VARIABLES) == 0) ? TDMT_DND_SYSTABLE_RETRIEVE
                                                                                : TDMT_MND_SYSTABLE_RETRIEVE;
D
dapan1121 已提交
1663

1664 1665 1666
      pMsgSendInfo->param = pOperator;
      pMsgSendInfo->msgInfo.pData = buf1;
      pMsgSendInfo->msgInfo.len = contLen;
D
dapan1121 已提交
1667
      pMsgSendInfo->msgType = msgType;
1668
      pMsgSendInfo->fp = loadSysTableCallback;
D
dapan1121 已提交
1669
      pMsgSendInfo->requestId = pTaskInfo->id.queryId;
H
Haojun Liao 已提交
1670

1671
      int64_t transporterId = 0;
1672 1673
      int32_t code =
          asyncSendMsgToServer(pInfo->readHandle.pMsgCb->clientRpc, &pInfo->epSet, &transporterId, pMsgSendInfo);
1674
      tsem_wait(&pInfo->ready);
H
Haojun Liao 已提交
1675

1676 1677 1678 1679 1680
      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 已提交
1681

1682 1683
      SRetrieveMetaTableRsp* pRsp = pInfo->pRsp;
      pInfo->req.showId = pRsp->handle;
H
Haojun Liao 已提交
1684

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

1690
        if (pRsp->numOfRows == 0) {
H
Haojun Liao 已提交
1691
          taosMemoryFree(pRsp);
1692 1693 1694
          return NULL;
        }
      }
H
Haojun Liao 已提交
1695

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

1699 1700
      // todo log the filter info
      doFilterResult(pInfo);
H
Haojun Liao 已提交
1701
      taosMemoryFree(pRsp);
1702 1703
      if (pInfo->pRes->info.rows > 0) {
        return pInfo->pRes;
D
dapan1121 已提交
1704 1705
      } else if (pOperator->status == OP_EXEC_DONE) {
        return NULL;
1706
      }
1707
    }
H
Haojun Liao 已提交
1708 1709 1710
  }
}

1711
int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity) {
1712
  SSDataBlock* p = buildSysTableMetaBlock();
1713
  blockDataEnsureCapacity(p, capacity);
1714

L
Liu Jicong 已提交
1715
  size_t               size = 0;
1716 1717 1718 1719 1720 1721 1722 1723
  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);

1724
  relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
1725
  pInfo->pRes->info.rows = p->info.rows;
1726 1727 1728
  blockDataDestroy(p);

  return pInfo->pRes->info.rows;
1729 1730
}

L
Liu Jicong 已提交
1731 1732 1733
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};
1734 1735
  int32_t numOfRows = p->info.rows;

L
Liu Jicong 已提交
1736
  for (int32_t i = 0; i < size; ++i) {
1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756
    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 已提交
1757
    for (int32_t j = 4; j <= 8; ++j) {
1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772
      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;
}

1773
SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode* pScanPhyNode,
1774
                                              const char* pUser, SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
1775 1776 1777
  SSysTableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SSysTableScanInfo));
  SOperatorInfo*     pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
1778
    goto _error;
H
Haojun Liao 已提交
1779 1780
  }

1781 1782 1783
  SScanPhysiNode* pScanNode = &pScanPhyNode->scan;

  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;
1784
  SSDataBlock*        pResBlock = createResDataBlock(pDescNode);
1785 1786

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

1789 1790
  pInfo->accountId = pScanPhyNode->accountId;
  pInfo->pUser = taosMemoryStrDup((void*)pUser);
1791
  pInfo->showRewrite = pScanPhyNode->showRewrite;
1792 1793 1794
  pInfo->pRes = pResBlock;
  pInfo->pCondition = pScanNode->node.pConditions;
  pInfo->scanCols = colList;
1795 1796

  initResultSizeInfo(pOperator, 4096);
H
Haojun Liao 已提交
1797

1798
  tNameAssign(&pInfo->name, &pScanNode->tableName);
1799
  const char* name = tNameGetTableName(&pInfo->name);
1800

1801
  if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0) {
L
Liu Jicong 已提交
1802
    pInfo->readHandle = *(SReadHandle*)readHandle;
1803
    blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
H
Haojun Liao 已提交
1804 1805
  } else {
    tsem_init(&pInfo->ready, 0, 0);
1806
    pInfo->epSet = pScanPhyNode->mgmtEpSet;
1807
    pInfo->readHandle = *(SReadHandle*)readHandle;
H
Haojun Liao 已提交
1808 1809
  }

1810
  pOperator->name = "SysTableScanOperator";
H
Haojun Liao 已提交
1811
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN;
1812 1813 1814
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
1815
  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pResBlock->pDataBlock);
1816
  pOperator->pTaskInfo = pTaskInfo;
1817

L
Liu Jicong 已提交
1818 1819
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doSysTableScan, NULL, NULL, destroySysScanOperator, NULL, NULL, NULL);
H
Haojun Liao 已提交
1820 1821

  return pOperator;
1822

1823
_error:
1824 1825 1826 1827
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
  return NULL;
H
Haojun Liao 已提交
1828
}
H
Haojun Liao 已提交
1829

1830
static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
1831 1832 1833 1834
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

1835 1836 1837
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

#if 0
H
Haojun Liao 已提交
1838 1839 1840 1841 1842 1843 1844 1845
  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);

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

1850
    SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[0];
H
Haojun Liao 已提交
1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909
    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.
1910
#endif
H
Haojun Liao 已提交
1911

1912
  STagScanInfo* pInfo = pOperator->info;
1913
  SExprInfo*    pExprInfo = &pOperator->exprSupp.pExprInfo[0];
1914
  SSDataBlock*  pRes = pInfo->pRes;
H
Haojun Liao 已提交
1915

wmmhello's avatar
wmmhello 已提交
1916 1917
  int32_t size = taosArrayGetSize(pInfo->pTableList->pTableList);
  if (size == 0) {
H
Haojun Liao 已提交
1918 1919 1920 1921
    setTaskStatus(pTaskInfo, TASK_COMPLETED);
    return NULL;
  }

1922 1923 1924
  char        str[512] = {0};
  int32_t     count = 0;
  SMetaReader mr = {0};
1925
  metaReaderInit(&mr, pInfo->readHandle.meta, 0);
H
Haojun Liao 已提交
1926

wmmhello's avatar
wmmhello 已提交
1927 1928
  while (pInfo->curPos < size && count < pOperator->resultInfo.capacity) {
    STableKeyInfo* item = taosArrayGet(pInfo->pTableList->pTableList, pInfo->curPos);
1929
    metaGetTableEntryByUid(&mr, item->uid);
H
Haojun Liao 已提交
1930

1931
    for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {
1932 1933 1934 1935 1936 1937
      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);
1938
      } else {  // it is a tag value
wmmhello's avatar
wmmhello 已提交
1939 1940 1941 1942
        STagVal val = {0};
        val.cid = pExprInfo[j].base.pParam[0].pCol->colId;
        const char* p = metaGetTableTagVal(&mr.me, pDst->info.type, &val);

1943 1944 1945 1946
        char* data = NULL;
        if (pDst->info.type != TSDB_DATA_TYPE_JSON && p != NULL) {
          data = tTagValToData((const STagVal*)p, false);
        } else {
wmmhello's avatar
wmmhello 已提交
1947 1948
          data = (char*)p;
        }
L
Liu Jicong 已提交
1949 1950
        colDataAppend(pDst, count, data,
                      (data == NULL) || (pDst->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data)));
1951

1952 1953
        if (pDst->info.type != TSDB_DATA_TYPE_JSON && p != NULL && IS_VAR_DATA_TYPE(((const STagVal*)p)->type) &&
            data != NULL) {
wmmhello's avatar
wmmhello 已提交
1954
          taosMemoryFree(data);
wmmhello's avatar
wmmhello 已提交
1955
        }
H
Haojun Liao 已提交
1956 1957 1958
      }
    }

1959
    count += 1;
wmmhello's avatar
wmmhello 已提交
1960
    if (++pInfo->curPos >= size) {
1961
      doSetOperatorCompleted(pOperator);
H
Haojun Liao 已提交
1962 1963 1964
    }
  }

1965 1966
  metaReaderClear(&mr);

1967
  // qDebug("QInfo:0x%"PRIx64" create tag values results completed, rows:%d", GET_TASKID(pRuntimeEnv), count);
H
Haojun Liao 已提交
1968
  if (pOperator->status == OP_EXEC_DONE) {
1969
    setTaskStatus(pTaskInfo, TASK_COMPLETED);
H
Haojun Liao 已提交
1970 1971 1972
  }

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

1975
  return (pRes->info.rows == 0) ? NULL : pInfo->pRes;
H
Haojun Liao 已提交
1976 1977 1978 1979 1980 1981 1982
}

static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
  STagScanInfo* pInfo = (STagScanInfo*)param;
  pInfo->pRes = blockDataDestroy(pInfo->pRes);
}

1983 1984
SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode,
                                         STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo) {
1985
  STagScanInfo*  pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo));
H
Haojun Liao 已提交
1986 1987 1988 1989 1990
  SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

1991 1992
  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;

1993
  int32_t    num = 0;
1994 1995
  int32_t    numOfExprs = 0;
  SExprInfo* pExprInfo = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &numOfExprs);
L
Liu Jicong 已提交
1996
  SArray*    colList = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID);
1997

1998 1999 2000 2001
  int32_t code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2002

2003 2004 2005 2006 2007
  pInfo->pTableList = pTableListInfo;
  pInfo->pColMatchInfo = colList;
  pInfo->pRes = createResDataBlock(pDescNode);
  pInfo->readHandle = *pReadHandle;
  pInfo->curPos = 0;
2008

2009
  pOperator->name = "TagScanOperator";
2010
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN;
2011

2012 2013 2014 2015
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
  pOperator->pTaskInfo = pTaskInfo;
2016

2017 2018 2019
  initResultSizeInfo(pOperator, 4096);
  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);

2020 2021
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doTagScan, NULL, NULL, destroyTagScanOperatorInfo, NULL, NULL, NULL);
H
Haojun Liao 已提交
2022 2023

  return pOperator;
2024

2025
_error:
H
Haojun Liao 已提交
2026 2027 2028 2029 2030
  taosMemoryFree(pInfo);
  taosMemoryFree(pOperator);
  terrno = TSDB_CODE_OUT_OF_MEMORY;
  return NULL;
}
2031 2032

typedef struct STableMergeScanInfo {
2033
  STableListInfo* tableListInfo;
S
slzhou 已提交
2034 2035 2036 2037
  int32_t         tableStartIndex;
  int32_t         tableEndIndex;
  bool            hasGroupId;
  uint64_t        groupId;
2038

2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050
  SArray*     dataReaders;  // array of tsdbReaderT*
  SReadHandle readHandle;

  int32_t  bufPageSize;
  uint32_t sortBufSize;  // max buffer size for in-memory sort

  SArray*      pSortInfo;
  SSortHandle* pSortHandle;

  SSDataBlock* pSortInputBlock;
  int64_t      startTs;  // sort start time

2051 2052 2053
  SArray*  sortSourceParams;
  uint64_t queryId;
  uint64_t taskId;
2054 2055 2056 2057 2058 2059 2060 2061 2062

  SFileBlockLoadRecorder readRecorder;
  int64_t                numOfRows;
  //  int32_t         prevGroupId;  // previous table group id
  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;
2063
  int32_t*        rowEntryInfoOffset;
2064 2065 2066 2067 2068 2069 2070 2071
  SExprInfo*      pExpr;
  SSDataBlock*    pResBlock;
  SArray*         pColMatchInfo;
  int32_t         numOfOutput;

  SExprInfo*      pPseudoExpr;
  int32_t         numOfPseudoExpr;
  SqlFunctionCtx* pPseudoCtx;
2072
  //  int32_t*        rowEntryInfoOffset;
2073 2074 2075 2076 2077 2078 2079 2080 2081 2082

  SQueryTableDataCond cond;
  int32_t             scanFlag;  // table scan flag to denote if it is a repeat/reverse/main scan
  int32_t             dataBlockLoadFlag;
  SInterval interval;  // 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.

  SSampleExecInfo sample;  // sample execution info
} STableMergeScanInfo;

2083
int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
H
Haojun Liao 已提交
2084
                                STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId) {
wmmhello's avatar
wmmhello 已提交
2085
  int32_t code = getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo);
2086
  if (code != TSDB_CODE_SUCCESS) {
2087
    return code;
2088 2089 2090 2091
  }

  if (taosArrayGetSize(pTableListInfo->pTableList) == 0) {
    qDebug("no table qualified for query, TID:0x%" PRIx64 ", QID:0x%" PRIx64, taskId, queryId);
2092 2093
    return TSDB_CODE_SUCCESS;
  }
2094
  code = generateGroupIdMap(pTableListInfo, pHandle, pTableScanNode->pGroupTags);
2095
  if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2096
    return code;
2097 2098 2099 2100 2101
  }

  return TSDB_CODE_SUCCESS;
}

S
slzhou 已提交
2102 2103 2104 2105 2106 2107 2108 2109 2110 2111 2112 2113 2114 2115 2116 2117
int32_t createMultipleDataReaders(SQueryTableDataCond* pQueryCond, SReadHandle* pHandle, STableListInfo* pTableListInfo,
                                  int32_t tableStartIdx, int32_t tableEndIdx, SArray* arrayReader, uint64_t queryId,
                                  uint64_t taskId) {
  for (int32_t i = tableStartIdx; i <= tableEndIdx; ++i) {
    SArray* subTableList = taosArrayInit(1, sizeof(STableKeyInfo));
    taosArrayPush(subTableList, taosArrayGet(pTableListInfo->pTableList, i));

    tsdbReaderT* pReader = tsdbReaderOpen(pHandle->vnode, pQueryCond, subTableList, queryId, taskId);
    taosArrayPush(arrayReader, &pReader);

    taosArrayDestroy(subTableList);
  }

  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
2118
// todo refactor
2119 2120
static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeScanInfo* pTableScanInfo,
                                         int32_t readerIdx, SSDataBlock* pBlock, uint32_t* status) {
L
Liu Jicong 已提交
2121
  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;
S
shenglian zhou 已提交
2122
  STableMergeScanInfo* pInfo = pOperator->info;
2123 2124 2125 2126 2127 2128 2129 2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144 2145 2146 2147 2148

  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
2149
    for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
2150 2151 2152 2153 2154 2155 2156 2157 2158 2159 2160 2161 2162 2163
      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;
    tsdbReaderT*     reader = taosArrayGetP(pTableScanInfo->dataReaders, readerIdx);
    tsdbRetrieveDataBlockStatisInfo(reader, &pColAgg, &allColumnsHaveAgg);

    if (allColumnsHaveAgg == true) {
2164
      int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
2165 2166 2167 2168 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190 2191 2192 2193 2194 2195 2196 2197 2198 2199 2200 2201 2202 2203 2204 2205 2206

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

  tsdbReaderT* reader = taosArrayGetP(pTableScanInfo->dataReaders, readerIdx);
  SArray*      pCols = tsdbRetrieveDataBlock(reader, NULL);
  if (pCols == NULL) {
    return terrno;
  }

2207
  relocateColumnData(pBlock, pTableScanInfo->pColMatchInfo, pCols, true);
2208 2209 2210 2211 2212 2213 2214 2215 2216 2217 2218 2219 2220 2221 2222 2223 2224 2225 2226 2227 2228 2229 2230 2231 2232

  // currently only the tbname pseudo column
  if (pTableScanInfo->numOfPseudoExpr > 0) {
    addTagPseudoColumnData(&pTableScanInfo->readHandle, pTableScanInfo->pPseudoExpr, pTableScanInfo->numOfPseudoExpr,
                           pBlock);
  }

  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;
2233
  SSDataBlock*   inputBlock;
2234 2235 2236 2237 2238 2239
} STableMergeScanSortSourceParam;

static SSDataBlock* getTableDataBlock(void* param) {
  STableMergeScanSortSourceParam* source = param;
  SOperatorInfo*                  pOperator = source->pOperator;
  int32_t                         readerIdx = source->readerIdx;
2240
  SSDataBlock*                    pBlock = source->inputBlock;
2241 2242 2243 2244
  STableMergeScanInfo*            pTableScanInfo = pOperator->info;

  int64_t st = taosGetTimestampUs();

2245 2246
  blockDataCleanup(pBlock);

2247 2248 2249 2250 2251 2252 2253 2254 2255 2256 2257 2258
  tsdbReaderT* reader = taosArrayGetP(pTableScanInfo->dataReaders, readerIdx);
  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;
    }

2259 2260 2261 2262 2263 2264 2265
    blockDataCleanup(pBlock);
    SDataBlockInfo binfo = pBlock->info;
    tsdbRetrieveDataBlockInfo(reader, &binfo);

    binfo.capacity = binfo.rows;
    blockDataEnsureCapacity(pBlock, binfo.capacity);
    pBlock->info = binfo;
2266 2267 2268 2269 2270 2271 2272 2273 2274 2275 2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303

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

SArray* generateSortByTsInfo(int32_t order) {
  SArray*         pList = taosArrayInit(1, sizeof(SBlockOrderInfo));
  SBlockOrderInfo bi = {0};
  bi.order = order;
  bi.slotId = 0;
  bi.nullFirst = NULL_ORDER_FIRST;

  taosArrayPush(pList, &bi);

  return pList;
}

2304
int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {
2305 2306 2307
  STableMergeScanInfo* pInfo = pOperator->info;
  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

S
slzhou 已提交
2308 2309 2310 2311 2312 2313 2314 2315 2316 2317 2318
  {
    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;
  }
2319

S
slzhou 已提交
2320 2321
  int32_t tableStartIdx = pInfo->tableStartIndex;
  int32_t tableEndIdx = pInfo->tableEndIndex;
2322

S
slzhou 已提交
2323 2324
  STableListInfo* tableListInfo = pInfo->tableListInfo;
  createMultipleDataReaders(&pInfo->cond, &pInfo->readHandle, tableListInfo, tableStartIdx, tableEndIdx,
2325
                            pInfo->dataReaders, pInfo->queryId, pInfo->taskId);
2326

2327 2328
  // 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 已提交
2329
  pInfo->sortBufSize = pInfo->bufPageSize * (tableEndIdx - tableStartIdx + 1 + 1);
2330
  int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
L
Liu Jicong 已提交
2331 2332
  pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage,
                                             pInfo->pSortInputBlock, pTaskInfo->id.str);
2333 2334 2335 2336

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

  size_t numReaders = taosArrayGetSize(pInfo->dataReaders);
2337 2338 2339 2340 2341 2342 2343 2344
  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);
  }

2345 2346
  for (int32_t i = 0; i < numReaders; ++i) {
    SSortSource*                    ps = taosMemoryCalloc(1, sizeof(SSortSource));
2347
    STableMergeScanSortSourceParam* param = taosArrayGet(pInfo->sortSourceParams, i);
2348 2349 2350 2351 2352 2353 2354 2355 2356 2357
    ps->param = param;
    tsortAddSource(pInfo->pSortHandle, ps);
  }

  int32_t code = tsortOpen(pInfo->pSortHandle);

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

2358 2359 2360 2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372
  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) {
    tsdbReaderT* reader = taosArrayGetP(pInfo->dataReaders, i);
    tsdbCleanupReadHandle(reader);
  }
  taosArrayDestroy(pInfo->dataReaders);
2373 2374 2375 2376

  return TSDB_CODE_SUCCESS;
}

2377
SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, int32_t capacity, SOperatorInfo* pOperator) {
2378 2379 2380 2381 2382 2383 2384 2385 2386 2387 2388
  STableMergeScanInfo* pInfo = pOperator->info;
  SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;

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

  blockDataEnsureCapacity(p, capacity);

  while (1) {
2389
    STupleHandle* pTupleHandle = tsortNextTuple(pHandle);
2390 2391 2392 2393
    if (pTupleHandle == NULL) {
      break;
    }

2394
    appendOneRowToDataBlock(p, pTupleHandle);
2395 2396 2397 2398 2399
    if (p->info.rows >= capacity) {
      break;
    }
  }

2400 2401
  qDebug("%s get sorted row blocks, rows:%d", GET_TASKID(pTaskInfo), p->info.rows);
  return (p->info.rows > 0) ? p : NULL;
2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415
}

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 已提交
2416 2417 2418
  size_t tableListSize = taosArrayGetSize(pInfo->tableListInfo->pTableList);
  if (!pInfo->hasGroupId) {
    pInfo->hasGroupId = true;
2419

S
slzhou 已提交
2420
    if (tableListSize == 0) {
2421 2422 2423
      doSetOperatorCompleted(pOperator);
      return NULL;
    }
S
slzhou 已提交
2424 2425
    pInfo->tableStartIndex = 0;
    pInfo->groupId = ((STableKeyInfo*)taosArrayGet(pInfo->tableListInfo->pTableList, pInfo->tableStartIndex))->groupId;
2426 2427
    startGroupTableMergeScan(pOperator);
  }
S
slzhou 已提交
2428 2429 2430 2431 2432 2433 2434 2435 2436 2437 2438 2439 2440 2441 2442 2443 2444 2445
  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 已提交
2446 2447
  }

2448 2449 2450 2451 2452
  return pBlock;
}

void destroyTableMergeScanOperatorInfo(void* param, int32_t numOfOutput) {
  STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;
2453
  cleanupQueryTableDataCond(&pTableScanInfo->cond);
2454 2455 2456 2457 2458 2459 2460 2461 2462 2463 2464

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

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

  taosArrayDestroy(pTableScanInfo->pSortInfo);
}

2465 2466
typedef struct STableMergeScanExecInfo {
  SFileBlockLoadRecorder blockRecorder;
L
Liu Jicong 已提交
2467
  SSortExecInfo          sortExecInfo;
2468 2469
} STableMergeScanExecInfo;

2470 2471
int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {
  ASSERT(pOptr != NULL);
2472 2473
  // TODO: merge these two info into one struct
  STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo));
L
Liu Jicong 已提交
2474
  STableMergeScanInfo*     pInfo = pOptr->info;
2475 2476 2477 2478 2479
  execInfo->blockRecorder = pInfo->readRecorder;
  execInfo->sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle);

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

2481 2482 2483
  return TSDB_CODE_SUCCESS;
}

S
slzhou 已提交
2484 2485 2486 2487 2488 2489
int32_t compareTableKeyInfoByGid(const void* p1, const void* p2) {
  const STableKeyInfo* info1 = p1;
  const STableKeyInfo* info2 = p2;
  return info1->groupId - info2->groupId;
}

2490 2491 2492
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo* pTableListInfo,
                                                SReadHandle* readHandle, SExecTaskInfo* pTaskInfo, uint64_t queryId,
                                                uint64_t taskId) {
2493 2494 2495 2496 2497
  STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));
  SOperatorInfo*       pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }
2498
  if (pTableScanNode->pGroupTags) {
S
slzhou 已提交
2499 2500
    taosArraySort(pTableListInfo->pTableList, compareTableKeyInfoByGid);
  }
2501 2502 2503 2504

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

  int32_t numOfCols = 0;
L
Liu Jicong 已提交
2505
  SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
2506 2507 2508 2509 2510 2511 2512 2513

  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);
2514
    pInfo->pPseudoCtx = createSqlFunctionCtx(pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, &pInfo->rowEntryInfoOffset);
2515 2516 2517 2518
  }

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

L
Liu Jicong 已提交
2519 2520
  pInfo->readHandle = *readHandle;
  pInfo->interval = extractIntervalInfo(pTableScanNode);
2521
  pInfo->sample.sampleRatio = pTableScanNode->ratio;
L
Liu Jicong 已提交
2522 2523 2524
  pInfo->sample.seed = taosGetTimestampSec();
  pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired;
  pInfo->pFilterNode = pTableScanNode->scan.node.pConditions;
2525
  pInfo->tableListInfo = pTableListInfo;
L
Liu Jicong 已提交
2526 2527
  pInfo->scanFlag = MAIN_SCAN;
  pInfo->pColMatchInfo = pColList;
2528 2529

  pInfo->pResBlock = createResDataBlock(pDescNode);
2530 2531 2532
  pInfo->dataReaders = taosArrayInit(64, POINTER_BYTES);
  pInfo->queryId = queryId;
  pInfo->taskId = taskId;
2533

2534
  pInfo->sortSourceParams = taosArrayInit(64, sizeof(STableMergeScanSortSourceParam));
2535

2536 2537
  pInfo->pSortInfo = generateSortByTsInfo(pInfo->cond.order);
  pInfo->pSortInputBlock = createOneDataBlock(pInfo->pResBlock, false);
2538

2539
  int32_t rowSize = pInfo->pResBlock->info.rowSize;
L
Liu Jicong 已提交
2540
  pInfo->bufPageSize = getProperSortPageSize(rowSize);
2541

L
Liu Jicong 已提交
2542
  pOperator->name = "TableMergeScanOperator";
2543
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN;
L
Liu Jicong 已提交
2544 2545 2546
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
L
Liu Jicong 已提交
2547
  pOperator->exprSupp.numOfExprs = numOfCols;
L
Liu Jicong 已提交
2548
  pOperator->pTaskInfo = pTaskInfo;
2549 2550 2551
  initResultSizeInfo(pOperator, 1024);

  pOperator->fpSet =
2552 2553
      createOperatorFpSet(operatorDummyOpenFn, doTableMergeScan, NULL, NULL, destroyTableMergeScanOperatorInfo, NULL,
                          NULL, getTableMergeScanExplainExecInfo);
2554 2555 2556 2557 2558 2559 2560 2561 2562
  pOperator->cost.openCost = 0;
  return pOperator;

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