executorimpl.h 35.7 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */
#ifndef TDENGINE_EXECUTORIMPL_H
#define TDENGINE_EXECUTORIMPL_H

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

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

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

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

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

5
54liuyao 已提交
49 50 51 52 53
#define IS_VALID_SESSION_WIN(winInfo)        ((winInfo).sessionWin.win.skey > 0)
#define SET_SESSION_WIN_INVALID(winInfo)     ((winInfo).sessionWin.win.skey = INT64_MIN)
#define IS_INVALID_SESSION_WIN_KEY(winKey)   ((winKey).win.skey <= 0)
#define SET_SESSION_WIN_KEY_INVALID(pWinKey) ((pWinKey)->win.skey = INT64_MIN)

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

58
  /* Task is over
59 60 61
   * 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
   */
62
  TASK_COMPLETED = 0x2u,
63 64 65 66 67 68
};

/**
 * 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 已提交
69 70 71 72 73
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 已提交
74
} SResultInfo;
75 76

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

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

86
typedef struct STableScanAnalyzeInfo SFileBlockLoadRecorder;
87

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

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

102 103
struct SOperatorInfo;

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

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

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

L
Liu Jicong 已提交
121 122
enum {
  STREAM_RECOVER_STEP__NONE = 0,
123 124
  STREAM_RECOVER_STEP__PREPARE1,
  STREAM_RECOVER_STEP__PREPARE2,
5
54liuyao 已提交
125 126
  STREAM_RECOVER_STEP__SCAN1,
  STREAM_RECOVER_STEP__SCAN2,
L
Liu Jicong 已提交
127 128
};

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

138
  SPackedData         submit;
L
Liu Jicong 已提交
139 140 141
  SSchemaWrapper*     schema;
  char                tbName[TSDB_TABLE_NAME_LEN];
  int8_t              recoverStep;
L
Liu Jicong 已提交
142
  int8_t              recoverScanFinished;
L
Liu Jicong 已提交
143
  SQueryTableDataCond tableCond;
L
Liu Jicong 已提交
144 145
  int64_t             fillHistoryVer1;
  int64_t             fillHistoryVer2;
146

L
Liu Jicong 已提交
147 148
  // int8_t        triggerSaved;
  // int64_t       deleteMarkSaved;
149
  SStreamState* pState;
L
Liu Jicong 已提交
150 151
} SStreamTaskInfo;

152 153 154 155 156
typedef struct {
  char*           tablename;
  char*           dbname;
  int32_t         tversion;
  SSchemaWrapper* sw;
157
  SSchemaWrapper* qsw;
158 159
} SSchemaInfo;

H
Haojun Liao 已提交
160
typedef struct SExchangeOpStopInfo {
L
Liu Jicong 已提交
161 162
  int32_t operatorType;
  int64_t refId;
D
dapan1121 已提交
163 164
} SExchangeOpStopInfo;

H
Haojun Liao 已提交
165
typedef struct STaskStopInfo {
D
dapan1121 已提交
166 167 168 169
  SRWLatch lock;
  SArray*  pStopInfo;
} STaskStopInfo;

H
Haojun Liao 已提交
170
struct SExecTaskInfo {
171 172 173 174 175 176
  STaskIdInfo   id;
  uint32_t      status;
  STimeWindow   window;
  STaskCostInfo cost;
  int64_t       owner;  // if it is in execution
  int32_t       code;
177
  int32_t       qbufQuota;  // total available buffer (in KB) during execution query
178

179
  int64_t               version;  // used for stream to record wal version, why not move to sschemainfo
L
Liu Jicong 已提交
180 181
  SStreamTaskInfo       streamInfo;
  SSchemaInfo           schemaInfo;
H
Haojun Liao 已提交
182
  STableListInfo*       pTableInfoList;  // this is a table list
L
Liu Jicong 已提交
183 184 185 186
  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
  SLocalFetch           localFetch;
L
Liu Jicong 已提交
189
  SArray*               pResultBlockList;  // result block list
D
dapan1121 已提交
190
  STaskStopInfo         stopInfo;
H
Haojun Liao 已提交
191
};
H
Haojun Liao 已提交
192

193
enum {
L
Liu Jicong 已提交
194 195
  OP_NOT_OPENED = 0x0,
  OP_OPENED = 0x1,
H
Haojun Liao 已提交
196
  OP_RES_TO_RETURN = 0x5,
L
Liu Jicong 已提交
197
  OP_EXEC_DONE = 0x9,
L
Liu Jicong 已提交
198
  OP_EXEC_RECV = 0x11,
199 200
};

201
typedef struct SOperatorFpSet {
L
Liu Jicong 已提交
202 203
  __optr_open_fn_t    _openFn;  // DO NOT invoke this function directly
  __optr_fn_t         getNextFn;
204
  __optr_fn_t         cleanupFn;  // call this function to release the allocated resources ASAP
L
Liu Jicong 已提交
205
  __optr_close_fn_t   closeFn;
L
Liu Jicong 已提交
206
  __optr_reqBuf_fn_t  reqBufFn;  // total used buffer for blocking operator
L
Liu Jicong 已提交
207 208 209
  __optr_encode_fn_t  encodeResultRow;
  __optr_decode_fn_t  decodeResultRow;
  __optr_explain_fn_t getExplainFn;
210 211
} SOperatorFpSet;

212 213
typedef struct SExprSupp {
  SExprInfo*      pExprInfo;
L
Liu Jicong 已提交
214
  int32_t         numOfExprs;  // the number of scalar expression in group operator
215 216
  SqlFunctionCtx* pCtx;
  int32_t*        rowEntryInfoOffset;  // offset value for each row result cell info
H
Haojun Liao 已提交
217
  SFilterInfo*    pFilterInfo;
218 219
} SExprSupp;

220
typedef struct SOperatorInfo {
221
  uint16_t               operatorType;
222
  int16_t                resultDataBlockId;
L
Liu Jicong 已提交
223 224 225 226 227 228 229 230 231 232 233
  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;
234 235
} SOperatorInfo;

236 237
typedef enum {
  EX_SOURCE_DATA_NOT_READY = 0x1,
L
Liu Jicong 已提交
238
  EX_SOURCE_DATA_READY = 0x2,
239 240
  EX_SOURCE_DATA_EXHAUSTED = 0x3,
} EX_SOURCE_STATUS;
241

242 243 244
#define COL_MATCH_FROM_COL_ID  0x1
#define COL_MATCH_FROM_SLOT_ID 0x2

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

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

261
typedef struct SExchangeInfo {
L
Liu Jicong 已提交
262 263 264 265
  SArray* pSources;
  SArray* pSourceDataInfo;
  tsem_t  ready;
  void*   pTransporter;
H
Haojun Liao 已提交
266

267 268
  // SArray<SSDataBlock*>, result block list, used to keep the multi-block that
  // passed by downstream operator
L
Liu Jicong 已提交
269 270 271 272 273
  SArray*      pResultBlockList;
  SArray*      pRecycledBlocks;  // build a pool for small data block to avoid to repeatly create and then destroy.
  SSDataBlock* pDummyBlock;      // dummy block, not keep data
  bool         seqLoadData;      // sequential load data or not, false by default
  int32_t      current;
H
Haojun Liao 已提交
274
  SLoadRemoteDataInfo loadInfo;
275
  uint64_t            self;
276
  SLimitInfo          limitInfo;
L
Liu Jicong 已提交
277
  int64_t             openedTs;  // start exec time stamp, todo: move to SLoadRemoteDataInfo
278 279
} SExchangeInfo;

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

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

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

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

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

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

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

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

344
typedef struct STableMergeScanInfo {
L
Liu Jicong 已提交
345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364
  int32_t         tableStartIndex;
  int32_t         tableEndIndex;
  bool            hasGroupId;
  uint64_t        groupId;
  SArray*         queryConds;  // array of queryTableDataCond
  STableScanBase  base;
  int32_t         bufPageSize;
  uint32_t        sortBufSize;  // max buffer size for in-memory sort
  SArray*         pSortInfo;
  SSortHandle*    pSortHandle;
  SSDataBlock*    pSortInputBlock;
  int64_t         startTs;  // sort start time
  SArray*         sortSourceParams;
  SLimitInfo      limitInfo;
  int64_t         numOfRows;
  SScanInfo       scanInfo;
  int32_t         scanTimes;
  SSDataBlock*    pResBlock;
  SSampleExecInfo sample;  // sample execution info
  SSortExecInfo   sortExecInfo;
365 366
} STableMergeScanInfo;

367
typedef struct STagScanInfo {
L
Liu Jicong 已提交
368 369 370 371 372
  SColumnInfo*  pCols;
  SSDataBlock*  pRes;
  SColMatchInfo matchInfo;
  int32_t       curPos;
  SReadHandle   readHandle;
373 374
} STagScanInfo;

5
54liuyao 已提交
375 376 377 378
typedef enum EStreamScanMode {
  STREAM_SCAN_FROM_READERHANDLE = 1,
  STREAM_SCAN_FROM_RES,
  STREAM_SCAN_FROM_UPDATERES,
379
  STREAM_SCAN_FROM_DELETE_DATA,
5
54liuyao 已提交
380
  STREAM_SCAN_FROM_DATAREADER_RETRIEVE,
L
Liu Jicong 已提交
381
  STREAM_SCAN_FROM_DATAREADER_RANGE,
5
54liuyao 已提交
382 383
} EStreamScanMode;

384 385 386 387 388
enum {
  PROJECT_RETRIEVE_CONTINUE = 0x1,
  PROJECT_RETRIEVE_DONE = 0x2,
};

5
54liuyao 已提交
389
typedef struct SStreamAggSupporter {
5
54liuyao 已提交
390 391 392
  int32_t         resultRowSize;  // the result buffer size for each result row, with the meta data size for each row
  SSDataBlock*    pScanBlock;
  SStreamState*   pState;
393 394
  int64_t         gap;        // stream session window gap
  SqlFunctionCtx* pDummyCtx;  // for combine
5
54liuyao 已提交
395 396 397 398
  SSHashObj*      pResultRows;
  int32_t         stateKeySize;
  int16_t         stateKeyType;
  SDiskbasedBuf*  pResultBuf;
5
54liuyao 已提交
399 400
} SStreamAggSupporter;

401
typedef struct SWindowSupporter {
5
54liuyao 已提交
402
  SStreamAggSupporter* pStreamAggSup;
L
Liu Jicong 已提交
403
  int64_t              gap;
404
  uint16_t             parentType;
405
  SAggSupporter*       pIntervalAggSup;
406
} SWindowSupporter;
407

408
typedef struct SPartitionBySupporter {
L
Liu Jicong 已提交
409 410 411 412
  SArray* pGroupCols;     // group by columns, SArray<SColumn>
  SArray* pGroupColVals;  // current group column values, SArray<SGroupKeys>
  char*   keyBuf;         // group by keys for hash
  bool    needCalc;       // partition by column
413 414 415
} SPartitionBySupporter;

typedef struct SPartitionDataInfo {
L
Liu Jicong 已提交
416
  uint64_t groupId;
417 418
  char*    tbname;
  SArray*  tags;
L
Liu Jicong 已提交
419
  SArray*  rowIds;
420
} SPartitionDataInfo;
421

5
54liuyao 已提交
422
typedef struct STimeWindowAggSupp {
L
Liu Jicong 已提交
423
  int8_t          calTrigger;
L
Liu Jicong 已提交
424
  int8_t          calTriggerSaved;
5
54liuyao 已提交
425
  int64_t         deleteMark;
L
Liu Jicong 已提交
426 427
  int64_t         deleteMarkSaved;
  int64_t         waterMark;
L
Liu Jicong 已提交
428
  TSKEY           maxTs;
5
54liuyao 已提交
429
  TSKEY           minTs;
L
Liu Jicong 已提交
430
  SColumnInfoData timeWindowData;  // query time window info for scalar function execution.
5
54liuyao 已提交
431 432
} STimeWindowAggSupp;

433
typedef struct SStreamScanInfo {
L
Liu Jicong 已提交
434 435 436 437 438 439 440 441 442
  uint64_t      tableUid;  // queried super table uid
  SExprInfo*    pPseudoExpr;
  int32_t       numOfPseudoExpr;
  SExprSupp     tbnameCalSup;
  SExprSupp     tagCalSup;
  int32_t       primaryTsIndex;  // primary time stamp slot id
  SReadHandle   readHandle;
  SInterval     interval;  // if the upstream is an interval operator, the interval info is also kept here.
  SColMatchInfo matchInfo;
L
Liu Jicong 已提交
443 444 445 446 447 448 449 450 451 452 453 454 455

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

  uint64_t     groupId;
  SUpdateInfo* pUpdateInfo;

L
Liu Jicong 已提交
456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471
  EStreamScanMode       scanMode;
  SOperatorInfo*        pStreamScanOp;
  SOperatorInfo*        pTableScanOp;
  SArray*               childIds;
  SWindowSupporter      windowSup;
  SPartitionBySupporter partitionSup;
  SExprSupp*            pPartScalarSup;
  bool                  assignBlockUid;  // assign block uid to groupId, temporarily used for generating rollup SMA.
  int32_t               scanWinIndex;    // for state operator
  int32_t               pullDataResIndex;
  SSDataBlock*          pPullDataRes;    // pull data SSDataBlock
  SSDataBlock*          pDeleteDataRes;  // delete data SSDataBlock
  int32_t               deleteDataIndex;
  STimeWindow           updateWin;
  STimeWindowAggSupp    twAggSup;
  SSDataBlock*          pUpdateDataRes;
L
Liu Jicong 已提交
472
  // status for tmq
L
Liu Jicong 已提交
473 474 475
  SNodeList* pGroupTags;
  SNode*     pTagCond;
  SNode*     pTagIndexCond;
L
Liu Jicong 已提交
476 477

  // recover
5
54liuyao 已提交
478 479 480
  int32_t      blockRecoverContiCnt;
  int32_t      blockRecoverTotCnt;
  SSDataBlock* pRecoverRes;
L
Liu Jicong 已提交
481

5
54liuyao 已提交
482
  SSDataBlock* pCreateTbRes;
X
Xiaoyu Wang 已提交
483 484
  int8_t       igCheckUpdate;
  int8_t       igExpired;
485
} SStreamScanInfo;
H
Haojun Liao 已提交
486

L
Liu Jicong 已提交
487 488 489 490 491 492 493 494 495 496 497 498
typedef struct {
  //  int8_t    subType;
  //  bool      withMeta;
  //  int64_t   suid;
  //  int64_t   snapVersion;
  //  void     *metaInfo;
  //  void     *dataInfo;
  SVnode*       vnode;
  SSDataBlock   pRes;  // result SSDataBlock
  STsdbReader*  dataReader;
  SSnapContext* sContext;
} SStreamRawScanInfo;
499

S
slzhou 已提交
500 501 502 503
typedef struct STableCountScanSupp {
  int16_t dbNameSlotId;
  int16_t stbNameSlotId;
  int16_t tbCountSlotId;
504 505
  bool    groupByDbName;
  bool    groupByStbName;
S
slzhou 已提交
506 507
  char    dbNameFilter[TSDB_DB_NAME_LEN];
  char    stbNameFilter[TSDB_TABLE_NAME_LEN];
S
slzhou 已提交
508 509 510 511 512 513 514 515
} STableCountScanSupp;

typedef struct STableCountScanOperatorInfo {
  SReadHandle  readHandle;
  SSDataBlock* pRes;

  STableCountScanSupp supp;

L
Liu Jicong 已提交
516 517
  int32_t currGrpIdx;
  SArray* stbUidList;  // when group by db_name and/or stable_name
S
slzhou 已提交
518
} STableCountScanOperatorInfo;
S
shenglian zhou 已提交
519

520
typedef struct SOptrBasicInfo {
L
Liu Jicong 已提交
521 522 523
  SResultRowInfo resultRowInfo;
  SSDataBlock*   pRes;
  bool           mergeResultBlock;
524 525
} SOptrBasicInfo;

526
typedef struct SIntervalAggOperatorInfo {
527
  SOptrBasicInfo     binfo;              // basic info
wmmhello's avatar
wmmhello 已提交
528
  SAggSupporter      aggSup;             // aggregate supporter
529
  SExprSupp          scalarSupp;         // supporter for perform scalar function
530 531 532 533 534
  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
535
  SArray*            pInterpCols;        // interpolation columns
536
  int32_t            resultTsOrder;      // result timestamp order
537
  int32_t            inputOrder;         // input data ts order
538 539
  EOPTR_EXEC_MODEL   execModel;          // operator execution model [batch model|stream model]
  STimeWindowAggSupp twAggSup;
L
Liu Jicong 已提交
540
  SArray*            pPrevValues;  //  SArray<SGroupKeys> used to keep the previous not null value for interpolation.
541
} SIntervalAggOperatorInfo;
542

543
typedef struct SMergeAlignedIntervalAggOperatorInfo {
L
Liu Jicong 已提交
544
  SIntervalAggOperatorInfo* intervalAggOperatorInfo;
545

546 547
  uint64_t     groupId;  // current groupId
  int64_t      curTs;    // current ts
548
  SSDataBlock* prefetchedBlock;
549
  SResultRow*  pResultRow;
550 551
} SMergeAlignedIntervalAggOperatorInfo;

552
typedef struct SStreamIntervalOperatorInfo {
553 554 555 556 557 558
  SOptrBasicInfo     binfo;           // basic info
  SAggSupporter      aggSup;          // aggregate supporter
  SExprSupp          scalarSupp;      // supporter for perform scalar function
  SGroupResInfo      groupResInfo;    // multiple results build supporter
  SInterval          interval;        // interval info
  int32_t            primaryTsIndex;  // primary time stamp slot id from result of downstream operator.
559 560 561
  STimeWindowAggSupp twAggSup;
  bool               invertible;
  bool               ignoreExpiredData;
562
  bool               ignoreExpiredDataSaved;
563
  SArray*            pDelWins;  // SWinRes
564 565
  int32_t            delIndex;
  SSDataBlock*       pDelRes;
566
  SPhysiNode*        pPhyNode;  // create new child
5
54liuyao 已提交
567
  SHashObj*          pPullDataMap;
568
  SArray*            pPullWins;  // SPullWindowInfo
5
54liuyao 已提交
569 570
  int32_t            pullIndex;
  SSDataBlock*       pPullDataRes;
571 572 573 574
  bool               isFinal;
  SArray*            pChildren;
  SStreamState*      pState;
  SWinKey            delKey;
5
54liuyao 已提交
575
  uint64_t           numOfDatapack;
5
54liuyao 已提交
576
  SArray*            pUpdated;
577
  SSHashObj*         pUpdatedMap;
578
} SStreamIntervalOperatorInfo;
5
54liuyao 已提交
579

H
Haojun Liao 已提交
580
typedef struct SDataGroupInfo {
L
Liu Jicong 已提交
581 582 583
  uint64_t groupId;
  int64_t  numOfRows;
  SArray*  pPageList;
H
Haojun Liao 已提交
584 585
} SDataGroupInfo;

586
typedef struct SWindowRowsSup {
dengyihao's avatar
dengyihao 已提交
587 588 589 590
  STimeWindow win;
  TSKEY       prevTs;
  int32_t     startRowIndex;
  int32_t     numOfRows;
591
  uint64_t    groupId;
592 593
} SWindowRowsSup;

5
54liuyao 已提交
594
typedef struct SResultWindowInfo {
595 596 597
  void*       pOutputBuf;
  SSessionKey sessionWin;
  bool        isOutput;
5
54liuyao 已提交
598 599
} SResultWindowInfo;

5
54liuyao 已提交
600 601
typedef struct SStateWindowInfo {
  SResultWindowInfo winInfo;
5
54liuyao 已提交
602
  SStateKeys*       pStateKey;
5
54liuyao 已提交
603 604
} SStateWindowInfo;

5
54liuyao 已提交
605
typedef struct SStreamSessionAggOperatorInfo {
L
Liu Jicong 已提交
606 607
  SOptrBasicInfo      binfo;
  SStreamAggSupporter streamAggSup;
608
  SExprSupp           scalarSupp;  // supporter for perform scalar function
L
Liu Jicong 已提交
609 610 611 612 613
  SGroupResInfo       groupResInfo;
  int32_t             primaryTsIndex;  // primary timestamp slot id
  int32_t             endTsIndex;      // window end timestamp slot id
  int32_t             order;           // current SSDataBlock scan order
  STimeWindowAggSupp  twAggSup;
614 615 616
  SSDataBlock*        pWinBlock;   // window result
  SSDataBlock*        pDelRes;     // delete result
  SSDataBlock*        pUpdateRes;  // update window
L
Liu Jicong 已提交
617
  bool                returnUpdate;
5
54liuyao 已提交
618
  SSHashObj*          pStDeleted;
L
Liu Jicong 已提交
619
  void*               pDelIterator;
620 621
  SArray*             pChildren;  // cache for children's result; final stream operator
  SPhysiNode*         pPhyNode;   // create new child
L
Liu Jicong 已提交
622 623
  bool                isFinal;
  bool                ignoreExpiredData;
624
  bool                ignoreExpiredDataSaved;
5
54liuyao 已提交
625 626
  SArray*             pUpdated;
  SSHashObj*          pStUpdated;
5
54liuyao 已提交
627 628
} SStreamSessionAggOperatorInfo;

5
54liuyao 已提交
629 630 631
typedef struct SStreamStateAggOperatorInfo {
  SOptrBasicInfo      binfo;
  SStreamAggSupporter streamAggSup;
632
  SExprSupp           scalarSupp;  // supporter for perform scalar function
5
54liuyao 已提交
633 634 635 636 637 638 639
  SGroupResInfo       groupResInfo;
  int32_t             primaryTsIndex;  // primary timestamp slot id
  STimeWindowAggSupp  twAggSup;
  SColumn             stateCol;
  SSDataBlock*        pDelRes;
  SSHashObj*          pSeDeleted;
  void*               pDelIterator;
640
  SArray*             pChildren;  // cache for children's result;
5
54liuyao 已提交
641
  bool                ignoreExpiredData;
642
  bool                ignoreExpiredDataSaved;
5
54liuyao 已提交
643 644
  SArray*             pUpdated;
  SSHashObj*          pSeUpdated;
5
54liuyao 已提交
645 646
} SStreamStateAggOperatorInfo;

647 648 649 650
typedef struct SStreamPartitionOperatorInfo {
  SOptrBasicInfo        binfo;
  SPartitionBySupporter partitionSup;
  SExprSupp             scalarSup;
651
  SExprSupp             tbnameCalSup;
L
Liu Jicong 已提交
652
  SExprSupp             tagCalSup;
653 654
  SHashObj*             pPartitions;
  void*                 parIte;
5
54liuyao 已提交
655
  void*                 pTbNameIte;
656 657
  SSDataBlock*          pInputDataBlock;
  int32_t               tsColIndex;
658
  SSDataBlock*          pDelRes;
5
54liuyao 已提交
659
  SSDataBlock*          pCreateTbRes;
660 661
} SStreamPartitionOperatorInfo;

5
54liuyao 已提交
662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678
typedef struct SStreamFillSupporter {
  int32_t        type;  // fill type
  SInterval      interval;
  SResultRowData prev;
  SResultRowData cur;
  SResultRowData next;
  SResultRowData nextNext;
  SFillColInfo*  pAllColInfo;  // fill exprs and not fill exprs
  SExprSupp      notFillExprSup;
  int32_t        numOfAllCols;  // number of all exprs, including the tags columns
  int32_t        numOfFillCols;
  int32_t        numOfNotFillCols;
  int32_t        rowSize;
  SSHashObj*     pResMap;
  bool           hasDelete;
} SStreamFillSupporter;

5
54liuyao 已提交
679 680
typedef struct SStreamFillOperatorInfo {
  SStreamFillSupporter* pFillSup;
681 682 683 684 685 686
  SSDataBlock*          pRes;
  SSDataBlock*          pSrcBlock;
  int32_t               srcRowIndex;
  SSDataBlock*          pSrcDelBlock;
  int32_t               srcDelRowIndex;
  SSDataBlock*          pDelRes;
L
Liu Jicong 已提交
687
  SColMatchInfo         matchInfo;
688 689 690
  int32_t               primaryTsCol;
  int32_t               primarySrcSlotId;
  SStreamFillInfo*      pFillInfo;
5
54liuyao 已提交
691 692
} SStreamFillOperatorInfo;

693 694 695
#define OPTR_IS_OPENED(_optr)  (((_optr)->status & OP_OPENED) == OP_OPENED)
#define OPTR_SET_OPENED(_optr) ((_optr)->status |= OP_OPENED)

H
Haojun Liao 已提交
696
SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup,
697 698
                                   __optr_close_fn_t closeFn, __optr_reqBuf_fn_t reqBufFn, __optr_explain_fn_t explain);
int32_t        optrDummyOpenFn(SOperatorInfo* pOperator);
H
Haojun Liao 已提交
699 700 701 702 703
int32_t        appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num);
void           setOperatorCompleted(SOperatorInfo* pOperator);
void           setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status,
                               void* pInfo, SExecTaskInfo* pTaskInfo);
void           destroyOperatorInfo(SOperatorInfo* pOperator);
704
int32_t        optrDefaultBufFn(SOperatorInfo* pOperator);
705

L
Liu Jicong 已提交
706 707
void initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
void cleanupBasicInfo(SOptrBasicInfo* pInfo);
H
Haojun Liao 已提交
708

709
int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
710
void    cleanupExprSupp(SExprSupp* pSup);
H
Haojun Liao 已提交
711

L
Liu Jicong 已提交
712
void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs);
H
Haojun Liao 已提交
713

H
Haojun Liao 已提交
714
int32_t initAggSup(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
L
Liu Jicong 已提交
715
                   const char* pkey, void* pState);
H
Haojun Liao 已提交
716 717
void    cleanupAggSup(SAggSupporter* pAggSup);

L
Liu Jicong 已提交
718
void initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows);
719 720 721

void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                           SDiskbasedBuf* pBuf);
722 723
void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                            SDiskbasedBuf* pBuf);
724

H
Haojun Liao 已提交
725
bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo);
726
bool hasSlimitOffsetInfo(SLimitInfo* pLimitInfo);
H
Haojun Liao 已提交
727
void initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo);
728
void resetLimitInfoForNextGroup(SLimitInfo* pLimitInfo);
729
bool applyLimitOffset(SLimitInfo* pLimitInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
730

H
Haojun Liao 已提交
731 732
void applyAggFunctionOnPartialTuples(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, SColumnInfoData* pTimeWindowData,
                                     int32_t offset, int32_t forwardStep, int32_t numOfTotal, int32_t numOfOutput);
733

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

738 739
STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order);

740
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag, bool inheritUsOrder);
741
int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz);
742

H
Haojun Liao 已提交
743
extern void doDestroyExchangeOperatorInfo(void* param);
H
Haojun Liao 已提交
744

H
Haojun Liao 已提交
745
void    doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo);
L
Liu Jicong 已提交
746 747
int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock,
                               int32_t rows, const char* idStr, STableMetaCacheInfo* pCache);
748

L
Liu Jicong 已提交
749
void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
H
Haojun Liao 已提交
750
void setTbNameColData(const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId, const char* name);
751

752
void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset);
753
void clearResultRowInitFlag(SqlFunctionCtx* pCtx, int32_t numOfOutput);
754

L
Liu Jicong 已提交
755 756
SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pResultRowInfo, char* pData,
                                   int16_t bytes, bool masterscan, uint64_t groupId, SExecTaskInfo* pTaskInfo,
D
dapan1121 已提交
757
                                   bool isIntervalQuery, SAggSupporter* pSup, bool keepGroup);
H
Haojun Liao 已提交
758 759
// operator creater functions
// clang-format off
760 761
SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo);

H
Haojun Liao 已提交
762 763
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);

H
Haojun Liao 已提交
764 765
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);

H
Haojun Liao 已提交
766 767 768
SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);

SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode* pScanPhyNode, const char* pUser, SExecTaskInfo* pTaskInfo);
769

770 771
SOperatorInfo* createTableCountScanOperatorInfo(SReadHandle* handle, STableCountScanPhysiNode* pNode, SExecTaskInfo* pTaskInfo);

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

H
Haojun Liao 已提交
774 775 776 777
SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, SExecTaskInfo* pTaskInfo);

SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo);

778
SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
779 780 781 782 783

SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams, SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo);

SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);

5
54liuyao 已提交
784
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
785 786 787 788 789 790 791 792 793

SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMergeIntervalPhysiNode* pIntervalPhyNode, SExecTaskInfo* pTaskInfo);

SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SMergeAlignedIntervalPhysiNode* pNode, SExecTaskInfo* pTaskInfo);

SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, int32_t numOfChild);

SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPhysiNode* pSessionNode, SExecTaskInfo* pTaskInfo);

794
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNode, SExecTaskInfo* pTaskInfo);
795

H
Haojun Liao 已提交
796 797 798
SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDistScanPhysiNode* pBlockScanNode, SExecTaskInfo* pTaskInfo);

SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond, SExecTaskInfo* pTaskInfo);
799

800 801
SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);

H
Haojun Liao 已提交
802 803 804
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode, SExecTaskInfo* pTaskInfo);

SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWinodwPhysiNode* pStateNode, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
805

H
Haojun Liao 已提交
806
SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
807

H
Haojun Liao 已提交
808
SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo);
809

810
SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
L
Liu Jicong 已提交
811

H
Haojun Liao 已提交
812 813 814 815 816 817 818 819 820
SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SSortMergeJoinPhysiNode* pJoinNode, SExecTaskInfo* pTaskInfo);

SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);

SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, int32_t numOfChild);

SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);

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

H
Haojun Liao 已提交
822
SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
823 824

SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
825 826

SOperatorInfo* createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
827
// clang-format on
5
54liuyao 已提交
828

829
int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
L
Liu Jicong 已提交
830
                              int32_t numOfOutput, SArray* pPseudoList);
831

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

834 835
int32_t checkForQueryBuf(size_t numOfTables);

L
Liu Jicong 已提交
836 837 838 839
bool isTaskKilled(SExecTaskInfo* pTaskInfo);
void setTaskKilled(SExecTaskInfo* pTaskInfo, int32_t rspCode);
void doDestroyTask(SExecTaskInfo* pTaskInfo);
void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
H
Haojun Liao 已提交
840

H
Haojun Liao 已提交
841 842
char* buildTaskId(uint64_t taskId, uint64_t queryId);

843 844
int32_t createExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
                               int32_t vgId, char* sql, EOPTR_EXEC_MODEL model);
L
Liu Jicong 已提交
845
int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, qTaskInfo_t* pTaskInfo, SReadHandle* readHandle);
H
Haojun Liao 已提交
846
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList);
dengyihao's avatar
dengyihao 已提交
847

848
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
849
                                int32_t order);
850
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
L
Liu Jicong 已提交
851
                                 __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
5
54liuyao 已提交
852
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
853
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, int32_t interBufSize);
5
54liuyao 已提交
854
void getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId, SSessionKey* pKey);
855 856 857 858 859 860 861 862
bool isInTimeWindow(STimeWindow* pWin, TSKEY ts, int64_t gap);
bool functionNeedToExecute(SqlFunctionCtx* pCtx);
bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup);
bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup);
bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup);
void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid,
                                      uint64_t* pGp, void* pTbName);
L
Liu Jicong 已提交
863
uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
864

865 866
int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup,
                           SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
867

L
Liu Jicong 已提交
868
bool    groupbyTbname(SNodeList* pGroupList);
869
int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
870
                                   SGroupResInfo* pGroupResInfo);
5
54liuyao 已提交
871
int32_t saveSessionDiscBuf(SStreamState* pState, SSessionKey* key, void* buf, int32_t size);
872
int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock,
5
54liuyao 已提交
873
                                    SExprSupp* pSup, SGroupResInfo* pGroupResInfo);
874 875
int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId,
                     SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup);
876 877
int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult);
int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize);
5
54liuyao 已提交
878
void    getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order);
L
Liu Jicong 已提交
879
int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo);
X
Xiaoyu Wang 已提交
880 881
int32_t getForwardStepsInBlock(int32_t numOfRows, __block_search_fn_t searchFn, TSKEY ekey, int32_t pos, int32_t order,
                               int64_t* pData);
5
54liuyao 已提交
882
void    appendCreateTableRow(SStreamState* pState, SExprSupp* pTableSup, SExprSupp* pTagSup, uint64_t groupId,
5
54liuyao 已提交
883 884 885 886
                             SSDataBlock* pSrcBlock, int32_t rowId, SSDataBlock* pDestBlock);

SSDataBlock* buildCreateTableBlock(SExprSupp* tbName, SExprSupp* tag);
SExprInfo*   createExpr(SNodeList* pNodeList, int32_t* numOfExprs);
887

H
Haojun Liao 已提交
888
void copyResultrowToDataBlock(SExprInfo* pExprInfo, int32_t numOfExprs, SResultRow* pRow, SqlFunctionCtx* pCtx,
X
Xiaoyu Wang 已提交
889 890
                              SSDataBlock* pBlock, const int32_t* rowEntryOffset, SExecTaskInfo* pTaskInfo);
void doUpdateNumOfRows(SqlFunctionCtx* pCtx, SResultRow* pRow, int32_t numOfExprs, const int32_t* rowEntryOffset);
891
void doClearBufferedBlocks(SStreamScanInfo* pInfo);
H
Haojun Liao 已提交
892

893 894 895 896
#ifdef __cplusplus
}
#endif

897
#endif  // TDENGINE_EXECUTORIMPL_H