qExecutor.h 13.9 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
/*
 * 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_QUERYEXECUTOR_H
#define TDENGINE_QUERYEXECUTOR_H

#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 "qTsbuf.h"
26
#include "query.h"
27
#include "taosdef.h"
H
Haojun Liao 已提交
28
#include "tarray.h"
B
Bomin Zhang 已提交
29
#include "tlockfree.h"
H
Haojun Liao 已提交
30
#include "tsdb.h"
31 32

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

H
Haojun Liao 已提交
36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65
#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 SET_STABLE_QUERY_OVER(_q) ((_q)->tableIndex = (int32_t)((_q)->tableqinfoGroupInfo.numOfTables))
#define IS_STASBLE_QUERY_OVER(_q) ((_q)->tableIndex >= (int32_t)((_q)->tableqinfoGroupInfo.numOfTables))

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

enum {
  // when query starts to execute, this status will set
      QUERY_NOT_COMPLETED = 0x1u,

  /* result output buffer is full, current query is paused.
   * this status is only exist in group-by clause and diff/add/division/multiply/ query.
   */
      QUERY_RESBUF_FULL = 0x2u,

  /* 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
   */
      QUERY_COMPLETED = 0x4u,

  /* when the result is not completed return to client, this status will be
   * usually used in case of interval query with interpolation option
   */
      QUERY_OVER = 0x8u,
};

H
Haojun Liao 已提交
66
typedef struct SResultRowPool {
67 68 69 70 71 72 73 74 75 76
  int32_t elemSize;
  int32_t blockSize;
  int32_t numOfElemPerBlock;

  struct {
    int32_t blockIndex;
    int32_t pos;
  } position;

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

79
typedef struct SSqlGroupbyExpr {
H
Haojun Liao 已提交
80 81 82 83 84
  int16_t tableIndex;
  SArray* columnInfo;  // SArray<SColIndex>, group by columns information
  int16_t numOfGroupCols;
  int16_t orderIndex;  // order by column index
  int16_t orderType;   // order by type: asc/desc
85 86
} SSqlGroupbyExpr;

H
Haojun Liao 已提交
87
typedef struct SResultRow {
88
  int32_t       pageId;      // pageId & rowId is the position of current result in disk-based output buffer
89 90 91 92 93
  int32_t       rowId:29;    // row index in buffer page
  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 已提交
94
  SResultRowCellInfo*  pCellInfo;  // For each result column, there is a resultInfo
95
  union {STimeWindow win; char* key;};  // start key of current time window
H
Haojun Liao 已提交
96
} SResultRow;
97

98
typedef struct SGroupResInfo {
H
Haojun Liao 已提交
99 100
  int32_t totalGroup;
  int32_t currentGroup;
101 102 103 104
  int32_t index;
  SArray* pRows;      // SArray<SResultRow*>
} SGroupResInfo;

H
Haojun Liao 已提交
105 106 107 108
/**
 * If the number of generated results is greater than this value,
 * query query will be halt and return results to client immediate.
 */
109
typedef struct SResultRec {
H
Haojun Liao 已提交
110 111 112 113
  int64_t total;      // total generated result size in rows
  int64_t rows;       // current result set size in rows
  int64_t capacity;   // capacity of current result output buffer
  int32_t threshold;  // result size threshold in rows.
114 115
} SResultRec;

H
Haojun Liao 已提交
116
typedef struct SResultRowInfo {
H
Haojun Liao 已提交
117 118 119 120 121 122
  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 已提交
123
} SResultRowInfo;
124 125 126 127 128 129 130 131

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

typedef struct SSingleColumnFilterInfo {
H
Haojun Liao 已提交
132
  void*              pData;
133
  int32_t            numOfFilters;
H
Haojun Liao 已提交
134
  SColumnInfo        info;
135 136 137
  SColumnFilterElem* pFilters;
} SSingleColumnFilterInfo;

H
Haojun Liao 已提交
138
typedef struct STableQueryInfo {
H
hjxilinx 已提交
139
  TSKEY       lastKey;
H
Haojun Liao 已提交
140
  int32_t     groupIndex;     // group id in table list
141
  int16_t     queryRangeSet;  // denote if the query range is set, only available for interval query
142
  tVariant    tag;
H
hjxilinx 已提交
143
  STimeWindow win;
144
  STSCursor   cur;
H
Haojun Liao 已提交
145
  void*       pTable;         // for retrieve the page id list
H
Haojun Liao 已提交
146
  SResultRowInfo resInfo;
147 148
} STableQueryInfo;

H
Haojun Liao 已提交
149 150 151 152 153 154 155 156 157
typedef struct SQueryCostInfo {
  uint64_t loadStatisTime;
  uint64_t loadFileBlockTime;
  uint64_t loadDataInCacheTime;
  uint64_t loadStatisSize;
  uint64_t loadFileBlockSize;
  uint64_t loadDataInCacheSize;
  
  uint64_t loadDataTime;
158 159 160 161
  uint64_t totalRows;
  uint64_t totalCheckedRows;
  uint32_t totalBlocks;
  uint32_t loadBlocks;
H
Haojun Liao 已提交
162 163
  uint32_t loadBlockStatis;
  uint32_t discardBlocks;
164
  uint64_t elapsedTime;
H
Haojun Liao 已提交
165
  uint64_t firstStageMergeTime;
H
Haojun Liao 已提交
166 167
  uint64_t winInfoSize;
  uint64_t tableInfoSize;
H
Haojun Liao 已提交
168
  uint64_t hashSize;
169
  uint64_t numOfTimeWindows;
H
Haojun Liao 已提交
170
} SQueryCostInfo;
H
hjxilinx 已提交
171

H
Haojun Liao 已提交
172 173 174 175 176
typedef struct {
  int64_t vgroupLimit;
  int64_t ts;
} SOrderedPrjQueryInfo;

177 178 179 180 181
typedef struct {
  char*   tags;
  SArray* pResult;  // SArray<SStddevInterResult>
} SInterResult;

H
Haojun Liao 已提交
182 183 184 185 186 187
typedef struct SSDataBlock {
  SDataStatis *pBlockStatis;
  SArray      *pDataBlock;
  SDataBlockInfo info;
} SSDataBlock;

188
typedef struct SQuery {
H
Haojun Liao 已提交
189 190
  SLimitVal        limit;

H
Haojun Liao 已提交
191 192 193 194 195 196 197 198 199
  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             queryWindowIdentical; // all query time windows are identical for all tables in one group
  bool             queryBlockDist;    // if query data block distribution
  bool             stabledev;        // super table stddev query
  int32_t          interBufSize;     // intermediate buffer sizse
H
Haojun Liao 已提交
200 201 202

  SOrderVal        order;

H
Haojun Liao 已提交
203 204
  int16_t          numOfCols;
  int16_t          numOfTags;
H
Haojun Liao 已提交
205

H
Haojun Liao 已提交
206
  STimeWindow      window;
207
  SInterval        interval;
208
  int16_t          precision;
H
Haojun Liao 已提交
209 210
  int16_t          numOfOutput;
  int16_t          fillType;
211 212 213 214 215 216
  int16_t          checkResultBuf;   // check if the buffer is full during scan each block

  int32_t          srcRowSize;       // todo extract struct
  int32_t          resultRowSize;
  int32_t          maxSrcColumnSize;
  int32_t          tagLen;           // tag value length of current query
H
Haojun Liao 已提交
217
  SSqlGroupbyExpr* pGroupbyExpr;
H
Haojun Liao 已提交
218
  SExprInfo*       pExpr1;
H
Haojun Liao 已提交
219 220
  SExprInfo*       pExpr2;
  int32_t          numOfExpr2;
H
Haojun Liao 已提交
221 222 223 224
  SColumnInfo*     colList;
  SColumnInfo*     tagColList;
  int32_t          numOfFilterCols;
  int64_t*         fillVal;
H
Haojun Liao 已提交
225 226 227
  SOrderedPrjQueryInfo prjInfo;        // limit value for each vgroup, only available in global order projection query.
  SSingleColumnFilterInfo* pFilterInfo;

228
  uint32_t         status;             // query status
H
Haojun Liao 已提交
229 230 231 232
  SResultRec       rec;
  int32_t          pos;
  tFilePage**      sdata;
  STableQueryInfo* current;
233
  int32_t          numOfCheckedBlocks; // number of check data blocks
H
Haojun Liao 已提交
234

H
Haojun Liao 已提交
235 236
  void*            tsdb;
  SMemRef          memRef;
H
Haojun Liao 已提交
237
  STableGroupInfo  tableGroupInfo;       // table <tid, last_key> list  SArray<STableKeyInfo>
H
Haojun Liao 已提交
238
  int32_t          vgId;
239 240
} SQuery;

H
Haojun Liao 已提交
241 242 243
typedef SSDataBlock* (*__operator_fn_t)(void* param);

typedef struct SOperatorInfo {
H
Haojun Liao 已提交
244 245 246 247 248 249 250
  char      *name;
  bool       blockingOptr;
  bool       completed;
  void      *optInfo;
  SExprInfo *pExpr;
  int32_t    numOfOutput;

H
Haojun Liao 已提交
251 252 253 254
  __operator_fn_t       exec;
  struct SOperatorInfo *upstream;
} SOperatorInfo;

255
typedef struct SQueryRuntimeEnv {
H
Haojun Liao 已提交
256
  jmp_buf              env;
H
Haojun Liao 已提交
257
  SQuery*              pQuery;
H
Haojun Liao 已提交
258
  void*                qinfo;
H
Haojun Liao 已提交
259

H
Haojun Liao 已提交
260
  SQLFunctionCtx*      pCtx;
261
  int32_t              numOfRowsPerPage;
H
Haojun Liao 已提交
262
  uint16_t*            offset;
H
Haojun Liao 已提交
263
  uint16_t             scanFlag;         // denotes reversed scan of data or not
H
Haojun Liao 已提交
264
  SFillInfo*           pFillInfo;
H
Haojun Liao 已提交
265
  SResultRowInfo       resultRowInfo;
H
Haojun Liao 已提交
266 267
  void*                pQueryHandle;
  void*                pSecQueryHandle;  // another thread for
H
Haojun Liao 已提交
268

H
Haojun Liao 已提交
269
  int32_t              prevGroupId;      // previous executed group id
H
Haojun Liao 已提交
270
  SDiskbasedResultBuf* pResultBuf;       // query result buffer based on blocked-wised disk file
H
Haojun Liao 已提交
271
  SHashObj*            pResultRowHashTable; // quick locate the window object for each result
H
Haojun Liao 已提交
272
  char*                keyBuf;           // window key buffer
273
  SResultRowPool*      pool;             // window result object pool
H
Haojun Liao 已提交
274 275

  int32_t*             rowCellInfoOffset;// offset value for each row result cell info
276
  char**               prevRow;
277

278 279 280 281 282
  SArray*              prevResult;       // intermediate result, SArray<SInterResult>
  STSBuf*              pTsBuf;           // timestamp filter list
  STSCursor            cur;

  char*                tagVal;           // tag value of current data block
283
  SArithmeticSupport  *sasArray;
H
Haojun Liao 已提交
284

H
Haojun Liao 已提交
285
  SOperatorInfo*   pi;
H
Haojun Liao 已提交
286
  SSDataBlock *outputBuf;
H
Haojun Liao 已提交
287 288

  int32_t          groupIndex;
H
Haojun Liao 已提交
289
  int32_t          tableIndex;
H
Haojun Liao 已提交
290
  STableGroupInfo  tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
H
Haojun Liao 已提交
291
  SOperatorInfo* proot;
292 293
} SQueryRuntimeEnv;

H
Haojun Liao 已提交
294 295 296 297 298
typedef struct {
  char* name;
  void* info;
} SQEStage;

299 300 301 302 303
enum {
  QUERY_RESULT_NOT_READY = 1,
  QUERY_RESULT_READY     = 2,
};

304
typedef struct SQInfo {
H
Haojun Liao 已提交
305
  void*            signature;
H
Haojun Liao 已提交
306
  int32_t          code;   // error code to returned to client
307
  int64_t          owner;  // if it is in execution
H
Haojun Liao 已提交
308

309
  SQueryRuntimeEnv runtimeEnv;
H
Haojun Liao 已提交
310 311
  SQuery           query;

312
  SHashObj*        arrTableIdInfo;
H
Haojun Liao 已提交
313

314 315 316 317
  /*
   * the query is executed position on which meter of the whole list.
   * when the index reaches the last one of the list, it means the query is completed.
   */
H
Haojun Liao 已提交
318
  SGroupResInfo    groupResInfo;
319
  void*            pBuf;        // allocated buffer for STableQueryInfo, sizeof(STableQueryInfo)*numOfTables;
H
Haojun Liao 已提交
320

321
  pthread_mutex_t  lock;        // used to synchronize the rsp/query threads
H
Haojun Liao 已提交
322
  tsem_t           ready;
323 324
  int32_t          dataReady;   // denote if query result is ready or not
  void*            rspContext;  // response context
325
  int64_t          startExecTs; // start to exec timestamp
326
  char*            sql;         // query sql string
H
Haojun Liao 已提交
327
  SQueryCostInfo   summary;
328 329
} SQInfo;

H
Haojun Liao 已提交
330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345
typedef struct SQueryParam {
  char            *sql;
  char            *tagCond;
  char            *tbnameCond;
  char            *prevResult;
  SArray          *pTableIdList;
  SSqlFuncMsg    **pExprMsg;
  SSqlFuncMsg    **pSecExprMsg;
  SExprInfo       *pExprs;
  SExprInfo       *pSecExprs;

  SColIndex       *pGroupColIndex;
  SColumnInfo     *pTagColumnInfo;
  SSqlGroupbyExpr *pGroupbyExpr;
} SQueryParam;

H
Haojun Liao 已提交
346
typedef struct STableScanInfo {
H
Haojun Liao 已提交
347
  SQueryRuntimeEnv *pRuntimeEnv;
H
Haojun Liao 已提交
348 349 350 351 352 353 354
  void        *pQueryHandle;
  int32_t      numOfBlocks;
  int32_t      numOfSkipped;
  int32_t      numOfBlockStatis;

  int64_t      numOfRows;

H
Haojun Liao 已提交
355 356
  int32_t      order;  // scan order
  int32_t      times;  // repeat counts
H
Haojun Liao 已提交
357 358 359
  int32_t      current;

  int32_t      reverseTimes; // 0 by default
H
Haojun Liao 已提交
360

H
Haojun Liao 已提交
361
  SSDataBlock  block;
H
Haojun Liao 已提交
362 363 364
  int64_t      elapsedTime;
} STableScanInfo;

H
Haojun Liao 已提交
365 366
SOperatorInfo optrList[5];

H
Haojun Liao 已提交
367 368 369 370
typedef struct SAggOperatorInfo {
  SResultRowInfo   *pResultRowInfo;
  STableQueryInfo  *pTableQueryInfo;
  SQueryRuntimeEnv *pRuntimeEnv;
H
Haojun Liao 已提交
371
  SQLFunctionCtx   *pCtx;
H
Haojun Liao 已提交
372 373
} SAggOperatorInfo;

H
Haojun Liao 已提交
374 375 376
typedef struct SArithOperatorInfo {
  STableQueryInfo  *pTableQueryInfo;
  SQueryRuntimeEnv *pRuntimeEnv;
H
Haojun Liao 已提交
377
  SQLFunctionCtx* pCtx;
H
Haojun Liao 已提交
378 379
} SArithOperatorInfo;

H
Haojun Liao 已提交
380 381 382 383 384 385 386 387 388 389 390 391
typedef struct SLimitOperatorInfo {
  int64_t limit;
  int64_t total;
  SQueryRuntimeEnv* pRuntimeEnv;
} SLimitOperatorInfo;

typedef struct SOffsetOperatorInfo {
  int64_t offset;
  int64_t currentOffset;
  SQueryRuntimeEnv* pRuntimeEnv;
} SOffsetOperatorInfo;

H
Haojun Liao 已提交
392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422
void freeParam(SQueryParam *param);
int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param);
int32_t createQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t numOfOutput, SExprInfo **pExprInfo, SSqlFuncMsg **pExprMsg,
                                   SColumnInfo* pTagCols);
SSqlGroupbyExpr *createGroupbyExprFromMsg(SQueryTableMsg *pQueryMsg, SColIndex *pColIndex, int32_t *code);
SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,
                        SExprInfo *pSecExprs, STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, bool stableQuery, char* sql);
int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQInfo *pQInfo, SQueryParam* param, bool isSTable);
void freeColumnFilterInfo(SColumnFilterInfo* pFilter, int32_t numOfFilters);

bool isQueryKilled(SQInfo *pQInfo);
int32_t checkForQueryBuf(size_t numOfTables);
bool doBuildResCheck(SQInfo* pQInfo);
void setQueryStatus(SQuery *pQuery, int8_t status);

bool onlyQueryTags(SQuery* pQuery);
void buildTagQueryResult(SQInfo *pQInfo);
void stableQueryImpl(SQInfo *pQInfo);
void buildTableBlockDistResult(SQInfo *pQInfo);
void tableQueryImpl(SQInfo *pQInfo);
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);

int32_t getMaximumIdleDurationSec();

423
#endif  // TDENGINE_QUERYEXECUTOR_H