executorimpl.h 25.6 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
/*
 * 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

#include "os.h"
#include "common.h"
#include "ttszip.h"
#include "tvariant.h"

H
Haojun Liao 已提交
23
#include "dataSinkMgt.h"
24
#include "executil.h"
H
Haojun Liao 已提交
25
#include "planner.h"
26 27 28
#include "taosdef.h"
#include "tarray.h"
#include "tfilter.h"
H
Haojun Liao 已提交
29
#include "thash.h"
30 31
#include "tlockfree.h"
#include "tpagedfile.h"
H
Haojun Liao 已提交
32
#include "executor.h"
H
Haojun Liao 已提交
33

34 35 36 37 38 39 40 41 42 43 44 45 46 47 48
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)
#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)))

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

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

enum {
49 50
  // when this task starts to execute, this status will set
  TASK_NOT_COMPLETED = 0x1u,
51

52
  /* Task is over
53 54 55
   * 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
   */
56
  TASK_COMPLETED = 0x2u,
57 58 59 60

  /* when the result is not completed return to client, this status will be
   * usually used in case of interval query with interpolation option
   */
61
  TASK_OVER = 0x4u,
62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97
};

typedef struct SResultRowCell {
  uint64_t     groupId;
  SResultRow  *pRow;
} SResultRowCell;

/**
 * If the number of generated results is greater than this value,
 * query query will be halt and return results to client immediate.
 */
typedef struct SRspResultInfo {
  int64_t total;      // total generated result size in rows
  int32_t capacity;   // capacity of current result output buffer
  int32_t threshold;  // result size threshold in rows.
} SRspResultInfo;

typedef struct SColumnFilterElem {
  int16_t           bytes;  // column length
  __filter_func_t   fp;
  SColumnFilterInfo filterInfo;
  void              *q;
} SColumnFilterElem;

typedef struct SSingleColumnFilterInfo {
  void*              pData;
  void*              pData2;  //used for nchar column
  int32_t            numOfFilters;
  SColumnInfo        info;
  SColumnFilterElem* pFilters;
} SSingleColumnFilterInfo;

typedef struct STableQueryInfo {
  TSKEY       lastKey;
  int32_t     groupIndex;     // group id in table list
  SVariant    tag;
H
Haojun Liao 已提交
98
  STimeWindow win;            // todo remove it later
99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125
  STSCursor   cur;
  void*       pTable;         // for retrieve the page id list
  SResultRowInfo resInfo;
} STableQueryInfo;

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

typedef struct {
  EQueryProfEventType eventType;
  int64_t eventTime;

  union {
    uint8_t operatorType; //for operator event
    int32_t abortCode; //for query abort event
  };
} SQueryProfEvent;

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

H
Haojun Liao 已提交
126
typedef struct STaskCostInfo {
127
  int64_t   created;
H
Haojun Liao 已提交
128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154
  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>
} STaskCostInfo;
155 156 157 158 159 160 161 162 163 164 165 166 167

typedef struct {
  int64_t vgroupLimit;
  int64_t ts;
} SOrderedPrjQueryInfo;

typedef struct {
  char*   tags;
  SArray* pResult;  // SArray<SStddevInterResult>
} SInterResult;

// The basic query information extracted from the SQueryInfo tree to support the
// execution of query in a data node.
H
Haojun Liao 已提交
168
typedef struct STaskAttr {
169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 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 226 227 228 229 230
  SLimit           limit;
  SLimit           slimit;

  // todo comment it
  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;
  SSessionWindow   sw;
  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
  SGroupbyExpr    *pGroupbyExpr;

  SExprInfo*       pExpr1;
  SExprInfo*       pExpr2;
  int32_t          numOfExpr2;
  SExprInfo*       pExpr3;
  int32_t          numOfExpr3;

  SColumnInfo*     tableCols;
  SColumnInfo*     tagColList;
  int32_t          numOfFilterCols;
  int64_t*         fillVal;
  SOrderedPrjQueryInfo prjInfo;        // limit value for each vgroup, only available in global order projection query.

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

typedef SSDataBlock* (*__operator_fn_t)(void* param, bool* newgroup);
typedef void (*__optr_cleanup_fn_t)(void* param, int32_t num);

struct SOperatorInfo;

H
Haojun Liao 已提交
238 239 240 241
typedef struct STaskIdInfo {
  uint64_t       queryId;    // this is also a request id
  uint64_t       subplanId;
  uint64_t       templateId;
242
  char          *str;
H
Haojun Liao 已提交
243 244
} STaskIdInfo;

245
typedef struct SExecTaskInfo {
H
Haojun Liao 已提交
246 247 248 249 250 251
  STaskIdInfo     id;
  char           *content;
  uint32_t        status;
  STimeWindow     window;
  STaskCostInfo   cost;
  int64_t         owner;       // if it is in execution
252
  int32_t         code;
H
Haojun Liao 已提交
253
  uint64_t        totalRows;   // total number of rows
H
Haojun Liao 已提交
254 255
  STableGroupInfo tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
  char           *sql;         // query sql string
256 257 258
  jmp_buf         env;         //
  struct SOperatorInfo  *pRoot;
} SExecTaskInfo;
H
Haojun Liao 已提交
259

H
Haojun Liao 已提交
260
typedef struct STaskRuntimeEnv {
261
  jmp_buf               env;
H
Haojun Liao 已提交
262
  STaskAttr*            pQueryAttr;
263 264 265
  uint32_t              status;           // query status
  void*                 qinfo;
  uint8_t               scanFlag;         // denotes reversed scan of data or not
266
  void*                 pTsdbReadHandle;
267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294

  int32_t               prevGroupId;      // previous executed group id
  bool                  enableGroupData;
  SDiskbasedResultBuf*  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
  SResultRowPool*       pool;             // The window result objects pool, all the resultRow Objects are allocated and managed by this object.
  char**                prevRow;

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

  char*                 tagVal;           // tag value of current data block
  struct SScalarFunctionSupport   * scalarSup;

  SSDataBlock          *outputBuf;
  STableGroupInfo       tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
  struct SOperatorInfo *proot;
  SGroupResInfo         groupResInfo;
  int64_t               currentOffset;   // dynamic offset value

  STableQueryInfo      *current;
  SRspResultInfo        resultInfo;
  SHashObj             *pTableRetrieveTsMap;
  struct SUdfInfo      *pUdfInfo;
H
Haojun Liao 已提交
295
} STaskRuntimeEnv;
296 297 298 299 300 301 302 303 304 305 306 307 308 309 310

enum {
  OP_IN_EXECUTING   = 1,
  OP_RES_TO_RETURN  = 2,
  OP_EXEC_DONE      = 3,
};

typedef struct SOperatorInfo {
  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;
311 312
  STaskRuntimeEnv      *pRuntimeEnv;   // todo remove it
  SExecTaskInfo        *pTaskInfo;
313

H
Haojun Liao 已提交
314 315
  struct SOperatorInfo **pDownstream;  // downstram pointer list
  int32_t               numOfDownstream; // number of downstream. The value is always ONE expect for join operator
316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336
  __operator_fn_t       exec;
  __optr_cleanup_fn_t   cleanup;
} SOperatorInfo;

enum {
  QUERY_RESULT_NOT_READY = 1,
  QUERY_RESULT_READY     = 2,
};

typedef struct {
  int32_t      numOfTags;
  int32_t      numOfCols;
  SColumnInfo *colList;
} SQueriedTableInfo;

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

H
Haojun Liao 已提交
337 338
  STaskRuntimeEnv runtimeEnv;
  STaskAttr       query;
339 340 341 342 343 344 345 346
  void*            pBuf;        // allocated buffer for STableQueryInfo, sizeof(STableQueryInfo)*numOfTables;

  pthread_mutex_t  lock;        // used to synchronize the rsp/query threads
  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
H
Haojun Liao 已提交
347
  STaskCostInfo    summary;
348 349
} SQInfo;

H
Haojun Liao 已提交
350
typedef struct STaskParam {
351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369
  char            *sql;
  char            *tagCond;
  char            *colCond;
  char            *tbnameCond;
  char            *prevResult;
  SArray          *pTableIdList;
  SSqlExpr       **pExpr;
  SSqlExpr       **pSecExpr;
  SExprInfo       *pExprs;
  SExprInfo       *pSecExprs;

  SFilterInfo     *pFilters;

  SColIndex       *pGroupColIndex;
  SColumnInfo     *pTagColumnInfo;
  SGroupbyExpr *pGroupbyExpr;
  int32_t          tableScanOperator;
  SArray          *pOperator;
  struct SUdfInfo        *pUdfInfo;
H
Haojun Liao 已提交
370
} STaskParam;
371

372
typedef struct SExchangeInfo {
H
Haojun Liao 已提交
373 374 375 376 377
  SArray            *pSources;
  tsem_t             ready;
  void              *pTransporter;
  SRetrieveTableRsp *pRsp;
  SSDataBlock       *pResult;
378 379
  int32_t            current;
  uint64_t           rowsOfCurrentSource;
H
Haojun Liao 已提交
380 381 382 383

  uint64_t           totalSize;   // total load bytes from remote
  uint64_t           totalRows;   // total number of rows
  uint64_t           totalElapsed;// total elapsed time
384 385
} SExchangeInfo;

386
typedef struct STableScanInfo {
387
  void           *pTsdbReadHandle;
388
  int32_t         numOfBlocks;     // extract basic running information.
389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404
  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
  SResultRowInfo *pResultRowInfo;
  int32_t        *rowCellInfoOffset;
  SExprInfo      *pExpr;
  SSDataBlock     block;
  int32_t         numOfOutput;
  int64_t         elapsedTime;
H
Haojun Liao 已提交
405 406 407
  int32_t         prevGroupId;  // previous table group id

  int32_t         scanFlag;     // table scan flag to denote if it is a repeat/reverse/main scan
408 409 410 411 412 413 414 415 416
} STableScanInfo;

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

H
Haojun Liao 已提交
417
typedef struct SStreamBlockScanInfo {
418 419 420 421 422
  SSDataBlock *pRes;        // result SSDataBlock
  SColumnInfo *pCols;       // the output column info
  uint64_t     numOfRows;   // total scanned rows
  uint64_t     numOfExec;   // execution times
  void        *readerHandle;// stream block reader handle
H
Haojun Liao 已提交
423 424
} SStreamBlockScanInfo;

425 426 427 428 429 430 431 432 433 434
typedef struct SOptrBasicInfo {
  SResultRowInfo    resultRowInfo;
  int32_t          *rowCellInfoOffset;  // offset value for each row result cell info
  SQLFunctionCtx   *pCtx;
  SSDataBlock      *pRes;
} SOptrBasicInfo;

typedef struct SOptrBasicInfo STableIntervalOperatorInfo;

typedef struct SAggOperatorInfo {
H
Haojun Liao 已提交
435 436 437
  SOptrBasicInfo        binfo;
  uint32_t              seed;
  SDiskbasedResultBuf  *pResultBuf;       // query result buffer based on blocked-wised disk file
H
Haojun Liao 已提交
438 439 440 441 442
  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
  SResultRowPool*       pool;             // The window result objects pool, all the resultRow Objects are allocated and managed by this object.
H
Haojun Liao 已提交
443
  STableQueryInfo      *current;
444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563
} SAggOperatorInfo;

typedef struct SProjectOperatorInfo {
  SOptrBasicInfo binfo;
  int32_t        bufCapacity;
  uint32_t       seed;

  SSDataBlock   *existDataBlock;
} SProjectOperatorInfo;

typedef struct SLimitOperatorInfo {
  int64_t   limit;
  int64_t   total;
} SLimitOperatorInfo;

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

typedef struct SFilterOperatorInfo {
  SSingleColumnFilterInfo *pFilterInfo;
  int32_t numOfFilterCols;
} SFilterOperatorInfo;

typedef struct SFillOperatorInfo {
  struct SFillInfo   *pFillInfo;
  SSDataBlock *pRes;
  int64_t      totalInputRows;
  void       **p;
  SSDataBlock *existNewGroupBlock;
  bool         multigroupResult;
} SFillOperatorInfo;

typedef struct SGroupbyOperatorInfo {
  SOptrBasicInfo binfo;
  int32_t        colIndex;
  char          *prevData;   // previous group by value
} SGroupbyOperatorInfo;

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
  bool           reptScan;    // next round scan
} SSWindowOperatorInfo;

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 
  bool           reptScan;
} SStateWindowOperatorInfo;

typedef struct SDistinctDataInfo {
  int32_t index;
  int32_t type;
  int32_t bytes;
} SDistinctDataInfo; 

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;
  int32_t           totalBytes; 
  char*             buf;
  SArray*           pDistinctDataInfo; 
} SDistinctOperatorInfo;

struct SGlobalMerger;

typedef struct SMultiwayMergeInfo {
  struct SGlobalMerger *pMerge;
  SOptrBasicInfo       binfo;
  int32_t              bufCapacity;
  int64_t              seed;
  char               **prevRow;
  SArray              *orderColumnList;
  int32_t              resultRowFactor;

  bool                 hasGroupColData;
  char               **currentGroupColData;
  SArray              *groupColumnList;
  bool                 hasDataBlockForNewGroup;
  SSDataBlock         *pExistBlock;

  SArray              *udfInfo;
  bool                 hasPrev;
  bool                 multiGroupResults;
} SMultiwayMergeInfo;

// todo support the disk-based sort
typedef struct SOrderOperatorInfo {
  int32_t      colIndex;
  int32_t      order;
  SSDataBlock *pDataBlock;
} SOrderOperatorInfo;

H
Haojun Liao 已提交
564
SOperatorInfo* createExchangeOperatorInfo(const SArray* pSources, const SArray* pSchema, SExecTaskInfo* pTaskInfo);
565

H
Haojun Liao 已提交
566
SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime, int32_t reverseTime, SExecTaskInfo* pTaskInfo);
567
SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
568
SOperatorInfo* createTableSeqScanOperator(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
H
Haojun Liao 已提交
569
SOperatorInfo* createSubmitBlockScanOperatorInfo(void *pSubmitBlockReadHandle, int32_t numOfOutput, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
570

H
Haojun Liao 已提交
571
SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SExecTaskInfo* pTaskInfo);
H
Haojun Liao 已提交
572 573 574 575 576 577 578 579 580 581
SOperatorInfo* createProjectOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createLimitOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream);
SOperatorInfo* createTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createAllTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createSWindowOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createFillOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput, bool multigroupResult);
SOperatorInfo* createGroupbyOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createMultiTableAggOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput);
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 已提交
582
SOperatorInfo* createTagScanOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput);
H
Haojun Liao 已提交
583 584
SOperatorInfo* createDistinctOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
H
Haojun Liao 已提交
585
SOperatorInfo* createMultiwaySortOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput,
586
                                              int32_t numOfRows, void* merger);
H
Haojun Liao 已提交
587 588 589 590
SOperatorInfo* createGlobalAggregateOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput, void* param, SArray* pUdfInfo, bool groupResultMixedUp);
SOperatorInfo* createStatewindowOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createSLimitOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput, void* merger, bool multigroupResult);
SOperatorInfo* createFilterOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr,
591 592
                                        int32_t numOfOutput, SColumnInfo* pCols, int32_t numOfFilter);

H
Haojun Liao 已提交
593 594
SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pdownstream, int32_t numOfDownstream, SSchema* pSchema, int32_t numOfOutput);
SOperatorInfo* createOrderOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfOutput, SOrder* pOrderVal);
595

596 597 598
//SSDataBlock* doGlobalAggregate(void* param, bool* newgroup);
//SSDataBlock* doMultiwayMergeSort(void* param, bool* newgroup);
//SSDataBlock* doSLimit(void* param, bool* newgroup);
599

600
//int32_t doCreateFilterInfo(SColumnInfo* pCols, int32_t numOfCols, int32_t numOfFilterCols, SSingleColumnFilterInfo** pFilterInfo, uint64_t qId);
601 602 603 604 605 606 607 608 609 610 611 612 613 614 615
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);

SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numOfRows);

void* destroyOutputBuf(SSDataBlock* pBlock);
void* doDestroyFilterInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols);

void setInputDataBlock(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order);
void finalizeQueryResult(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SResultRowInfo* pResultRowInfo, int32_t* rowCellInfoOffset);
void updateOutputBuf(SOptrBasicInfo* pBInfo, int32_t *bufCapacity, int32_t numOfInputRows);
void clearOutputBuf(SOptrBasicInfo* pBInfo, int32_t *bufCapacity);
void copyTsColoum(SSDataBlock* pRes, SQLFunctionCtx* pCtx, int32_t numOfOutput);

H
Haojun Liao 已提交
616
void freeParam(STaskParam *param);
617 618 619
int32_t createQueryFunc(SQueriedTableInfo* pTableInfo, int32_t numOfOutput, SExprInfo** pExprInfo,
                        SSqlExpr** pExprMsg, SColumnInfo* pTagCols, int32_t queryType, void* pMsg, struct SUdfInfo* pUdfInfo);

S
Shengliang Guan 已提交
620
int32_t createIndirectQueryFuncExprFromMsg(SQueryTableReq *pQueryMsg, int32_t numOfOutput, SExprInfo **pExprInfo,
621 622 623 624
                                           SSqlExpr **pExpr, SExprInfo *prevExpr, struct SUdfInfo *pUdfInfo);

int32_t createQueryFilter(char *data, uint16_t len, SFilterInfo** pFilters);

S
Shengliang Guan 已提交
625
SGroupbyExpr *createGroupbyExprFromMsg(SQueryTableReq *pQueryMsg, SColIndex *pColIndex, int32_t *code);
626

H
Haojun Liao 已提交
627
int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, void* sourceOptr, SQInfo* pQInfo, STaskParam* param, char* start,
628 629
                  int32_t prevResultLen, void* merger);

H
Haojun Liao 已提交
630
int32_t createFilterInfo(STaskAttr* pQueryAttr, uint64_t qId);
631 632
void freeColumnFilterInfo(SColumnFilterInfo* pFilter, int32_t numOfFilters);

H
Haojun Liao 已提交
633
STableQueryInfo *createTableQueryInfo(STaskAttr* pQueryAttr, void* pTable, bool groupbyColumn, STimeWindow win, void* buf);
634 635 636 637
STableQueryInfo* createTmpTableQueryInfo(STimeWindow win);

int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQueryMsg);

638
bool isTaskKilled(SExecTaskInfo *pTaskInfo);
639 640
int32_t checkForQueryBuf(size_t numOfTables);
bool checkNeedToCompressQueryCol(SQInfo *pQInfo);
H
Haojun Liao 已提交
641
void setQueryStatus(STaskRuntimeEnv *pRuntimeEnv, int8_t status);
642

H
Haojun Liao 已提交
643
bool onlyQueryTags(STaskAttr* pQueryAttr);
644
//void destroyUdfInfo(struct SUdfInfo* pUdfInfo);
645 646 647 648

int32_t doDumpQueryResult(SQInfo *pQInfo, char *data, int8_t compressed, int32_t *compLen);

size_t getResultSize(SQInfo *pQInfo, int64_t *numOfRows);
H
Haojun Liao 已提交
649
void setTaskKilled(SExecTaskInfo *pTaskInfo);
650 651

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

654
void calculateOperatorProfResults(SQInfo* pQInfo);
H
Haojun Liao 已提交
655
void queryCostStatis(SExecTaskInfo *pTaskInfo);
656

H
Haojun Liao 已提交
657
void doDestroyTask(SExecTaskInfo *pTaskInfo);
H
Haojun Liao 已提交
658
void freeQueryAttr(STaskAttr *pQuery);
659 660 661 662

int32_t getMaximumIdleDurationSec();

void doInvokeUdf(struct SUdfInfo* pUdfInfo, SQLFunctionCtx *pCtx, int32_t idx, int32_t type);
663
void setTaskStatus(SExecTaskInfo *pTaskInfo, int8_t status);
H
Haojun Liao 已提交
664
int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* readerHandle, uint64_t taskId);
665 666

#endif  // TDENGINE_EXECUTORIMPL_H