executorimpl.h 41.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;

H
Haojun Liao 已提交
166
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;
H
Haojun Liao 已提交
185
};
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
  int32_t            index;
  SRetrieveTableRsp* pRsp;
  uint64_t           totalRows;
242
  int64_t            startTime;
L
Liu Jicong 已提交
243 244 245
  int32_t            code;
  EX_SOURCE_STATUS   status;
  const char*        taskId;
246 247
} SSourceDataInfo;

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

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

264
typedef struct SExchangeInfo {
L
Liu Jicong 已提交
265 266 267 268
  SArray* pSources;
  SArray* pSourceDataInfo;
  tsem_t  ready;
  void*   pTransporter;
269 270 271
  // SArray<SSDataBlock*>, result block list, used to keep the multi-block that
  // passed by downstream operator
  SArray*             pResultBlockList;
L
Liu Jicong 已提交
272 273 274
  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 已提交
275
  int32_t             current;
H
Haojun Liao 已提交
276
  SLoadRemoteDataInfo loadInfo;
277
  uint64_t            self;
278
  SLimitInfo          limitInfo;
L
Liu Jicong 已提交
279
  int64_t             openedTs;  // start exec time stamp
280 281
} SExchangeInfo;

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

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

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

297
typedef struct SAggSupporter {
H
Haojun Liao 已提交
298
  SSHashObj*     pResultRowHashTable;  // quick locate the window object for each result
299 300
  char*          keyBuf;               // window key buffer
  SDiskbasedBuf* pResultBuf;           // query result buffer based on blocked-wised disk file
L
Liu Jicong 已提交
301 302
  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
303 304 305
} SAggSupporter;

typedef struct {
L
Liu Jicong 已提交
306 307 308 309 310
  // 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
311 312
} SAggOptrPushDownInfo;

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

H
Haojun Liao 已提交
319
typedef struct STableScanBase {
320
  STsdbReader*           dataReader;
H
Haojun Liao 已提交
321 322 323 324
  SFileBlockLoadRecorder readRecorder;
  SQueryTableDataCond    cond;
  SAggOptrPushDownInfo   pdInfo;
  SColMatchInfo          matchInfo;
325
  SReadHandle            readHandle;
H
Haojun Liao 已提交
326 327 328
  SExprSupp              pseudoSup;
  STableMetaCacheInfo    metaCache;
  int32_t                scanFlag;  // table scan flag to denote if it is a repeat/reverse/main scan
H
Haojun Liao 已提交
329 330
  int32_t                dataBlockLoadFlag;
  SLimitInfo             limitInfo;
H
Haojun Liao 已提交
331 332 333 334
} STableScanBase;

typedef struct STableScanInfo {
  STableScanBase         base;
L
Liu Jicong 已提交
335 336
  SScanInfo              scanInfo;
  int32_t                scanTimes;
337 338 339 340 341 342
  SSDataBlock*           pResBlock;
  SSampleExecInfo        sample;  // sample execution info
  int32_t                currentGroupId;
  int32_t                currentTable;
  int8_t                 scanMode;
  int8_t                 assignBlockUid;
343 344
} STableScanInfo;

345
typedef struct STableMergeScanInfo {
dengyihao's avatar
opt mem  
dengyihao 已提交
346 347 348 349
  int32_t                tableStartIndex;
  int32_t                tableEndIndex;
  bool                   hasGroupId;
  uint64_t               groupId;
L
Liu Jicong 已提交
350
  SArray*                queryConds;  // array of queryTableDataCond
H
Haojun Liao 已提交
351
  STableScanBase         base;
dengyihao's avatar
opt mem  
dengyihao 已提交
352 353 354 355 356 357 358 359
  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;
360 361 362 363
  int64_t                numOfRows;
  SScanInfo              scanInfo;
  int32_t                scanTimes;
  SSDataBlock*           pResBlock;
H
Haojun Liao 已提交
364 365
  SSampleExecInfo        sample;  // sample execution info
  SSortExecInfo          sortExecInfo;
366 367
} STableMergeScanInfo;

368
typedef struct STagScanInfo {
L
Liu Jicong 已提交
369 370
  SColumnInfo*    pCols;
  SSDataBlock*    pRes;
H
Haojun Liao 已提交
371
  SColMatchInfo   matchInfo;
L
Liu Jicong 已提交
372 373 374
  int32_t         curPos;
  SReadHandle     readHandle;
  STableListInfo* pTableList;
375 376
} STagScanInfo;

377
typedef struct SLastrowScanInfo {
H
Haojun Liao 已提交
378 379 380 381 382 383 384 385 386 387 388
  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;
389 390
} SLastrowScanInfo;

5
54liuyao 已提交
391 392 393 394
typedef enum EStreamScanMode {
  STREAM_SCAN_FROM_READERHANDLE = 1,
  STREAM_SCAN_FROM_RES,
  STREAM_SCAN_FROM_UPDATERES,
395
  STREAM_SCAN_FROM_DELETE_DATA,
5
54liuyao 已提交
396
  STREAM_SCAN_FROM_DATAREADER_RETRIEVE,
L
Liu Jicong 已提交
397
  STREAM_SCAN_FROM_DATAREADER_RANGE,
5
54liuyao 已提交
398 399
} EStreamScanMode;

400 401 402 403 404
enum {
  PROJECT_RETRIEVE_CONTINUE = 0x1,
  PROJECT_RETRIEVE_DONE = 0x2,
};

5
54liuyao 已提交
405
typedef struct SStreamAggSupporter {
5
54liuyao 已提交
406 407 408
  int32_t         resultRowSize;  // the result buffer size for each result row, with the meta data size for each row
  SSDataBlock*    pScanBlock;
  SStreamState*   pState;
409 410
  int64_t         gap;        // stream session window gap
  SqlFunctionCtx* pDummyCtx;  // for combine
5
54liuyao 已提交
411 412 413 414
  SSHashObj*      pResultRows;
  int32_t         stateKeySize;
  int16_t         stateKeyType;
  SDiskbasedBuf*  pResultBuf;
5
54liuyao 已提交
415 416
} SStreamAggSupporter;

417
typedef struct SWindowSupporter {
5
54liuyao 已提交
418
  SStreamAggSupporter* pStreamAggSup;
L
Liu Jicong 已提交
419
  int64_t              gap;
420
  uint16_t             parentType;
421
  SAggSupporter*       pIntervalAggSup;
422
} SWindowSupporter;
423

424
typedef struct SPartitionBySupporter {
L
Liu Jicong 已提交
425 426 427 428
  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
429 430 431
} SPartitionBySupporter;

typedef struct SPartitionDataInfo {
L
Liu Jicong 已提交
432
  uint64_t groupId;
433 434
  char*    tbname;
  SArray*  tags;
L
Liu Jicong 已提交
435
  SArray*  rowIds;
436
} SPartitionDataInfo;
437

5
54liuyao 已提交
438
typedef struct STimeWindowAggSupp {
L
Liu Jicong 已提交
439
  int8_t          calTrigger;
L
Liu Jicong 已提交
440
  int8_t          calTriggerSaved;
5
54liuyao 已提交
441
  int64_t         deleteMark;
L
Liu Jicong 已提交
442 443
  int64_t         deleteMarkSaved;
  int64_t         waterMark;
L
Liu Jicong 已提交
444
  TSKEY           maxTs;
5
54liuyao 已提交
445
  TSKEY           minTs;
L
Liu Jicong 已提交
446
  SColumnInfoData timeWindowData;  // query time window info for scalar function execution.
5
54liuyao 已提交
447 448
} STimeWindowAggSupp;

449
typedef struct SStreamScanInfo {
L
Liu Jicong 已提交
450 451 452 453 454 455 456 457 458
  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 已提交
459 460 461 462 463 464 465 466 467 468 469 470 471

  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 已提交
472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487
  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 已提交
488
  // status for tmq
L
Liu Jicong 已提交
489 490 491
  SNodeList* pGroupTags;
  SNode*     pTagCond;
  SNode*     pTagIndexCond;
492
} SStreamScanInfo;
H
Haojun Liao 已提交
493

L
Liu Jicong 已提交
494 495 496 497 498 499 500 501 502 503 504 505
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;
506

dengyihao's avatar
dengyihao 已提交
507
typedef struct SSysTableIndex {
508 509 510
  int8_t  init;
  SArray* uids;
  int32_t lastIdx;
dengyihao's avatar
dengyihao 已提交
511 512
} SSysTableIndex;

513
typedef struct SSysTableScanInfo {
dengyihao's avatar
dengyihao 已提交
514 515 516 517
  SRetrieveMetaTableRsp* pRsp;
  SRetrieveTableReq      req;
  SEpSet                 epSet;
  tsem_t                 ready;
518 519
  SReadHandle            readHandle;
  int32_t                accountId;
520
  const char*            pUser;
521
  bool                   sysInfo;
522 523 524
  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.
525
  SSysTableIndex*        pIdx;        // idx for local table meta
H
Haojun Liao 已提交
526
  SColMatchInfo          matchInfo;
527 528 529 530
  SName                  name;
  SSDataBlock*           pRes;
  int64_t                numOfBlocks;  // extract basic running information.
  SLoadRemoteDataInfo    loadInfo;
531 532
} SSysTableScanInfo;

S
shenglian zhou 已提交
533 534 535 536 537 538 539 540 541 542 543
typedef struct STableCountScanInfo {
  SReadHandle readHandle;
  SSDataBlock*  pRes;
  SExprSupp     pseudoSup;
  SNode*        pCondition;
  SName         name;
  uint64_t      suid;
  uint64_t      uid;
  int8_t        tableType;
} STableCountScanInfo;

544
typedef struct SBlockDistInfo {
L
Liu Jicong 已提交
545 546 547 548
  SSDataBlock* pResBlock;
  STsdbReader* pHandle;
  SReadHandle  readHandle;
  uint64_t     uid;  // table uid
549 550
} SBlockDistInfo;

551
// todo remove this
552
typedef struct SOptrBasicInfo {
L
Liu Jicong 已提交
553 554 555
  SResultRowInfo resultRowInfo;
  SSDataBlock*   pRes;
  bool           mergeResultBlock;
556 557
} SOptrBasicInfo;

558
typedef struct SIntervalAggOperatorInfo {
559
  SOptrBasicInfo     binfo;              // basic info
wmmhello's avatar
wmmhello 已提交
560
  SAggSupporter      aggSup;             // aggregate supporter
561
  SExprSupp          scalarSupp;         // supporter for perform scalar function
562 563 564 565 566
  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
567
  SArray*            pInterpCols;        // interpolation columns
568
  int32_t            resultTsOrder;      // result timestamp order
569
  int32_t            inputOrder;         // input data ts order
570 571
  EOPTR_EXEC_MODEL   execModel;          // operator execution model [batch model|stream model]
  STimeWindowAggSupp twAggSup;
L
Liu Jicong 已提交
572
  SArray*            pPrevValues;  //  SArray<SGroupKeys> used to keep the previous not null value for interpolation.
573
} SIntervalAggOperatorInfo;
574

575
typedef struct SMergeAlignedIntervalAggOperatorInfo {
L
Liu Jicong 已提交
576
  SIntervalAggOperatorInfo* intervalAggOperatorInfo;
577

578 579
  uint64_t     groupId;  // current groupId
  int64_t      curTs;    // current ts
580
  SSDataBlock* prefetchedBlock;
581
  SResultRow*  pResultRow;
582 583
} SMergeAlignedIntervalAggOperatorInfo;

584
typedef struct SStreamIntervalOperatorInfo {
585 586 587 588 589 590
  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.
591 592 593
  STimeWindowAggSupp twAggSup;
  bool               invertible;
  bool               ignoreExpiredData;
594
  SArray*            pDelWins;  // SWinRes
595 596
  int32_t            delIndex;
  SSDataBlock*       pDelRes;
597
  SPhysiNode*        pPhyNode;  // create new child
5
54liuyao 已提交
598
  SHashObj*          pPullDataMap;
599
  SArray*            pPullWins;  // SPullWindowInfo
5
54liuyao 已提交
600 601
  int32_t            pullIndex;
  SSDataBlock*       pPullDataRes;
602 603 604 605 606
  bool               isFinal;
  SArray*            pChildren;
  SStreamState*      pState;
  SWinKey            delKey;
} SStreamIntervalOperatorInfo;
5
54liuyao 已提交
607

608
typedef struct SAggOperatorInfo {
H
Haojun Liao 已提交
609 610
  SOptrBasicInfo   binfo;
  SAggSupporter    aggSup;
L
Liu Jicong 已提交
611 612 613 614
  STableQueryInfo* current;
  uint64_t         groupId;
  SGroupResInfo    groupResInfo;
  SExprSupp        scalarExprSup;
615 616 617
} SAggOperatorInfo;

typedef struct SProjectOperatorInfo {
L
Liu Jicong 已提交
618 619 620 621 622 623
  SOptrBasicInfo binfo;
  SAggSupporter  aggSup;
  SArray*        pPseudoColInfo;
  SLimitInfo     limitInfo;
  bool           mergeDataBlocks;
  SSDataBlock*   pFinalRes;
624 625
} SProjectOperatorInfo;

H
Haojun Liao 已提交
626
typedef struct SIndefOperatorInfo {
L
Liu Jicong 已提交
627 628 629 630 631
  SOptrBasicInfo binfo;
  SAggSupporter  aggSup;
  SArray*        pPseudoColInfo;
  SExprSupp      scalarSup;
  uint64_t       groupId;
632
  SSDataBlock*   pNextGroupRes;
H
Haojun Liao 已提交
633 634
} SIndefOperatorInfo;

635
typedef struct SFillOperatorInfo {
636 637
  struct SFillInfo* pFillInfo;
  SSDataBlock*      pRes;
H
Haojun Liao 已提交
638
  SSDataBlock*      pFinalRes;
639 640 641
  int64_t           totalInputRows;
  void**            p;
  SSDataBlock*      existNewGroupBlock;
642
  STimeWindow       win;
H
Haojun Liao 已提交
643
  SColMatchInfo     matchInfo;
H
Haojun Liao 已提交
644
  int32_t           primaryTsCol;
H
Haojun Liao 已提交
645
  int32_t           primarySrcSlotId;
L
Liu Jicong 已提交
646
  uint64_t          curGroupId;  // current handled group id
647 648
  SExprInfo*        pExprInfo;
  int32_t           numOfExpr;
649
  SExprSupp         noFillExprSupp;
650 651 652
} SFillOperatorInfo;

typedef struct SGroupbyOperatorInfo {
L
Liu Jicong 已提交
653 654
  SOptrBasicInfo binfo;
  SAggSupporter  aggSup;
H
Haojun Liao 已提交
655 656
  SArray*        pGroupCols;     // group by columns, SArray<SColumn>
  SArray*        pGroupColVals;  // current group column values, SArray<SGroupKeys>
L
Liu Jicong 已提交
657 658 659
  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 已提交
660 661
  SGroupResInfo  groupResInfo;
  SExprSupp      scalarSup;
662 663
} SGroupbyOperatorInfo;

H
Haojun Liao 已提交
664
typedef struct SDataGroupInfo {
L
Liu Jicong 已提交
665 666 667
  uint64_t groupId;
  int64_t  numOfRows;
  SArray*  pPageList;
H
Haojun Liao 已提交
668 669 670 671
} SDataGroupInfo;

// The sort in partition may be needed later.
typedef struct SPartitionOperatorInfo {
dengyihao's avatar
dengyihao 已提交
672 673 674 675 676 677 678
  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 已提交
679 680 681 682 683 684
  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
685
  SExprSupp      scalarSup;
H
Haojun Liao 已提交
686
} SPartitionOperatorInfo;
687

688
typedef struct SWindowRowsSup {
dengyihao's avatar
dengyihao 已提交
689 690 691 692
  STimeWindow win;
  TSKEY       prevTs;
  int32_t     startRowIndex;
  int32_t     numOfRows;
693
  uint64_t    groupId;
694 695
} SWindowRowsSup;

H
Haojun Liao 已提交
696
typedef struct SSessionAggOperatorInfo {
L
Liu Jicong 已提交
697 698
  SOptrBasicInfo binfo;
  SAggSupporter  aggSup;
wmmhello's avatar
wmmhello 已提交
699

700 701
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
L
Liu Jicong 已提交
702 703 704
  bool               reptScan;  // next round scan
  int64_t            gap;       // session window gap
  int32_t            tsSlotId;  // primary timestamp slot id
705
  STimeWindowAggSupp twAggSup;
H
Haojun Liao 已提交
706
} SSessionAggOperatorInfo;
707

5
54liuyao 已提交
708
typedef struct SResultWindowInfo {
709 710 711
  void*       pOutputBuf;
  SSessionKey sessionWin;
  bool        isOutput;
5
54liuyao 已提交
712 713
} SResultWindowInfo;

5
54liuyao 已提交
714 715
typedef struct SStateWindowInfo {
  SResultWindowInfo winInfo;
5
54liuyao 已提交
716
  SStateKeys*       pStateKey;
5
54liuyao 已提交
717 718
} SStateWindowInfo;

5
54liuyao 已提交
719
typedef struct SStreamSessionAggOperatorInfo {
L
Liu Jicong 已提交
720 721
  SOptrBasicInfo      binfo;
  SStreamAggSupporter streamAggSup;
722
  SExprSupp           scalarSupp;  // supporter for perform scalar function
L
Liu Jicong 已提交
723 724 725 726 727
  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;
728 729 730
  SSDataBlock*        pWinBlock;   // window result
  SSDataBlock*        pDelRes;     // delete result
  SSDataBlock*        pUpdateRes;  // update window
L
Liu Jicong 已提交
731
  bool                returnUpdate;
5
54liuyao 已提交
732
  SSHashObj*          pStDeleted;
L
Liu Jicong 已提交
733
  void*               pDelIterator;
734 735
  SArray*             pChildren;  // cache for children's result; final stream operator
  SPhysiNode*         pPhyNode;   // create new child
L
Liu Jicong 已提交
736 737
  bool                isFinal;
  bool                ignoreExpiredData;
5
54liuyao 已提交
738 739
} SStreamSessionAggOperatorInfo;

5
54liuyao 已提交
740 741 742
typedef struct SStreamStateAggOperatorInfo {
  SOptrBasicInfo      binfo;
  SStreamAggSupporter streamAggSup;
743
  SExprSupp           scalarSupp;  // supporter for perform scalar function
5
54liuyao 已提交
744 745 746 747 748 749 750
  SGroupResInfo       groupResInfo;
  int32_t             primaryTsIndex;  // primary timestamp slot id
  STimeWindowAggSupp  twAggSup;
  SColumn             stateCol;
  SSDataBlock*        pDelRes;
  SSHashObj*          pSeDeleted;
  void*               pDelIterator;
751
  SArray*             pChildren;  // cache for children's result;
5
54liuyao 已提交
752 753 754
  bool                ignoreExpiredData;
} SStreamStateAggOperatorInfo;

755 756 757 758
typedef struct SStreamPartitionOperatorInfo {
  SOptrBasicInfo        binfo;
  SPartitionBySupporter partitionSup;
  SExprSupp             scalarSup;
759
  SExprSupp             tbnameCalSup;
L
Liu Jicong 已提交
760
  SExprSupp             tagCalSup;
761 762 763 764
  SHashObj*             pPartitions;
  void*                 parIte;
  SSDataBlock*          pInputDataBlock;
  int32_t               tsColIndex;
765
  SSDataBlock*          pDelRes;
766 767
} SStreamPartitionOperatorInfo;

5
54liuyao 已提交
768 769
typedef struct SStreamFillOperatorInfo {
  SStreamFillSupporter* pFillSup;
770 771 772 773 774 775 776
  SSDataBlock*          pRes;
  SSDataBlock*          pSrcBlock;
  int32_t               srcRowIndex;
  SSDataBlock*          pPrevSrcBlock;
  SSDataBlock*          pSrcDelBlock;
  int32_t               srcDelRowIndex;
  SSDataBlock*          pDelRes;
L
Liu Jicong 已提交
777
  SColMatchInfo         matchInfo;
778 779 780
  int32_t               primaryTsCol;
  int32_t               primarySrcSlotId;
  SStreamFillInfo*      pFillInfo;
5
54liuyao 已提交
781 782
} SStreamFillOperatorInfo;

783
typedef struct STimeSliceOperatorInfo {
L
Liu Jicong 已提交
784 785 786 787 788 789 790 791 792 793 794 795 796
  SSDataBlock*         pRes;
  STimeWindow          win;
  SInterval            interval;
  int64_t              current;
  SArray*              pPrevRow;     // SArray<SGroupValue>
  SArray*              pNextRow;     // SArray<SGroupValue>
  SArray*              pLinearInfo;  // SArray<SFillLinearInfo>
  bool                 isPrevRowSet;
  bool                 isNextRowSet;
  int32_t              fillType;      // fill type
  SColumn              tsCol;         // primary timestamp column
  SExprSupp            scalarSup;     // scalar calculation
  struct SFillColInfo* pFillColInfo;  // fill column info
797 798
} STimeSliceOperatorInfo;

799
typedef struct SStateWindowOperatorInfo {
wmmhello's avatar
wmmhello 已提交
800
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
L
Liu Jicong 已提交
801 802
  SOptrBasicInfo binfo;
  SAggSupporter  aggSup;
803
  SExprSupp      scalarSup;
wmmhello's avatar
wmmhello 已提交
804

805 806
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
807
  SColumn            stateCol;  // start row index
808 809
  bool               hasKey;
  SStateKeys         stateKey;
810
  int32_t            tsSlotId;  // primary timestamp column slot id
811
  STimeWindowAggSupp twAggSup;
812 813
} SStateWindowOperatorInfo;

H
Haojun Liao 已提交
814
typedef struct SSortOperatorInfo {
815
  SOptrBasicInfo binfo;
L
Liu Jicong 已提交
816 817 818
  uint32_t       sortBufSize;  // max buffer size for in-memory sort
  SArray*        pSortInfo;
  SSortHandle*   pSortHandle;
H
Haojun Liao 已提交
819
  SColMatchInfo  matchInfo;
L
Liu Jicong 已提交
820
  int32_t        bufPageSize;
821 822 823
  int64_t        startTs;      // sort start time
  uint64_t       sortElapsed;  // sort elapsed time, time to flush to disk not included.
  SLimitInfo     limitInfo;
H
Haojun Liao 已提交
824
} SSortOperatorInfo;
825

826
typedef struct SJoinOperatorInfo {
L
Liu Jicong 已提交
827 828 829 830 831 832 833 834 835 836 837 838
  SSDataBlock* pRes;
  int32_t      joinType;
  int32_t      inputOrder;

  SSDataBlock* pLeft;
  int32_t      leftPos;
  SColumnInfo  leftCol;

  SSDataBlock* pRight;
  int32_t      rightPos;
  SColumnInfo  rightCol;
  SNode*       pCondAfterMerge;
839 840
} SJoinOperatorInfo;

841 842 843
#define OPTR_IS_OPENED(_optr)  (((_optr)->status & OP_OPENED) == OP_OPENED)
#define OPTR_SET_OPENED(_optr) ((_optr)->status |= OP_OPENED)

H
Haojun Liao 已提交
844 845
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);
846

H
Haojun Liao 已提交
847 848
int32_t operatorDummyOpenFn(SOperatorInfo* pOperator);
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num);
849 850 851

void    initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
void    cleanupBasicInfo(SOptrBasicInfo* pInfo);
852
int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
853
void    cleanupExprSupp(SExprSupp* pSup);
854
void    destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs);
L
Liu Jicong 已提交
855
int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
856
                    const char* pkey);
L
Liu Jicong 已提交
857
void    initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows);
858 859 860

void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                           SDiskbasedBuf* pBuf);
861 862
void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                            SDiskbasedBuf* pBuf);
863

L
Liu Jicong 已提交
864
int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf);
865 866
bool    hasLimitOffsetInfo(SLimitInfo* pLimitInfo);
void    initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo);
867
void applyLimitOffset(SLimitInfo* pLimitInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo, SOperatorInfo* pOperator);
868

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

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

876 877
STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order);

L
Liu Jicong 已提交
878
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag);
879
int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz);
880

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

H
Haojun Liao 已提交
883
void    setOperatorCompleted(SOperatorInfo* pOperator);
L
Liu Jicong 已提交
884 885
void    setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status,
                        void* pInfo, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
886
void    doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo);
L
Liu Jicong 已提交
887 888
int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock,
                               int32_t rows, const char* idStr, STableMetaCacheInfo* pCache);
889

L
Liu Jicong 已提交
890 891
void cleanupAggSup(SAggSupporter* pAggSup);
void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
H
Haojun Liao 已提交
892
void setTbNameColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, const char* name);
893

L
Liu Jicong 已提交
894
int32_t doPrepareScan(SOperatorInfo* pOperator, uint64_t uid, int64_t ts);
895
int32_t doGetScanStatus(SOperatorInfo* pOperator, uint64_t* uid, int64_t* ts);
896

897
SSDataBlock* loadNextDataBlock(void* param);
898

899
void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset);
900

L
Liu Jicong 已提交
901 902 903
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 已提交
904

905 906
SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo);

L
Liu Jicong 已提交
907 908
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
                                           SExecTaskInfo* pTaskInfo);
dengyihao's avatar
dengyihao 已提交
909 910
SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode,
                                         STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
911 912
SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode* pScanPhyNode,
                                              const char* pUser, SExecTaskInfo* pTaskInfo);
913

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

L
Liu Jicong 已提交
916 917 918 919
SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode,
                                                 SExecTaskInfo* pTaskInfo);
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode,
                                         SExecTaskInfo* pTaskInfo);
920
SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
921 922 923 924
SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams,
                                               SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
                                           SExecTaskInfo* pTaskInfo);
925

H
Haojun Liao 已提交
926
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPhysiNode* pPhyNode,
L
Liu Jicong 已提交
927
                                          SExecTaskInfo* pTaskInfo, bool isStream);
928 929 930 931
SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMergeIntervalPhysiNode* pIntervalPhyNode,
                                               SExecTaskInfo* pTaskInfo);
SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SMergeAlignedIntervalPhysiNode* pNode,
                                                      SExecTaskInfo* pTaskInfo);
932 933
SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                     SExecTaskInfo* pTaskInfo, int32_t numOfChild);
H
Haojun Liao 已提交
934 935
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPhysiNode* pSessionNode,
                                            SExecTaskInfo* pTaskInfo);
936
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNode, SExecTaskInfo* pTaskInfo);
937 938
SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDistScanPhysiNode* pBlockScanNode,
                                               SExecTaskInfo* pTaskInfo);
939

940
SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond,
941
                                            SExecTaskInfo* pTaskInfo);
942

943 944
SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);

L
Liu Jicong 已提交
945 946 947 948 949
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode,
                                      SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWinodwPhysiNode* pStateNode,
                                             SExecTaskInfo* pTaskInfo);
SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode,
950
                                           SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
951

952
SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode,
L
Liu Jicong 已提交
953
                                                 SExecTaskInfo* pTaskInfo);
954

955
SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
956 957
SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream,
                                           SSortMergeJoinPhysiNode* pJoinNode, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
958

L
Liu Jicong 已提交
959 960 961 962
SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                  SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                       SExecTaskInfo* pTaskInfo, int32_t numOfChild);
963 964
SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                SExecTaskInfo* pTaskInfo);
5
54liuyao 已提交
965

L
Liu Jicong 已提交
966 967
SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                SExecTaskInfo* pTaskInfo);
5
54liuyao 已提交
968 969
SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode,
                                            SExecTaskInfo* pTaskInfo);
5
54liuyao 已提交
970

971
int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
L
Liu Jicong 已提交
972
                              int32_t numOfOutput, SArray* pPseudoList);
973

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

976
bool    isTaskKilled(SExecTaskInfo* pTaskInfo);
977 978
int32_t checkForQueryBuf(size_t numOfTables);

L
Liu Jicong 已提交
979 980
void setTaskKilled(SExecTaskInfo* pTaskInfo);
void queryCostStatis(SExecTaskInfo* pTaskInfo);
981

dengyihao's avatar
dengyihao 已提交
982
void    doDestroyTask(SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
983
void    destroyOperatorInfo(SOperatorInfo* pOperator);
984 985
int32_t getMaximumIdleDurationSec();

wmmhello's avatar
wmmhello 已提交
986 987 988 989
/*
 * 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 已提交
990
 * nOptrWithVal: *nOptrWithVal save the number of optr with value
wmmhello's avatar
wmmhello 已提交
991 992
 * return:  result code, 0 means success
 */
L
Liu Jicong 已提交
993
int32_t encodeOperator(SOperatorInfo* ops, char** data, int32_t* length, int32_t* nOptrWithVal);
wmmhello's avatar
wmmhello 已提交
994 995 996 997 998 999 1000

/*
 * 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 已提交
1001
int32_t decodeOperator(SOperatorInfo* ops, const char* data, int32_t length);
wmmhello's avatar
wmmhello 已提交
1002

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

1009
int32_t aggDecodeResultRow(SOperatorInfo* pOperator, char* result);
wmmhello's avatar
wmmhello 已提交
1010 1011
int32_t aggEncodeResultRow(SOperatorInfo* pOperator, char** result, int32_t* length);

1012
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
1013
                                int32_t order);
1014
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
L
Liu Jicong 已提交
1015
                                 __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
5
54liuyao 已提交
1016
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
1017
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, int32_t interBufSize);
5
54liuyao 已提交
1018
void getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId, SSessionKey* pKey);
1019 1020 1021 1022 1023 1024 1025 1026 1027
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);
1028
uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
L
Liu Jicong 已提交
1029
void     calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock);
1030

1031 1032
int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup,
                           SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
1033

S
shenglian zhou 已提交
1034 1035
SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode,
                                           SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
1036 1037
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
                                                SExecTaskInfo* pTaskInfo);
1038

1039
void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);
1040

L
Liu Jicong 已提交
1041 1042
bool    groupbyTbname(SNodeList* pGroupList);
void*   destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput);
1043
int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
1044
                                   SGroupResInfo* pGroupResInfo);
5
54liuyao 已提交
1045
int32_t saveSessionDiscBuf(SStreamState* pState, SSessionKey* key, void* buf, int32_t size);
1046
int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock,
5
54liuyao 已提交
1047
                                    SExprSupp* pSup, SGroupResInfo* pGroupResInfo);
1048 1049
int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId,
                     SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup);
1050 1051
int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult);
int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize);
5
54liuyao 已提交
1052
void    getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order);
L
Liu Jicong 已提交
1053
int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo);
1054

1055 1056 1057 1058
#ifdef __cplusplus
}
#endif

1059
#endif  // TDENGINE_EXECUTORIMPL_H