executorimpl.h 30.2 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"
H
Haojun Liao 已提交
23
#include "tsort.h"
S
common  
Shengliang Guan 已提交
24
#include "tcommon.h"
25
#include "tlosertree.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"
H
Haojun Liao 已提交
36
#include "thash.h"
37
#include "tlockfree.h"
H
Haojun Liao 已提交
38
#include "tpagedbuf.h"
D
dapan1121 已提交
39
#include "tmsg.h"
H
Haojun Liao 已提交
40

41 42 43 44 45
struct SColumnFilterElem;

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

#define IS_QUERY_KILLED(_q) ((_q)->code == TSDB_CODE_TSC_QUERY_CANCELLED)
46
#define Q_STATUS_EQUAL(p, s) (((p) & (s)) != 0u)
47 48
#define QUERY_IS_ASC_QUERY(q) (GET_FORWARD_DIRECTION_FACTOR((q)->order.order) == QUERY_ASC_FORWARD_STEP)

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

51
#define GET_NUM_OF_RESULTS(_r) (((_r)->outputBuf) == NULL ? 0 : ((_r)->outputBuf)->info.rows)
52

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

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

59
  /* Task is over
60 61 62
   * 1. this status is used in one row result query process, e.g., count/sum/first/last/ avg...etc.
   * 2. when all data within queried time window, it is also denoted as query_completed
   */
63
  TASK_COMPLETED = 0x2u,
64 65 66 67

  /* when the result is not completed return to client, this status will be
   * usually used in case of interval query with interpolation option
   */
68
  TASK_OVER = 0x4u,
69 70 71
};

typedef struct SResultRowCell {
72
  uint64_t    groupId;
H
Haojun Liao 已提交
73
  SResultRowPosition pos;
74 75 76 77 78 79
} SResultRowCell;

/**
 * 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 已提交
80 81 82 83 84 85
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.
} SResultInfo;
86 87 88 89 90

typedef struct SColumnFilterElem {
  int16_t           bytes;  // column length
  __filter_func_t   fp;
  SColumnFilterInfo filterInfo;
91
  void*             q;
92 93 94 95
} SColumnFilterElem;

typedef struct SSingleColumnFilterInfo {
  void*              pData;
96
  void*              pData2;  // used for nchar column
97 98 99 100 101 102
  int32_t            numOfFilters;
  SColumnInfo        info;
  SColumnFilterElem* pFilters;
} SSingleColumnFilterInfo;

typedef struct STableQueryInfo {
H
Haojun Liao 已提交
103 104
  TSKEY          lastKey;     // last check ts
  uint64_t       uid;         // table uid
105
  int32_t        groupIndex;  // group id in table list
106
//  SVariant       tag;
H
Haojun Liao 已提交
107
  SResultRowInfo resInfo;     // result info
108 109 110 111 112 113 114 115 116 117
} STableQueryInfo;

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

typedef struct {
  EQueryProfEventType eventType;
118
  int64_t             eventTime;
119 120

  union {
121 122
    uint8_t operatorType;  // for operator event
    int32_t abortCode;     // for query abort event
123 124 125 126 127 128 129 130 131
  };
} SQueryProfEvent;

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

H
Haojun Liao 已提交
132 133 134 135 136
typedef struct SLimit {
  int64_t limit;
  int64_t offset;
} SLimit;

H
Haojun Liao 已提交
137
typedef struct STaskCostInfo {
138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164
  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;
  uint64_t totalRows;
  uint64_t totalCheckedRows;
  uint32_t totalBlocks;
  uint32_t loadBlocks;
  uint32_t loadBlockStatis;
  uint32_t discardBlocks;
  uint64_t elapsedTime;
  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 已提交
165
} STaskCostInfo;
166

H
Haojun Liao 已提交
167 168
typedef struct SOperatorCostInfo {
  uint64_t openCost;
D
dapan1121 已提交
169
  uint64_t totalCost;
H
Haojun Liao 已提交
170 171
} SOperatorCostInfo;

H
Haojun Liao 已提交
172 173 174 175 176
typedef struct SOrder {
  uint32_t order;
  SColumn  col;
} SOrder;

177 178
// The basic query information extracted from the SQueryInfo tree to support the
// execution of query in a data node.
H
Haojun Liao 已提交
179
typedef struct STaskAttr {
180 181
  SLimit limit;
  SLimit slimit;
182 183

  // todo comment it
184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225
  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
  bool    tsCompQuery;        // is tscomp query
  bool    diffQuery;          // is diff query
  bool    simpleAgg;
  bool    pointInterpQuery;      // point interpolation query
  bool    needReverseScan;       // need reverse scan
  bool    distinct;              // distinct  query or not
  bool    stateWindow;           // window State on sub/normal table
  bool    createFilterOperator;  // if filter operator is needed
  bool    multigroupResult;      // multigroup result can exist in one SSDataBlock
  int32_t interBufSize;          // intermediate buffer sizse

  int32_t havingNum;  // having expr number

  SOrder  order;
  int16_t numOfCols;
  int16_t numOfTags;

  STimeWindow    window;
  SInterval      interval;
  int16_t        precision;
  int16_t        numOfOutput;
  int16_t        fillType;

  int32_t       srcRowSize;  // todo extract struct
  int32_t       resultRowSize;
  int32_t       intermediateResultRowSize;  // intermediate result row size, in case of top-k query.
  int32_t       maxTableColumnWidth;
  int32_t       tagLen;  // tag value length of current query

  SExprInfo* pExpr1;

  SColumnInfo*         tableCols;
  SColumnInfo*         tagColList;
  int32_t              numOfFilterCols;
  int64_t*             fillVal;
226 227

  SSingleColumnFilterInfo* pFilterInfo;
228 229 230 231
  void*           tsdb;
  STableGroupInfo tableGroupInfo;  // table <tid, last_key> list  SArray<STableKeyInfo>
  int32_t         vgId;
  SArray*         pUdfInfo;  // no need to free
H
Haojun Liao 已提交
232
} STaskAttr;
233 234 235

struct SOperatorInfo;

wmmhello's avatar
wmmhello 已提交
236 237
typedef void (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, char **result, int32_t *length);
typedef bool (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, char *result, int32_t length);
238

239 240
typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* pOptr);
typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* pOptr, bool* newgroup);
H
Haojun Liao 已提交
241
typedef void (*__optr_close_fn_t)(void* param, int32_t num);
D
dapan1121 已提交
242
typedef int32_t (*__optr_get_explain_fn_t)(struct SOperatorInfo* pOptr, void **pOptrExplain);
H
Haojun Liao 已提交
243

H
Haojun Liao 已提交
244
typedef struct STaskIdInfo {
245 246 247 248
  uint64_t queryId;  // this is also a request id
  uint64_t subplanId;
  uint64_t templateId;
  char*    str;
H
Haojun Liao 已提交
249 250
} STaskIdInfo;

251
typedef struct SExecTaskInfo {
H
Haojun Liao 已提交
252
  STaskIdInfo     id;
253
  char*           content;
H
Haojun Liao 已提交
254 255 256
  uint32_t        status;
  STimeWindow     window;
  STaskCostInfo   cost;
257
  int64_t         owner;  // if it is in execution
258
  int32_t         code;
259
  uint64_t        totalRows;            // total number of rows
H
Haojun Liao 已提交
260
  STableGroupInfo tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
261
  char*           sql;                  // query sql string
262 263
  jmp_buf         env;                  // jump to this position when error happens.
  EOPTR_EXEC_MODEL execModel;            // operator execution model [batch model|stream model]
264
  struct SOperatorInfo* pRoot;
265
} SExecTaskInfo;
H
Haojun Liao 已提交
266

H
Haojun Liao 已提交
267
typedef struct STaskRuntimeEnv {
H
Haojun Liao 已提交
268 269 270 271 272 273 274 275 276 277 278 279 280 281
  jmp_buf         env;
  STaskAttr*      pQueryAttr;
  uint32_t        status;  // query status
  void*           qinfo;
  uint8_t         scanFlag;  // denotes reversed scan of data or not
  void*           pTsdbReadHandle;

  int32_t         prevGroupId;  // previous executed group id
  bool            enableGroupData;
  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 已提交
282
  // The window result objects pool, all the resultRow Objects are allocated and managed by this object.
H
Haojun Liao 已提交
283 284 285 286
  char**          prevRow;
  SArray*         prevResult;  // intermediate result, SArray<SInterResult>
  STSBuf*         pTsBuf;      // timestamp filter list
  STSCursor       cur;
287

H
Haojun Liao 已提交
288
  char*           tagVal;  // tag value of current data block
289 290 291 292 293
  struct SScalarFunctionSupport* scalarSup;

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

297
  STableQueryInfo* current;
H
Haojun Liao 已提交
298
  SResultInfo   resultInfo;
299 300
  SHashObj*        pTableRetrieveTsMap;
  struct SUdfInfo* pUdfInfo;
H
Haojun Liao 已提交
301
} STaskRuntimeEnv;
302 303

enum {
H
Haojun Liao 已提交
304 305 306 307
  OP_NOT_OPENED    = 0x0,
  OP_OPENED        = 0x1,
  OP_RES_TO_RETURN = 0x5,
  OP_EXEC_DONE     = 0x9,
308 309 310
};

typedef struct SOperatorInfo {
D
dapan1121 已提交
311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331
  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;
  STaskRuntimeEnv*        pRuntimeEnv;   // todo remove it
  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
  __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.
  __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;
332 333 334 335 336
} SOperatorInfo;

typedef struct {
  int32_t      numOfTags;
  int32_t      numOfCols;
337
  SColumnInfo* colList;
338 339 340
} SQueriedTableInfo;

typedef struct SQInfo {
341 342 343 344
  void*    signature;
  uint64_t qId;
  int32_t  code;   // error code to returned to client
  int64_t  owner;  // if it is in execution
345

H
Haojun Liao 已提交
346 347
  STaskRuntimeEnv runtimeEnv;
  STaskAttr       query;
348 349
  void*           pBuf;  // allocated buffer for STableQueryInfo, sizeof(STableQueryInfo)*numOfTables;

wafwerar's avatar
wafwerar 已提交
350
  TdThreadMutex lock;  // used to synchronize the rsp/query threads
351 352 353 354 355 356
  tsem_t          ready;
  int32_t         dataReady;    // denote if query result is ready or not
  void*           rspContext;   // response context
  int64_t         startExecTs;  // start to exec timestamp
  char*           sql;          // query sql string
  STaskCostInfo   summary;
357 358
} SQInfo;

359 360 361 362 363
typedef enum {
  EX_SOURCE_DATA_NOT_READY = 0x1,
  EX_SOURCE_DATA_READY     = 0x2,
  EX_SOURCE_DATA_EXHAUSTED = 0x3,
} EX_SOURCE_STATUS;
364 365 366

typedef struct SSourceDataInfo {
  struct SExchangeInfo *pEx;
367 368 369
  int32_t               index;
  SRetrieveTableRsp    *pRsp;
  uint64_t              totalRows;
H
Haojun Liao 已提交
370
  int32_t               code;
371
  EX_SOURCE_STATUS      status;
372 373
} SSourceDataInfo;

H
Haojun Liao 已提交
374 375 376 377 378 379
typedef struct SLoadRemoteDataInfo {
  uint64_t           totalSize;     // total load bytes from remote
  uint64_t           totalRows;     // total number of rows
  uint64_t           totalElapsed;  // total elapsed time
} SLoadRemoteDataInfo;

380
typedef struct SExchangeInfo {
381
  SArray*            pSources;
382
  SArray*            pSourceDataInfo;
H
Haojun Liao 已提交
383
  tsem_t             ready;
384 385
  void*              pTransporter;
  SSDataBlock*       pResult;
H
Haojun Liao 已提交
386
  bool               seqLoadData;   // sequential load data or not, false by default
387
  int32_t            current;
H
Haojun Liao 已提交
388
  SLoadRemoteDataInfo loadInfo;
389 390
} SExchangeInfo;

H
Haojun Liao 已提交
391 392 393 394 395 396
typedef struct SColMatchInfo {
  int32_t colId;
  int32_t targetSlotId;
  bool    output;
} SColMatchInfo;

397
typedef struct STableScanInfo {
H
Haojun Liao 已提交
398
  void*           dataReader;
399 400 401 402 403 404 405 406
  int32_t         numOfBlocks;  // extract basic running information.
  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
H
Haojun Liao 已提交
407
  SNode*          pFilterNode;   // filter operator info
408 409 410 411
  SqlFunctionCtx* pCtx;  // next operator query context
  SResultRowInfo* pResultRowInfo;
  int32_t*        rowCellInfoOffset;
  SExprInfo*      pExpr;
412
  SSDataBlock     block;
H
Haojun Liao 已提交
413
  SArray*         pColMatchInfo;
414 415
  int32_t         numOfOutput;
  int64_t         elapsedTime;
H
Haojun Liao 已提交
416
  int32_t         prevGroupId;  // previous table group id
417
  int32_t         scanFlag;  // table scan flag to denote if it is a repeat/reverse/main scan
418 419 420 421 422 423 424 425 426
} STableScanInfo;

typedef struct STagScanInfo {
  SColumnInfo* pCols;
  SSDataBlock* pRes;
  int32_t      totalTables;
  int32_t      curPos;
} STagScanInfo;

H
Haojun Liao 已提交
427
typedef struct SStreamBlockScanInfo {
H
Haojun Liao 已提交
428
  SArray*      pBlockLists;   // multiple SSDatablock.
429
  SSDataBlock* pRes;          // result SSDataBlock
430
  int32_t      blockType;     // current block type
H
Haojun Liao 已提交
431
  int32_t      validBlockIndex;    // Is current data has returned?
432 433 434 435
  SColumnInfo* pCols;         // the output column info
  uint64_t     numOfRows;     // total scanned rows
  uint64_t     numOfExec;     // execution times
  void*        readerHandle;  // stream block reader handle
436
  SArray*      pColMatchInfo; //
H
Haojun Liao 已提交
437 438
} SStreamBlockScanInfo;

439
typedef struct SSysTableScanInfo {
440 441 442 443 444
  union {
    void* pTransporter;
    void* readHandle;
  };

H
Haojun Liao 已提交
445
  SRetrieveMetaTableRsp *pRsp;
H
Haojun Liao 已提交
446 447 448
  SRetrieveTableReq   req;
  SEpSet              epSet;
  tsem_t              ready;
H
Haojun Liao 已提交
449

D
dapan1121 已提交
450 451
  int32_t             accountId;
  bool                showRewrite;
H
Haojun Liao 已提交
452
  SNode*              pCondition; // db_name filter condition, to discard data that are not in current database
453 454 455 456
  void               *pCur;       // cursor for iterate the local table meta store.
  SArray             *scanCols;   // SArray<int16_t> scan column id list

  int32_t             type;       // show type, TODO remove it
H
Haojun Liao 已提交
457
  SName               name;
H
Haojun Liao 已提交
458 459 460
  SSDataBlock*        pRes;
  int32_t             capacity;
  int64_t             numOfBlocks;  // extract basic running information.
H
Haojun Liao 已提交
461
  SLoadRemoteDataInfo loadInfo;
462 463
} SSysTableScanInfo;

464
typedef struct SOptrBasicInfo {
H
Haojun Liao 已提交
465 466 467 468
  SResultRowInfo     resultRowInfo;
  int32_t*           rowCellInfoOffset;  // offset value for each row result cell info
  SqlFunctionCtx*    pCtx;
  SSDataBlock*       pRes;
469
  int32_t            capacity;  // TODO remove it
470 471
} SOptrBasicInfo;

472
//TODO move the resultrowsiz together with SOptrBasicInfo:rowCellInfoOffset
473
typedef struct SAggSupporter {
H
Haojun Liao 已提交
474 475 476 477
  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 已提交
478
  SDiskbasedBuf     *pResultBuf;           // query result buffer based on blocked-wised disk file
H
Haojun Liao 已提交
479
  int32_t            resultRowSize;        // the result buffer size for each result row, with the meta data size for each row
480 481
} SAggSupporter;

H
Haojun Liao 已提交
482
typedef struct STableIntervalOperatorInfo {
483 484 485
  SOptrBasicInfo     binfo;                // basic info
  SGroupResInfo      groupResInfo;         // multiple results build supporter
  SInterval          interval;             // interval info
486
  int32_t            primaryTsIndex;       // primary time stamp slot id from result of downstream operator.
487 488 489 490 491 492
  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
493
  EOPTR_EXEC_MODEL   execModel;            // operator execution model [batch model|stream model]
494
  SArray            *pUpdatedWindow;       // updated time window due to the input data block from the downstream operator.
495
  SColumnInfoData    timeWindowData;       // query time window info for scalar function execution.
H
Haojun Liao 已提交
496
} STableIntervalOperatorInfo;
497 498

typedef struct SAggOperatorInfo {
H
Haojun Liao 已提交
499 500 501 502 503 504 505
  SOptrBasicInfo     binfo;
  SDiskbasedBuf     *pResultBuf;           // query result buffer based on blocked-wised disk file
  SAggSupporter      aggSup;
  STableQueryInfo   *current;
  uint32_t           groupId;
  SGroupResInfo      groupResInfo;
  STableQueryInfo   *pTableQueryInfo;
506 507 508 509
} SAggOperatorInfo;

typedef struct SProjectOperatorInfo {
  SOptrBasicInfo binfo;
H
Haojun Liao 已提交
510
  SAggSupporter  aggSup;
H
Haojun Liao 已提交
511
  SSDataBlock   *existDataBlock;
H
Haojun Liao 已提交
512
  SArray        *pPseudoColInfo;
H
Haojun Liao 已提交
513
  SLimit         limit;
H
Haojun Liao 已提交
514 515 516 517 518 519
  SLimit         slimit;

  uint64_t       groupId;
  int64_t        curSOffset;
  int64_t        curGroupOutput;

H
Haojun Liao 已提交
520 521
  int64_t        curOffset;
  int64_t        curOutput;
522 523 524
} SProjectOperatorInfo;

typedef struct SSLimitOperatorInfo {
H
Haojun Liao 已提交
525 526 527 528 529 530 531 532 533 534 535 536 537 538 539
  int64_t            groupTotal;
  int64_t            currentGroupOffset;
  int64_t            rowsTotal;
  int64_t            currentOffset;
  SLimit             limit;
  SLimit             slimit;
  char**             prevRow;
  SArray*            orderColumnList;
  bool               hasPrev;
  bool               ignoreCurrentGroup;
  bool               multigroupResult;
  SSDataBlock*       pRes;  // result buffer
  SSDataBlock*       pPrevBlock;
  int64_t            capacity;
  int64_t            threshold;
540 541 542
} SSLimitOperatorInfo;

typedef struct SFillOperatorInfo {
543 544 545 546 547 548
  struct SFillInfo* pFillInfo;
  SSDataBlock*      pRes;
  int64_t           totalInputRows;
  void**            p;
  SSDataBlock*      existNewGroupBlock;
  bool              multigroupResult;
H
Haojun Liao 已提交
549 550
  SInterval         intervalInfo;
  int32_t           capacity;
551 552
} SFillOperatorInfo;

553
typedef struct {
554 555 556 557
  char   *pData;
  bool    isNull;
  int16_t type;
  int32_t bytes;
558
} SGroupKeys, SStateKeys;
559

560 561
typedef struct SGroupbyOperatorInfo {
  SOptrBasicInfo binfo;
H
Haojun Liao 已提交
562
  SArray*        pGroupCols;
563
  SArray*        pGroupColVals; // current group column values, SArray<SGroupKeys>
564
  SNode*         pCondition;
565 566 567
  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
H
Haojun Liao 已提交
568
  SGroupResInfo  groupResInfo;
H
Haojun Liao 已提交
569
  SAggSupporter  aggSup;
570 571
} SGroupbyOperatorInfo;

H
Haojun Liao 已提交
572 573 574 575 576 577 578 579 580 581 582 583 584 585
typedef struct SDataGroupInfo {
  uint64_t groupId;
  int64_t  numOfRows;
  SArray  *pPageList;
} SDataGroupInfo;

// The sort in partition may be needed later.
typedef struct SPartitionOperatorInfo {
  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
H
Haojun Liao 已提交
586

H
Haojun Liao 已提交
587
  SDiskbasedBuf* pBuf;          // query result buffer based on blocked-wised disk file
H
Haojun Liao 已提交
588 589
  int32_t        rowCapacity;   // maximum number of rows for each buffer page
  int32_t*       columnOffset;  // start position for each column data
H
Haojun Liao 已提交
590 591 592 593

  void*          pGroupIter;    // group iterator
  int32_t        pageIndex;     // page index of current group
} SPartitionOperatorInfo;
594

595 596 597 598 599 600 601
typedef struct SWindowRowsSup {
  STimeWindow      win;
  TSKEY            prevTs;
  int32_t          startRowIndex;
  int32_t          numOfRows;
} SWindowRowsSup;

H
Haojun Liao 已提交
602
typedef struct SSessionAggOperatorInfo {
603 604 605
  SOptrBasicInfo   binfo;
  SAggSupporter    aggSup;
  SGroupResInfo    groupResInfo;
606
  SWindowRowsSup   winSup;
607 608 609
  bool             reptScan;         // next round scan
  int64_t          gap;              // session window gap
  SColumnInfoData  timeWindowData;   // query time window info for scalar function execution.
H
Haojun Liao 已提交
610
} SSessionAggOperatorInfo;
611 612

typedef struct SStateWindowOperatorInfo {
613 614 615 616 617 618 619 620 621
  SOptrBasicInfo   binfo;
  SAggSupporter    aggSup;
  SGroupResInfo    groupResInfo;
  SWindowRowsSup   winSup;
  int32_t          colIndex;   // start row index
  bool             hasKey;
  SStateKeys       stateKey;
  SColumnInfoData  timeWindowData;   // query time window info for scalar function execution.
//  bool             reptScan;
622 623
} SStateWindowOperatorInfo;

624
typedef struct SSortedMergeOperatorInfo {
H
Haojun Liao 已提交
625
  SOptrBasicInfo     binfo;
626 627
  bool               hasVarCol;
  
H
Haojun Liao 已提交
628
  SArray*            pSortInfo;
H
Haojun Liao 已提交
629
  int32_t            numOfSources;
630

631 632 633 634
  SSortHandle       *pSortHandle;

  int32_t            bufPageSize;
  uint32_t           sortBufSize;  // max buffer size for in-memory sort
H
Haojun Liao 已提交
635 636

  int32_t            resultRowFactor;
637 638 639 640 641 642 643 644
  bool               hasGroupVal;

  SDiskbasedBuf     *pTupleStore;  // keep the final results
  int32_t            numOfResPerPage;

  char**             groupVal;
  SArray            *groupInfo;
  SAggSupporter      aggSup;
645
} SSortedMergeOperatorInfo;
H
Haojun Liao 已提交
646

H
Haojun Liao 已提交
647
typedef struct SSortOperatorInfo {
H
Haojun Liao 已提交
648 649
  uint32_t           sortBufSize;  // max buffer size for in-memory sort
  SSDataBlock       *pDataBlock;
H
Haojun Liao 已提交
650
  SArray*            pSortInfo;
H
Haojun Liao 已提交
651
  SSortHandle       *pSortHandle;
652
  SArray*            inputSlotMap;  // for index map from table scan output
H
Haojun Liao 已提交
653 654
  int32_t            bufPageSize;
  int32_t            numOfRowsInRes;
H
Haojun Liao 已提交
655

H
Haojun Liao 已提交
656
  // TODO extact struct
H
Haojun Liao 已提交
657 658 659 660 661
  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 已提交
662
} SSortOperatorInfo;
663

H
Haojun Liao 已提交
664 665 666 667 668 669 670 671 672 673 674 675 676 677
int32_t operatorDummyOpenFn(SOperatorInfo* pOperator);
void operatorDummyCloseFn(void* param, int32_t numOfCols);
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num);
int32_t initAggInfo(SOptrBasicInfo* pBasicInfo, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols,
                    int32_t numOfRows, SSDataBlock* pResultBlock, const char* pkey);
void toSDatablock(SGroupResInfo* pGroupResInfo, SDiskbasedBuf* pBuf, SSDataBlock* pBlock, int32_t rowCapacity, int32_t* rowCellOffset);
void finalizeMultiTupleQueryResult(SqlFunctionCtx* pCtx, int32_t numOfOutput, SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int32_t* rowCellInfoOffset);
void doApplyFunctions(SqlFunctionCtx* pCtx, STimeWindow* pWin, SColumnInfoData* pTimeWindowData, int32_t offset, int32_t forwardStep, TSKEY* tsCol, int32_t numOfTotal, int32_t numOfOutput, int32_t order);
int32_t setGroupResultOutputBuf_rv(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);
678 679
void doSetOperatorCompleted(SOperatorInfo* pOperator);
void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock);
H
Haojun Liao 已提交
680 681
SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, int32_t capacity);
SSDataBlock* loadNextDataBlock(void* param);
H
Haojun Liao 已提交
682

H
Haojun Liao 已提交
683
SOperatorInfo* createExchangeOperatorInfo(const SNodeList* pSources, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
684
SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfCols, int32_t repeatTime,
H
Haojun Liao 已提交
685
                                           int32_t reverseTime, SArray* pColMatchInfo, SNode* pCondition, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
686 687 688
SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock,
                                           SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
SOperatorInfo* createMultiTableAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
689

wmmhello's avatar
wmmhello 已提交
690
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t num, SSDataBlock* pResBlock, SLimit* pLimit, SLimit* pSlimit, SExecTaskInfo* pTaskInfo);
691
SOperatorInfo *createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pResBlock, SArray* pSortInfo, SArray* pIndexMap, SExecTaskInfo* pTaskInfo);
692

H
Haojun Liao 已提交
693
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pSortInfo, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
X
Xiaoyu Wang 已提交
694
SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataBlock* pResBlock, const SName* pName,
D
dapan1121 已提交
695
                                              SNode* pCondition, SEpSet epset, SArray* colList, SExecTaskInfo* pTaskInfo, bool showRewrite, int32_t accountId);
696
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlot,
H
Haojun Liao 已提交
697
                                          const STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
698
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, int64_t gap, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
699
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock,
700
                                       SArray* pGroupColList, SNode* pCondition, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
H
Haojun Liao 已提交
701
SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
702
SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList, SArray* pTableIdList, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
703

H
Haojun Liao 已提交
704 705
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SInterval* pInterval, SSDataBlock* pResBlock,
                                      int32_t fillType, char* fillVal, bool multigroupResult, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
706
SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
707

708 709
SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SArray* pGroupColList,
                                           SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
H
Haojun Liao 已提交
710
#if 0
H
Haojun Liao 已提交
711
SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
712 713
SOperatorInfo* createAllTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
                                                 SExprInfo* pExpr, int32_t numOfOutput);
H
Haojun Liao 已提交
714

715 716 717 718
SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
                                                        SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createAllMultiTableTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
                                                           SExprInfo* pExpr, int32_t numOfOutput);
H
Haojun Liao 已提交
719
SOperatorInfo* createTagScanOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput);
H
Haojun Liao 已提交
720

721 722
SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pdownstream, int32_t numOfDownstream, SSchema* pSchema,
                                      int32_t numOfOutput);
H
Haojun Liao 已提交
723
#endif
724

H
Haojun Liao 已提交
725
void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order);
X
Xiaoyu Wang 已提交
726

727
void finalizeQueryResult(SqlFunctionCtx* pCtx, int32_t numOfOutput);
H
Haojun Liao 已提交
728
void copyTsColoum(SSDataBlock* pRes, SqlFunctionCtx* pCtx, int32_t numOfOutput);
729

H
Haojun Liao 已提交
730
STableQueryInfo* createTableQueryInfo(void* buf, bool groupbyColumn, STimeWindow win);
731

732
bool    isTaskKilled(SExecTaskInfo* pTaskInfo);
733 734
int32_t checkForQueryBuf(size_t numOfTables);

735
void   setTaskKilled(SExecTaskInfo* pTaskInfo);
736 737

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

740
void calculateOperatorProfResults(SQInfo* pQInfo);
741
void queryCostStatis(SExecTaskInfo* pTaskInfo);
742

743
void doDestroyTask(SExecTaskInfo* pTaskInfo);
744 745
int32_t getMaximumIdleDurationSec();

746 747
void    doInvokeUdf(struct SUdfInfo* pUdfInfo, SqlFunctionCtx* pCtx, int32_t idx, int32_t type);
void    setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
748
int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId, EOPTR_EXEC_MODEL model);
D
dapan1121 已提交
749
int32_t getOperatorExplainExecInfo(SOperatorInfo *operatorInfo, SExplainExecInfo **pRes, int32_t *capacity, int32_t *resNum);
750

751 752 753 754
#ifdef __cplusplus
}
#endif

755
#endif  // TDENGINE_EXECUTORIMPL_H