executil.h 6.0 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_QUERYUTIL_H
#define TDENGINE_QUERYUTIL_H

H
Haojun Liao 已提交
18
#include <libs/function/function.h>
19
#include "tbuffer.h"
H
Haojun Liao 已提交
20
#include "tcommon.h"
H
Haojun Liao 已提交
21
#include "tpagedbuf.h"
22

H
Haojun Liao 已提交
23 24 25 26 27 28 29
#define SET_RES_WINDOW_KEY(_k, _ori, _len, _uid)     \
  do {                                               \
    assert(sizeof(_uid) == sizeof(uint64_t));        \
    *(uint64_t *)(_k) = (_uid);                      \
    memcpy((_k) + sizeof(uint64_t), (_ori), (_len)); \
  } while (0)

W
fix bug  
wpan 已提交
30 31 32
#define SET_RES_EXT_WINDOW_KEY(_k, _ori, _len, _uid, _buf)             \
  do {                                                                 \
    assert(sizeof(_uid) == sizeof(uint64_t));                          \
W
fix bug  
wpan 已提交
33
    *(void **)(_k) = (_buf);                                             \
W
fix bug  
wpan 已提交
34 35 36 37 38
    *(uint64_t *)((_k) + POINTER_BYTES) = (_uid);                      \
    memcpy((_k) + POINTER_BYTES + sizeof(uint64_t), (_ori), (_len));   \
  } while (0)


H
Haojun Liao 已提交
39
#define GET_RES_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t))
W
fix bug  
wpan 已提交
40 41
#define GET_RES_EXT_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t) + POINTER_BYTES)

42
#define GET_TASKID(_t)  (((SExecTaskInfo*)(_t))->id.str)
H
Haojun Liao 已提交
43

44 45 46 47
typedef struct SGroupResInfo {
  int32_t totalGroup;
  int32_t currentGroup;
  int32_t index;
48
  SArray* pRows;      // SArray<SResultRowPosition*>
49 50 51 52 53 54 55 56 57 58 59 60
  bool    ordered;
  int32_t position;
} SGroupResInfo;

typedef struct SResultRow {
  int32_t       pageId;      // pageId & rowId is the position of current result in disk-based output buffer
  int32_t       offset: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
  STimeWindow   win;
H
Haojun Liao 已提交
61 62
  struct SResultRowEntryInfo pEntryInfo[];  // For each result column, there is a resultInfo
//  char         *key;         // start key of current result row
63 64
} SResultRow;

65 66 67 68 69
typedef struct SResultRowPosition {
  int32_t pageId;
  int32_t offset;
} SResultRowPosition;

70 71 72 73 74 75
typedef struct SResKeyPos {
  SResultRowPosition pos;
  uint64_t  groupId;
  char      key[];
} SResKeyPos;

76
typedef struct SResultRowInfo {
77
  SResultRowPosition *pPosition;  // todo remove this
78
  int32_t      size;       // number of result set
79
  int32_t      capacity;   // max capacity
80
  SResultRowPosition cur;
81 82
} SResultRowInfo;

H
Haojun Liao 已提交
83 84
struct STaskAttr;
struct STaskRuntimeEnv;
85
struct SUdfInfo;
86
struct SqlFunctionCtx;
87

H
Haojun Liao 已提交
88
int32_t getOutputInterResultBufSize(struct STaskAttr* pQueryAttr);
89

90
size_t getResultRowSize(struct SqlFunctionCtx* pCtx, int32_t numOfOutput);
91
int32_t initResultRowInfo(SResultRowInfo* pResultRowInfo, int32_t size);
H
Haojun Liao 已提交
92
void    cleanupResultRowInfo(SResultRowInfo* pResultRowInfo);
93

H
Haojun Liao 已提交
94
void    resetResultRowInfo(struct STaskRuntimeEnv* pRuntimeEnv, SResultRowInfo* pResultRowInfo);
H
Haojun Liao 已提交
95 96
int32_t numOfClosedResultRows(SResultRowInfo* pResultRowInfo);
void    closeAllResultRows(SResultRowInfo* pResultRowInfo);
97

H
Haojun Liao 已提交
98 99 100
void    initResultRow(SResultRow *pResultRow);
void    closeResultRow(SResultRow* pResultRow);
bool    isResultRowClosed(SResultRow* pResultRow);
101

102
struct SResultRowEntryInfo* getResultCell(const SResultRow* pRow, int32_t index, int32_t* offset);
H
Haojun Liao 已提交
103

H
Haojun Liao 已提交
104
int32_t getRowNumForMultioutput(struct STaskAttr* pQueryAttr, bool topBottomQuery, bool stable);
H
Haojun Liao 已提交
105

H
Haojun Liao 已提交
106 107 108 109
static FORCE_INLINE SResultRow *getResultRow(SDiskbasedBuf* pBuf, SResultRowInfo *pResultRowInfo, int32_t slot) {
  ASSERT(pResultRowInfo != NULL && slot >= 0 && slot < pResultRowInfo->size);
  SResultRowPosition* pos = &pResultRowInfo->pPosition[slot];

110
  SFilePage*  bufPage = (SFilePage*) getBufPage(pBuf, pos->pageId);
H
Haojun Liao 已提交
111 112 113 114 115
  SResultRow* pRow = (SResultRow*)((char*)bufPage + pos->offset);
  return pRow;
}

static FORCE_INLINE SResultRow *getResultRowByPos(SDiskbasedBuf* pBuf, SResultRowPosition* pos) {
116
  SFilePage*  bufPage = (SFilePage*) getBufPage(pBuf, pos->pageId);
H
Haojun Liao 已提交
117 118
  SResultRow* pRow = (SResultRow*)((char*)bufPage + pos->offset);
  return pRow;
H
Haojun Liao 已提交
119
}
H
Haojun Liao 已提交
120

H
Haojun Liao 已提交
121
static FORCE_INLINE char* getPosInResultPage(struct STaskAttr* pQueryAttr, SFilePage* page, int32_t rowOffset,
122
                                             int32_t offset) {
H
Haojun Liao 已提交
123
  assert(rowOffset >= 0 && pQueryAttr != NULL);
H
Haojun Liao 已提交
124

125 126
  // int32_t numOfRows = (int32_t)getRowNumForMultioutput(pQueryAttr, pQueryAttr->topBotQuery, pQueryAttr->stableQuery);
  return ((char *)page->data);  
wmmhello's avatar
wmmhello 已提交
127

H
Haojun Liao 已提交
128
}
129

130 131 132 133
static FORCE_INLINE char* getPosInResultPage_rv(SFilePage* page, int32_t rowOffset, int32_t offset) {
  assert(rowOffset >= 0);

  int32_t numOfRows = 1;//(int32_t)getRowNumForMultioutput(pQueryAttr, pQueryAttr->topBotQuery, pQueryAttr->stableQuery);
134
  return (char*) page + rowOffset + offset * numOfRows;
135 136
}

137 138 139 140 141
typedef struct {
  SArray* pResult;     // SArray<SResPair>
  int32_t colId;
} SStddevInterResult;

142
void    initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order);
H
Haojun Liao 已提交
143 144
void    initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList);

H
Haojun Liao 已提交
145
void    cleanupGroupResInfo(SGroupResInfo* pGroupResInfo);
H
Haojun Liao 已提交
146
bool    hasRemainDataInCurrentGroup(SGroupResInfo* pGroupResInfo);
H
Haojun Liao 已提交
147
bool    hasRemainData(SGroupResInfo* pGroupResInfo);
H
Haojun Liao 已提交
148

H
Haojun Liao 已提交
149 150 151
bool    incNextGroup(SGroupResInfo* pGroupResInfo);
int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo);

H
Haojun Liao 已提交
152
int32_t mergeIntoGroupResult(SGroupResInfo* pGroupResInfo, struct STaskRuntimeEnv *pRuntimeEnv, int32_t* offset);
H
Haojun Liao 已提交
153

154
//int32_t initUdfInfo(struct SUdfInfo* pUdfInfo);
D
dapan1121 已提交
155

156
#endif  // TDENGINE_QUERYUTIL_H