executorimpl.h 40.1 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14
/*
 * 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/>.
 */
H
Hongze Cheng 已提交
15
// clang-format off
16 17 18
#ifndef TDENGINE_EXECUTORIMPL_H
#define TDENGINE_EXECUTORIMPL_H

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

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

H
Haojun Liao 已提交
30
#include "dataSinkMgt.h"
31
#include "executil.h"
H
Haojun Liao 已提交
32
#include "executor.h"
H
Haojun Liao 已提交
33
#include "planner.h"
D
dapan1121 已提交
34
#include "scalar.h"
35 36
#include "taosdef.h"
#include "tarray.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"
5
54liuyao 已提交
41
#include "tstreamUpdate.h"
L
Liu Jicong 已提交
42
#include "tstream.h"
H
Haojun Liao 已提交
43

L
Liu Jicong 已提交
44
#include "vnode.h"
45
#include "executorInt.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);

dengyihao's avatar
dengyihao 已提交
49 50
#define IS_QUERY_KILLED(_q)   ((_q)->code == TSDB_CODE_TSC_QUERY_CANCELLED)
#define Q_STATUS_EQUAL(p, s)  (((p) & (s)) != 0u)
51 52
#define QUERY_IS_ASC_QUERY(q) (GET_FORWARD_DIRECTION_FACTOR((q)->order.order) == QUERY_ASC_FORWARD_STEP)

53
#define NEEDTO_COMPRESS_QUERY(size) ((size) > tsCompressColData ? 1 : 0)
54

55 56 57 58 59 60
#define START_TS_COLUMN_INDEX             0
#define END_TS_COLUMN_INDEX               1
#define UID_COLUMN_INDEX                  2
#define GROUPID_COLUMN_INDEX              3
#define CALCULATE_START_TS_COLUMN_INDEX   4
#define CALCULATE_END_TS_COLUMN_INDEX     5
61

62
enum {
63 64
  // when this task starts to execute, this status will set
  TASK_NOT_COMPLETED = 0x1u,
65

66
  /* Task is over
67 68 69
   * 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
   */
70
  TASK_COMPLETED = 0x2u,
71 72 73 74 75 76
};

/**
 * 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 已提交
77 78 79 80 81
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 已提交
82
} SResultInfo;
83 84

typedef struct STableQueryInfo {
L
Liu Jicong 已提交
85 86
  TSKEY              lastKey;  // last check ts, todo remove it later
  SResultRowPosition pos;      // current active time window
87 88
} STableQueryInfo;

H
Haojun Liao 已提交
89 90 91 92 93
typedef struct SLimit {
  int64_t limit;
  int64_t offset;
} SLimit;

94
typedef struct STableScanAnalyzeInfo SFileBlockLoadRecorder;
95

H
Haojun Liao 已提交
96
typedef struct STaskCostInfo {
97 98
  int64_t  created;
  int64_t  start;
99 100 101 102 103 104 105 106
  uint64_t loadStatisTime;
  uint64_t loadFileBlockTime;
  uint64_t loadDataInCacheTime;
  uint64_t loadStatisSize;
  uint64_t loadFileBlockSize;
  uint64_t loadDataInCacheSize;

  uint64_t loadDataTime;
107 108

  SFileBlockLoadRecorder* pRecoder;
L
Liu Jicong 已提交
109
  uint64_t                elapsedTime;
110

111 112 113 114 115 116 117 118
  uint64_t firstStageMergeTime;
  uint64_t winInfoSize;
  uint64_t tableInfoSize;
  uint64_t hashSize;
  uint64_t numOfTimeWindows;

  SArray*   queryProfEvents;      // SArray<SQueryProfEvent>
  SHashObj* operatorProfResults;  // map<operator_type, SQueryProfEvent>
H
Haojun Liao 已提交
119
} STaskCostInfo;
120

H
Haojun Liao 已提交
121
typedef struct SOperatorCostInfo {
L
Liu Jicong 已提交
122 123
  double openCost;
  double totalCost;
H
Haojun Liao 已提交
124 125
} SOperatorCostInfo;

126 127
struct SOperatorInfo;

wmmhello's avatar
wmmhello 已提交
128
typedef int32_t (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, char** result, int32_t* length);
129
typedef int32_t (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, char* result);
130

131
typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* pOptr);
132
typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* pOptr);
H
Haojun Liao 已提交
133
typedef void (*__optr_close_fn_t)(void* param, int32_t num);
134
typedef int32_t (*__optr_explain_fn_t)(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len);
H
Haojun Liao 已提交
135

H
Haojun Liao 已提交
136
typedef struct STaskIdInfo {
137 138 139 140
  uint64_t queryId;  // this is also a request id
  uint64_t subplanId;
  uint64_t templateId;
  char*    str;
H
Haojun Liao 已提交
141 142
} STaskIdInfo;

L
Liu Jicong 已提交
143 144 145 146 147 148
enum {
  STREAM_RECOVER_STEP__NONE = 0,
  STREAM_RECOVER_STEP__PREPARE,
  STREAM_RECOVER_STEP__SCAN,
};

L
Liu Jicong 已提交
149
typedef struct {
L
Liu Jicong 已提交
150
  //TODO remove prepareStatus
L
Liu Jicong 已提交
151 152 153 154 155
  STqOffsetVal   prepareStatus; // for tmq
  STqOffsetVal   lastStatus;    // for tmq
  void*          metaBlk;       // for tmq fetching meta
  SSDataBlock*   pullOverBlk;   // for streaming
  SWalFilterCond cond;
L
Liu Jicong 已提交
156
  int64_t        lastScanUid;
L
Liu Jicong 已提交
157 158 159 160
  int8_t         recoverStep;
  SQueryTableDataCond tableCond;
  int64_t recoverStartVer;
  int64_t recoverEndVer;
L
Liu Jicong 已提交
161 162
} SStreamTaskInfo;

163 164 165 166 167
typedef struct {
  char*           tablename;
  char*           dbname;
  int32_t         tversion;
  SSchemaWrapper* sw;
168
  SSchemaWrapper* qsw;
169 170
} SSchemaInfo;

171
typedef struct SExecTaskInfo {
172 173 174 175 176 177 178
  STaskIdInfo      id;
  uint32_t         status;
  STimeWindow      window;
  STaskCostInfo    cost;
  int64_t          owner;  // if it is in execution
  int32_t          code;

179
  int64_t          version; // used for stream to record wal version
180 181 182 183 184 185 186
  SStreamTaskInfo  streamInfo;
  SSchemaInfo      schemaInfo;
  STableListInfo   tableqinfoList;  // this is a table list
  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;
187
  struct SOperatorInfo* pRoot;
188
} SExecTaskInfo;
H
Haojun Liao 已提交
189

190
enum {
191 192
  OP_NOT_OPENED    = 0x0,
  OP_OPENED        = 0x1,
H
Haojun Liao 已提交
193
  OP_RES_TO_RETURN = 0x5,
194
  OP_EXEC_DONE     = 0x9,
195 196
};

197
typedef struct SOperatorFpSet {
L
Liu Jicong 已提交
198 199 200 201 202 203 204 205
  __optr_open_fn_t    _openFn;  // DO NOT invoke this function directly
  __optr_fn_t         getNextFn;
  __optr_fn_t         getStreamResFn;  // execute the aggregate in the stream model, todo remove it
  __optr_fn_t         cleanupFn;       // call this function to release the allocated resources ASAP
  __optr_close_fn_t   closeFn;
  __optr_encode_fn_t  encodeResultRow;
  __optr_decode_fn_t  decodeResultRow;
  __optr_explain_fn_t getExplainFn;
206 207
} SOperatorFpSet;

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

215
typedef struct SOperatorInfo {
L
Liu Jicong 已提交
216 217 218 219 220 221 222 223 224 225 226 227
  uint8_t                operatorType;
  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;
228 229
} SOperatorInfo;

230 231
typedef enum {
  EX_SOURCE_DATA_NOT_READY = 0x1,
232
  EX_SOURCE_DATA_READY     = 0x2,
233 234
  EX_SOURCE_DATA_EXHAUSTED = 0x3,
} EX_SOURCE_STATUS;
235

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

239
typedef struct SSourceDataInfo {
L
Liu Jicong 已提交
240 241 242 243 244 245
  int32_t            index;
  SRetrieveTableRsp* pRsp;
  uint64_t           totalRows;
  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 {
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 {
dengyihao's avatar
dengyihao 已提交
265 266 267 268 269 270 271
  SArray*             pSources;
  SArray*             pSourceDataInfo;
  tsem_t              ready;
  void*               pTransporter;
  SSDataBlock*        pResult;
  bool                seqLoadData;  // sequential load data or not, false by default
  int32_t             current;
H
Haojun Liao 已提交
272
  SLoadRemoteDataInfo loadInfo;
273
  uint64_t            self;
274
  SLimitInfo          limitInfo;
275 276
} SExchangeInfo;

H
Haojun Liao 已提交
277
typedef struct SColMatchInfo {
278
  int32_t srcSlotId;     // source slot id
H
Haojun Liao 已提交
279 280
  int32_t colId;
  int32_t targetSlotId;
281
  bool    output;        // todo remove this?
282
  bool    reserved;
283
  int32_t matchType;     // determinate the source according to col id or slot id
H
Haojun Liao 已提交
284 285
} SColMatchInfo;

286
typedef struct SScanInfo {
287 288
  int32_t         numOfAsc;
  int32_t         numOfDesc;
289 290
} SScanInfo;

291 292 293 294 295
typedef struct SSampleExecInfo {
  double          sampleRatio;  // data block sample ratio, 1 by default
  uint32_t        seed;         // random seed value
} SSampleExecInfo;

L
Liu Jicong 已提交
296 297
enum {
  TABLE_SCAN__TABLE_ORDER = 1,
298
  TABLE_SCAN__BLOCK_ORDER = 2,
L
Liu Jicong 已提交
299 300
};

301
typedef struct STableScanInfo {
H
Hongze Cheng 已提交
302
  STsdbReader*    dataReader;
303 304
  SReadHandle     readHandle;

305
  SFileBlockLoadRecorder readRecorder;
306
  SScanInfo       scanInfo;
307 308
  int32_t         scanTimes;
  SNode*          pFilterNode;  // filter info, which is push down by optimizer
309

310
  SSDataBlock*    pResBlock;
H
Haojun Liao 已提交
311
  SArray*         pColMatchInfo;
312
  SExprSupp       pseudoSup;
313
  SQueryTableDataCond cond;
314
  int32_t         scanFlag;     // table scan flag to denote if it is a repeat/reverse/main scan
315
  int32_t         dataBlockLoadFlag;
316
  SInterval       interval;     // if the upstream is an interval operator, the interval info is also kept here to get the time window to check if current data block needs to be loaded.
317
  SSampleExecInfo sample;       // sample execution info
wmmhello's avatar
wmmhello 已提交
318
  int32_t         currentGroupId;
L
Liu Jicong 已提交
319
  int32_t         currentTable;
320 321
  int8_t          scanMode;
  int8_t          noTable;
322 323 324
} STableScanInfo;

typedef struct STagScanInfo {
325 326 327 328 329
  SColumnInfo     *pCols;
  SSDataBlock     *pRes;
  SArray          *pColMatchInfo;
  int32_t          curPos;
  SReadHandle      readHandle;
wmmhello's avatar
wmmhello 已提交
330
  STableListInfo  *pTableList;
331 332
} STagScanInfo;

333 334 335
typedef struct SLastrowScanInfo {
  SSDataBlock    *pRes;
  SReadHandle     readHandle;
336 337 338
  void           *pLastrowReader;
  SArray         *pColMatchInfo;
  int32_t        *pSlotIds;
339
  SExprSupp       pseudoExprSup;
340 341 342 343 344
  int32_t         retrieveType;
  int32_t         currentGroupIndex;
  SSDataBlock    *pBufferredRes;
  SArray         *pUidList;
  int32_t         indexOfBufferedRes;
345 346
} SLastrowScanInfo;

5
54liuyao 已提交
347 348 349 350
typedef enum EStreamScanMode {
  STREAM_SCAN_FROM_READERHANDLE = 1,
  STREAM_SCAN_FROM_RES,
  STREAM_SCAN_FROM_UPDATERES,
5
54liuyao 已提交
351
  STREAM_SCAN_FROM_DATAREADER_RETRIEVE,
L
Liu Jicong 已提交
352
  STREAM_SCAN_FROM_DATAREADER_RANGE,
5
54liuyao 已提交
353 354
} EStreamScanMode;

5
54liuyao 已提交
355
typedef struct SCatchSupporter {
L
Liu Jicong 已提交
356 357 358 359
  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 已提交
360 361
} SCatchSupporter;

5
54liuyao 已提交
362
typedef struct SStreamAggSupporter {
363 364 365
  SHashObj*      pResultRows;
  SArray*        pCurWins;
  int32_t        valueSize;
5
54liuyao 已提交
366 367 368 369
  int32_t        keySize;
  char*          pKeyBuf;              // window key buffer
  SDiskbasedBuf* pResultBuf;           // query result buffer based on blocked-wised disk file
  int32_t        resultRowSize;        // the result buffer size for each result row, with the meta data size for each row
370
  SSDataBlock*   pScanBlock;
5
54liuyao 已提交
371 372 373 374
} SStreamAggSupporter;

typedef struct SessionWindowSupporter {
  SStreamAggSupporter* pStreamAggSup;
L
Liu Jicong 已提交
375 376
  int64_t              gap;
  uint8_t              parentType;
5
54liuyao 已提交
377
} SessionWindowSupporter;
378

5
54liuyao 已提交
379 380 381 382 383 384 385
typedef struct STimeWindowSupp {
  int8_t           calTrigger;
  int64_t          waterMark;
  TSKEY            maxTs;
  SColumnInfoData  timeWindowData;     // query time window info for scalar function execution.
} STimeWindowAggSupp;

386
typedef struct SStreamScanInfo {
L
Liu Jicong 已提交
387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411
  uint64_t    tableUid;  // queried super table uid
  SExprInfo*  pPseudoExpr;
  int32_t     numOfPseudoExpr;
  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.
  SArray*     pColMatchInfo;  //
  SNode*      pCondition;

  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;

  EStreamScanMode        scanMode;
  SOperatorInfo*         pStreamScanOp;
  SOperatorInfo*         pTableScanOp;
  SArray*                childIds;
5
54liuyao 已提交
412
  SessionWindowSupporter sessionSup;
L
Liu Jicong 已提交
413 414 415 416 417 418
  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;
5
54liuyao 已提交
419
  STimeWindow            updateWin;
5
54liuyao 已提交
420
  STimeWindowAggSupp     twAggSup;
421
  SSDataBlock*           pUpdateDataRes;
L
Liu Jicong 已提交
422
  // status for tmq
L
Liu Jicong 已提交
423
  // SSchemaWrapper schema;
424
  STqOffset              offset;
425
  SNodeList*             pGroupTags;
426 427
  SNode*                 pTagCond;
  SNode*                 pTagIndexCond;
428
} SStreamScanInfo;
H
Haojun Liao 已提交
429

430
typedef struct SSysTableScanInfo {
dengyihao's avatar
dengyihao 已提交
431 432 433 434
  SRetrieveMetaTableRsp* pRsp;
  SRetrieveTableReq      req;
  SEpSet                 epSet;
  tsem_t                 ready;
435 436
  SReadHandle            readHandle;
  int32_t                accountId;
437
  const char*            pUser;
438 439 440 441 442 443 444 445
  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.
  SArray*                scanCols;    // SArray<int16_t> scan column id list
  SName                  name;
  SSDataBlock*           pRes;
  int64_t                numOfBlocks;  // extract basic running information.
  SLoadRemoteDataInfo    loadInfo;
446 447
} SSysTableScanInfo;

448 449 450
typedef struct SBlockDistInfo {
  SSDataBlock* pResBlock;
  void*        pHandle;
451 452
  SReadHandle  readHandle;
  uint64_t     uid;        // table uid
453 454
} SBlockDistInfo;

455
// todo remove this
456
typedef struct SOptrBasicInfo {
dengyihao's avatar
dengyihao 已提交
457 458
  SResultRowInfo  resultRowInfo;
  SSDataBlock*    pRes;
459 460
} SOptrBasicInfo;

461
typedef struct SAggSupporter {
dengyihao's avatar
dengyihao 已提交
462 463 464
  SHashObj*      pResultRowHashTable;  // quick locate the window object for each result
  char*          keyBuf;               // window key buffer
  SDiskbasedBuf* pResultBuf;           // query result buffer based on blocked-wised disk file
H
Haojun Liao 已提交
465
  int32_t        resultRowSize;        // the result buffer size for each result row, with the meta data size for each row
466 467
} SAggSupporter;

468
typedef struct SIntervalAggOperatorInfo {
wmmhello's avatar
wmmhello 已提交
469
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
470
  SOptrBasicInfo     binfo;              // basic info
wmmhello's avatar
wmmhello 已提交
471
  SAggSupporter      aggSup;             // aggregate supporter
472
  SExprSupp          scalarSupp;         // supporter for perform scalar function
473 474 475 476 477
  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
478
  SArray*            pInterpCols;        // interpolation columns
479 480 481
  int32_t            order;              // current SSDataBlock scan order
  EOPTR_EXEC_MODEL   execModel;          // operator execution model [batch model|stream model]
  STimeWindowAggSupp twAggSup;
5
54liuyao 已提交
482
  bool               invertible;
483
  SArray*            pPrevValues;        //  SArray<SGroupKeys> used to keep the previous not null value for interpolation.
5
54liuyao 已提交
484
  bool               ignoreExpiredData;
485 486 487 488
  SArray*            pRecycledPages;
  SArray*            pDelWins;           // SWinRes
  int32_t            delIndex;
  SSDataBlock*       pDelRes;
489 490

  SNode *pCondition;
491
} SIntervalAggOperatorInfo;
492

493 494 495 496 497 498 499 500 501 502 503
typedef struct SMergeAlignedIntervalAggOperatorInfo {
  SIntervalAggOperatorInfo *intervalAggOperatorInfo;

  bool         hasGroupId;
  uint64_t     groupId;
  SSDataBlock* prefetchedBlock;
  bool         inputBlocksFinished;

  SNode*       pCondition;
} SMergeAlignedIntervalAggOperatorInfo;

5
54liuyao 已提交
504
typedef struct SStreamFinalIntervalOperatorInfo {
wmmhello's avatar
wmmhello 已提交
505
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
5
54liuyao 已提交
506
  SOptrBasicInfo     binfo;              // basic info
wmmhello's avatar
wmmhello 已提交
507
  SAggSupporter      aggSup;             // aggregate supporter
5
54liuyao 已提交
508
  SExprSupp          scalarSupp;         // supporter for perform scalar function
5
54liuyao 已提交
509 510 511 512 513
  SGroupResInfo      groupResInfo;       // multiple results build supporter
  SInterval          interval;           // interval info
  int32_t            primaryTsIndex;     // primary time stamp slot id from result of downstream operator.
  int32_t            order;              // current SSDataBlock scan order
  STimeWindowAggSupp twAggSup;
514
  SArray*            pChildren;
5
54liuyao 已提交
515
  SSDataBlock*       pUpdateRes;
5
54liuyao 已提交
516
  bool               returnUpdate;
5
54liuyao 已提交
517
  SPhysiNode*        pPhyNode;           // create new child
5
54liuyao 已提交
518
  bool               isFinal;
5
54liuyao 已提交
519 520 521 522
  SHashObj*          pPullDataMap;
  SArray*            pPullWins;          // SPullWindowInfo
  int32_t            pullIndex;
  SSDataBlock*       pPullDataRes;
5
54liuyao 已提交
523
  bool               ignoreExpiredData;
524 525 526 527
  SArray*            pRecycledPages;
  SArray*            pDelWins;           // SWinRes
  int32_t            delIndex;
  SSDataBlock*       pDelRes;
5
54liuyao 已提交
528 529
} SStreamFinalIntervalOperatorInfo;

530
typedef struct SAggOperatorInfo {
wmmhello's avatar
wmmhello 已提交
531
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
H
Haojun Liao 已提交
532 533
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
wmmhello's avatar
wmmhello 已提交
534

H
Haojun Liao 已提交
535
  STableQueryInfo   *current;
H
Haojun Liao 已提交
536
  uint64_t           groupId;
H
Haojun Liao 已提交
537
  SGroupResInfo      groupResInfo;
538
  SExprSupp          scalarExprSup;
S
slzhou 已提交
539 540

  SNode             *pCondition;
541 542 543
} SAggOperatorInfo;

typedef struct SProjectOperatorInfo {
wmmhello's avatar
wmmhello 已提交
544
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
545 546
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
547
  SNode*             pFilterNode;  // filter info, which is push down by optimizer
548
  SArray*            pPseudoColInfo;
549
  SLimitInfo         limitInfo;
H
Haojun Liao 已提交
550 551
  bool               mergeDataBlocks;
  SSDataBlock*       pFinalRes;
552 553
} SProjectOperatorInfo;

H
Haojun Liao 已提交
554 555 556 557
typedef struct SIndefOperatorInfo {
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
  SArray*            pPseudoColInfo;
558
  SExprSupp          scalarSup;
S
slzhou 已提交
559
  SNode*             pCondition;
560 561 562
  uint64_t           groupId;

  SSDataBlock*       pNextGroupRes;
H
Haojun Liao 已提交
563 564
} SIndefOperatorInfo;

565
typedef struct SFillOperatorInfo {
566 567 568 569 570 571
  struct SFillInfo* pFillInfo;
  SSDataBlock*      pRes;
  int64_t           totalInputRows;
  void**            p;
  SSDataBlock*      existNewGroupBlock;
  bool              multigroupResult;
572
  STimeWindow       win;
S
slzhou 已提交
573
  SNode*            pCondition;
574
  SArray*           pColMatchColInfo;
H
Haojun Liao 已提交
575
  int32_t           primaryTsCol;
576
  uint64_t          curGroupId;       // current handled group id
577 578 579
} SFillOperatorInfo;

typedef struct SGroupbyOperatorInfo {
wmmhello's avatar
wmmhello 已提交
580
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
dengyihao's avatar
dengyihao 已提交
581
  SOptrBasicInfo  binfo;
wmmhello's avatar
wmmhello 已提交
582 583
  SAggSupporter   aggSup;

584
  SArray*         pGroupCols;     // group by columns, SArray<SColumn>
dengyihao's avatar
dengyihao 已提交
585 586 587 588 589 590
  SArray*         pGroupColVals;  // current group column values, SArray<SGroupKeys>
  SNode*          pCondition;
  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
  SGroupResInfo   groupResInfo;
591
  SExprSupp       scalarSup;
592 593
} SGroupbyOperatorInfo;

H
Haojun Liao 已提交
594
typedef struct SDataGroupInfo {
595 596 597
  uint64_t        groupId;
  int64_t         numOfRows;
  SArray*         pPageList;
H
Haojun Liao 已提交
598 599 600 601
} SDataGroupInfo;

// The sort in partition may be needed later.
typedef struct SPartitionOperatorInfo {
dengyihao's avatar
dengyihao 已提交
602 603 604 605 606 607 608 609 610 611
  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

  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
612 613
  SArray*        sortedGroupArray;   // SDataGroupInfo sorted by group id
  int32_t        groupIndex;  // group index
614
  int32_t        pageIndex;   // page index of current group
615
  SSDataBlock*   pUpdateRes;
616
  SExprSupp      scalarSup;
H
Haojun Liao 已提交
617
} SPartitionOperatorInfo;
618

619
typedef struct SWindowRowsSup {
dengyihao's avatar
dengyihao 已提交
620 621 622 623
  STimeWindow win;
  TSKEY       prevTs;
  int32_t     startRowIndex;
  int32_t     numOfRows;
624 625
} SWindowRowsSup;

H
Haojun Liao 已提交
626
typedef struct SSessionAggOperatorInfo {
wmmhello's avatar
wmmhello 已提交
627
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
628 629
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
wmmhello's avatar
wmmhello 已提交
630

631 632 633 634
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
  bool               reptScan;        // next round scan
  int64_t            gap;             // session window gap
635
  int32_t            tsSlotId;        // primary timestamp slot id
636
  STimeWindowAggSupp twAggSup;
L
Liu Jicong 已提交
637
  const SNode* pCondition;
H
Haojun Liao 已提交
638
} SSessionAggOperatorInfo;
639

5
54liuyao 已提交
640 641 642 643
typedef struct SResultWindowInfo {
  SResultRowPosition pos;
  STimeWindow win;
  bool isOutput;
5
54liuyao 已提交
644
  bool isClosed;
5
54liuyao 已提交
645 646
} SResultWindowInfo;

5
54liuyao 已提交
647 648 649 650 651
typedef struct SStateWindowInfo {
  SResultWindowInfo winInfo;
  SStateKeys stateKey;
} SStateWindowInfo;

5
54liuyao 已提交
652 653 654
typedef struct SStreamSessionAggOperatorInfo {
  SOptrBasicInfo       binfo;
  SStreamAggSupporter  streamAggSup;
5
54liuyao 已提交
655
  SExprSupp            scalarSupp;         // supporter for perform scalar function
5
54liuyao 已提交
656 657 658
  SGroupResInfo        groupResInfo;
  int64_t              gap;             // session window gap
  int32_t              primaryTsIndex;  // primary timestamp slot id
5
54liuyao 已提交
659
  int32_t              endTsIndex;      // window end timestamp slot id
5
54liuyao 已提交
660 661 662 663
  int32_t              order;           // current SSDataBlock scan order
  STimeWindowAggSupp   twAggSup;
  SSDataBlock*         pWinBlock;       // window result
  SqlFunctionCtx*      pDummyCtx;       // for combine
5
54liuyao 已提交
664
  SSDataBlock*         pDelRes;         // delete result
5
54liuyao 已提交
665
  bool                 returnDelete;
5
54liuyao 已提交
666
  SSDataBlock*         pUpdateRes;      // update window
5
54liuyao 已提交
667 668
  SHashObj*            pStDeleted;
  void*                pDelIterator;
5
54liuyao 已提交
669
  SArray*              pChildren;       // cache for children's result; final stream operator
5
54liuyao 已提交
670 671
  SPhysiNode*          pPhyNode;           // create new child
  bool                 isFinal;
5
54liuyao 已提交
672
  bool                 ignoreExpiredData;
5
54liuyao 已提交
673 674
} SStreamSessionAggOperatorInfo;

675
typedef struct STimeSliceOperatorInfo {
676
  SSDataBlock*   pRes;
H
Haojun Liao 已提交
677
  STimeWindow    win;
dengyihao's avatar
dengyihao 已提交
678
  SInterval      interval;
H
Haojun Liao 已提交
679 680
  int64_t        current;
  SArray*        pPrevRow;      // SArray<SGroupValue>
H
Haojun Liao 已提交
681
  int32_t        fillType;      // fill type
682 683
  SColumn        tsCol;         // primary timestamp column
  SExprSupp      scalarSup;     // scalar calculation
H
Haojun Liao 已提交
684
  struct SFillColInfo*  pFillColInfo;  // fill column info
685 686
} STimeSliceOperatorInfo;

687
typedef struct SStateWindowOperatorInfo {
wmmhello's avatar
wmmhello 已提交
688
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
689 690
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
wmmhello's avatar
wmmhello 已提交
691

692 693
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
694
  SColumn            stateCol;  // start row index
695 696
  bool               hasKey;
  SStateKeys         stateKey;
697
  int32_t            tsSlotId;  // primary timestamp column slot id
698
  STimeWindowAggSupp twAggSup;
dengyihao's avatar
dengyihao 已提交
699
  //  bool             reptScan;
L
Liu Jicong 已提交
700
  const SNode* pCondition;
701 702
} SStateWindowOperatorInfo;

5
54liuyao 已提交
703 704 705
typedef struct SStreamStateAggOperatorInfo {
  SOptrBasicInfo       binfo;
  SStreamAggSupporter  streamAggSup;
5
54liuyao 已提交
706
  SExprSupp            scalarSupp;      // supporter for perform scalar function
5
54liuyao 已提交
707 708 709 710
  SGroupResInfo        groupResInfo;
  int32_t              primaryTsIndex;  // primary timestamp slot id
  int32_t              order;           // current SSDataBlock scan order
  STimeWindowAggSupp   twAggSup;
5
54liuyao 已提交
711
  SColumn              stateCol;
5
54liuyao 已提交
712 713 714 715 716
  SqlFunctionCtx*      pDummyCtx;       // for combine
  SSDataBlock*         pDelRes;
  SHashObj*            pSeDeleted;
  void*                pDelIterator;
  SArray*              pChildren;       // cache for children's result;
5
54liuyao 已提交
717
  bool                 ignoreExpiredData;
5
54liuyao 已提交
718 719
} SStreamStateAggOperatorInfo;

720
typedef struct SSortedMergeOperatorInfo {
wmmhello's avatar
wmmhello 已提交
721
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
722
  SOptrBasicInfo   binfo;
wmmhello's avatar
wmmhello 已提交
723 724
  SAggSupporter    aggSup;

725 726 727 728 729 730 731 732 733 734 735
  SArray*          pSortInfo;
  int32_t          numOfSources;
  SSortHandle     *pSortHandle;
  int32_t          bufPageSize;
  uint32_t         sortBufSize;  // max buffer size for in-memory sort
  int32_t          resultRowFactor;
  bool             hasGroupVal;
  SDiskbasedBuf   *pTupleStore;  // keep the final results
  int32_t          numOfResPerPage;
  char**           groupVal;
  SArray          *groupInfo;
736
} SSortedMergeOperatorInfo;
H
Haojun Liao 已提交
737

H
Haojun Liao 已提交
738
typedef struct SSortOperatorInfo {
739
  SOptrBasicInfo binfo;
740
  uint32_t     sortBufSize;    // max buffer size for in-memory sort
dengyihao's avatar
dengyihao 已提交
741 742
  SArray*      pSortInfo;
  SSortHandle* pSortHandle;
743
  SArray*      pColMatchInfo;  // for index map from table scan output
dengyihao's avatar
dengyihao 已提交
744
  int32_t      bufPageSize;
H
Haojun Liao 已提交
745

746 747
  int64_t      startTs;       // sort start time
  uint64_t     sortElapsed;   // sort elapsed time, time to flush to disk not included.
748 749
  SLimitInfo   limitInfo;
  SNode*       pCondition;
H
Haojun Liao 已提交
750
} SSortOperatorInfo;
751

dengyihao's avatar
dengyihao 已提交
752 753 754
typedef struct STagFilterOperatorInfo {
  SOptrBasicInfo binfo;
} STagFilterOperatorInfo;
755

756 757 758 759 760 761 762 763 764 765 766
typedef struct SJoinOperatorInfo {
  SSDataBlock       *pRes;
  int32_t            joinType;

  SSDataBlock       *pLeft;
  int32_t            leftPos;
  SColumnInfo        leftCol;

  SSDataBlock       *pRight;
  int32_t            rightPos;
  SColumnInfo        rightCol;
767
  SNode             *pCondAfterMerge;
768 769
} SJoinOperatorInfo;

770 771 772
#define OPTR_IS_OPENED(_optr)  (((_optr)->status & OP_OPENED) == OP_OPENED)
#define OPTR_SET_OPENED(_optr) ((_optr)->status |= OP_OPENED)

773 774
void doDestroyExchangeOperatorInfo(void* param);

775 776
SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t streamFn,
    __optr_fn_t cleanup, __optr_close_fn_t closeFn, __optr_encode_fn_t encode,
777
    __optr_decode_fn_t decode, __optr_explain_fn_t explain);
778

H
Haojun Liao 已提交
779
int32_t operatorDummyOpenFn(SOperatorInfo* pOperator);
dengyihao's avatar
dengyihao 已提交
780
void    operatorDummyCloseFn(void* param, int32_t numOfCols);
H
Haojun Liao 已提交
781
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num);
782 783 784

void    initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
void    cleanupBasicInfo(SOptrBasicInfo* pInfo);
785
int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
786
void    cleanupExprSupp(SExprSupp* pSup);
787 788
int32_t initAggInfo(SExprSupp *pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
                    const char* pkey);
789
void    initResultSizeInfo(SResultInfo * pResultInfo, int32_t numOfRows);
790
void    doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo, SDiskbasedBuf* pBuf);
791 792 793
int32_t handleLimitOffset(SOperatorInfo *pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf);
bool    hasLimitOffsetInfo(SLimitInfo* pLimitInfo);
void    initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo);
794

795
void    doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, STimeWindow* pWin, SColumnInfoData* pTimeWindowData, int32_t offset,
dengyihao's avatar
dengyihao 已提交
796
                         int32_t forwardStep, TSKEY* tsCol, int32_t numOfTotal, int32_t numOfOutput, int32_t order);
797 798

int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData,
dengyihao's avatar
dengyihao 已提交
799 800
                                  int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total,
                                  SArray* pColList);
801
STimeWindow getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t key);
802 803
STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order);

804
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t *order, int32_t* scanFlag);
805
int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz);
806

dengyihao's avatar
dengyihao 已提交
807
void    doSetOperatorCompleted(SOperatorInfo* pOperator);
808
void    doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, const SArray* pColMatchInfo);
809 810
int32_t addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr,
                               SSDataBlock* pBlock, const char* idStr);
811

812
void    cleanupAggSup(SAggSupporter* pAggSup);
813 814
void    destroyBasicOperatorInfo(void* param, int32_t numOfOutput);
void    appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
815
void    setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId);
816

L
Liu Jicong 已提交
817
int32_t doPrepareScan(SOperatorInfo* pOperator, uint64_t uid, int64_t ts);
818
int32_t doGetScanStatus(SOperatorInfo* pOperator, uint64_t* uid, int64_t* ts);
819

820
SSDataBlock* loadNextDataBlock(void* param);
821

822
void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset);
823 824 825 826

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 已提交
827

828 829
SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo);

H
Haojun Liao 已提交
830
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);
dengyihao's avatar
dengyihao 已提交
831 832
SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode,
                                         STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
833
SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode *pScanPhyNode, const char* pUser, SExecTaskInfo* pTaskInfo);
834

S
slzhou 已提交
835
SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SNode* pCondition, SExprInfo* pScalarExprInfo,
wmmhello's avatar
wmmhello 已提交
836
                                           int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo);
837

H
Haojun Liao 已提交
838
SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode *pNode, SExecTaskInfo* pTaskInfo);
839
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo);
840
SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo);
841
SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams, SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
842
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pSortInfo, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
843
SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
844

dengyihao's avatar
dengyihao 已提交
845
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
846
                                          SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
847
                                          STimeWindowAggSupp* pTwAggSupp, SIntervalPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, bool isStream);
S
shenglian zhou 已提交
848

849 850 851 852
SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                          SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
                                          SExecTaskInfo* pTaskInfo);

853
SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
S
shenglian zhou 已提交
854
                                          SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
855
                                          SNode* pCondition, SExecTaskInfo* pTaskInfo);
S
shenglian zhou 已提交
856

5
54liuyao 已提交
857 858
SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
    SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, int32_t numOfChild);
859 860
SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                                SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
wmmhello's avatar
wmmhello 已提交
861
                                                STimeWindowAggSupp *pTwAggSupp, SExecTaskInfo* pTaskInfo);
dengyihao's avatar
dengyihao 已提交
862
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
863
                                            SSDataBlock* pResBlock, int64_t gap, int32_t tsSlotId, STimeWindowAggSupp* pTwAggSupp,
864
                                            SNode* pCondition, SExecTaskInfo* pTaskInfo);
dengyihao's avatar
dengyihao 已提交
865 866
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                       SSDataBlock* pResultBlock, SArray* pGroupColList, SNode* pCondition,
wmmhello's avatar
wmmhello 已提交
867
                                       SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo);
868 869
SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SReadHandle* readHandle, uint64_t uid, SBlockDistScanPhysiNode* pBlockScanNode,
                                               SExecTaskInfo* pTaskInfo);
870

871
SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond,
872
                                            SExecTaskInfo* pTaskInfo);
873

H
Haojun Liao 已提交
874
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
875

dengyihao's avatar
dengyihao 已提交
876
SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
877 878
                                             SSDataBlock* pResBlock, STimeWindowAggSupp *pTwAggSupp, int32_t tsSlotId,
                                             SColumn* pStateKeyCol, SNode* pCondition, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
879

880
SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo);
881

882
SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
883

884 885
SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SJoinPhysiNode* pJoinNode,
                                           SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
886

887 888 889 890
SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream,
    SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream,
    SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, int32_t numOfChild);
5
54liuyao 已提交
891

892
SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
5
54liuyao 已提交
893

H
Haojun Liao 已提交
894
#if 0
H
Haojun Liao 已提交
895
SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
H
Haojun Liao 已提交
896
#endif
897

898
int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
dengyihao's avatar
dengyihao 已提交
899
                           int32_t numOfOutput, SArray* pPseudoList);
900

901
void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, bool createDummyCol);
X
Xiaoyu Wang 已提交
902

903
bool    isTaskKilled(SExecTaskInfo* pTaskInfo);
904 905
int32_t checkForQueryBuf(size_t numOfTables);

906 907
void    setTaskKilled(SExecTaskInfo* pTaskInfo);
void    queryCostStatis(SExecTaskInfo* pTaskInfo);
908

dengyihao's avatar
dengyihao 已提交
909
void    doDestroyTask(SExecTaskInfo* pTaskInfo);
910 911
int32_t getMaximumIdleDurationSec();

wmmhello's avatar
wmmhello 已提交
912 913 914 915
/*
 * 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 已提交
916
 * nOptrWithVal: *nOptrWithVal save the number of optr with value
wmmhello's avatar
wmmhello 已提交
917 918
 * return:  result code, 0 means success
 */
C
Cary Xu 已提交
919
int32_t encodeOperator(SOperatorInfo* ops, char** data, int32_t *length, int32_t *nOptrWithVal);
wmmhello's avatar
wmmhello 已提交
920 921 922 923 924 925 926

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

929
void    setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
dengyihao's avatar
dengyihao 已提交
930
int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
931
                               const char* sql, EOPTR_EXEC_MODEL model);
D
dapan1121 已提交
932
int32_t createDataSinkParam(SDataSinkNode *pNode, void **pParam, qTaskInfo_t* pTaskInfo, SReadHandle* readHandle);                               
dengyihao's avatar
dengyihao 已提交
933 934 935
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SExplainExecInfo** pRes, int32_t* capacity,
                                   int32_t* resNum);

936
int32_t aggDecodeResultRow(SOperatorInfo* pOperator, char* result);
wmmhello's avatar
wmmhello 已提交
937 938
int32_t aggEncodeResultRow(SOperatorInfo* pOperator, char** result, int32_t* length);

939
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
940
                                int32_t order);
941 942
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
    __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
5
54liuyao 已提交
943
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
944 945
int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput,
    int32_t size);
946
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, int32_t interBufSize);
5
54liuyao 已提交
947 948
SResultWindowInfo* getSessionTimeWindow(SStreamAggSupporter* pAggSup, TSKEY startTs,
    TSKEY endTs, uint64_t groupId, int64_t gap, int32_t* pIndex);
5
54liuyao 已提交
949 950 951
SResultWindowInfo* getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs,
    TSKEY endTs, uint64_t groupId, int64_t gap, int32_t* pIndex);
bool isInTimeWindow(STimeWindow* pWin, TSKEY ts, int64_t gap);
5
54liuyao 已提交
952 953
int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pStartTs,
    TSKEY* pEndTs, int32_t rows, int32_t start, int64_t gap, SHashObj* pStDeleted);
5
54liuyao 已提交
954
bool functionNeedToExecute(SqlFunctionCtx* pCtx);
5
54liuyao 已提交
955
bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup);
956
void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid);
957

958
int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition,
959 960
                                       SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset,
                                       SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
961

962
int32_t createScanTableListInfo(SScanPhysiNode *pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
963
                                STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond, const char* idstr);
964

S
shenglian zhou 已提交
965 966
SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode,
                                           SExecTaskInfo* pTaskInfo);
967
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo *pTableListInfo,
968
                                                SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
969

970
void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);
971

wmmhello's avatar
wmmhello 已提交
972
int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, SNodeList* groupKey);
973
SSDataBlock* createSpecialDataBlock(EStreamType type);
974

975 976 977 978
#ifdef __cplusplus
}
#endif

979
#endif  // TDENGINE_EXECUTORIMPL_H