qExecutor.h 22.3 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
Haojun Liao 已提交
15 16
#ifndef TDENGINE_QEXECUTOR_H
#define TDENGINE_QEXECUTOR_H
17 18 19 20

#include "os.h"

#include "hash.h"
H
Haojun Liao 已提交
21
#include "qAggMain.h"
H
Haojun Liao 已提交
22 23
#include "qFill.h"
#include "qResultbuf.h"
24
#include "qSqlparser.h"
H
Haojun Liao 已提交
25
#include "qTableMeta.h"
H
Haojun Liao 已提交
26
#include "qTsbuf.h"
27
#include "query.h"
28
#include "taosdef.h"
H
Haojun Liao 已提交
29
#include "tarray.h"
B
Bomin Zhang 已提交
30
#include "tlockfree.h"
H
Haojun Liao 已提交
31
#include "tsdb.h"
32 33

struct SColumnFilterElem;
34
typedef bool (*__filter_func_t)(struct SColumnFilterElem* pFilter, const char* val1, const char* val2, int16_t type);
H
hjxilinx 已提交
35
typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int32_t order);
36

H
Haojun Liao 已提交
37 38 39 40 41 42
#define IS_QUERY_KILLED(_q) ((_q)->code == TSDB_CODE_TSC_QUERY_CANCELLED)
#define Q_STATUS_EQUAL(p, s)  (((p) & (s)) != 0u)
#define QUERY_IS_ASC_QUERY(q) (GET_FORWARD_DIRECTION_FACTOR((q)->order.order) == QUERY_ASC_FORWARD_STEP)

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

H
Haojun Liao 已提交
43 44
#define GET_NUM_OF_RESULTS(_r) (((_r)->outputBuf) == NULL? 0:((_r)->outputBuf)->info.rows)

H
Haojun Liao 已提交
45 46 47 48 49 50 51 52
enum {
  // when query starts to execute, this status will set
      QUERY_NOT_COMPLETED = 0x1u,

  /* query is over
   * 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
   */
H
Haojun Liao 已提交
53
      QUERY_COMPLETED = 0x2u,
H
Haojun Liao 已提交
54 55 56 57

  /* when the result is not completed return to client, this status will be
   * usually used in case of interval query with interpolation option
   */
H
Haojun Liao 已提交
58
      QUERY_OVER = 0x4u,
H
Haojun Liao 已提交
59 60
};

H
Haojun Liao 已提交
61
typedef struct SResultRowPool {
62 63 64 65 66 67 68 69 70 71
  int32_t elemSize;
  int32_t blockSize;
  int32_t numOfElemPerBlock;

  struct {
    int32_t blockIndex;
    int32_t pos;
  } position;

  SArray* pData;    // SArray<void*>
H
Haojun Liao 已提交
72
} SResultRowPool;
73

H
Haojun Liao 已提交
74
typedef struct SResultRow {
75
  int32_t       pageId;      // pageId & rowId is the position of current result in disk-based output buffer
H
Haojun Liao 已提交
76
  int32_t       offset:29;    // row index in buffer page
77 78 79 80
  bool          startInterp; // the time window start timestamp has done the interpolation already.
  bool          endInterp;   // the time window end timestamp has done the interpolation already.
  bool          closed;      // this result status: closed or opened
  uint32_t      numOfRows;   // number of rows of current time window
H
Haojun Liao 已提交
81
  SResultRowCellInfo*  pCellInfo;  // For each result column, there is a resultInfo
H
Haojun Liao 已提交
82
  STimeWindow win;
D
dapan1121 已提交
83
  char* key;                 // start key of current result row
H
Haojun Liao 已提交
84
} SResultRow;
85

86
typedef struct SGroupResInfo {
H
Haojun Liao 已提交
87 88
  int32_t totalGroup;
  int32_t currentGroup;
89 90 91 92
  int32_t index;
  SArray* pRows;      // SArray<SResultRow*>
} SGroupResInfo;

H
Haojun Liao 已提交
93 94 95 96
/**
 * If the number of generated results is greater than this value,
 * query query will be halt and return results to client immediate.
 */
H
Haojun Liao 已提交
97
typedef struct SRspResultInfo {
H
Haojun Liao 已提交
98
  int64_t total;      // total generated result size in rows
H
Haojun Liao 已提交
99
  int32_t capacity;   // capacity of current result output buffer
H
Haojun Liao 已提交
100
  int32_t threshold;  // result size threshold in rows.
H
Haojun Liao 已提交
101
} SRspResultInfo;
102

H
Haojun Liao 已提交
103
typedef struct SResultRowInfo {
H
Haojun Liao 已提交
104 105 106 107 108 109
  SResultRow** pResult;    // result list
  int16_t      type:8;     // data type for hash key
  int32_t      size:24;    // number of result set
  int32_t      capacity;   // max capacity
  int32_t      curIndex;   // current start active index
  int64_t      prevSKey;   // previous (not completed) sliding window start key
H
Haojun Liao 已提交
110
} SResultRowInfo;
111 112 113 114 115

typedef struct SColumnFilterElem {
  int16_t           bytes;  // column length
  __filter_func_t   fp;
  SColumnFilterInfo filterInfo;
Y
yihaoDeng 已提交
116
  void              *q;
117 118 119
} SColumnFilterElem;

typedef struct SSingleColumnFilterInfo {
H
Haojun Liao 已提交
120
  void*              pData;
121
  int32_t            numOfFilters;
H
Haojun Liao 已提交
122
  SColumnInfo        info;
123 124 125
  SColumnFilterElem* pFilters;
} SSingleColumnFilterInfo;

H
Haojun Liao 已提交
126
typedef struct STableQueryInfo {
H
hjxilinx 已提交
127
  TSKEY       lastKey;
H
Haojun Liao 已提交
128
  int32_t     groupIndex;     // group id in table list
129
  tVariant    tag;
H
hjxilinx 已提交
130
  STimeWindow win;
131
  STSCursor   cur;
H
Haojun Liao 已提交
132
  void*       pTable;         // for retrieve the page id list
H
Haojun Liao 已提交
133
  SResultRowInfo resInfo;
134 135
} STableQueryInfo;

H
Haojun Liao 已提交
136 137 138 139 140 141 142 143 144
typedef struct SQueryCostInfo {
  uint64_t loadStatisTime;
  uint64_t loadFileBlockTime;
  uint64_t loadDataInCacheTime;
  uint64_t loadStatisSize;
  uint64_t loadFileBlockSize;
  uint64_t loadDataInCacheSize;
  
  uint64_t loadDataTime;
145 146 147 148
  uint64_t totalRows;
  uint64_t totalCheckedRows;
  uint32_t totalBlocks;
  uint32_t loadBlocks;
H
Haojun Liao 已提交
149 150
  uint32_t loadBlockStatis;
  uint32_t discardBlocks;
151
  uint64_t elapsedTime;
H
Haojun Liao 已提交
152
  uint64_t firstStageMergeTime;
H
Haojun Liao 已提交
153 154
  uint64_t winInfoSize;
  uint64_t tableInfoSize;
H
Haojun Liao 已提交
155
  uint64_t hashSize;
156
  uint64_t numOfTimeWindows;
H
Haojun Liao 已提交
157
} SQueryCostInfo;
H
hjxilinx 已提交
158

H
Haojun Liao 已提交
159 160 161 162 163
typedef struct {
  int64_t vgroupLimit;
  int64_t ts;
} SOrderedPrjQueryInfo;

164 165 166 167 168
typedef struct {
  char*   tags;
  SArray* pResult;  // SArray<SStddevInterResult>
} SInterResult;

H
Haojun Liao 已提交
169 170 171 172 173 174
typedef struct SSDataBlock {
  SDataStatis *pBlockStatis;
  SArray      *pDataBlock;
  SDataBlockInfo info;
} SSDataBlock;

H
Haojun Liao 已提交
175
// The basic query information extracted from the SQueryInfo tree to support the
H
Haojun Liao 已提交
176
// execution of query in a data node.
H
Haojun Liao 已提交
177
typedef struct SQueryAttr {
H
Haojun Liao 已提交
178
  SLimitVal        limit;
H
Haojun Liao 已提交
179
  SLimitVal        slimit;
H
Haojun Liao 已提交
180

H
Haojun Liao 已提交
181 182 183 184 185 186 187
  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             hasTagResults;    // if there are tag values in final result or not
  bool             timeWindowInterpo;// if the time window start/end required interpolation
  bool             queryBlockDist;    // if query data block distribution
  bool             stabledev;        // super table stddev query
H
Haojun Liao 已提交
188
  bool             tsCompQuery;      // is tscomp query
189
  bool             diffQuery;        // is diff query
H
Haojun Liao 已提交
190 191 192
  bool             simpleAgg;
  bool             pointInterpQuery; // point interpolation query
  bool             needReverseScan;  // need reverse scan
193
  bool             distinctTag;      // distinct tag query
194
  bool             stateWindow;       // window State on sub/normal table
H
Haojun Liao 已提交
195
  int32_t          interBufSize;     // intermediate buffer sizse
H
Haojun Liao 已提交
196

D
dapan1121 已提交
197 198
  int32_t          havingNum;        // having expr number

H
Haojun Liao 已提交
199
  SOrderVal        order;
H
Haojun Liao 已提交
200 201
  int16_t          numOfCols;
  int16_t          numOfTags;
H
Haojun Liao 已提交
202

H
Haojun Liao 已提交
203
  STimeWindow      window;
204
  SInterval        interval;
205
  SSessionWindow   sw;
206
  int16_t          precision;
H
Haojun Liao 已提交
207 208
  int16_t          numOfOutput;
  int16_t          fillType;
209 210 211

  int32_t          srcRowSize;       // todo extract struct
  int32_t          resultRowSize;
H
Haojun Liao 已提交
212
  int32_t          intermediateResultRowSize; // intermediate result row size, in case of top-k query.
H
Haojun Liao 已提交
213
  int32_t          maxTableColumnWidth;
214
  int32_t          tagLen;           // tag value length of current query
H
Haojun Liao 已提交
215
  SGroupbyExpr    *pGroupbyExpr;
H
Haojun Liao 已提交
216

H
Haojun Liao 已提交
217
  SExprInfo*       pExpr1;
H
Haojun Liao 已提交
218 219
  SExprInfo*       pExpr2;
  int32_t          numOfExpr2;
H
Haojun Liao 已提交
220 221 222
  SExprInfo*       pExpr3;
  int32_t          numOfExpr3;

H
Haojun Liao 已提交
223
  SColumnInfo*     tableCols;
H
Haojun Liao 已提交
224 225 226
  SColumnInfo*     tagColList;
  int32_t          numOfFilterCols;
  int64_t*         fillVal;
227
  SOrderedPrjQueryInfo prjInfo;        // limit value for each vgroup, only available in global order projection query.
228
  SSingleColumnFilterInfo* pFilterInfo;
H
Haojun Liao 已提交
229 230 231

  void*            tsdb;
  SMemRef          memRef;
H
Haojun Liao 已提交
232
  STableGroupInfo  tableGroupInfo;       // table <tid, last_key> list  SArray<STableKeyInfo>
H
Haojun Liao 已提交
233
  int32_t          vgId;
H
Haojun Liao 已提交
234
} SQueryAttr;
235

H
Haojun Liao 已提交
236
typedef SSDataBlock* (*__operator_fn_t)(void* param, bool* newgroup);
H
Haojun Liao 已提交
237
typedef void (*__optr_cleanup_fn_t)(void* param, int32_t num);
H
Haojun Liao 已提交
238

H
Haojun Liao 已提交
239
struct SOperatorInfo;
H
Haojun Liao 已提交
240

241
typedef struct SQueryRuntimeEnv {
H
Haojun Liao 已提交
242
  jmp_buf               env;
H
Haojun Liao 已提交
243
  SQueryAttr*           pQueryAttr;
H
Haojun Liao 已提交
244
  uint32_t              status;           // query status
H
Haojun Liao 已提交
245
  void*                 qinfo;
H
Haojun Liao 已提交
246
  uint8_t               scanFlag;         // denotes reversed scan of data or not
H
Haojun Liao 已提交
247 248 249
  void*                 pQueryHandle;

  int32_t               prevGroupId;      // previous executed group id
250
  bool                  enableGroupData;
H
Haojun Liao 已提交
251 252 253 254 255 256 257 258 259 260 261 262 263 264 265
  SDiskbasedResultBuf*  pResultBuf;       // query result buffer based on blocked-wised disk file
  SHashObj*             pResultRowHashTable; // quick locate the window object for each result
  char*                 keyBuf;           // window key buffer
  SResultRowPool*       pool;             // window result object pool
  char**                prevRow;

  SArray*               prevResult;       // intermediate result, SArray<SInterResult>
  STSBuf*               pTsBuf;           // timestamp filter list
  STSCursor             cur;

  char*                 tagVal;           // tag value of current data block
  SArithmeticSupport   *sasArray;

  SSDataBlock          *outputBuf;
  STableGroupInfo       tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
H
Haojun Liao 已提交
266 267 268
  struct SOperatorInfo *proot;
  SGroupResInfo         groupResInfo;
  int64_t               currentOffset;   // dynamic offset value
H
Haojun Liao 已提交
269

H
Haojun Liao 已提交
270
  STableQueryInfo      *current;
H
Haojun Liao 已提交
271
  SRspResultInfo        resultInfo;
H
Haojun Liao 已提交
272
  SHashObj             *pTableRetrieveTsMap;
273 274
} SQueryRuntimeEnv;

H
Haojun Liao 已提交
275 276 277 278 279 280
enum {
  OP_IN_EXECUTING   = 1,
  OP_RES_TO_RETURN  = 2,
  OP_EXEC_DONE      = 3,
};

H
Haojun Liao 已提交
281 282 283 284 285
enum OPERATOR_TYPE_E {
  OP_TableScan         = 1,
  OP_DataBlocksOptScan = 2,
  OP_TableSeqScan      = 3,
  OP_TagScan           = 4,
H
Haojun Liao 已提交
286 287 288 289 290
  OP_TableBlockInfoScan= 5,
  OP_Aggregate         = 6,
  OP_Arithmetic        = 7,
  OP_Groupby           = 8,
  OP_Limit             = 9,
H
Haojun Liao 已提交
291 292 293 294 295 296 297
  OP_SLimit            = 10,
  OP_TimeWindow        = 11,
  OP_SessionWindow     = 12,
  OP_Fill              = 13,
  OP_MultiTableAggregate     = 14,
  OP_MultiTableTimeInterval  = 15,
  OP_DummyInput        = 16,   //TODO remove it after fully refactor.
298
  OP_MultiwayMergeSort      = 17,   // multi-way data merge into one input stream.
H
Haojun Liao 已提交
299
  OP_GlobalAggregate   = 18,   // global merge for the multi-way data sources.
300 301
  OP_Filter            = 19,
  OP_Distinct          = 20,
H
Haojun Liao 已提交
302
  OP_Join              = 21,
Y
yihaoDeng 已提交
303
  OP_StateWindow       = 22,
H
Haojun Liao 已提交
304 305
};

H
Haojun Liao 已提交
306
typedef struct SOperatorInfo {
H
Haojun Liao 已提交
307 308 309 310 311 312 313 314
  uint8_t               operatorType;
  bool                  blockingOptr;  // block operator or not
  uint8_t               status;        // denote if current operator is completed
  int32_t               numOfOutput;   // number of columns of the current operator results
  char                 *name;          // name, used to show the query execution plan
  void                 *info;          // extension attribution
  SExprInfo            *pExpr;
  SQueryRuntimeEnv     *pRuntimeEnv;
H
Haojun Liao 已提交
315

H
Haojun Liao 已提交
316 317
  struct SOperatorInfo **upstream;     // upstream pointer list
  int32_t               numOfUpstream; // number of upstream. The value is always ONE expect for join operator
H
Haojun Liao 已提交
318
  __operator_fn_t       exec;
H
Haojun Liao 已提交
319
  __optr_cleanup_fn_t   cleanup;
H
Haojun Liao 已提交
320 321
} SOperatorInfo;

322 323 324 325 326
enum {
  QUERY_RESULT_NOT_READY = 1,
  QUERY_RESULT_READY     = 2,
};

H
Haojun Liao 已提交
327 328 329 330 331 332
typedef struct {
  int32_t      numOfTags;
  int32_t      numOfCols;
  SColumnInfo *colList;
} SQueriedTableInfo;

333
typedef struct SQInfo {
H
Haojun Liao 已提交
334
  void*            signature;
D
fix bug  
dapan1121 已提交
335
  uint64_t         qId;
H
Haojun Liao 已提交
336 337
  int32_t          code;        // error code to returned to client
  int64_t          owner;       // if it is in execution
H
Haojun Liao 已提交
338

339
  SQueryRuntimeEnv runtimeEnv;
H
Haojun Liao 已提交
340
  SQueryAttr       query;
341
  void*            pBuf;        // allocated buffer for STableQueryInfo, sizeof(STableQueryInfo)*numOfTables;
H
Haojun Liao 已提交
342

343
  pthread_mutex_t  lock;        // used to synchronize the rsp/query threads
H
Haojun Liao 已提交
344
  tsem_t           ready;
345 346
  int32_t          dataReady;   // denote if query result is ready or not
  void*            rspContext;  // response context
347
  int64_t          startExecTs; // start to exec timestamp
348
  char*            sql;         // query sql string
H
Haojun Liao 已提交
349
  SQueryCostInfo   summary;
350 351
} SQInfo;

H
Haojun Liao 已提交
352 353 354 355 356 357
typedef struct SQueryParam {
  char            *sql;
  char            *tagCond;
  char            *tbnameCond;
  char            *prevResult;
  SArray          *pTableIdList;
H
Haojun Liao 已提交
358 359
  SSqlExpr       **pExpr;
  SSqlExpr       **pSecExpr;
H
Haojun Liao 已提交
360 361 362 363 364
  SExprInfo       *pExprs;
  SExprInfo       *pSecExprs;

  SColIndex       *pGroupColIndex;
  SColumnInfo     *pTagColumnInfo;
H
Haojun Liao 已提交
365
  SGroupbyExpr *pGroupbyExpr;
H
Haojun Liao 已提交
366 367
  int32_t          tableScanOperator;
  SArray          *pOperator;
H
Haojun Liao 已提交
368 369
} SQueryParam;

H
Haojun Liao 已提交
370
typedef struct STableScanInfo {
H
Haojun Liao 已提交
371 372 373 374 375 376 377 378 379 380 381 382
  void           *pQueryHandle;
  int32_t         numOfBlocks;
  int32_t         numOfSkipped;
  int32_t         numOfBlockStatis;
  int64_t         numOfRows;
                 
  int32_t         order;        // scan order
  int32_t         times;        // repeat counts
  int32_t         current;
  int32_t         reverseTimes; // 0 by default

  SQLFunctionCtx *pCtx;         // next operator query context
H
Haojun Liao 已提交
383
  SResultRowInfo *pResultRowInfo;
H
Haojun Liao 已提交
384 385
  int32_t        *rowCellInfoOffset;
  SExprInfo      *pExpr;
H
Haojun Liao 已提交
386 387
  SSDataBlock     block;
  bool            loadExternalRows; // load external rows (prev & next rows)
H
Haojun Liao 已提交
388
  int32_t         numOfOutput;
H
Haojun Liao 已提交
389
  int64_t         elapsedTime;
H
Haojun Liao 已提交
390

H
Haojun Liao 已提交
391
  int32_t         tableIndex;
392
  int32_t         prevGroupId;     // previous table group id
H
Haojun Liao 已提交
393 394
} STableScanInfo;

395 396 397
typedef struct STagScanInfo {
  SColumnInfo* pCols;
  SSDataBlock* pRes;
H
Haojun Liao 已提交
398 399
  int32_t      totalTables;
  int32_t      currentIndex;
400 401
} STagScanInfo;

H
Haojun Liao 已提交
402
typedef struct SOptrBasicInfo {
H
Haojun Liao 已提交
403
  SResultRowInfo    resultRowInfo;
H
Haojun Liao 已提交
404
  int32_t          *rowCellInfoOffset;  // offset value for each row result cell info
H
Haojun Liao 已提交
405
  SQLFunctionCtx   *pCtx;
H
Haojun Liao 已提交
406
  SSDataBlock      *pRes;
H
Haojun Liao 已提交
407 408
} SOptrBasicInfo;

H
Haojun Liao 已提交
409 410 411 412 413 414
typedef struct SOptrBasicInfo STableIntervalOperatorInfo;

typedef struct SAggOperatorInfo {
  SOptrBasicInfo binfo;
  uint32_t       seed;
} SAggOperatorInfo;
H
Haojun Liao 已提交
415

H
Haojun Liao 已提交
416
typedef struct SArithOperatorInfo {
H
Haojun Liao 已提交
417 418
  SOptrBasicInfo binfo;
  int32_t        bufCapacity;
H
Haojun Liao 已提交
419
  uint32_t       seed;
H
Haojun Liao 已提交
420 421

  SSDataBlock   *existDataBlock;
H
Haojun Liao 已提交
422 423
} SArithOperatorInfo;

H
Haojun Liao 已提交
424
typedef struct SLimitOperatorInfo {
H
Haojun Liao 已提交
425 426
  int64_t   limit;
  int64_t   total;
H
Haojun Liao 已提交
427 428 429 430 431 432 433 434 435 436 437
} SLimitOperatorInfo;

typedef struct SSLimitOperatorInfo {
  int64_t   groupTotal;
  int64_t   currentGroupOffset;

  int64_t   rowsTotal;
  int64_t   currentOffset;
  SLimitVal limit;
  SLimitVal slimit;

H
Haojun Liao 已提交
438 439 440
  char    **prevRow;
  SArray   *orderColumnList;
} SSLimitOperatorInfo;
H
Haojun Liao 已提交
441

H
Haojun Liao 已提交
442
typedef struct SFilterOperatorInfo {
443 444
  SSingleColumnFilterInfo *pFilterInfo;
  int32_t numOfFilterCols;
H
Haojun Liao 已提交
445
} SFilterOperatorInfo;
D
dapan1121 已提交
446

H
Haojun Liao 已提交
447
typedef struct SFillOperatorInfo {
H
Haojun Liao 已提交
448
  SFillInfo   *pFillInfo;
H
Haojun Liao 已提交
449 450
  SSDataBlock *pRes;
  int64_t      totalInputRows;
H
Haojun Liao 已提交
451 452

  SSDataBlock *existNewGroupBlock;
H
Haojun Liao 已提交
453 454
} SFillOperatorInfo;

H
Haojun Liao 已提交
455 456 457
typedef struct SGroupbyOperatorInfo {
  SOptrBasicInfo binfo;
  int32_t        colIndex;
H
Haojun Liao 已提交
458
  char          *prevData;   // previous group by value
H
Haojun Liao 已提交
459
} SGroupbyOperatorInfo;
H
Haojun Liao 已提交
460

461 462 463 464 465 466
typedef struct SSWindowOperatorInfo {
  SOptrBasicInfo binfo;
  STimeWindow    curWindow;  // current time window
  TSKEY          prevTs;     // previous timestamp
  int32_t        numOfRows;  // number of rows
  int32_t        start;      // start row index
467
  bool           reptScan;    // next round scan
468 469
} SSWindowOperatorInfo;

Y
TD-2570  
yihaoDeng 已提交
470 471 472 473 474 475 476
typedef struct SStateWindowOperatorInfo {
  SOptrBasicInfo binfo;
  STimeWindow    curWindow;  // current time window
  int32_t        numOfRows;  // number of rows
  int32_t        colIndex;      // start row index
  int32_t        start;
  char*          prevData;    // previous data 
477
  bool           reptScan;
Y
TD-2570  
yihaoDeng 已提交
478 479
} SStateWindowOperatorInfo ;

480 481 482 483 484 485 486 487
typedef struct SDistinctOperatorInfo {
  SHashObj         *pSet;
  SSDataBlock      *pRes;
  bool              recordNullVal;  //has already record the null value, no need to try again
  int64_t           threshold;
  int64_t           outputCapacity;
} SDistinctOperatorInfo;

H
Haojun Liao 已提交
488
struct SGlobalMerger;
H
Haojun Liao 已提交
489 490

typedef struct SMultiwayMergeInfo {
H
Haojun Liao 已提交
491
  struct SGlobalMerger *pMerge;
H
Haojun Liao 已提交
492
  SOptrBasicInfo       binfo;
H
Haojun Liao 已提交
493
  int32_t              bufCapacity;
H
Haojun Liao 已提交
494 495 496
  int64_t              seed;
  char               **prevRow;
  SArray              *orderColumnList;
497
  int32_t              resultRowFactor;
H
Haojun Liao 已提交
498

H
Haojun Liao 已提交
499 500
  bool                 hasGroupColData;
  char               **currentGroupColData;
H
Haojun Liao 已提交
501 502 503 504 505 506
  SArray              *groupColumnList;
  bool                 hasDataBlockForNewGroup;
  SSDataBlock         *pExistBlock;

  bool                 hasPrev;
  bool                 groupMix;
H
Haojun Liao 已提交
507 508
} SMultiwayMergeInfo;

H
Haojun Liao 已提交
509 510
void appendUpstream(SOperatorInfo* p, SOperatorInfo* pUpstream);

H
Haojun Liao 已提交
511 512 513 514 515 516 517 518 519 520 521 522 523 524
SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, SQueryRuntimeEnv* pRuntimeEnv, int32_t repeatTime, int32_t reverseTime);
SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv* pRuntimeEnv, int32_t repeatTime);
SOperatorInfo* createTableSeqScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv* pRuntimeEnv);

SOperatorInfo* createAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createArithOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream);
SOperatorInfo* createTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createSWindowOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createGroupbyOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createMultiTableAggOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createTagScanOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput);
525
SOperatorInfo* createDistinctOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
H
Haojun Liao 已提交
526
SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv* pRuntimeEnv);
H
Haojun Liao 已提交
527
SOperatorInfo* createMultiwaySortOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput,
H
Haojun Liao 已提交
528
                                              int32_t numOfRows, void* merger, bool groupMix);
Y
TD-2570  
yihaoDeng 已提交
529
SOperatorInfo* createStatewindowOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
H
Haojun Liao 已提交
530 531
SOperatorInfo* createGlobalAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, void* param);
SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, void* merger);
H
Haojun Liao 已提交
532 533
SOperatorInfo* createFilterOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr,
                                        int32_t numOfOutput, SColumnInfo* pCols, int32_t numOfFilter);
H
Haojun Liao 已提交
534

535
SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pUpstream, int32_t numOfUpstream, SSchema* pSchema, int32_t numOfOutput);
H
Haojun Liao 已提交
536

H
Haojun Liao 已提交
537
SSDataBlock* doGlobalAggregate(void* param, bool* newgroup);
H
Haojun Liao 已提交
538
SSDataBlock* doMultiwayMergeSort(void* param, bool* newgroup);
H
Haojun Liao 已提交
539
SSDataBlock* doSLimit(void* param, bool* newgroup);
H
Haojun Liao 已提交
540

541 542 543 544 545
int32_t doCreateFilterInfo(SColumnInfo* pCols, int32_t numOfCols, int32_t numOfFilterCols, SSingleColumnFilterInfo** pFilterInfo, uint64_t qId);
void doSetFilterColumnInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols, SSDataBlock* pBlock);
bool doFilterDataBlock(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols, int32_t numOfRows, int8_t* p);
void doCompactSDataBlock(SSDataBlock* pBlock, int32_t numOfRows, int8_t* p);

H
Haojun Liao 已提交
546
SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numOfRows);
547 548
void* destroyOutputBuf(SSDataBlock* pBlock);

H
Haojun Liao 已提交
549 550 551
void setInputDataBlock(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order);
int32_t getNumOfResult(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx* pCtx, int32_t numOfOutput);
void finalizeQueryResult(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SResultRowInfo* pResultRowInfo, int32_t* rowCellInfoOffset);
H
Haojun Liao 已提交
552
void updateOutputBuf(SOptrBasicInfo* pBInfo, int32_t *bufCapacity, int32_t numOfInputRows);
H
Haojun Liao 已提交
553

H
Haojun Liao 已提交
554 555
void freeParam(SQueryParam *param);
int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param);
H
Haojun Liao 已提交
556 557 558
int32_t createQueryFunc(SQueriedTableInfo* pTableInfo, int32_t numOfOutput, SExprInfo** pExprInfo,
                        SSqlExpr** pExprMsg, SColumnInfo* pTagCols, int32_t queryType, void* pMsg);

H
Haojun Liao 已提交
559
int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t numOfOutput, SExprInfo **pExprInfo,
H
Haojun Liao 已提交
560
                                           SSqlExpr **pExpr, SExprInfo *prevExpr);
H
Haojun Liao 已提交
561

H
Haojun Liao 已提交
562 563
SGroupbyExpr *createGroupbyExprFromMsg(SQueryTableMsg *pQueryMsg, SColIndex *pColIndex, int32_t *code);
SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,
H
Haojun Liao 已提交
564 565
                        SExprInfo *pSecExprs, STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, int32_t vgId, char* sql, uint64_t *qId);

H
Haojun Liao 已提交
566
int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, void* sourceOptr, SQInfo* pQInfo, SQueryParam* param, char* start,
H
Haojun Liao 已提交
567
                  int32_t prevResultLen, void* merger);
H
Haojun Liao 已提交
568

H
Haojun Liao 已提交
569
int32_t createFilterInfo(SQueryAttr* pQueryAttr, uint64_t qId);
H
Haojun Liao 已提交
570 571
void freeColumnFilterInfo(SColumnFilterInfo* pFilter, int32_t numOfFilters);

H
Haojun Liao 已提交
572
STableQueryInfo *createTableQueryInfo(SQueryAttr* pQueryAttr, void* pTable, bool groupbyColumn, STimeWindow win, void* buf);
H
Haojun Liao 已提交
573 574
STableQueryInfo* createTmpTableQueryInfo(STimeWindow win);

575
int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQueryMsg);
H
Haojun Liao 已提交
576

H
Haojun Liao 已提交
577 578 579
bool isQueryKilled(SQInfo *pQInfo);
int32_t checkForQueryBuf(size_t numOfTables);
bool doBuildResCheck(SQInfo* pQInfo);
H
Haojun Liao 已提交
580
void setQueryStatus(SQueryRuntimeEnv *pRuntimeEnv, int8_t status);
H
Haojun Liao 已提交
581

H
Haojun Liao 已提交
582
bool onlyQueryTags(SQueryAttr* pQueryAttr);
H
Haojun Liao 已提交
583 584 585 586 587 588 589 590
bool isValidQInfo(void *param);

int32_t doDumpQueryResult(SQInfo *pQInfo, char *data);

size_t getResultSize(SQInfo *pQInfo, int64_t *numOfRows);
void setQueryKilled(SQInfo *pQInfo);
void queryCostStatis(SQInfo *pQInfo);
void freeQInfo(SQInfo *pQInfo);
591
void freeQueryAttr(SQueryAttr *pQuery);
H
Haojun Liao 已提交
592 593 594

int32_t getMaximumIdleDurationSec();

H
Haojun Liao 已提交
595
#endif  // TDENGINE_QEXECUTOR_H