executorimpl.h 32.8 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"
H
Haojun Liao 已提交
42

H
Hongze Cheng 已提交
43
#include "vnode.h"
44
#include "executorInt.h"
H
Hongze Cheng 已提交
45

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

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

52
#define GET_TABLEGROUP(q, _index) ((SArray*)taosArrayGetP((q)->tableqinfoGroupInfo.pGroupList, (_index)))
53

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

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

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

typedef struct SResultRowCell {
dengyihao's avatar
dengyihao 已提交
68
  uint64_t           groupId;
H
Haojun Liao 已提交
69
  SResultRowPosition pos;
70 71 72 73 74 75
} SResultRowCell;

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

typedef struct STableQueryInfo {
84 85
  TSKEY              lastKey;     // last check ts, todo remove it later
  SResultRowPosition pos;       // current active time window
86
//  SVariant       tag;
87 88 89 90 91 92 93 94 95 96
} STableQueryInfo;

typedef enum {
  QUERY_PROF_BEFORE_OPERATOR_EXEC = 0,
  QUERY_PROF_AFTER_OPERATOR_EXEC,
  QUERY_PROF_QUERY_ABORT
} EQueryProfEventType;

typedef struct {
  EQueryProfEventType eventType;
97
  int64_t             eventTime;
98 99

  union {
100 101
    uint8_t operatorType;  // for operator event
    int32_t abortCode;     // for query abort event
102 103 104 105 106 107 108 109 110
  };
} SQueryProfEvent;

typedef struct {
  uint8_t operatorType;
  int64_t sumSelfTime;
  int64_t sumRunTimes;
} SOperatorProfResult;

H
Haojun Liao 已提交
111 112 113 114 115
typedef struct SLimit {
  int64_t limit;
  int64_t offset;
} SLimit;

116 117 118 119 120 121 122 123 124 125 126
typedef struct SFileBlockLoadRecorder {
  uint64_t totalRows;
  uint64_t totalCheckedRows;
  uint32_t totalBlocks;
  uint32_t loadBlocks;
  uint32_t loadBlockStatis;
  uint32_t skipBlocks;
  uint32_t filterOutBlocks;
  uint64_t elapsedTime;
} SFileBlockLoadRecorder;

H
Haojun Liao 已提交
127
typedef struct STaskCostInfo {
128 129 130 131 132 133 134 135 136 137 138 139
  int64_t created;
  int64_t start;
  int64_t end;

  uint64_t loadStatisTime;
  uint64_t loadFileBlockTime;
  uint64_t loadDataInCacheTime;
  uint64_t loadStatisSize;
  uint64_t loadFileBlockSize;
  uint64_t loadDataInCacheSize;

  uint64_t loadDataTime;
140 141

  SFileBlockLoadRecorder* pRecoder;
142
  uint64_t elapsedTime;
143

144 145 146 147 148 149 150 151
  uint64_t firstStageMergeTime;
  uint64_t winInfoSize;
  uint64_t tableInfoSize;
  uint64_t hashSize;
  uint64_t numOfTimeWindows;

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

H
Haojun Liao 已提交
154 155
typedef struct SOperatorCostInfo {
  uint64_t openCost;
D
dapan1121 已提交
156
  uint64_t totalCost;
H
Haojun Liao 已提交
157 158
} SOperatorCostInfo;

159 160
// The basic query information extracted from the SQueryInfo tree to support the
// execution of query in a data node.
H
Haojun Liao 已提交
161
typedef struct STaskAttr {
dengyihao's avatar
dengyihao 已提交
162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182
  SLimit      limit;
  SLimit      slimit;
  bool        stableQuery;        // super table query or not
  bool        topBotQuery;        // TODO used bitwise flag
  bool        groupbyColumn;      // denote if this is a groupby normal column query
  bool        timeWindowInterpo;  // if the time window start/end required interpolation
  bool        tsCompQuery;        // is tscomp query
  bool        diffQuery;          // is diff query
  bool        pointInterpQuery;   // point interpolation query
  int32_t     havingNum;          // having expr number
  int16_t     numOfCols;
  int16_t     numOfTags;
  STimeWindow window;
  SInterval   interval;
  int16_t     precision;
  int16_t     numOfOutput;
  int16_t     fillType;
  int32_t     resultRowSize;
  int32_t     tagLen;  // tag value length of current query

  SExprInfo*      pExpr1;
183 184 185
  SColumnInfo*    tagColList;
  int32_t         numOfFilterCols;
  int64_t*        fillVal;
186 187 188
  void*           tsdb;
  STableGroupInfo tableGroupInfo;  // table <tid, last_key> list  SArray<STableKeyInfo>
  int32_t         vgId;
H
Haojun Liao 已提交
189
} STaskAttr;
190 191

struct SOperatorInfo;
192 193
struct SAggSupporter;
struct SOptrBasicInfo;
194

dengyihao's avatar
dengyihao 已提交
195 196 197 198
typedef void (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, struct SAggSupporter* pSup,
                                   struct SOptrBasicInfo* pInfo, char** result, int32_t* length);
typedef bool (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, struct SAggSupporter* pSup,
                                   struct SOptrBasicInfo* pInfo, char* result, int32_t length);
199

200
typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* pOptr);
201
typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* pOptr);
H
Haojun Liao 已提交
202
typedef void (*__optr_close_fn_t)(void* param, int32_t num);
dengyihao's avatar
dengyihao 已提交
203
typedef int32_t (*__optr_get_explain_fn_t)(struct SOperatorInfo* pOptr, void** pOptrExplain);
H
Haojun Liao 已提交
204

H
Haojun Liao 已提交
205
typedef struct STaskIdInfo {
206 207 208 209
  uint64_t queryId;  // this is also a request id
  uint64_t subplanId;
  uint64_t templateId;
  char*    str;
H
Haojun Liao 已提交
210 211
} STaskIdInfo;

212
typedef struct SExecTaskInfo {
dengyihao's avatar
dengyihao 已提交
213 214 215 216 217 218 219 220 221 222 223
  STaskIdInfo      id;
  uint32_t         status;
  STimeWindow      window;
  STaskCostInfo    cost;
  int64_t          owner;  // if it is in execution
  int32_t          code;
  uint64_t         totalRows;            // total number of rows
  STableGroupInfo  tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
  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]
224
  struct SOperatorInfo* pRoot;
225
} SExecTaskInfo;
H
Haojun Liao 已提交
226

H
Haojun Liao 已提交
227
typedef struct STaskRuntimeEnv {
H
Haojun Liao 已提交
228 229 230 231 232 233 234 235
  STaskAttr*      pQueryAttr;
  uint32_t        status;  // query status
  uint8_t         scanFlag;  // denotes reversed scan of data or not
  SDiskbasedBuf*  pResultBuf;           // query result buffer based on blocked-wised disk file
  SHashObj*       pResultRowHashTable;  // quick locate the window object for each result
  SHashObj*       pResultRowListSet;    // used to check if current ResultRowInfo has ResultRow object or not
  SArray*         pResultRowArrayList;  // The array list that contains the Result rows
  char*           keyBuf;               // window key buffer
H
Haojun Liao 已提交
236
  // The window result objects pool, all the resultRow Objects are allocated and managed by this object.
dengyihao's avatar
dengyihao 已提交
237 238 239
  char**    prevRow;
  STSBuf*   pTsBuf;      // timestamp filter list
  STSCursor cur;
240

dengyihao's avatar
dengyihao 已提交
241
  char*                          tagVal;  // tag value of current data block
242 243
  STableGroupInfo tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
  struct SOperatorInfo* proot;
dengyihao's avatar
dengyihao 已提交
244 245
  SGroupResInfo         groupResInfo;
  int64_t               currentOffset;  // dynamic offset value
246

247
  STableQueryInfo* current;
248
  SResultInfo      resultInfo;
249
  struct SUdfInfo* pUdfInfo;
H
Haojun Liao 已提交
250
} STaskRuntimeEnv;
251 252

enum {
dengyihao's avatar
dengyihao 已提交
253 254
  OP_NOT_OPENED = 0x0,
  OP_OPENED = 0x1,
H
Haojun Liao 已提交
255
  OP_RES_TO_RETURN = 0x5,
dengyihao's avatar
dengyihao 已提交
256
  OP_EXEC_DONE = 0x9,
257 258
};

259 260 261 262 263 264 265 266 267 268 269
typedef struct SOperatorFpSet {
  __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_get_explain_fn_t getExplainFn;
} SOperatorFpSet;

270
typedef struct SOperatorInfo {
D
dapan1121 已提交
271
  uint8_t                 operatorType;
272
  bool                    blocking;      // block operator or not
D
dapan1121 已提交
273
  uint8_t                 status;        // denote if current operator is completed
274
  int32_t                 numOfExprs;   // number of columns of the current operator results
D
dapan1121 已提交
275 276 277 278 279 280 281 282
  char*                   name;          // name, used to show the query execution plan
  void*                   info;          // extension attribution
  SExprInfo*              pExpr;
  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
283
  SOperatorFpSet          fpSet;
284 285
} SOperatorInfo;

286 287
typedef enum {
  EX_SOURCE_DATA_NOT_READY = 0x1,
288
  EX_SOURCE_DATA_READY     = 0x2,
289 290
  EX_SOURCE_DATA_EXHAUSTED = 0x3,
} EX_SOURCE_STATUS;
291 292

typedef struct SSourceDataInfo {
dengyihao's avatar
dengyihao 已提交
293
  struct SExchangeInfo* pEx;
294
  int32_t               index;
dengyihao's avatar
dengyihao 已提交
295
  SRetrieveTableRsp*    pRsp;
296
  uint64_t              totalRows;
H
Haojun Liao 已提交
297
  int32_t               code;
298
  EX_SOURCE_STATUS      status;
299 300
} SSourceDataInfo;

H
Haojun Liao 已提交
301
typedef struct SLoadRemoteDataInfo {
dengyihao's avatar
dengyihao 已提交
302 303 304
  uint64_t totalSize;     // total load bytes from remote
  uint64_t totalRows;     // total number of rows
  uint64_t totalElapsed;  // total elapsed time
H
Haojun Liao 已提交
305 306
} SLoadRemoteDataInfo;

307
typedef struct SExchangeInfo {
dengyihao's avatar
dengyihao 已提交
308 309 310 311 312 313 314
  SArray*             pSources;
  SArray*             pSourceDataInfo;
  tsem_t              ready;
  void*               pTransporter;
  SSDataBlock*        pResult;
  bool                seqLoadData;  // sequential load data or not, false by default
  int32_t             current;
H
Haojun Liao 已提交
315
  SLoadRemoteDataInfo loadInfo;
316 317
} SExchangeInfo;

318 319 320
#define COL_MATCH_FROM_COL_ID  0x1
#define COL_MATCH_FROM_SLOT_ID 0x2

H
Haojun Liao 已提交
321
typedef struct SColMatchInfo {
322
  int32_t srcSlotId;     // source slot id
H
Haojun Liao 已提交
323 324 325
  int32_t colId;
  int32_t targetSlotId;
  bool    output;
326
  int32_t matchType;     // determinate the source according to col id or slot id
H
Haojun Liao 已提交
327 328
} SColMatchInfo;

329 330 331 332 333
typedef struct SScanInfo {
  int32_t numOfAsc;
  int32_t numOfDesc;
} SScanInfo;

334
typedef struct STableScanInfo {
H
Haojun Liao 已提交
335
  void*           dataReader;
336 337
  SReadHandle     readHandle;

338
  SFileBlockLoadRecorder readRecorder;
339
  int64_t         numOfRows;
340
  int64_t         elapsedTime;
341
//  int32_t         prevGroupId;  // previous table group id
342
  SScanInfo       scanInfo;
343 344 345
  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
346 347 348
  SResultRowInfo* pResultRowInfo;
  int32_t*        rowCellInfoOffset;
  SExprInfo*      pExpr;
349
  SSDataBlock*    pResBlock;
H
Haojun Liao 已提交
350
  SArray*         pColMatchInfo;
351
  int32_t         numOfOutput;
352

353 354 355 356 357
  SExprInfo*      pPseudoExpr;
  int32_t         numOfPseudoExpr;
  SqlFunctionCtx* pPseudoCtx;
//  int32_t*        rowCellInfoOffset;

358
  SQueryTableDataCond cond;
359
  int32_t         scanFlag;     // table scan flag to denote if it is a repeat/reverse/main scan
360
  int32_t         dataBlockLoadFlag;
361
  double          sampleRatio;  // data block sample ratio, 1 by default
362
  SInterval       interval;     // if the upstream is an interval operator, the interval info is also kept here to get the time window to check if current data block needs to be loaded.
363 364 365
} STableScanInfo;

typedef struct STagScanInfo {
366 367 368 369 370 371
  SColumnInfo     *pCols;
  SSDataBlock     *pRes;
  SArray          *pColMatchInfo;
  int32_t          curPos;
  SReadHandle      readHandle;
  STableGroupInfo *pTableGroups;
372 373
} STagScanInfo;

5
54liuyao 已提交
374 375 376 377 378 379 380
typedef enum EStreamScanMode {
  STREAM_SCAN_FROM_READERHANDLE = 1,
  STREAM_SCAN_FROM_RES,
  STREAM_SCAN_FROM_UPDATERES,
  STREAM_SCAN_FROM_DATAREADER,
} EStreamScanMode;

H
Haojun Liao 已提交
381
typedef struct SStreamBlockScanInfo {
dengyihao's avatar
dengyihao 已提交
382 383
  SArray*      pBlockLists;      // multiple SSDatablock.
  SSDataBlock* pRes;             // result SSDataBlock
5
54liuyao 已提交
384 385
  SSDataBlock* pUpdateRes;       // update SSDataBlock
  int32_t      updateResIndex;
dengyihao's avatar
dengyihao 已提交
386 387 388 389 390 391 392
  int32_t      blockType;        // current block type
  int32_t      validBlockIndex;  // Is current data has returned?
  SColumnInfo* pCols;            // the output column info
  uint64_t     numOfRows;        // total scanned rows
  uint64_t     numOfExec;        // execution times
  void*        readerHandle;     // stream block reader handle
  SArray*      pColMatchInfo;    //
5
54liuyao 已提交
393
  SNode*       pCondition;
5
54liuyao 已提交
394
  SArray*      tsArray;
5
54liuyao 已提交
395
  SUpdateInfo* pUpdateInfo;
5
54liuyao 已提交
396
  int32_t      primaryTsIndex;    // primary time stamp slot id
5
54liuyao 已提交
397 398 399 400
  void*        pDataReader;
  EStreamScanMode scanMode;
  SOperatorInfo* pOperatorDumy;
  SInterval      interval;     // if the upstream is an interval operator, the interval info is also kept here.
H
Haojun Liao 已提交
401 402
} SStreamBlockScanInfo;

403
typedef struct SSysTableScanInfo {
404
  SReadHandle readHandle;
405

dengyihao's avatar
dengyihao 已提交
406 407 408 409
  SRetrieveMetaTableRsp* pRsp;
  SRetrieveTableReq      req;
  SEpSet                 epSet;
  tsem_t                 ready;
H
Haojun Liao 已提交
410

H
Haojun Liao 已提交
411 412 413
  int32_t             accountId;
  bool                showRewrite;
  SNode*              pCondition;  // db_name filter condition, to discard data that are not in current database
H
Hongze Cheng 已提交
414
  SMTbCursor*         pCur;        // cursor for iterate the local table meta store.
H
Haojun Liao 已提交
415
  SArray*             scanCols;    // SArray<int16_t> scan column id list
H
Haojun Liao 已提交
416
  SName               name;
dengyihao's avatar
dengyihao 已提交
417
  SSDataBlock*        pRes;
H
Haojun Liao 已提交
418
  int64_t             numOfBlocks;  // extract basic running information.
H
Haojun Liao 已提交
419
  SLoadRemoteDataInfo loadInfo;
420 421
} SSysTableScanInfo;

422
typedef struct SOptrBasicInfo {
dengyihao's avatar
dengyihao 已提交
423 424 425 426
  SResultRowInfo  resultRowInfo;
  int32_t*        rowCellInfoOffset;  // offset value for each row result cell info
  SqlFunctionCtx* pCtx;
  SSDataBlock*    pRes;
427 428
} SOptrBasicInfo;

dengyihao's avatar
dengyihao 已提交
429
// TODO move the resultrowsiz together with SOptrBasicInfo:rowCellInfoOffset
430
typedef struct SAggSupporter {
dengyihao's avatar
dengyihao 已提交
431 432 433
  SHashObj*      pResultRowHashTable;  // quick locate the window object for each result
  char*          keyBuf;               // window key buffer
  SDiskbasedBuf* pResultBuf;           // query result buffer based on blocked-wised disk file
434
  int32_t        resultRowSize;        // the result buffer size for each result row, with the meta data size for each row
435 436
} SAggSupporter;

437 438 439
typedef struct STimeWindowSupp {
  int8_t           calTrigger;
  int64_t          waterMark;
dengyihao's avatar
dengyihao 已提交
440
  SColumnInfoData  timeWindowData;     // query time window info for scalar function execution.
441 442
} STimeWindowAggSupp;

443
typedef struct SIntervalAggOperatorInfo {
444 445 446 447 448 449 450 451 452 453 454 455 456
  SOptrBasicInfo     binfo;              // basic info
  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
  char**             pRow;               // previous row/tuple of already processed datablock
  SAggSupporter      aggSup;             // aggregate supporter
  STableQueryInfo*   pCurrent;           // current tableQueryInfo struct
  int32_t            order;              // current SSDataBlock scan order
  EOPTR_EXEC_MODEL   execModel;          // operator execution model [batch model|stream model]
  SArray*            pUpdatedWindow;     // updated time window due to the input data block from the downstream operator.
  STimeWindowAggSupp twAggSup;
457
  struct SFillInfo*  pFillInfo;          // fill info
5
54liuyao 已提交
458
  bool               invertible;
459
} SIntervalAggOperatorInfo;
460 461

typedef struct SAggOperatorInfo {
H
Haojun Liao 已提交
462 463 464
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
  STableQueryInfo   *current;
H
Haojun Liao 已提交
465
  uint64_t           groupId;
H
Haojun Liao 已提交
466 467
  SGroupResInfo      groupResInfo;
  STableQueryInfo   *pTableQueryInfo;
468 469 470 471 472

  SExprInfo         *pScalarExprInfo;
  int32_t            numOfScalarExpr;      // the number of scalar expression before the aggregate function can be applied
  SqlFunctionCtx    *pScalarCtx;                 // scalar function requried sql function struct.
  int32_t           *rowCellInfoOffset;  // offset value for each row result cell info
473 474 475
} SAggOperatorInfo;

typedef struct SProjectOperatorInfo {
476 477 478 479 480 481 482 483 484 485 486 487 488
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
  SSDataBlock*       existDataBlock;
  SArray*            pPseudoColInfo;
  SLimit             limit;
  SLimit             slimit;

  uint64_t           groupId;
  int64_t            curSOffset;
  int64_t            curGroupOutput;

  int64_t            curOffset;
  int64_t            curOutput;
489 490 491
} SProjectOperatorInfo;

typedef struct SFillOperatorInfo {
492 493 494 495 496 497
  struct SFillInfo* pFillInfo;
  SSDataBlock*      pRes;
  int64_t           totalInputRows;
  void**            p;
  SSDataBlock*      existNewGroupBlock;
  bool              multigroupResult;
498 499 500
} SFillOperatorInfo;

typedef struct SGroupbyOperatorInfo {
dengyihao's avatar
dengyihao 已提交
501
  SOptrBasicInfo  binfo;
502
  SArray*         pGroupCols;     // group by columns, SArray<SColumn>
dengyihao's avatar
dengyihao 已提交
503 504 505 506 507 508 509 510 511 512
  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;
  SAggSupporter   aggSup;
  SExprInfo*      pScalarExprInfo;
  int32_t         numOfScalarExpr;  // the number of scalar expression in group operator
  SqlFunctionCtx* pScalarFuncCtx;
513
  int32_t*        rowCellInfoOffset;  // offset value for each row result cell info
514 515
} SGroupbyOperatorInfo;

H
Haojun Liao 已提交
516
typedef struct SDataGroupInfo {
517 518 519
  uint64_t        groupId;
  int64_t         numOfRows;
  SArray*         pPageList;
H
Haojun Liao 已提交
520 521 522 523
} SDataGroupInfo;

// The sort in partition may be needed later.
typedef struct SPartitionOperatorInfo {
dengyihao's avatar
dengyihao 已提交
524 525 526 527 528 529 530 531 532 533
  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
534 535
  void*          pGroupIter;  // group iterator
  int32_t        pageIndex;   // page index of current group
H
Haojun Liao 已提交
536
} SPartitionOperatorInfo;
537

538
typedef struct SWindowRowsSup {
dengyihao's avatar
dengyihao 已提交
539 540 541 542
  STimeWindow win;
  TSKEY       prevTs;
  int32_t     startRowIndex;
  int32_t     numOfRows;
543 544
} SWindowRowsSup;

H
Haojun Liao 已提交
545
typedef struct SSessionAggOperatorInfo {
546 547 548 549 550 551
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
  bool               reptScan;        // next round scan
  int64_t            gap;             // session window gap
552
  int32_t            tsSlotId;        // primary timestamp slot id
553
  STimeWindowAggSupp twAggSup;
H
Haojun Liao 已提交
554
} SSessionAggOperatorInfo;
555

556
typedef struct STimeSliceOperatorInfo {
dengyihao's avatar
dengyihao 已提交
557 558 559
  SOptrBasicInfo binfo;
  SInterval      interval;
  SGroupResInfo  groupResInfo;  // multiple results build supporter
560 561
} STimeSliceOperatorInfo;

562
typedef struct SStateWindowOperatorInfo {
563 564 565 566
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
567
  SColumn            stateCol;  // start row index
568 569
  bool               hasKey;
  SStateKeys         stateKey;
570
  int32_t            tsSlotId;  // primary timestamp column slot id
571
  STimeWindowAggSupp twAggSup;
dengyihao's avatar
dengyihao 已提交
572
  //  bool             reptScan;
573 574
} SStateWindowOperatorInfo;

575
typedef struct SSortedMergeOperatorInfo {
576

577 578 579 580 581 582 583 584 585 586 587 588 589
  SOptrBasicInfo   binfo;
  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;
  SAggSupporter    aggSup;
590
} SSortedMergeOperatorInfo;
H
Haojun Liao 已提交
591

H
Haojun Liao 已提交
592
typedef struct SSortOperatorInfo {
593
  SOptrBasicInfo binfo;
dengyihao's avatar
dengyihao 已提交
594 595 596
  uint32_t     sortBufSize;  // max buffer size for in-memory sort
  SArray*      pSortInfo;
  SSortHandle* pSortHandle;
597
  SArray*      pColMatchInfo;  // for index map from table scan output
dengyihao's avatar
dengyihao 已提交
598
  int32_t      bufPageSize;
H
Haojun Liao 已提交
599

H
Haojun Liao 已提交
600
  // TODO extact struct
dengyihao's avatar
dengyihao 已提交
601 602 603 604 605
  int64_t  startTs;       // sort start time
  uint64_t sortElapsed;   // sort elapsed time, time to flush to disk not included.
  uint64_t totalSize;     // total load bytes from remote
  uint64_t totalRows;     // total number of rows
  uint64_t totalElapsed;  // total elapsed time
H
Haojun Liao 已提交
606
} SSortOperatorInfo;
607

dengyihao's avatar
dengyihao 已提交
608 609 610
typedef struct STagFilterOperatorInfo {
  SOptrBasicInfo binfo;
} STagFilterOperatorInfo;
611

612 613 614 615 616 617 618 619 620 621 622 623 624 625
typedef struct SJoinOperatorInfo {
  SSDataBlock       *pRes;
  int32_t            joinType;

  SSDataBlock       *pLeft;
  int32_t            leftPos;
  SColumnInfo        leftCol;

  SSDataBlock       *pRight;
  int32_t            rightPos;
  SColumnInfo        rightCol;
  SNode             *pOnCondition;
} SJoinOperatorInfo;

626 627 628
#define OPTR_IS_OPENED(_optr)  (((_optr)->status & OP_OPENED) == OP_OPENED)
#define OPTR_SET_OPENED(_optr) ((_optr)->status |= OP_OPENED)

629 630 631 632
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,
    __optr_decode_fn_t decode, __optr_get_explain_fn_t explain);

H
Haojun Liao 已提交
633
int32_t operatorDummyOpenFn(SOperatorInfo* pOperator);
dengyihao's avatar
dengyihao 已提交
634
void    operatorDummyCloseFn(void* param, int32_t numOfCols);
H
Haojun Liao 已提交
635 636
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num);
int32_t initAggInfo(SOptrBasicInfo* pBasicInfo, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols,
637 638
                    SSDataBlock* pResultBlock, size_t keyBufSize, const char* pkey);
void    initResultSizeInfo(SOperatorInfo* pOperator, int32_t numOfRows);
639
void    doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo, SDiskbasedBuf* pBuf);
640

641
void    finalizeMultiTupleQueryResult(int32_t numOfOutput, SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int32_t* rowCellInfoOffset);
642
void    doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, STimeWindow* pWin, SColumnInfoData* pTimeWindowData, int32_t offset,
dengyihao's avatar
dengyihao 已提交
643 644 645 646 647 648 649
                         int32_t forwardStep, TSKEY* tsCol, int32_t numOfTotal, int32_t numOfOutput, int32_t order);
int32_t setGroupResultOutputBuf(SOptrBasicInfo* binfo, int32_t numOfCols, char* pData, int16_t type, int16_t bytes,
                                int32_t groupId, SDiskbasedBuf* pBuf, SExecTaskInfo* pTaskInfo, SAggSupporter* pAggSup);
void    doDestroyBasicInfo(SOptrBasicInfo* pInfo, int32_t numOfOutput);
int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData,
                                  int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total,
                                  SArray* pColList);
650
void    getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t key, STimeWindow* win);
651
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t *order, int32_t* scanFlag);
652

dengyihao's avatar
dengyihao 已提交
653 654
void    doSetOperatorCompleted(SOperatorInfo* pOperator);
void    doFilter(const SNode* pFilterNode, SSDataBlock* pBlock);
655
SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowCellInfoOffset);
656
void    relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols);
657 658
void    initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow);
void    cleanupAggSup(SAggSupporter* pAggSup);
659 660
void    destroyBasicOperatorInfo(void* param, int32_t numOfOutput);
void    appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
661
SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode);
662
SColumn extractColumnFromColumnNode(SColumnNode* pColNode);
663

664
SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, int32_t capacity, SArray* pColMatchInfo);
665
SSDataBlock* loadNextDataBlock(void* param);
666

667
void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowCellInfoOffset);
668

669 670 671 672 673 674
SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,
                            int32_t type);
SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t* numOfExprs);
SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode);
int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode);

675 676 677
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 已提交
678

679
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, tsdbReaderT pDataReader, SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);
680

681 682
SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SExprInfo* pScalarExprInfo,
                                           int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
683

wmmhello's avatar
wmmhello 已提交
684
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t num, SSDataBlock* pResBlock, SLimit* pLimit, SLimit* pSlimit, SExecTaskInfo* pTaskInfo);
685 686
SOperatorInfo *createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pResBlock, SArray* pSortInfo, SExprInfo* pExprInfo, int32_t numOfCols,
                                      SArray* pIndexMap, SExecTaskInfo* pTaskInfo);
687

H
Haojun Liao 已提交
688
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pSortInfo, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
689

X
Xiaoyu Wang 已提交
690
SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataBlock* pResBlock, const SName* pName,
dengyihao's avatar
dengyihao 已提交
691 692 693
                                              SNode* pCondition, SEpSet epset, SArray* colList,
                                              SExecTaskInfo* pTaskInfo, bool showRewrite, int32_t accountId);
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
694
                                          SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
695
                                          STimeWindowAggSupp *pTwAggSupp, const STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
696
SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
697

698 699
                                                SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
                                                STimeWindowAggSupp *pTwAggSupp, const STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
dengyihao's avatar
dengyihao 已提交
700
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
701 702
                                            SSDataBlock* pResBlock, int64_t gap, int32_t tsSlotId, STimeWindowAggSupp* pTwAggSupp,
                                            SExecTaskInfo* pTaskInfo);
dengyihao's avatar
dengyihao 已提交
703 704 705 706
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                       SSDataBlock* pResultBlock, SArray* pGroupColList, SNode* pCondition,
                                       SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo,
                                       const STableGroupInfo* pTableGroupInfo);
H
Haojun Liao 已提交
707
SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo);
5
54liuyao 已提交
708 709
SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataReader, SSDataBlock* pResBlock,
                                             SArray* pColList, SArray* pTableIdList, SExecTaskInfo* pTaskInfo,
5
54liuyao 已提交
710
                                             SNode* pConditions, SOperatorInfo* pOperatorDumy);
H
Haojun Liao 已提交
711

dengyihao's avatar
dengyihao 已提交
712
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
713
                                      SInterval* pInterval, STimeWindow* pWindow, SSDataBlock* pResBlock, int32_t fillType, SNodeListNode* fillVal,
dengyihao's avatar
dengyihao 已提交
714 715
                                      bool multigroupResult, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
716
                                             SSDataBlock* pResBlock, STimeWindowAggSupp *pTwAggSupp, int32_t tsSlotId, SColumn* pStateKeyCol, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
717

dengyihao's avatar
dengyihao 已提交
718 719 720
SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                           SSDataBlock* pResultBlock, SArray* pGroupColList, SExecTaskInfo* pTaskInfo,
                                           const STableGroupInfo* pTableGroupInfo);
721

dengyihao's avatar
dengyihao 已提交
722 723
SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                           SSDataBlock* pResultBlock, SExecTaskInfo* pTaskInfo);
724

H
Haojun Liao 已提交
725
SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SNode* pOnCondition, SExecTaskInfo* pTaskInfo);
726
SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, SExprInfo* pExpr, int32_t numOfOutput, SSDataBlock* pResBlock, SArray* pColMatchInfo, STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
727

H
Haojun Liao 已提交
728
#if 0
H
Haojun Liao 已提交
729
SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
H
Haojun Liao 已提交
730
#endif
731

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

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

H
Haojun Liao 已提交
737
void copyTsColoum(SSDataBlock* pRes, SqlFunctionCtx* pCtx, int32_t numOfOutput);
738

739
STableQueryInfo* createTableQueryInfo(void* buf, STimeWindow win);
740

741
bool    isTaskKilled(SExecTaskInfo* pTaskInfo);
742 743
int32_t checkForQueryBuf(size_t numOfTables);

dengyihao's avatar
dengyihao 已提交
744
void setTaskKilled(SExecTaskInfo* pTaskInfo);
745 746

void publishOperatorProfEvent(SOperatorInfo* operatorInfo, EQueryProfEventType eventType);
747
void publishQueryAbortEvent(SExecTaskInfo* pTaskInfo, int32_t code);
748

749
void queryCostStatis(SExecTaskInfo* pTaskInfo);
750

dengyihao's avatar
dengyihao 已提交
751
void    doDestroyTask(SExecTaskInfo* pTaskInfo);
752 753
int32_t getMaximumIdleDurationSec();

754
void    setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
dengyihao's avatar
dengyihao 已提交
755 756 757 758 759 760 761 762 763
int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
                               EOPTR_EXEC_MODEL model);
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SExplainExecInfo** pRes, int32_t* capacity,
                                   int32_t* resNum);

bool aggDecodeResultRow(SOperatorInfo* pOperator, SAggSupporter* pSup, SOptrBasicInfo* pInfo, char* result,
                        int32_t length);
void aggEncodeResultRow(SOperatorInfo* pOperator, SAggSupporter* pSup, SOptrBasicInfo* pInfo, char** result,
                        int32_t* length);
5
54liuyao 已提交
764 765 766 767 768 769 770 771 772
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts,
                                       SInterval* pInterval, int32_t precision, STimeWindow* win);
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos,
                                        TSKEY ekey, __block_search_fn_t searchFn, STableQueryInfo* item,
                                        int32_t order);
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);

void doClearWindow(SIntervalAggOperatorInfo* pInfo, char* pData, int16_t bytes,
    uint64_t groupId, int32_t numOfOutput);
773

774 775 776 777
#ifdef __cplusplus
}
#endif

778
#endif  // TDENGINE_EXECUTORIMPL_H