executorimpl.h 42.7 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * 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/>.
 */
#ifndef TDENGINE_EXECUTORIMPL_H
#define TDENGINE_EXECUTORIMPL_H

18 19 20 21
#ifdef __cplusplus
extern "C" {
#endif

22
#include "os.h"
S
common  
Shengliang Guan 已提交
23
#include "tcommon.h"
24
#include "tlosertree.h"
dengyihao's avatar
dengyihao 已提交
25
#include "tsort.h"
26 27 28
#include "ttszip.h"
#include "tvariant.h"

H
Haojun Liao 已提交
29
#include "dataSinkMgt.h"
30
#include "executil.h"
H
Haojun Liao 已提交
31
#include "executor.h"
H
Haojun Liao 已提交
32
#include "planner.h"
D
dapan1121 已提交
33
#include "scalar.h"
34 35
#include "taosdef.h"
#include "tarray.h"
5
54liuyao 已提交
36
#include "tfill.h"
H
Haojun Liao 已提交
37
#include "thash.h"
38
#include "tlockfree.h"
D
dapan1121 已提交
39
#include "tmsg.h"
dengyihao's avatar
dengyihao 已提交
40
#include "tpagedbuf.h"
L
Liu Jicong 已提交
41
#include "tstream.h"
L
Liu Jicong 已提交
42
#include "tstreamUpdate.h"
H
Haojun Liao 已提交
43

44
#include "executorInt.h"
L
Liu Jicong 已提交
45
#include "vnode.h"
H
Hongze Cheng 已提交
46

47 48
typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int32_t order);

L
Liu Jicong 已提交
49
#define Q_STATUS_EQUAL(p, s)                 (((p) & (s)) != 0u)
5
54liuyao 已提交
50 51 52 53 54
#define IS_VALID_SESSION_WIN(winInfo)        ((winInfo).sessionWin.win.skey > 0)
#define SET_SESSION_WIN_INVALID(winInfo)     ((winInfo).sessionWin.win.skey = INT64_MIN)
#define IS_INVALID_SESSION_WIN_KEY(winKey)   ((winKey).win.skey <= 0)
#define SET_SESSION_WIN_KEY_INVALID(pWinKey) ((pWinKey)->win.skey = INT64_MIN)

55
enum {
56 57
  // when this task starts to execute, this status will set
  TASK_NOT_COMPLETED = 0x1u,
58

59
  /* Task is over
60 61 62
   * 1. this status is used in one row result query process, e.g., count/sum/first/last/ avg...etc.
   * 2. when all data within queried time window, it is also denoted as query_completed
   */
63
  TASK_COMPLETED = 0x2u,
64 65 66 67 68 69
};

/**
 * If the number of generated results is greater than this value,
 * query query will be halt and return results to client immediate.
 */
dengyihao's avatar
dengyihao 已提交
70 71 72 73 74
typedef struct SResultInfo {  // TODO refactor
  int64_t totalRows;          // total generated result size in rows
  int64_t totalBytes;         // total results in bytes.
  int32_t capacity;           // capacity of current result output buffer
  int32_t threshold;          // result size threshold in rows.
H
Haojun Liao 已提交
75
} SResultInfo;
76 77

typedef struct STableQueryInfo {
L
Liu Jicong 已提交
78 79
  TSKEY              lastKey;  // last check ts, todo remove it later
  SResultRowPosition pos;      // current active time window
80 81
} STableQueryInfo;

H
Haojun Liao 已提交
82 83 84 85 86
typedef struct SLimit {
  int64_t limit;
  int64_t offset;
} SLimit;

87
typedef struct STableScanAnalyzeInfo SFileBlockLoadRecorder;
88

H
Haojun Liao 已提交
89
typedef struct STaskCostInfo {
90 91 92 93 94
  int64_t                 created;
  int64_t                 start;
  uint64_t                elapsedTime;
  double                  extractListTime;
  double                  groupIdMapTime;
95
  SFileBlockLoadRecorder* pRecoder;
H
Haojun Liao 已提交
96
} STaskCostInfo;
97

H
Haojun Liao 已提交
98
typedef struct SOperatorCostInfo {
L
Liu Jicong 已提交
99 100
  double openCost;
  double totalCost;
H
Haojun Liao 已提交
101 102
} SOperatorCostInfo;

103 104
struct SOperatorInfo;

wmmhello's avatar
wmmhello 已提交
105
typedef int32_t (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, char** result, int32_t* length);
106
typedef int32_t (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, char* result);
107

108
typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* pOptr);
109
typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* pOptr);
110
typedef void (*__optr_close_fn_t)(void* param);
111
typedef int32_t (*__optr_explain_fn_t)(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len);
H
Haojun Liao 已提交
112

H
Haojun Liao 已提交
113
typedef struct STaskIdInfo {
114 115 116 117
  uint64_t queryId;  // this is also a request id
  uint64_t subplanId;
  uint64_t templateId;
  char*    str;
H
Haojun Liao 已提交
118 119
} STaskIdInfo;

L
Liu Jicong 已提交
120 121
enum {
  STREAM_RECOVER_STEP__NONE = 0,
122 123
  STREAM_RECOVER_STEP__PREPARE1,
  STREAM_RECOVER_STEP__PREPARE2,
L
Liu Jicong 已提交
124 125 126
  STREAM_RECOVER_STEP__SCAN,
};

L
Liu Jicong 已提交
127
typedef struct {
L
Liu Jicong 已提交
128
  // TODO remove prepareStatus
129 130 131 132 133 134
  STqOffsetVal      prepareStatus;  // for tmq
  STqOffsetVal      lastStatus;     // for tmq
  SMqMetaRsp        metaRsp;        // for tmq fetching meta
  int8_t            returned;
  int64_t           snapshotVer;
  const SSubmitReq* pReq;
L
Liu Jicong 已提交
135 136 137 138

  SSchemaWrapper*     schema;
  char                tbName[TSDB_TABLE_NAME_LEN];
  int8_t              recoverStep;
L
Liu Jicong 已提交
139
  SQueryTableDataCond tableCond;
L
Liu Jicong 已提交
140 141
  int64_t             fillHistoryVer1;
  int64_t             fillHistoryVer2;
142

L
Liu Jicong 已提交
143 144
  // int8_t        triggerSaved;
  // int64_t       deleteMarkSaved;
145
  SStreamState* pState;
L
Liu Jicong 已提交
146 147
} SStreamTaskInfo;

148 149 150 151 152
typedef struct {
  char*           tablename;
  char*           dbname;
  int32_t         tversion;
  SSchemaWrapper* sw;
153
  SSchemaWrapper* qsw;
154 155
} SSchemaInfo;

D
dapan1121 已提交
156
typedef struct {
L
Liu Jicong 已提交
157 158
  int32_t operatorType;
  int64_t refId;
D
dapan1121 已提交
159 160 161 162 163 164 165
} SExchangeOpStopInfo;

typedef struct {
  SRWLatch lock;
  SArray*  pStopInfo;
} STaskStopInfo;

166
typedef struct SExecTaskInfo {
167 168 169 170 171 172 173
  STaskIdInfo   id;
  uint32_t      status;
  STimeWindow   window;
  STaskCostInfo cost;
  int64_t       owner;  // if it is in execution
  int32_t       code;

L
Liu Jicong 已提交
174 175 176
  int64_t               version;  // used for stream to record wal version
  SStreamTaskInfo       streamInfo;
  SSchemaInfo           schemaInfo;
H
Haojun Liao 已提交
177
  STableListInfo*       pTableInfoList;  // this is a table list
L
Liu Jicong 已提交
178 179 180 181
  const char*           sql;             // query sql string
  jmp_buf               env;             // jump to this position when error happens.
  EOPTR_EXEC_MODEL      execModel;       // operator execution model [batch model|stream model]
  SSubplan*             pSubplan;
182
  struct SOperatorInfo* pRoot;
183
  SLocalFetch           localFetch;
D
dapan1121 已提交
184
  STaskStopInfo         stopInfo;
185
} SExecTaskInfo;
H
Haojun Liao 已提交
186

187
enum {
L
Liu Jicong 已提交
188 189
  OP_NOT_OPENED = 0x0,
  OP_OPENED = 0x1,
H
Haojun Liao 已提交
190
  OP_RES_TO_RETURN = 0x5,
L
Liu Jicong 已提交
191
  OP_EXEC_DONE = 0x9,
L
Liu Jicong 已提交
192
  OP_EXEC_RECV = 0x11,
193 194
};

195
typedef struct SOperatorFpSet {
L
Liu Jicong 已提交
196 197
  __optr_open_fn_t    _openFn;  // DO NOT invoke this function directly
  __optr_fn_t         getNextFn;
198
  __optr_fn_t         cleanupFn;  // call this function to release the allocated resources ASAP
L
Liu Jicong 已提交
199 200 201 202
  __optr_close_fn_t   closeFn;
  __optr_encode_fn_t  encodeResultRow;
  __optr_decode_fn_t  decodeResultRow;
  __optr_explain_fn_t getExplainFn;
203 204
} SOperatorFpSet;

205 206
typedef struct SExprSupp {
  SExprInfo*      pExprInfo;
L
Liu Jicong 已提交
207
  int32_t         numOfExprs;  // the number of scalar expression in group operator
208 209
  SqlFunctionCtx* pCtx;
  int32_t*        rowEntryInfoOffset;  // offset value for each row result cell info
H
Haojun Liao 已提交
210
  SFilterInfo*    pFilterInfo;
211 212
} SExprSupp;

213
typedef struct SOperatorInfo {
214
  uint16_t               operatorType;
215
  int16_t                resultDataBlockId;
L
Liu Jicong 已提交
216 217 218 219 220 221 222 223 224 225 226
  bool                   blocking;  // block operator or not
  uint8_t                status;    // denote if current operator is completed
  char*                  name;      // name, for debug purpose
  void*                  info;      // extension attribution
  SExprSupp              exprSupp;
  SExecTaskInfo*         pTaskInfo;
  SOperatorCostInfo      cost;
  SResultInfo            resultInfo;
  struct SOperatorInfo** pDownstream;      // downstram pointer list
  int32_t                numOfDownstream;  // number of downstream. The value is always ONE expect for join operator
  SOperatorFpSet         fpSet;
227 228
} SOperatorInfo;

229 230
typedef enum {
  EX_SOURCE_DATA_NOT_READY = 0x1,
L
Liu Jicong 已提交
231
  EX_SOURCE_DATA_READY = 0x2,
232 233
  EX_SOURCE_DATA_EXHAUSTED = 0x3,
} EX_SOURCE_STATUS;
234

235 236 237
#define COL_MATCH_FROM_COL_ID  0x1
#define COL_MATCH_FROM_SLOT_ID 0x2

238
typedef struct SSourceDataInfo {
L
Liu Jicong 已提交
239 240 241 242 243 244
  int32_t            index;
  SRetrieveTableRsp* pRsp;
  uint64_t           totalRows;
  int32_t            code;
  EX_SOURCE_STATUS   status;
  const char*        taskId;
245 246
} SSourceDataInfo;

H
Haojun Liao 已提交
247
typedef struct SLoadRemoteDataInfo {
dengyihao's avatar
dengyihao 已提交
248 249 250
  uint64_t totalSize;     // total load bytes from remote
  uint64_t totalRows;     // total number of rows
  uint64_t totalElapsed;  // total elapsed time
H
Haojun Liao 已提交
251 252
} SLoadRemoteDataInfo;

253
typedef struct SLimitInfo {
L
Liu Jicong 已提交
254 255 256 257 258 259 260
  SLimit   limit;
  SLimit   slimit;
  uint64_t currentGroupId;
  int64_t  remainGroupOffset;
  int64_t  numOfOutputGroups;
  int64_t  remainOffset;
  int64_t  numOfOutputRows;
261 262
} SLimitInfo;

263
typedef struct SExchangeInfo {
L
Liu Jicong 已提交
264 265 266 267
  SArray* pSources;
  SArray* pSourceDataInfo;
  tsem_t  ready;
  void*   pTransporter;
268 269 270
  // SArray<SSDataBlock*>, result block list, used to keep the multi-block that
  // passed by downstream operator
  SArray*             pResultBlockList;
L
Liu Jicong 已提交
271 272 273
  int32_t             rspBlockIndex;  // indicate the return block index in pResultBlockList
  SSDataBlock*        pDummyBlock;    // dummy block, not keep data
  bool                seqLoadData;    // sequential load data or not, false by default
dengyihao's avatar
dengyihao 已提交
274
  int32_t             current;
H
Haojun Liao 已提交
275
  SLoadRemoteDataInfo loadInfo;
276
  uint64_t            self;
277
  SLimitInfo          limitInfo;
L
Liu Jicong 已提交
278
  int64_t             openedTs;  // start exec time stamp
279 280
} SExchangeInfo;

281
typedef struct SScanInfo {
L
Liu Jicong 已提交
282 283
  int32_t numOfAsc;
  int32_t numOfDesc;
284 285
} SScanInfo;

286
typedef struct SSampleExecInfo {
L
Liu Jicong 已提交
287 288
  double   sampleRatio;  // data block sample ratio, 1 by default
  uint32_t seed;         // random seed value
289 290
} SSampleExecInfo;

L
Liu Jicong 已提交
291 292
enum {
  TABLE_SCAN__TABLE_ORDER = 1,
293
  TABLE_SCAN__BLOCK_ORDER = 2,
L
Liu Jicong 已提交
294 295
};

296
typedef struct SAggSupporter {
H
Haojun Liao 已提交
297
  SSHashObj*     pResultRowHashTable;  // quick locate the window object for each result
298 299
  char*          keyBuf;               // window key buffer
  SDiskbasedBuf* pResultBuf;           // query result buffer based on blocked-wised disk file
L
Liu Jicong 已提交
300 301
  int32_t        resultRowSize;  // the result buffer size for each result row, with the meta data size for each row
  int32_t        currentPageId;  // current write page id
302 303 304
} SAggSupporter;

typedef struct {
L
Liu Jicong 已提交
305 306 307 308 309
  // if the upstream is an interval operator, the interval info is also kept here to get the time window to check if
  // current data block needs to be loaded.
  SInterval      interval;
  SAggSupporter* pAggSup;
  SExprSupp*     pExprSup;  // expr supporter of aggregate operator
310 311
} SAggOptrPushDownInfo;

312
typedef struct STableMetaCacheInfo {
L
Liu Jicong 已提交
313 314 315
  SLRUCache* pTableMetaEntryCache;  // 100 by default
  uint64_t   metaFetch;
  uint64_t   cacheHit;
316 317
} STableMetaCacheInfo;

318
typedef struct STableScanInfo {
319 320 321
  STsdbReader*           dataReader;
  SReadHandle            readHandle;
  SLimitInfo             limitInfo;
322
  SFileBlockLoadRecorder readRecorder;
L
Liu Jicong 已提交
323 324
  SScanInfo              scanInfo;
  int32_t                scanTimes;
325 326 327 328 329 330 331 332 333 334 335 336
  SSDataBlock*           pResBlock;
  SColMatchInfo          matchInfo;
  SExprSupp              pseudoSup;
  SQueryTableDataCond    cond;
  int32_t                scanFlag;  // table scan flag to denote if it is a repeat/reverse/main scan
  int32_t                dataBlockLoadFlag;
  SSampleExecInfo        sample;  // sample execution info
  int32_t                currentGroupId;
  int32_t                currentTable;
  int8_t                 scanMode;
  SAggOptrPushDownInfo   pdInfo;
  int8_t                 assignBlockUid;
337
  STableMetaCacheInfo    metaCache;
338 339
} STableScanInfo;

340
typedef struct STableMergeScanInfo {
dengyihao's avatar
opt mem  
dengyihao 已提交
341 342 343 344 345
  STableListInfo*        tableListInfo;
  int32_t                tableStartIndex;
  int32_t                tableEndIndex;
  bool                   hasGroupId;
  uint64_t               groupId;
L
Liu Jicong 已提交
346
  SArray*                queryConds;  // array of queryTableDataCond
dengyihao's avatar
opt mem  
dengyihao 已提交
347 348 349 350 351 352 353 354 355 356
  STsdbReader*           pReader;
  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
  SArray*                sortSourceParams;
  SLimitInfo             limitInfo;
357 358 359 360
  SFileBlockLoadRecorder readRecorder;
  int64_t                numOfRows;
  SScanInfo              scanInfo;
  int32_t                scanTimes;
L
Liu Jicong 已提交
361
  SqlFunctionCtx*        pCtx;  // which belongs to the direct upstream operator operator query context
362 363 364 365
  SResultRowInfo*        pResultRowInfo;
  int32_t*               rowEntryInfoOffset;
  SExprInfo*             pExpr;
  SSDataBlock*           pResBlock;
H
Haojun Liao 已提交
366
  SColMatchInfo          matchInfo;
367
  int32_t                numOfOutput;
H
Haojun Liao 已提交
368 369 370 371
  SExprSupp              pseudoSup;
  SQueryTableDataCond    cond;
  int32_t                scanFlag;  // table scan flag to denote if it is a repeat/reverse/main scan
  int32_t                dataBlockLoadFlag;
372

373 374 375 376
  // if the upstream is an interval operator, the interval info is also kept here to get the time
  // window to check if current data block needs to be loaded.
  SInterval       interval;
  SSampleExecInfo sample;  // sample execution info
L
Liu Jicong 已提交
377
  SSortExecInfo   sortExecInfo;
378 379
} STableMergeScanInfo;

380
typedef struct STagScanInfo {
L
Liu Jicong 已提交
381 382
  SColumnInfo*    pCols;
  SSDataBlock*    pRes;
H
Haojun Liao 已提交
383
  SColMatchInfo   matchInfo;
L
Liu Jicong 已提交
384 385 386
  int32_t         curPos;
  SReadHandle     readHandle;
  STableListInfo* pTableList;
387 388
} STagScanInfo;

389
typedef struct SLastrowScanInfo {
L
Liu Jicong 已提交
390 391 392 393 394 395 396 397 398 399 400
  SSDataBlock*  pRes;
  SReadHandle   readHandle;
  void*         pLastrowReader;
  SColMatchInfo matchInfo;
  int32_t*      pSlotIds;
  SExprSupp     pseudoExprSup;
  int32_t       retrieveType;
  int32_t       currentGroupIndex;
  SSDataBlock*  pBufferredRes;
  SArray*       pUidList;
  int32_t       indexOfBufferedRes;
401 402
} SLastrowScanInfo;

5
54liuyao 已提交
403 404 405 406
typedef enum EStreamScanMode {
  STREAM_SCAN_FROM_READERHANDLE = 1,
  STREAM_SCAN_FROM_RES,
  STREAM_SCAN_FROM_UPDATERES,
407
  STREAM_SCAN_FROM_DELETE_DATA,
5
54liuyao 已提交
408
  STREAM_SCAN_FROM_DATAREADER_RETRIEVE,
L
Liu Jicong 已提交
409
  STREAM_SCAN_FROM_DATAREADER_RANGE,
5
54liuyao 已提交
410 411
} EStreamScanMode;

412 413 414 415 416
enum {
  PROJECT_RETRIEVE_CONTINUE = 0x1,
  PROJECT_RETRIEVE_DONE = 0x2,
};

5
54liuyao 已提交
417
typedef struct SCatchSupporter {
L
Liu Jicong 已提交
418 419 420 421
  SHashObj*      pWindowHashTable;  // quick locate the window object for each window
  SDiskbasedBuf* pDataBuf;          // buffer based on blocked-wised disk file
  int32_t        keySize;
  int64_t*       pKeyBuf;
5
54liuyao 已提交
422 423
} SCatchSupporter;

5
54liuyao 已提交
424
typedef struct SStreamAggSupporter {
5
54liuyao 已提交
425 426 427
  int32_t         resultRowSize;  // the result buffer size for each result row, with the meta data size for each row
  SSDataBlock*    pScanBlock;
  SStreamState*   pState;
428 429
  int64_t         gap;        // stream session window gap
  SqlFunctionCtx* pDummyCtx;  // for combine
5
54liuyao 已提交
430 431 432 433
  SSHashObj*      pResultRows;
  int32_t         stateKeySize;
  int16_t         stateKeyType;
  SDiskbasedBuf*  pResultBuf;
5
54liuyao 已提交
434 435
} SStreamAggSupporter;

436
typedef struct SWindowSupporter {
5
54liuyao 已提交
437
  SStreamAggSupporter* pStreamAggSup;
L
Liu Jicong 已提交
438
  int64_t              gap;
439
  uint16_t             parentType;
440
  SAggSupporter*       pIntervalAggSup;
441
} SWindowSupporter;
442

443
typedef struct SPartitionBySupporter {
L
Liu Jicong 已提交
444 445 446 447
  SArray* pGroupCols;     // group by columns, SArray<SColumn>
  SArray* pGroupColVals;  // current group column values, SArray<SGroupKeys>
  char*   keyBuf;         // group by keys for hash
  bool    needCalc;       // partition by column
448 449 450
} SPartitionBySupporter;

typedef struct SPartitionDataInfo {
L
Liu Jicong 已提交
451
  uint64_t groupId;
452 453
  char*    tbname;
  SArray*  tags;
L
Liu Jicong 已提交
454
  SArray*  rowIds;
455
} SPartitionDataInfo;
456

5
54liuyao 已提交
457
typedef struct STimeWindowAggSupp {
L
Liu Jicong 已提交
458
  int8_t          calTrigger;
L
Liu Jicong 已提交
459
  int8_t          calTriggerSaved;
5
54liuyao 已提交
460
  int64_t         deleteMark;
L
Liu Jicong 已提交
461 462
  int64_t         deleteMarkSaved;
  int64_t         waterMark;
L
Liu Jicong 已提交
463
  TSKEY           maxTs;
5
54liuyao 已提交
464
  TSKEY           minTs;
L
Liu Jicong 已提交
465
  SColumnInfoData timeWindowData;  // query time window info for scalar function execution.
5
54liuyao 已提交
466 467
} STimeWindowAggSupp;

468
typedef struct SStreamScanInfo {
L
Liu Jicong 已提交
469 470 471 472 473 474 475 476 477
  uint64_t      tableUid;  // queried super table uid
  SExprInfo*    pPseudoExpr;
  int32_t       numOfPseudoExpr;
  SExprSupp     tbnameCalSup;
  SExprSupp     tagCalSup;
  int32_t       primaryTsIndex;  // primary time stamp slot id
  SReadHandle   readHandle;
  SInterval     interval;  // if the upstream is an interval operator, the interval info is also kept here.
  SColMatchInfo matchInfo;
L
Liu Jicong 已提交
478 479 480 481 482 483 484 485 486 487 488 489 490

  SArray*      pBlockLists;  // multiple SSDatablock.
  SSDataBlock* pRes;         // result SSDataBlock
  SSDataBlock* pUpdateRes;   // update SSDataBlock
  int32_t      updateResIndex;
  int32_t      blockType;        // current block type
  int32_t      validBlockIndex;  // Is current data has returned?
  uint64_t     numOfExec;        // execution times
  STqReader*   tqReader;

  uint64_t     groupId;
  SUpdateInfo* pUpdateInfo;

L
Liu Jicong 已提交
491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506
  EStreamScanMode       scanMode;
  SOperatorInfo*        pStreamScanOp;
  SOperatorInfo*        pTableScanOp;
  SArray*               childIds;
  SWindowSupporter      windowSup;
  SPartitionBySupporter partitionSup;
  SExprSupp*            pPartScalarSup;
  bool                  assignBlockUid;  // assign block uid to groupId, temporarily used for generating rollup SMA.
  int32_t               scanWinIndex;    // for state operator
  int32_t               pullDataResIndex;
  SSDataBlock*          pPullDataRes;    // pull data SSDataBlock
  SSDataBlock*          pDeleteDataRes;  // delete data SSDataBlock
  int32_t               deleteDataIndex;
  STimeWindow           updateWin;
  STimeWindowAggSupp    twAggSup;
  SSDataBlock*          pUpdateDataRes;
L
Liu Jicong 已提交
507
  SHashObj*             pGroupIdTbNameMap;
L
Liu Jicong 已提交
508
  // status for tmq
L
Liu Jicong 已提交
509 510 511
  SNodeList* pGroupTags;
  SNode*     pTagCond;
  SNode*     pTagIndexCond;
512
} SStreamScanInfo;
H
Haojun Liao 已提交
513

L
Liu Jicong 已提交
514 515 516 517 518 519 520 521 522 523 524 525
typedef struct {
  //  int8_t    subType;
  //  bool      withMeta;
  //  int64_t   suid;
  //  int64_t   snapVersion;
  //  void     *metaInfo;
  //  void     *dataInfo;
  SVnode*       vnode;
  SSDataBlock   pRes;  // result SSDataBlock
  STsdbReader*  dataReader;
  SSnapContext* sContext;
} SStreamRawScanInfo;
526

dengyihao's avatar
dengyihao 已提交
527
typedef struct SSysTableIndex {
528 529 530
  int8_t  init;
  SArray* uids;
  int32_t lastIdx;
dengyihao's avatar
dengyihao 已提交
531 532
} SSysTableIndex;

533
typedef struct SSysTableScanInfo {
dengyihao's avatar
dengyihao 已提交
534 535 536 537
  SRetrieveMetaTableRsp* pRsp;
  SRetrieveTableReq      req;
  SEpSet                 epSet;
  tsem_t                 ready;
538 539
  SReadHandle            readHandle;
  int32_t                accountId;
540
  const char*            pUser;
541
  bool                   sysInfo;
542 543 544
  bool                   showRewrite;
  SNode*                 pCondition;  // db_name filter condition, to discard data that are not in current database
  SMTbCursor*            pCur;        // cursor for iterate the local table meta store.
545
  SSysTableIndex*        pIdx;        // idx for local table meta
H
Haojun Liao 已提交
546
  SColMatchInfo          matchInfo;
547 548 549 550
  SName                  name;
  SSDataBlock*           pRes;
  int64_t                numOfBlocks;  // extract basic running information.
  SLoadRemoteDataInfo    loadInfo;
551 552
} SSysTableScanInfo;

553
typedef struct SBlockDistInfo {
L
Liu Jicong 已提交
554 555 556 557
  SSDataBlock* pResBlock;
  STsdbReader* pHandle;
  SReadHandle  readHandle;
  uint64_t     uid;  // table uid
558 559
} SBlockDistInfo;

560
// todo remove this
561
typedef struct SOptrBasicInfo {
L
Liu Jicong 已提交
562 563 564
  SResultRowInfo resultRowInfo;
  SSDataBlock*   pRes;
  bool           mergeResultBlock;
565 566
} SOptrBasicInfo;

567
typedef struct SIntervalAggOperatorInfo {
568
  SOptrBasicInfo     binfo;              // basic info
wmmhello's avatar
wmmhello 已提交
569
  SAggSupporter      aggSup;             // aggregate supporter
570
  SExprSupp          scalarSupp;         // supporter for perform scalar function
571 572 573 574 575
  SGroupResInfo      groupResInfo;       // multiple results build supporter
  SInterval          interval;           // interval info
  int32_t            primaryTsIndex;     // primary time stamp slot id from result of downstream operator.
  STimeWindow        win;                // query time range
  bool               timeWindowInterpo;  // interpolation needed or not
576
  SArray*            pInterpCols;        // interpolation columns
577
  int32_t            resultTsOrder;      // result timestamp order
578
  int32_t            inputOrder;         // input data ts order
579 580
  EOPTR_EXEC_MODEL   execModel;          // operator execution model [batch model|stream model]
  STimeWindowAggSupp twAggSup;
L
Liu Jicong 已提交
581
  SArray*            pPrevValues;  //  SArray<SGroupKeys> used to keep the previous not null value for interpolation.
582
} SIntervalAggOperatorInfo;
583

584
typedef struct SMergeAlignedIntervalAggOperatorInfo {
L
Liu Jicong 已提交
585
  SIntervalAggOperatorInfo* intervalAggOperatorInfo;
586

587 588
  uint64_t     groupId;  // current groupId
  int64_t      curTs;    // current ts
589
  SSDataBlock* prefetchedBlock;
590
  SResultRow*  pResultRow;
591 592
} SMergeAlignedIntervalAggOperatorInfo;

593
typedef struct SStreamIntervalOperatorInfo {
594 595 596 597 598 599
  SOptrBasicInfo     binfo;           // basic info
  SAggSupporter      aggSup;          // aggregate supporter
  SExprSupp          scalarSupp;      // supporter for perform scalar function
  SGroupResInfo      groupResInfo;    // multiple results build supporter
  SInterval          interval;        // interval info
  int32_t            primaryTsIndex;  // primary time stamp slot id from result of downstream operator.
600 601 602
  STimeWindowAggSupp twAggSup;
  bool               invertible;
  bool               ignoreExpiredData;
603
  SArray*            pDelWins;  // SWinRes
604 605
  int32_t            delIndex;
  SSDataBlock*       pDelRes;
606
  SPhysiNode*        pPhyNode;  // create new child
5
54liuyao 已提交
607
  SHashObj*          pPullDataMap;
608
  SArray*            pPullWins;  // SPullWindowInfo
5
54liuyao 已提交
609 610
  int32_t            pullIndex;
  SSDataBlock*       pPullDataRes;
611 612 613 614
  bool               isFinal;
  SArray*            pChildren;
  SStreamState*      pState;
  SWinKey            delKey;
615
  SHashObj*          pGroupIdTbNameMap;  // uint64_t -> char[TSDB_TABLE_NAME_LEN]
616
} SStreamIntervalOperatorInfo;
5
54liuyao 已提交
617

618
typedef struct SAggOperatorInfo {
H
Haojun Liao 已提交
619 620
  SOptrBasicInfo   binfo;
  SAggSupporter    aggSup;
L
Liu Jicong 已提交
621 622 623 624
  STableQueryInfo* current;
  uint64_t         groupId;
  SGroupResInfo    groupResInfo;
  SExprSupp        scalarExprSup;
625 626 627
} SAggOperatorInfo;

typedef struct SProjectOperatorInfo {
L
Liu Jicong 已提交
628 629 630 631 632 633
  SOptrBasicInfo binfo;
  SAggSupporter  aggSup;
  SArray*        pPseudoColInfo;
  SLimitInfo     limitInfo;
  bool           mergeDataBlocks;
  SSDataBlock*   pFinalRes;
634 635
} SProjectOperatorInfo;

H
Haojun Liao 已提交
636
typedef struct SIndefOperatorInfo {
L
Liu Jicong 已提交
637 638 639 640 641
  SOptrBasicInfo binfo;
  SAggSupporter  aggSup;
  SArray*        pPseudoColInfo;
  SExprSupp      scalarSup;
  uint64_t       groupId;
642
  SSDataBlock*   pNextGroupRes;
H
Haojun Liao 已提交
643 644
} SIndefOperatorInfo;

645
typedef struct SFillOperatorInfo {
646 647
  struct SFillInfo* pFillInfo;
  SSDataBlock*      pRes;
H
Haojun Liao 已提交
648
  SSDataBlock*      pFinalRes;
649 650 651
  int64_t           totalInputRows;
  void**            p;
  SSDataBlock*      existNewGroupBlock;
652
  STimeWindow       win;
H
Haojun Liao 已提交
653
  SColMatchInfo     matchInfo;
H
Haojun Liao 已提交
654
  int32_t           primaryTsCol;
H
Haojun Liao 已提交
655
  int32_t           primarySrcSlotId;
L
Liu Jicong 已提交
656
  uint64_t          curGroupId;  // current handled group id
657 658
  SExprInfo*        pExprInfo;
  int32_t           numOfExpr;
659
  SExprSupp         noFillExprSupp;
660 661 662
} SFillOperatorInfo;

typedef struct SGroupbyOperatorInfo {
L
Liu Jicong 已提交
663 664
  SOptrBasicInfo binfo;
  SAggSupporter  aggSup;
H
Haojun Liao 已提交
665 666
  SArray*        pGroupCols;     // group by columns, SArray<SColumn>
  SArray*        pGroupColVals;  // current group column values, SArray<SGroupKeys>
L
Liu Jicong 已提交
667 668 669
  bool           isInit;         // denote if current val is initialized or not
  char*          keyBuf;         // group by keys for hash
  int32_t        groupKeyLen;    // total group by column width
H
Haojun Liao 已提交
670 671
  SGroupResInfo  groupResInfo;
  SExprSupp      scalarSup;
672 673
} SGroupbyOperatorInfo;

H
Haojun Liao 已提交
674
typedef struct SDataGroupInfo {
L
Liu Jicong 已提交
675 676 677
  uint64_t groupId;
  int64_t  numOfRows;
  SArray*  pPageList;
H
Haojun Liao 已提交
678 679 680 681
} SDataGroupInfo;

// The sort in partition may be needed later.
typedef struct SPartitionOperatorInfo {
dengyihao's avatar
dengyihao 已提交
682 683 684 685 686 687 688
  SOptrBasicInfo binfo;
  SArray*        pGroupCols;
  SArray*        pGroupColVals;  // current group column values, SArray<SGroupKeys>
  char*          keyBuf;         // group by keys for hash
  int32_t        groupKeyLen;    // total group by column width
  SHashObj*      pGroupSet;      // quick locate the window object for each result

L
Liu Jicong 已提交
689 690 691 692 693 694
  SDiskbasedBuf* pBuf;              // query result buffer based on blocked-wised disk file
  int32_t        rowCapacity;       // maximum number of rows for each buffer page
  int32_t*       columnOffset;      // start position for each column data
  SArray*        sortedGroupArray;  // SDataGroupInfo sorted by group id
  int32_t        groupIndex;        // group index
  int32_t        pageIndex;         // page index of current group
695
  SExprSupp      scalarSup;
H
Haojun Liao 已提交
696
} SPartitionOperatorInfo;
697

698
typedef struct SWindowRowsSup {
dengyihao's avatar
dengyihao 已提交
699 700 701 702
  STimeWindow win;
  TSKEY       prevTs;
  int32_t     startRowIndex;
  int32_t     numOfRows;
703
  uint64_t    groupId;
704 705
} SWindowRowsSup;

H
Haojun Liao 已提交
706
typedef struct SSessionAggOperatorInfo {
L
Liu Jicong 已提交
707 708
  SOptrBasicInfo binfo;
  SAggSupporter  aggSup;
wmmhello's avatar
wmmhello 已提交
709

710 711
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
L
Liu Jicong 已提交
712 713 714
  bool               reptScan;  // next round scan
  int64_t            gap;       // session window gap
  int32_t            tsSlotId;  // primary timestamp slot id
715
  STimeWindowAggSupp twAggSup;
H
Haojun Liao 已提交
716
} SSessionAggOperatorInfo;
717

5
54liuyao 已提交
718
typedef struct SResultWindowInfo {
719 720 721
  void*       pOutputBuf;
  SSessionKey sessionWin;
  bool        isOutput;
5
54liuyao 已提交
722 723
} SResultWindowInfo;

5
54liuyao 已提交
724 725
typedef struct SStateWindowInfo {
  SResultWindowInfo winInfo;
5
54liuyao 已提交
726
  SStateKeys*       pStateKey;
5
54liuyao 已提交
727 728
} SStateWindowInfo;

5
54liuyao 已提交
729
typedef struct SStreamSessionAggOperatorInfo {
L
Liu Jicong 已提交
730 731
  SOptrBasicInfo      binfo;
  SStreamAggSupporter streamAggSup;
732
  SExprSupp           scalarSupp;  // supporter for perform scalar function
L
Liu Jicong 已提交
733 734 735 736 737
  SGroupResInfo       groupResInfo;
  int32_t             primaryTsIndex;  // primary timestamp slot id
  int32_t             endTsIndex;      // window end timestamp slot id
  int32_t             order;           // current SSDataBlock scan order
  STimeWindowAggSupp  twAggSup;
738 739 740
  SSDataBlock*        pWinBlock;   // window result
  SSDataBlock*        pDelRes;     // delete result
  SSDataBlock*        pUpdateRes;  // update window
L
Liu Jicong 已提交
741
  bool                returnUpdate;
5
54liuyao 已提交
742
  SSHashObj*          pStDeleted;
L
Liu Jicong 已提交
743
  void*               pDelIterator;
744 745
  SArray*             pChildren;  // cache for children's result; final stream operator
  SPhysiNode*         pPhyNode;   // create new child
L
Liu Jicong 已提交
746 747
  bool                isFinal;
  bool                ignoreExpiredData;
748
  SHashObj*           pGroupIdTbNameMap;
5
54liuyao 已提交
749 750
} SStreamSessionAggOperatorInfo;

5
54liuyao 已提交
751 752 753
typedef struct SStreamStateAggOperatorInfo {
  SOptrBasicInfo      binfo;
  SStreamAggSupporter streamAggSup;
754
  SExprSupp           scalarSupp;  // supporter for perform scalar function
5
54liuyao 已提交
755 756 757 758 759 760 761
  SGroupResInfo       groupResInfo;
  int32_t             primaryTsIndex;  // primary timestamp slot id
  STimeWindowAggSupp  twAggSup;
  SColumn             stateCol;
  SSDataBlock*        pDelRes;
  SSHashObj*          pSeDeleted;
  void*               pDelIterator;
762
  SArray*             pChildren;  // cache for children's result;
5
54liuyao 已提交
763 764 765 766
  bool                ignoreExpiredData;
  SHashObj*           pGroupIdTbNameMap;
} SStreamStateAggOperatorInfo;

767 768 769 770
typedef struct SStreamPartitionOperatorInfo {
  SOptrBasicInfo        binfo;
  SPartitionBySupporter partitionSup;
  SExprSupp             scalarSup;
771
  SExprSupp             tbnameCalSup;
L
Liu Jicong 已提交
772
  SExprSupp             tagCalSup;
773 774 775 776
  SHashObj*             pPartitions;
  void*                 parIte;
  SSDataBlock*          pInputDataBlock;
  int32_t               tsColIndex;
777
  SSDataBlock*          pDelRes;
778 779
} SStreamPartitionOperatorInfo;

5
54liuyao 已提交
780 781
typedef struct SStreamFillOperatorInfo {
  SStreamFillSupporter* pFillSup;
782 783 784 785 786 787 788
  SSDataBlock*          pRes;
  SSDataBlock*          pSrcBlock;
  int32_t               srcRowIndex;
  SSDataBlock*          pPrevSrcBlock;
  SSDataBlock*          pSrcDelBlock;
  int32_t               srcDelRowIndex;
  SSDataBlock*          pDelRes;
L
Liu Jicong 已提交
789
  SColMatchInfo         matchInfo;
790 791 792
  int32_t               primaryTsCol;
  int32_t               primarySrcSlotId;
  SStreamFillInfo*      pFillInfo;
5
54liuyao 已提交
793 794
} SStreamFillOperatorInfo;

795
typedef struct STimeSliceOperatorInfo {
L
Liu Jicong 已提交
796 797 798 799 800 801 802 803 804 805 806 807 808 809
  SSDataBlock*         pRes;
  STimeWindow          win;
  SInterval            interval;
  int64_t              current;
  SArray*              pPrevRow;     // SArray<SGroupValue>
  SArray*              pNextRow;     // SArray<SGroupValue>
  SArray*              pLinearInfo;  // SArray<SFillLinearInfo>
  bool                 fillLastPoint;
  bool                 isPrevRowSet;
  bool                 isNextRowSet;
  int32_t              fillType;      // fill type
  SColumn              tsCol;         // primary timestamp column
  SExprSupp            scalarSup;     // scalar calculation
  struct SFillColInfo* pFillColInfo;  // fill column info
810 811
} STimeSliceOperatorInfo;

812
typedef struct SStateWindowOperatorInfo {
wmmhello's avatar
wmmhello 已提交
813
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
L
Liu Jicong 已提交
814 815
  SOptrBasicInfo binfo;
  SAggSupporter  aggSup;
816
  SExprSupp      scalarSup;
wmmhello's avatar
wmmhello 已提交
817

818 819
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
820
  SColumn            stateCol;  // start row index
821 822
  bool               hasKey;
  SStateKeys         stateKey;
823
  int32_t            tsSlotId;  // primary timestamp column slot id
824
  STimeWindowAggSupp twAggSup;
825 826
} SStateWindowOperatorInfo;

H
Haojun Liao 已提交
827
typedef struct SSortOperatorInfo {
828
  SOptrBasicInfo binfo;
L
Liu Jicong 已提交
829 830 831
  uint32_t       sortBufSize;  // max buffer size for in-memory sort
  SArray*        pSortInfo;
  SSortHandle*   pSortHandle;
H
Haojun Liao 已提交
832
  SColMatchInfo  matchInfo;
L
Liu Jicong 已提交
833
  int32_t        bufPageSize;
834 835 836
  int64_t        startTs;      // sort start time
  uint64_t       sortElapsed;  // sort elapsed time, time to flush to disk not included.
  SLimitInfo     limitInfo;
H
Haojun Liao 已提交
837
} SSortOperatorInfo;
838

839
typedef struct SJoinOperatorInfo {
L
Liu Jicong 已提交
840 841 842 843 844 845 846 847 848 849 850 851
  SSDataBlock* pRes;
  int32_t      joinType;
  int32_t      inputOrder;

  SSDataBlock* pLeft;
  int32_t      leftPos;
  SColumnInfo  leftCol;

  SSDataBlock* pRight;
  int32_t      rightPos;
  SColumnInfo  rightCol;
  SNode*       pCondAfterMerge;
852 853
} SJoinOperatorInfo;

854 855 856
#define OPTR_IS_OPENED(_optr)  (((_optr)->status & OP_OPENED) == OP_OPENED)
#define OPTR_SET_OPENED(_optr) ((_optr)->status |= OP_OPENED)

H
Haojun Liao 已提交
857 858
SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup,
                                   __optr_close_fn_t closeFn, __optr_explain_fn_t explain);
859

H
Haojun Liao 已提交
860 861
int32_t operatorDummyOpenFn(SOperatorInfo* pOperator);
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num);
862 863 864

void    initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
void    cleanupBasicInfo(SOptrBasicInfo* pInfo);
865
int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
866
void    cleanupExprSupp(SExprSupp* pSup);
867
void    destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs);
L
Liu Jicong 已提交
868
int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
869
                    const char* pkey);
L
Liu Jicong 已提交
870
void    initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows);
871 872 873

void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                           SDiskbasedBuf* pBuf);
874 875
void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                            SDiskbasedBuf* pBuf);
876

L
Liu Jicong 已提交
877
int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf);
878 879
bool    hasLimitOffsetInfo(SLimitInfo* pLimitInfo);
void    initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo);
880
void applyLimitOffset(SLimitInfo* pLimitInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo, SOperatorInfo* pOperator);
881

L
Liu Jicong 已提交
882 883
void doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, SColumnInfoData* pTimeWindowData, int32_t offset,
                      int32_t forwardStep, int32_t numOfTotal, int32_t numOfOutput);
884

H
Haojun Liao 已提交
885
int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pColList, char** pNextStart);
L
Liu Jicong 已提交
886 887
void    updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int32_t numOfRows, int32_t dataLen, int64_t startTs,
                             SOperatorInfo* pOperator);
888

889 890
STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order);

L
Liu Jicong 已提交
891
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag);
892
int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz);
893

L
Liu Jicong 已提交
894
void doDestroyExchangeOperatorInfo(void* param);
H
Haojun Liao 已提交
895

H
Haojun Liao 已提交
896
void    setOperatorCompleted(SOperatorInfo* pOperator);
L
Liu Jicong 已提交
897 898
void    setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status,
                        void* pInfo, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
899
void    doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo);
L
Liu Jicong 已提交
900 901
int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock,
                               int32_t rows, const char* idStr, STableMetaCacheInfo* pCache);
902

L
Liu Jicong 已提交
903 904
void cleanupAggSup(SAggSupporter* pAggSup);
void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
H
Haojun Liao 已提交
905
void setTbNameColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, const char* name);
906

L
Liu Jicong 已提交
907
int32_t doPrepareScan(SOperatorInfo* pOperator, uint64_t uid, int64_t ts);
908
int32_t doGetScanStatus(SOperatorInfo* pOperator, uint64_t* uid, int64_t* ts);
909

910
SSDataBlock* loadNextDataBlock(void* param);
911

912
void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset);
913

L
Liu Jicong 已提交
914 915 916
SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pResultRowInfo, char* pData,
                                   int16_t bytes, bool masterscan, uint64_t groupId, SExecTaskInfo* pTaskInfo,
                                   bool isIntervalQuery, SAggSupporter* pSup);
H
Haojun Liao 已提交
917

918 919
SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo);

L
Liu Jicong 已提交
920 921
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
                                           SExecTaskInfo* pTaskInfo);
dengyihao's avatar
dengyihao 已提交
922 923
SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode,
                                         STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
924 925
SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode* pScanPhyNode,
                                              const char* pUser, SExecTaskInfo* pTaskInfo);
926

H
Haojun Liao 已提交
927
SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
928

L
Liu Jicong 已提交
929 930 931 932
SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode,
                                                 SExecTaskInfo* pTaskInfo);
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode,
                                         SExecTaskInfo* pTaskInfo);
933
SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
934 935 936 937
SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams,
                                               SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
                                           SExecTaskInfo* pTaskInfo);
938

H
Haojun Liao 已提交
939
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPhysiNode* pPhyNode,
L
Liu Jicong 已提交
940
                                          SExecTaskInfo* pTaskInfo, bool isStream);
941 942 943 944
SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMergeIntervalPhysiNode* pIntervalPhyNode,
                                               SExecTaskInfo* pTaskInfo);
SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SMergeAlignedIntervalPhysiNode* pNode,
                                                      SExecTaskInfo* pTaskInfo);
945 946
SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                     SExecTaskInfo* pTaskInfo, int32_t numOfChild);
H
Haojun Liao 已提交
947 948
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPhysiNode* pSessionNode,
                                            SExecTaskInfo* pTaskInfo);
949
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNode, SExecTaskInfo* pTaskInfo);
950 951
SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDistScanPhysiNode* pBlockScanNode,
                                               SExecTaskInfo* pTaskInfo);
952

953
SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond,
954
                                            SExecTaskInfo* pTaskInfo);
955

956 957
SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);

L
Liu Jicong 已提交
958 959 960 961 962
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode,
                                      SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWinodwPhysiNode* pStateNode,
                                             SExecTaskInfo* pTaskInfo);
SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode,
963
                                           SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
964

965
SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode,
L
Liu Jicong 已提交
966
                                                 SExecTaskInfo* pTaskInfo);
967

968
SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
969 970
SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream,
                                           SSortMergeJoinPhysiNode* pJoinNode, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
971

L
Liu Jicong 已提交
972 973 974 975
SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                  SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                       SExecTaskInfo* pTaskInfo, int32_t numOfChild);
976 977
SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                SExecTaskInfo* pTaskInfo);
5
54liuyao 已提交
978

L
Liu Jicong 已提交
979 980
SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                SExecTaskInfo* pTaskInfo);
5
54liuyao 已提交
981 982
SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode,
                                            SExecTaskInfo* pTaskInfo);
5
54liuyao 已提交
983

984
int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
L
Liu Jicong 已提交
985
                              int32_t numOfOutput, SArray* pPseudoList);
986

987
void setInputDataBlock(SExprSupp* pExprSupp, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, bool createDummyCol);
X
Xiaoyu Wang 已提交
988

989
bool    isTaskKilled(SExecTaskInfo* pTaskInfo);
990 991
int32_t checkForQueryBuf(size_t numOfTables);

L
Liu Jicong 已提交
992 993
void setTaskKilled(SExecTaskInfo* pTaskInfo);
void queryCostStatis(SExecTaskInfo* pTaskInfo);
994

dengyihao's avatar
dengyihao 已提交
995
void    doDestroyTask(SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
996
void    destroyOperatorInfo(SOperatorInfo* pOperator);
997 998
int32_t getMaximumIdleDurationSec();

wmmhello's avatar
wmmhello 已提交
999 1000 1001 1002
/*
 * ops:     root operator
 * data:    *data save the result of encode, need to be freed by caller
 * length:  *length save the length of *data
C
Cary Xu 已提交
1003
 * nOptrWithVal: *nOptrWithVal save the number of optr with value
wmmhello's avatar
wmmhello 已提交
1004 1005
 * return:  result code, 0 means success
 */
L
Liu Jicong 已提交
1006
int32_t encodeOperator(SOperatorInfo* ops, char** data, int32_t* length, int32_t* nOptrWithVal);
wmmhello's avatar
wmmhello 已提交
1007 1008 1009 1010 1011 1012 1013

/*
 * ops:    root operator, created by caller
 * data:   save the result of decode
 * length: the length of data
 * return: result code, 0 means success
 */
H
Haojun Liao 已提交
1014
int32_t decodeOperator(SOperatorInfo* ops, const char* data, int32_t length);
wmmhello's avatar
wmmhello 已提交
1015

1016
void    setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
dengyihao's avatar
dengyihao 已提交
1017
int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
D
dapan1121 已提交
1018
                               char* sql, EOPTR_EXEC_MODEL model);
L
Liu Jicong 已提交
1019
int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, qTaskInfo_t* pTaskInfo, SReadHandle* readHandle);
H
Haojun Liao 已提交
1020
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList);
dengyihao's avatar
dengyihao 已提交
1021

1022
int32_t aggDecodeResultRow(SOperatorInfo* pOperator, char* result);
wmmhello's avatar
wmmhello 已提交
1023 1024
int32_t aggEncodeResultRow(SOperatorInfo* pOperator, char** result, int32_t* length);

1025
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
1026
                                int32_t order);
1027
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
L
Liu Jicong 已提交
1028
                                 __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
5
54liuyao 已提交
1029
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
1030
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, int32_t interBufSize);
5
54liuyao 已提交
1031
void getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId, SSessionKey* pKey);
1032 1033 1034 1035 1036 1037 1038 1039 1040
bool isInTimeWindow(STimeWindow* pWin, TSKEY ts, int64_t gap);
bool functionNeedToExecute(SqlFunctionCtx* pCtx);
bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup);
bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup);
bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup);
void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid,
                                      uint64_t* pGp, void* pTbName);
void printDataBlock(SSDataBlock* pBlock, const char* flag);
1041
uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
L
Liu Jicong 已提交
1042
void     calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock);
1043

1044 1045
int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup,
                           SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
1046

S
shenglian zhou 已提交
1047 1048
SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode,
                                           SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
1049
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo* pTableListInfo,
1050
                                                SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
1051

1052
void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);
1053

L
Liu Jicong 已提交
1054 1055
bool    groupbyTbname(SNodeList* pGroupList);
void*   destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput);
1056
int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
1057
                                   SGroupResInfo* pGroupResInfo);
5
54liuyao 已提交
1058
int32_t saveSessionDiscBuf(SStreamState* pState, SSessionKey* key, void* buf, int32_t size);
1059
int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock,
5
54liuyao 已提交
1060
                                    SExprSupp* pSup, SGroupResInfo* pGroupResInfo);
1061 1062
int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId,
                     SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup);
1063 1064
int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult);
int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize);
5
54liuyao 已提交
1065
void    getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order);
L
Liu Jicong 已提交
1066
int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo);
1067

1068 1069 1070 1071
#ifdef __cplusplus
}
#endif

1072
#endif  // TDENGINE_EXECUTORIMPL_H