executorimpl.h 42.2 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

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
  int64_t  created;
  int64_t  start;
92 93 94 95 96 97 98 99
  uint64_t loadStatisTime;
  uint64_t loadFileBlockTime;
  uint64_t loadDataInCacheTime;
  uint64_t loadStatisSize;
  uint64_t loadFileBlockSize;
  uint64_t loadDataInCacheSize;

  uint64_t loadDataTime;
100 101

  SFileBlockLoadRecorder* pRecoder;
L
Liu Jicong 已提交
102
  uint64_t                elapsedTime;
103

104 105 106 107 108 109 110
  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 已提交
111
} STaskCostInfo;
112

H
Haojun Liao 已提交
113
typedef struct SOperatorCostInfo {
L
Liu Jicong 已提交
114 115
  double openCost;
  double totalCost;
H
Haojun Liao 已提交
116 117
} SOperatorCostInfo;

118 119
struct SOperatorInfo;

wmmhello's avatar
wmmhello 已提交
120
typedef int32_t (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, char** result, int32_t* length);
121
typedef int32_t (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, char* result);
122

123
typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* pOptr);
124
typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* pOptr);
125
typedef void (*__optr_close_fn_t)(void* param);
126
typedef int32_t (*__optr_explain_fn_t)(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len);
H
Haojun Liao 已提交
127

H
Haojun Liao 已提交
128
typedef struct STaskIdInfo {
129 130 131 132
  uint64_t queryId;  // this is also a request id
  uint64_t subplanId;
  uint64_t templateId;
  char*    str;
H
Haojun Liao 已提交
133 134
} STaskIdInfo;

L
Liu Jicong 已提交
135 136 137 138 139 140
enum {
  STREAM_RECOVER_STEP__NONE = 0,
  STREAM_RECOVER_STEP__PREPARE,
  STREAM_RECOVER_STEP__SCAN,
};

L
Liu Jicong 已提交
141
typedef struct {
L
Liu Jicong 已提交
142
  //TODO remove prepareStatus
L
Liu Jicong 已提交
143 144
  STqOffsetVal   prepareStatus; // for tmq
  STqOffsetVal   lastStatus;    // for tmq
wmmhello's avatar
wmmhello 已提交
145
  SMqMetaRsp     metaRsp;       // for tmq fetching meta
wmmhello's avatar
wmmhello 已提交
146 147
  SSchemaWrapper *schema;
  char           tbName[TSDB_TABLE_NAME_LEN];
L
Liu Jicong 已提交
148 149
  SSDataBlock*   pullOverBlk;   // for streaming
  SWalFilterCond cond;
L
Liu Jicong 已提交
150
  int64_t        lastScanUid;
L
Liu Jicong 已提交
151 152 153 154
  int8_t         recoverStep;
  SQueryTableDataCond tableCond;
  int64_t recoverStartVer;
  int64_t recoverEndVer;
H
Haojun Liao 已提交
155
  SStreamState* pState;
L
Liu Jicong 已提交
156 157
} SStreamTaskInfo;

158 159 160 161 162
typedef struct {
  char*           tablename;
  char*           dbname;
  int32_t         tversion;
  SSchemaWrapper* sw;
163
  SSchemaWrapper* qsw;
164 165
} SSchemaInfo;

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

174
  int64_t          version; // used for stream to record wal version
175 176 177 178 179 180 181
  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;
182
  struct SOperatorInfo* pRoot;
183
} SExecTaskInfo;
H
Haojun Liao 已提交
184

185
enum {
186 187
  OP_NOT_OPENED    = 0x0,
  OP_OPENED        = 0x1,
H
Haojun Liao 已提交
188
  OP_RES_TO_RETURN = 0x5,
189
  OP_EXEC_DONE     = 0x9,
190 191
};

192
typedef struct SOperatorFpSet {
L
Liu Jicong 已提交
193 194 195 196 197 198 199 200
  __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;
201 202
} SOperatorFpSet;

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

210
typedef struct SOperatorInfo {
211
  uint16_t               operatorType;
212
  int16_t                resultDataBlockId;
L
Liu Jicong 已提交
213 214 215 216 217 218 219 220 221 222 223
  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;
224 225
} SOperatorInfo;

226 227
typedef enum {
  EX_SOURCE_DATA_NOT_READY = 0x1,
228
  EX_SOURCE_DATA_READY     = 0x2,
229 230
  EX_SOURCE_DATA_EXHAUSTED = 0x3,
} EX_SOURCE_STATUS;
231

232 233 234
#define COL_MATCH_FROM_COL_ID  0x1
#define COL_MATCH_FROM_SLOT_ID 0x2

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

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

250
typedef struct SLimitInfo {
251 252 253 254 255 256 257
  SLimit              limit;
  SLimit              slimit;
  uint64_t            currentGroupId;
  int64_t             remainGroupOffset;
  int64_t             numOfOutputGroups;
  int64_t             remainOffset;
  int64_t             numOfOutputRows;
258 259
} SLimitInfo;

260
typedef struct SExchangeInfo {
dengyihao's avatar
dengyihao 已提交
261 262 263 264
  SArray*             pSources;
  SArray*             pSourceDataInfo;
  tsem_t              ready;
  void*               pTransporter;
265 266 267 268 269
  // SArray<SSDataBlock*>, result block list, used to keep the multi-block that
  // passed by downstream operator
  SArray*             pResultBlockList;
  int32_t             rspBlockIndex;    // indicate the return block index in pResultBlockList
  SSDataBlock*        pDummyBlock;  // dummy block, not keep data
dengyihao's avatar
dengyihao 已提交
270 271
  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 SAggSupporter {
H
Haojun Liao 已提交
302
  SSHashObj*     pResultRowHashTable;  // quick locate the window object for each result
303 304 305 306
  char*          keyBuf;               // 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
  int32_t        currentPageId;        // current write page id
307 308 309 310 311 312 313 314 315
} SAggSupporter;

typedef struct {
    // 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
} SAggOptrPushDownInfo;

316
typedef struct STableScanInfo {
H
Hongze Cheng 已提交
317
  STsdbReader*    dataReader;
318 319
  SReadHandle     readHandle;

320
  SFileBlockLoadRecorder readRecorder;
321
  SScanInfo       scanInfo;
322 323
  int32_t         scanTimes;
  SNode*          pFilterNode;  // filter info, which is push down by optimizer
324

325
  SSDataBlock*    pResBlock;
H
Haojun Liao 已提交
326
  SArray*         pColMatchInfo;
327
  SExprSupp       pseudoSup;
328
  SQueryTableDataCond cond;
329
  int32_t         scanFlag;     // table scan flag to denote if it is a repeat/reverse/main scan
330
  int32_t         dataBlockLoadFlag;
331
  SSampleExecInfo sample;       // sample execution info
wmmhello's avatar
wmmhello 已提交
332
  int32_t         currentGroupId;
L
Liu Jicong 已提交
333
  int32_t         currentTable;
334 335
  int8_t          scanMode;
  int8_t          noTable;
336
  SAggOptrPushDownInfo pdInfo;
337
  int8_t          assignBlockUid;
338 339
} STableScanInfo;

340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369
typedef struct STableMergeScanInfo {
  STableListInfo* tableListInfo;
  int32_t         tableStartIndex;
  int32_t         tableEndIndex;
  bool            hasGroupId;
  uint64_t        groupId;
  SArray*         dataReaders;  // array of tsdbReaderT*
  SReadHandle     readHandle;
  int32_t         bufPageSize;
  uint32_t        sortBufSize;  // max buffer size for in-memory sort
  SArray*         pSortInfo;
  SSortHandle*    pSortHandle;

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

  SFileBlockLoadRecorder readRecorder;
  int64_t                numOfRows;
  SScanInfo              scanInfo;
  int32_t                scanTimes;
  SNode*                 pFilterNode;  // filter info, which is push down by optimizer
  SqlFunctionCtx*        pCtx;         // which belongs to the direct upstream operator operator query context
  SResultRowInfo*        pResultRowInfo;
  int32_t*               rowEntryInfoOffset;
  SExprInfo*             pExpr;
  SSDataBlock*           pResBlock;
  SArray*                pColMatchInfo;
  int32_t                numOfOutput;

X
Xiaoyu Wang 已提交
370
  SExprSupp       pseudoSup;
371 372 373 374 375 376 377 378

  SQueryTableDataCond cond;
  int32_t             scanFlag;  // table scan flag to denote if it is a repeat/reverse/main scan
  int32_t             dataBlockLoadFlag;
  // if the upstream is an interval operator, the interval info is also kept here to get the time
  // window to check if current data block needs to be loaded.
  SInterval       interval;
  SSampleExecInfo sample;  // sample execution info
379 380

  SSortExecInfo sortExecInfo;
381 382
} STableMergeScanInfo;

383
typedef struct STagScanInfo {
384 385 386 387 388
  SColumnInfo     *pCols;
  SSDataBlock     *pRes;
  SArray          *pColMatchInfo;
  int32_t          curPos;
  SReadHandle      readHandle;
wmmhello's avatar
wmmhello 已提交
389
  STableListInfo  *pTableList;
390 391
} STagScanInfo;

392 393 394
typedef struct SLastrowScanInfo {
  SSDataBlock    *pRes;
  SReadHandle     readHandle;
395 396 397
  void           *pLastrowReader;
  SArray         *pColMatchInfo;
  int32_t        *pSlotIds;
398
  SExprSupp       pseudoExprSup;
399 400 401 402 403
  int32_t         retrieveType;
  int32_t         currentGroupIndex;
  SSDataBlock    *pBufferredRes;
  SArray         *pUidList;
  int32_t         indexOfBufferedRes;
404 405
} SLastrowScanInfo;

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

414 415 416 417 418
enum {
  PROJECT_RETRIEVE_CONTINUE = 0x1,
  PROJECT_RETRIEVE_DONE = 0x2,
};

5
54liuyao 已提交
419
typedef struct SCatchSupporter {
L
Liu Jicong 已提交
420 421 422 423
  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 已提交
424 425
} SCatchSupporter;

5
54liuyao 已提交
426
typedef struct SStreamAggSupporter {
427 428 429
  SHashObj*      pResultRows;
  SArray*        pCurWins;
  int32_t        valueSize;
5
54liuyao 已提交
430 431 432 433
  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
434
  int32_t        currentPageId;        // buffer page that is active
435
  SSDataBlock*   pScanBlock;
5
54liuyao 已提交
436 437 438 439
} SStreamAggSupporter;

typedef struct SessionWindowSupporter {
  SStreamAggSupporter* pStreamAggSup;
L
Liu Jicong 已提交
440
  int64_t              gap;
441
  uint16_t             parentType;
442
  SAggSupporter*       pIntervalAggSup;
5
54liuyao 已提交
443
} SessionWindowSupporter;
444

5
54liuyao 已提交
445 446 447 448 449 450 451
typedef struct STimeWindowSupp {
  int8_t           calTrigger;
  int64_t          waterMark;
  TSKEY            maxTs;
  SColumnInfoData  timeWindowData;     // query time window info for scalar function execution.
} STimeWindowAggSupp;

452
typedef struct SStreamScanInfo {
L
Liu Jicong 已提交
453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477
  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 已提交
478
  SessionWindowSupporter sessionSup;
L
Liu Jicong 已提交
479 480 481 482 483 484
  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 已提交
485
  STimeWindow            updateWin;
5
54liuyao 已提交
486
  STimeWindowAggSupp     twAggSup;
487
  SSDataBlock*           pUpdateDataRes;
L
Liu Jicong 已提交
488
  // status for tmq
L
Liu Jicong 已提交
489
  // SSchemaWrapper schema;
490
  SNodeList*             pGroupTags;
491 492
  SNode*                 pTagCond;
  SNode*                 pTagIndexCond;
493
} SStreamScanInfo;
H
Haojun Liao 已提交
494

495 496 497 498 499 500 501
typedef struct SStreamRawScanInfo{
//  int8_t    subType;
//  bool      withMeta;
//  int64_t   suid;
//  int64_t   snapVersion;
//  void     *metaInfo;
//  void     *dataInfo;
wmmhello's avatar
wmmhello 已提交
502
  SVnode*         vnode;
503 504 505 506 507
  SSDataBlock     pRes;         // result SSDataBlock
  STsdbReader*    dataReader;
  SSnapContext*   sContext;
}SStreamRawScanInfo;

508
typedef struct SSysTableScanInfo {
dengyihao's avatar
dengyihao 已提交
509 510 511 512
  SRetrieveMetaTableRsp* pRsp;
  SRetrieveTableReq      req;
  SEpSet                 epSet;
  tsem_t                 ready;
513 514
  SReadHandle            readHandle;
  int32_t                accountId;
515
  const char*            pUser;
516
  bool                   sysInfo;
517 518 519 520 521 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.
  SArray*                scanCols;    // SArray<int16_t> scan column id list
  SName                  name;
  SSDataBlock*           pRes;
  int64_t                numOfBlocks;  // extract basic running information.
  SLoadRemoteDataInfo    loadInfo;
525 526
} SSysTableScanInfo;

527 528 529
typedef struct SBlockDistInfo {
  SSDataBlock* pResBlock;
  void*        pHandle;
530 531
  SReadHandle  readHandle;
  uint64_t     uid;        // table uid
532 533
} SBlockDistInfo;

534
// todo remove this
535
typedef struct SOptrBasicInfo {
dengyihao's avatar
dengyihao 已提交
536 537
  SResultRowInfo  resultRowInfo;
  SSDataBlock*    pRes;
538
  bool            mergeResultBlock;
539 540
} SOptrBasicInfo;

541
typedef struct SIntervalAggOperatorInfo {
wmmhello's avatar
wmmhello 已提交
542
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
543
  SOptrBasicInfo     binfo;              // basic info
wmmhello's avatar
wmmhello 已提交
544
  SAggSupporter      aggSup;             // aggregate supporter
545
  SExprSupp          scalarSupp;         // supporter for perform scalar function
546 547 548 549 550
  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
551
  SArray*            pInterpCols;        // interpolation columns
552
  int32_t            resultTsOrder;      // result timestamp order
553
  int32_t            inputOrder;         // input data ts order
554 555
  EOPTR_EXEC_MODEL   execModel;          // operator execution model [batch model|stream model]
  STimeWindowAggSupp twAggSup;
5
54liuyao 已提交
556
  bool               invertible;
557
  SArray*            pPrevValues;        //  SArray<SGroupKeys> used to keep the previous not null value for interpolation.
5
54liuyao 已提交
558
  bool               ignoreExpiredData;
559 560 561 562
  SArray*            pRecycledPages;
  SArray*            pDelWins;           // SWinRes
  int32_t            delIndex;
  SSDataBlock*       pDelRes;
563
  SNode*             pCondition;
564
} SIntervalAggOperatorInfo;
565

566 567 568 569
typedef struct SMergeAlignedIntervalAggOperatorInfo {
  SIntervalAggOperatorInfo *intervalAggOperatorInfo;

  bool         hasGroupId;
570 571
  uint64_t     groupId;  // current groupId
  int64_t      curTs;    // current ts
572 573 574 575
  SSDataBlock* prefetchedBlock;
  SNode*       pCondition;
} SMergeAlignedIntervalAggOperatorInfo;

5
54liuyao 已提交
576
typedef struct SStreamFinalIntervalOperatorInfo {
wmmhello's avatar
wmmhello 已提交
577
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
5
54liuyao 已提交
578
  SOptrBasicInfo     binfo;              // basic info
wmmhello's avatar
wmmhello 已提交
579
  SAggSupporter      aggSup;             // aggregate supporter
5
54liuyao 已提交
580
  SExprSupp          scalarSupp;         // supporter for perform scalar function
5
54liuyao 已提交
581 582 583 584 585
  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;
586
  SArray*            pChildren;
5
54liuyao 已提交
587
  SSDataBlock*       pUpdateRes;
5
54liuyao 已提交
588
  bool               returnUpdate;
5
54liuyao 已提交
589
  SPhysiNode*        pPhyNode;           // create new child
5
54liuyao 已提交
590
  bool               isFinal;
5
54liuyao 已提交
591 592 593 594
  SHashObj*          pPullDataMap;
  SArray*            pPullWins;          // SPullWindowInfo
  int32_t            pullIndex;
  SSDataBlock*       pPullDataRes;
5
54liuyao 已提交
595
  bool               ignoreExpiredData;
596 597 598 599
  SArray*            pRecycledPages;
  SArray*            pDelWins;           // SWinRes
  int32_t            delIndex;
  SSDataBlock*       pDelRes;
5
54liuyao 已提交
600 601
} SStreamFinalIntervalOperatorInfo;

602
typedef struct SAggOperatorInfo {
wmmhello's avatar
wmmhello 已提交
603
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
H
Haojun Liao 已提交
604 605
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
wmmhello's avatar
wmmhello 已提交
606

H
Haojun Liao 已提交
607
  STableQueryInfo   *current;
H
Haojun Liao 已提交
608
  uint64_t           groupId;
H
Haojun Liao 已提交
609
  SGroupResInfo      groupResInfo;
610
  SExprSupp          scalarExprSup;
S
slzhou 已提交
611
  SNode             *pCondition;
612 613 614
} SAggOperatorInfo;

typedef struct SProjectOperatorInfo {
wmmhello's avatar
wmmhello 已提交
615
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
616 617
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
618
  SNode*             pFilterNode;  // filter info, which is push down by optimizer
619
  SArray*            pPseudoColInfo;
620
  SLimitInfo         limitInfo;
H
Haojun Liao 已提交
621 622
  bool               mergeDataBlocks;
  SSDataBlock*       pFinalRes;
623
  SNode*             pCondition;
624 625
} SProjectOperatorInfo;

H
Haojun Liao 已提交
626 627 628 629
typedef struct SIndefOperatorInfo {
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
  SArray*            pPseudoColInfo;
630
  SExprSupp          scalarSup;
S
slzhou 已提交
631
  SNode*             pCondition;
632 633 634
  uint64_t           groupId;

  SSDataBlock*       pNextGroupRes;
H
Haojun Liao 已提交
635 636
} SIndefOperatorInfo;

637
typedef struct SFillOperatorInfo {
638 639
  struct SFillInfo* pFillInfo;
  SSDataBlock*      pRes;
H
Haojun Liao 已提交
640
  SSDataBlock*      pFinalRes;
641 642 643
  int64_t           totalInputRows;
  void**            p;
  SSDataBlock*      existNewGroupBlock;
644
  STimeWindow       win;
S
slzhou 已提交
645
  SNode*            pCondition;
646
  SArray*           pColMatchColInfo;
H
Haojun Liao 已提交
647
  int32_t           primaryTsCol;
H
Haojun Liao 已提交
648
  int32_t           primarySrcSlotId;
649
  uint64_t          curGroupId;       // current handled group id
650 651 652 653
  SExprInfo*        pExprInfo;
  int32_t           numOfExpr;
  SExprInfo*        pNotFillExprInfo;
  int32_t           numOfNotFillExpr;
654 655 656
} SFillOperatorInfo;

typedef struct SGroupbyOperatorInfo {
wmmhello's avatar
wmmhello 已提交
657
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
dengyihao's avatar
dengyihao 已提交
658
  SOptrBasicInfo  binfo;
wmmhello's avatar
wmmhello 已提交
659 660
  SAggSupporter   aggSup;

661
  SArray*         pGroupCols;     // group by columns, SArray<SColumn>
dengyihao's avatar
dengyihao 已提交
662 663 664 665 666 667
  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;
668
  SExprSupp       scalarSup;
669 670
} SGroupbyOperatorInfo;

H
Haojun Liao 已提交
671
typedef struct SDataGroupInfo {
672 673 674
  uint64_t        groupId;
  int64_t         numOfRows;
  SArray*         pPageList;
H
Haojun Liao 已提交
675 676 677 678
} SDataGroupInfo;

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

  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
689 690
  SArray*        sortedGroupArray;   // SDataGroupInfo sorted by group id
  int32_t        groupIndex;  // group index
691
  int32_t        pageIndex;   // page index of current group
692
  SSDataBlock*   pUpdateRes;
693
  SExprSupp      scalarSup;
H
Haojun Liao 已提交
694
} SPartitionOperatorInfo;
695

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

H
Haojun Liao 已提交
704
typedef struct SSessionAggOperatorInfo {
wmmhello's avatar
wmmhello 已提交
705
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
706 707
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
wmmhello's avatar
wmmhello 已提交
708

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

5
54liuyao 已提交
718 719 720
typedef struct SResultWindowInfo {
  SResultRowPosition pos;
  STimeWindow win;
5
54liuyao 已提交
721
  uint64_t groupId;
5
54liuyao 已提交
722
  bool isOutput;
5
54liuyao 已提交
723
  bool isClosed;
5
54liuyao 已提交
724 725
} SResultWindowInfo;

5
54liuyao 已提交
726 727 728 729 730
typedef struct SStateWindowInfo {
  SResultWindowInfo winInfo;
  SStateKeys stateKey;
} SStateWindowInfo;

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

754
typedef struct STimeSliceOperatorInfo {
755 756 757 758 759 760 761
  SSDataBlock*            pRes;
  STimeWindow             win;
  SInterval               interval;
  int64_t                 current;
  SArray*                 pPrevRow;      // SArray<SGroupValue>
  SArray*                 pNextRow;      // SArray<SGroupValue>
  SArray*                 pLinearInfo;   // SArray<SFillLinearInfo>
762
  bool                    fillLastPoint;
763 764 765 766 767 768
  bool                    isPrevRowSet;
  bool                    isNextRowSet;
  int32_t                 fillType;      // fill type
  SColumn                 tsCol;         // primary timestamp column
  SExprSupp               scalarSup;     // scalar calculation
  struct SFillColInfo*    pFillColInfo;  // fill column info
769 770
} STimeSliceOperatorInfo;

771
typedef struct SStateWindowOperatorInfo {
wmmhello's avatar
wmmhello 已提交
772
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
773 774
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
wmmhello's avatar
wmmhello 已提交
775

776 777
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
778
  SColumn            stateCol;  // start row index
779 780
  bool               hasKey;
  SStateKeys         stateKey;
781
  int32_t            tsSlotId;  // primary timestamp column slot id
782
  STimeWindowAggSupp twAggSup;
dengyihao's avatar
dengyihao 已提交
783
  //  bool             reptScan;
L
Liu Jicong 已提交
784
  const SNode* pCondition;
785 786
} SStateWindowOperatorInfo;

5
54liuyao 已提交
787 788 789
typedef struct SStreamStateAggOperatorInfo {
  SOptrBasicInfo       binfo;
  SStreamAggSupporter  streamAggSup;
5
54liuyao 已提交
790
  SExprSupp            scalarSupp;      // supporter for perform scalar function
5
54liuyao 已提交
791 792 793 794
  SGroupResInfo        groupResInfo;
  int32_t              primaryTsIndex;  // primary timestamp slot id
  int32_t              order;           // current SSDataBlock scan order
  STimeWindowAggSupp   twAggSup;
5
54liuyao 已提交
795
  SColumn              stateCol;
5
54liuyao 已提交
796 797 798 799 800
  SqlFunctionCtx*      pDummyCtx;       // for combine
  SSDataBlock*         pDelRes;
  SHashObj*            pSeDeleted;
  void*                pDelIterator;
  SArray*              pChildren;       // cache for children's result;
5
54liuyao 已提交
801
  bool                 ignoreExpiredData;
5
54liuyao 已提交
802 803
} SStreamStateAggOperatorInfo;

804
typedef struct SSortedMergeOperatorInfo {
wmmhello's avatar
wmmhello 已提交
805
  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
806
  SOptrBasicInfo   binfo;
wmmhello's avatar
wmmhello 已提交
807 808
  SAggSupporter    aggSup;

809 810 811 812 813 814 815 816 817 818 819
  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;
820
} SSortedMergeOperatorInfo;
H
Haojun Liao 已提交
821

H
Haojun Liao 已提交
822
typedef struct SSortOperatorInfo {
823
  SOptrBasicInfo binfo;
824
  uint32_t     sortBufSize;    // max buffer size for in-memory sort
dengyihao's avatar
dengyihao 已提交
825 826
  SArray*      pSortInfo;
  SSortHandle* pSortHandle;
827
  SArray*      pColMatchInfo;  // for index map from table scan output
dengyihao's avatar
dengyihao 已提交
828
  int32_t      bufPageSize;
H
Haojun Liao 已提交
829

830 831
  int64_t      startTs;       // sort start time
  uint64_t     sortElapsed;   // sort elapsed time, time to flush to disk not included.
832 833
  SLimitInfo   limitInfo;
  SNode*       pCondition;
H
Haojun Liao 已提交
834
} SSortOperatorInfo;
835

dengyihao's avatar
dengyihao 已提交
836 837 838
typedef struct STagFilterOperatorInfo {
  SOptrBasicInfo binfo;
} STagFilterOperatorInfo;
839

840 841 842
typedef struct SJoinOperatorInfo {
  SSDataBlock       *pRes;
  int32_t            joinType;
843
  int32_t            inputOrder;
844 845 846 847 848 849 850 851

  SSDataBlock       *pLeft;
  int32_t            leftPos;
  SColumnInfo        leftCol;

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

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

858 859
void doDestroyExchangeOperatorInfo(void* param);

860 861
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,
862
    __optr_decode_fn_t decode, __optr_explain_fn_t explain);
863

H
Haojun Liao 已提交
864
int32_t operatorDummyOpenFn(SOperatorInfo* pOperator);
dengyihao's avatar
dengyihao 已提交
865
void    operatorDummyCloseFn(void* param, int32_t numOfCols);
H
Haojun Liao 已提交
866
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num);
867 868 869

void    initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
void    cleanupBasicInfo(SOptrBasicInfo* pInfo);
870
int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
871
void    cleanupExprSupp(SExprSupp* pSup);
872
void    destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs);
873 874
int32_t initAggInfo(SExprSupp *pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
                    const char* pkey);
875
void    initResultSizeInfo(SResultInfo * pResultInfo, int32_t numOfRows);
876
void    doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo, SDiskbasedBuf* pBuf);
877 878 879
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);
880

881 882
void    doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, SColumnInfoData* pTimeWindowData, int32_t offset,
                         int32_t forwardStep, int32_t numOfTotal, int32_t numOfOutput);
883

884 885 886 887
int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, int32_t numOfOutput, SArray* pColList, char** pNextStart);
void updateLoadRemoteInfo(SLoadRemoteDataInfo *pInfo, int32_t numOfRows, int32_t dataLen, int64_t startTs,
                          SOperatorInfo* pOperator);

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

890
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t *order, int32_t* scanFlag);
891
int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz);
892

dengyihao's avatar
dengyihao 已提交
893
void    doSetOperatorCompleted(SOperatorInfo* pOperator);
894
void    doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, const SArray* pColMatchInfo);
895 896
int32_t addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr,
                               SSDataBlock* pBlock, const char* idStr);
897

898
void    cleanupAggSup(SAggSupporter* pAggSup);
899 900
void    destroyBasicOperatorInfo(void* param, int32_t numOfOutput);
void    appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
901
void    setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId);
902

L
Liu Jicong 已提交
903
int32_t doPrepareScan(SOperatorInfo* pOperator, uint64_t uid, int64_t ts);
904
int32_t doGetScanStatus(SOperatorInfo* pOperator, uint64_t* uid, int64_t* ts);
905

906
SSDataBlock* loadNextDataBlock(void* param);
907

908
void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset);
909 910 911 912

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

914 915
SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo);

H
Haojun Liao 已提交
916
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);
dengyihao's avatar
dengyihao 已提交
917 918
SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode,
                                         STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
919
SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode *pScanPhyNode, const char* pUser, SExecTaskInfo* pTaskInfo);
920

S
slzhou 已提交
921
SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SNode* pCondition, SExprInfo* pScalarExprInfo,
922
                                           int32_t numOfScalarExpr, bool mergeResult, SExecTaskInfo* pTaskInfo);
923

H
Haojun Liao 已提交
924
SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode *pNode, SExecTaskInfo* pTaskInfo);
925
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo);
926
SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo);
927
SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams, SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo);
928
SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
929

dengyihao's avatar
dengyihao 已提交
930
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
931
                                          SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
932
                                          STimeWindowAggSupp* pTwAggSupp, SIntervalPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, bool isStream);
933 934 935 936
SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMergeIntervalPhysiNode* pIntervalPhyNode,
                                               SExecTaskInfo* pTaskInfo);
SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SMergeAlignedIntervalPhysiNode* pNode,
                                                      SExecTaskInfo* pTaskInfo);
937 938
SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                     SExecTaskInfo* pTaskInfo, int32_t numOfChild);
H
Haojun Liao 已提交
939 940
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPhysiNode* pSessionNode,
                                            SExecTaskInfo* pTaskInfo);
dengyihao's avatar
dengyihao 已提交
941 942
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                       SSDataBlock* pResultBlock, SArray* pGroupColList, SNode* pCondition,
wmmhello's avatar
wmmhello 已提交
943
                                       SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo);
944 945
SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SReadHandle* readHandle, uint64_t uid,
                                               SBlockDistScanPhysiNode* pBlockScanNode, SExecTaskInfo* pTaskInfo);
946

947
SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond,
948
                                            SExecTaskInfo* pTaskInfo);
949

950 951
SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);

H
Haojun Liao 已提交
952
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode, SExecTaskInfo* pTaskInfo);
953
SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWinodwPhysiNode* pStateNode, SExecTaskInfo* pTaskInfo);
954
SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo);
955
SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
956
SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SSortMergeJoinPhysiNode* pJoinNode,
957
                                           SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
958

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

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

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

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

971
bool    isTaskKilled(SExecTaskInfo* pTaskInfo);
972 973
int32_t checkForQueryBuf(size_t numOfTables);

974 975
void    setTaskKilled(SExecTaskInfo* pTaskInfo);
void    queryCostStatis(SExecTaskInfo* pTaskInfo);
976

dengyihao's avatar
dengyihao 已提交
977
void    doDestroyTask(SExecTaskInfo* pTaskInfo);
978 979
int32_t getMaximumIdleDurationSec();

wmmhello's avatar
wmmhello 已提交
980 981 982 983
/*
 * 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 已提交
984
 * nOptrWithVal: *nOptrWithVal save the number of optr with value
wmmhello's avatar
wmmhello 已提交
985 986
 * return:  result code, 0 means success
 */
C
Cary Xu 已提交
987
int32_t encodeOperator(SOperatorInfo* ops, char** data, int32_t *length, int32_t *nOptrWithVal);
wmmhello's avatar
wmmhello 已提交
988 989 990 991 992 993 994

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

997
void    setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
dengyihao's avatar
dengyihao 已提交
998
int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
D
dapan1121 已提交
999
                               char* sql, EOPTR_EXEC_MODEL model);
H
Haojun Liao 已提交
1000 1001
int32_t createDataSinkParam(SDataSinkNode *pNode, void **pParam, qTaskInfo_t* pTaskInfo, SReadHandle* readHandle);
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList);
dengyihao's avatar
dengyihao 已提交
1002

1003
int32_t aggDecodeResultRow(SOperatorInfo* pOperator, char* result);
wmmhello's avatar
wmmhello 已提交
1004 1005
int32_t aggEncodeResultRow(SOperatorInfo* pOperator, char** result, int32_t* length);

1006
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
1007
                                int32_t order);
1008 1009
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
    __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
5
54liuyao 已提交
1010
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
1011 1012
int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput,
    int32_t size);
1013
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, int32_t interBufSize);
5
54liuyao 已提交
1014 1015
SResultWindowInfo* getSessionTimeWindow(SStreamAggSupporter* pAggSup, TSKEY startTs,
    TSKEY endTs, uint64_t groupId, int64_t gap, int32_t* pIndex);
5
54liuyao 已提交
1016 1017 1018
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 已提交
1019
bool functionNeedToExecute(SqlFunctionCtx* pCtx);
5
54liuyao 已提交
1020
bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup);
5
54liuyao 已提交
1021
bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup);
1022
bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
H
Haojun Liao 已提交
1023
void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, int32_t uidCol, uint64_t* pID);
5
54liuyao 已提交
1024
void printDataBlock(SSDataBlock* pBlock, const char* flag);
1025

1026
int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition,
1027 1028
                                       SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset,
                                       SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
1029

1030
int32_t createScanTableListInfo(SScanPhysiNode *pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
1031
                                STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond, const char* idstr);
1032

S
shenglian zhou 已提交
1033 1034
SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode,
                                           SExecTaskInfo* pTaskInfo);
1035
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo *pTableListInfo,
1036
                                                SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
1037

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

1040
bool groupbyTbname(SNodeList* pGroupList);
wmmhello's avatar
wmmhello 已提交
1041
int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, SNodeList* groupKey);
1042
SSDataBlock* createSpecialDataBlock(EStreamType type);
5
54liuyao 已提交
1043
void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput);
1044

1045 1046 1047 1048
#ifdef __cplusplus
}
#endif

1049
#endif  // TDENGINE_EXECUTORIMPL_H