Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1193
Star 22018
Fork 4786
T
TDengine
未验证 提交 096c292d 编写于 作者: X Xiaoyu Wang 提交者： GitHub
浏览文件
操作

Merge pull request #14179 from taosdata/feature/3.0_debug_wxy 

feat: sql command 'show consumers' and 'show subscriptions'
上级 ef554e98 28d66616
展开全部 隐藏空白更改
内联 并排
Showing with 3206 addition and 2976 deletion
include/common/tmsg.h
浏览文件 @ 096c292d
......@@ -2463,6 +2463,8 @@ typedef struct {
int64_t interval;
int64_t offset;
int64_t sliding;
int64_t maxDelay;
int64_t watermark;
int32_t exprLen; // strlen + 1
int32_t tagsFilterLen; // strlen + 1
int32_t sqlLen; // strlen + 1
......
include/common/ttokendef.h
浏览文件 @ 096c292d
......@@ -158,107 +158,109 @@
#define TK_CLUSTER 140
#define TK_TRANSACTIONS 141
#define TK_DISTRIBUTED 142
#define TK_LIKE 143
#define TK_INDEX 144
#define TK_FUNCTION 145
#define TK_INTERVAL 146
#define TK_TOPIC 147
#define TK_AS 148
#define TK_WITH 149
#define TK_META 150
#define TK_CONSUMER 151
#define TK_GROUP 152
#define TK_DESC 153
#define TK_DESCRIBE 154
#define TK_RESET 155
#define TK_QUERY 156
#define TK_CACHE 157
#define TK_EXPLAIN 158
#define TK_ANALYZE 159
#define TK_VERBOSE 160
#define TK_NK_BOOL 161
#define TK_RATIO 162
#define TK_NK_FLOAT 163
#define TK_COMPACT 164
#define TK_VNODES 165
#define TK_IN 166
#define TK_OUTPUTTYPE 167
#define TK_AGGREGATE 168
#define TK_BUFSIZE 169
#define TK_STREAM 170
#define TK_INTO 171
#define TK_TRIGGER 172
#define TK_AT_ONCE 173
#define TK_WINDOW_CLOSE 174
#define TK_KILL 175
#define TK_CONNECTION 176
#define TK_TRANSACTION 177
#define TK_BALANCE 178
#define TK_VGROUP 179
#define TK_MERGE 180
#define TK_REDISTRIBUTE 181
#define TK_SPLIT 182
#define TK_SYNCDB 183
#define TK_DELETE 184
#define TK_NULL 185
#define TK_NK_QUESTION 186
#define TK_NK_ARROW 187
#define TK_ROWTS 188
#define TK_TBNAME 189
#define TK_QSTARTTS 190
#define TK_QENDTS 191
#define TK_WSTARTTS 192
#define TK_WENDTS 193
#define TK_WDURATION 194
#define TK_CAST 195
#define TK_NOW 196
#define TK_TODAY 197
#define TK_TIMEZONE 198
#define TK_COUNT 199
#define TK_LAST_ROW 200
#define TK_BETWEEN 201
#define TK_IS 202
#define TK_NK_LT 203
#define TK_NK_GT 204
#define TK_NK_LE 205
#define TK_NK_GE 206
#define TK_NK_NE 207
#define TK_MATCH 208
#define TK_NMATCH 209
#define TK_CONTAINS 210
#define TK_JOIN 211
#define TK_INNER 212
#define TK_SELECT 213
#define TK_DISTINCT 214
#define TK_WHERE 215
#define TK_PARTITION 216
#define TK_BY 217
#define TK_SESSION 218
#define TK_STATE_WINDOW 219
#define TK_SLIDING 220
#define TK_FILL 221
#define TK_VALUE 222
#define TK_NONE 223
#define TK_PREV 224
#define TK_LINEAR 225
#define TK_NEXT 226
#define TK_HAVING 227
#define TK_RANGE 228
#define TK_EVERY 229
#define TK_ORDER 230
#define TK_SLIMIT 231
#define TK_SOFFSET 232
#define TK_LIMIT 233
#define TK_OFFSET 234
#define TK_ASC 235
#define TK_NULLS 236
#define TK_ID 237
#define TK_NK_BITNOT 238
#define TK_INSERT 239
#define TK_VALUES 240
#define TK_IMPORT 241
#define TK_NK_SEMI 242
#define TK_FILE 243
#define TK_CONSUMERS 143
#define TK_SUBSCRIPTIONS 144
#define TK_LIKE 145
#define TK_INDEX 146
#define TK_FUNCTION 147
#define TK_INTERVAL 148
#define TK_TOPIC 149
#define TK_AS 150
#define TK_WITH 151
#define TK_META 152
#define TK_CONSUMER 153
#define TK_GROUP 154
#define TK_DESC 155
#define TK_DESCRIBE 156
#define TK_RESET 157
#define TK_QUERY 158
#define TK_CACHE 159
#define TK_EXPLAIN 160
#define TK_ANALYZE 161
#define TK_VERBOSE 162
#define TK_NK_BOOL 163
#define TK_RATIO 164
#define TK_NK_FLOAT 165
#define TK_COMPACT 166
#define TK_VNODES 167
#define TK_IN 168
#define TK_OUTPUTTYPE 169
#define TK_AGGREGATE 170
#define TK_BUFSIZE 171
#define TK_STREAM 172
#define TK_INTO 173
#define TK_TRIGGER 174
#define TK_AT_ONCE 175
#define TK_WINDOW_CLOSE 176
#define TK_KILL 177
#define TK_CONNECTION 178
#define TK_TRANSACTION 179
#define TK_BALANCE 180
#define TK_VGROUP 181
#define TK_MERGE 182
#define TK_REDISTRIBUTE 183
#define TK_SPLIT 184
#define TK_SYNCDB 185
#define TK_DELETE 186
#define TK_NULL 187
#define TK_NK_QUESTION 188
#define TK_NK_ARROW 189
#define TK_ROWTS 190
#define TK_TBNAME 191
#define TK_QSTARTTS 192
#define TK_QENDTS 193
#define TK_WSTARTTS 194
#define TK_WENDTS 195
#define TK_WDURATION 196
#define TK_CAST 197
#define TK_NOW 198
#define TK_TODAY 199
#define TK_TIMEZONE 200
#define TK_COUNT 201
#define TK_LAST_ROW 202
#define TK_BETWEEN 203
#define TK_IS 204
#define TK_NK_LT 205
#define TK_NK_GT 206
#define TK_NK_LE 207
#define TK_NK_GE 208
#define TK_NK_NE 209
#define TK_MATCH 210
#define TK_NMATCH 211
#define TK_CONTAINS 212
#define TK_JOIN 213
#define TK_INNER 214
#define TK_SELECT 215
#define TK_DISTINCT 216
#define TK_WHERE 217
#define TK_PARTITION 218
#define TK_BY 219
#define TK_SESSION 220
#define TK_STATE_WINDOW 221
#define TK_SLIDING 222
#define TK_FILL 223
#define TK_VALUE 224
#define TK_NONE 225
#define TK_PREV 226
#define TK_LINEAR 227
#define TK_NEXT 228
#define TK_HAVING 229
#define TK_RANGE 230
#define TK_EVERY 231
#define TK_ORDER 232
#define TK_SLIMIT 233
#define TK_SOFFSET 234
#define TK_LIMIT 235
#define TK_OFFSET 236
#define TK_ASC 237
#define TK_NULLS 238
#define TK_ID 239
#define TK_NK_BITNOT 240
#define TK_INSERT 241
#define TK_VALUES 242
#define TK_IMPORT 243
#define TK_NK_SEMI 244
#define TK_FILE 245
#define TK_NK_SPACE 300
#define TK_NK_COMMENT 301
......
include/libs/function/functionMgt.h
浏览文件 @ 096c292d
......@@ -123,6 +123,7 @@ typedef enum EFunctionType {
FUNCTION_TYPE_SELECT_VALUE,
FUNCTION_TYPE_BLOCK_DIST, // block distribution aggregate function
FUNCTION_TYPE_BLOCK_DIST_INFO, // block distribution pseudo column function
FUNCTION_TYPE_TO_COLUMN,
// distributed splitting functions
FUNCTION_TYPE_APERCENTILE_PARTIAL = 4000,
......
include/libs/nodes/cmdnodes.h
浏览文件 @ 096c292d
......@@ -28,19 +28,18 @@ extern "C" {
#define DESCRIBE_RESULT_TYPE_LEN (20 + VARSTR_HEADER_SIZE)
#define DESCRIBE_RESULT_NOTE_LEN (8 + VARSTR_HEADER_SIZE)
#define SHOW_CREATE_DB_RESULT_COLS 2
#define SHOW_CREATE_DB_RESULT_COLS 2
#define SHOW_CREATE_DB_RESULT_FIELD1_LEN (TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE)
#define SHOW_CREATE_DB_RESULT_FIELD2_LEN (TSDB_MAX_BINARY_LEN + VARSTR_HEADER_SIZE)
#define SHOW_CREATE_TB_RESULT_COLS 2
#define SHOW_CREATE_TB_RESULT_COLS 2
#define SHOW_CREATE_TB_RESULT_FIELD1_LEN (TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE)
#define SHOW_CREATE_TB_RESULT_FIELD2_LEN (TSDB_MAX_BINARY_LEN + VARSTR_HEADER_SIZE)
#define SHOW_LOCAL_VARIABLES_RESULT_COLS 2
#define SHOW_LOCAL_VARIABLES_RESULT_COLS 2
#define SHOW_LOCAL_VARIABLES_RESULT_FIELD1_LEN (TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE)
#define SHOW_LOCAL_VARIABLES_RESULT_FIELD2_LEN (TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE)
#define PRIVILEGE_TYPE_MASK(n) (1 << n)
#define PRIVILEGE_TYPE_ALL PRIVILEGE_TYPE_MASK(0)
......@@ -234,10 +233,10 @@ typedef struct SShowCreateDatabaseStmt {
} SShowCreateDatabaseStmt;
typedef struct SShowCreateTableStmt {
ENodeType type;
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
void* pCfg; // STableCfg
ENodeType type;
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
void* pCfg; // STableCfg
} SShowCreateTableStmt;
typedef struct SShowTableDistributedStmt {
......@@ -259,6 +258,7 @@ typedef struct SIndexOptions {
SNode* pInterval;
SNode* pOffset;
SNode* pSliding;
SNode* pStreamOptions;
} SIndexOptions;
typedef struct SCreateIndexStmt {
......
include/libs/nodes/nodes.h
浏览文件 @ 096c292d
......@@ -188,6 +188,7 @@ typedef enum ENodeType {
QUERY_NODE_SHOW_CREATE_STABLE_STMT,
QUERY_NODE_SHOW_TRANSACTIONS_STMT,
QUERY_NODE_SHOW_TABLE_DISTRIBUTED_STMT,
QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT,
QUERY_NODE_KILL_CONNECTION_STMT,
QUERY_NODE_KILL_QUERY_STMT,
QUERY_NODE_KILL_TRANSACTION_STMT,
......@@ -284,6 +285,7 @@ int32_t nodesListPushFront(SNodeList* pList, SNode* pNode);
SListCell* nodesListErase(SNodeList* pList, SListCell* pCell);
void nodesListInsertList(SNodeList* pTarget, SListCell* pPos, SNodeList* pSrc);
SNode* nodesListGetNode(SNodeList* pList, int32_t index);
SListCell* nodesListGetCell(SNodeList* pList, int32_t index);
void nodesDestroyList(SNodeList* pList);
// Only clear the linked list structure, without releasing the elements inside
void nodesClearList(SNodeList* pList);
......
include/libs/nodes/plannodes.h
浏览文件 @ 096c292d
......@@ -32,6 +32,8 @@ typedef struct SLogicNode {
struct SLogicNode* pParent;
int32_t optimizedFlag;
uint8_t precision;
SNode* pLimit;
SNode* pSlimit;
} SLogicNode;
typedef enum EScanType {
......@@ -92,10 +94,6 @@ typedef struct SProjectLogicNode {
SLogicNode node;
SNodeList* pProjections;
char stmtName[TSDB_TABLE_NAME_LEN];
int64_t limit;
int64_t offset;
int64_t slimit;
int64_t soffset;
} SProjectLogicNode;
typedef struct SIndefRowsFuncLogicNode {
......@@ -248,6 +246,8 @@ typedef struct SPhysiNode {
SNode* pConditions;
SNodeList* pChildren;
struct SPhysiNode* pParent;
SNode* pLimit;
SNode* pSlimit;
} SPhysiNode;
typedef struct SScanPhysiNode {
......@@ -297,10 +297,6 @@ typedef STableScanPhysiNode SStreamScanPhysiNode;
typedef struct SProjectPhysiNode {
SPhysiNode node;
SNodeList* pProjections;
int64_t limit;
int64_t offset;
int64_t slimit;
int64_t soffset;
} SProjectPhysiNode;
typedef struct SIndefRowsFuncPhysiNode {
......@@ -378,7 +374,7 @@ typedef struct SIntervalPhysiNode {
int8_t slidingUnit;
} SIntervalPhysiNode;
typedef SIntervalPhysiNode SMergeIntervalPhysiNode;
typedef SIntervalPhysiNode SMergeIntervalPhysiNode;
typedef SIntervalPhysiNode SMergeAlignedIntervalPhysiNode;
typedef SIntervalPhysiNode SStreamIntervalPhysiNode;
typedef SIntervalPhysiNode SStreamFinalIntervalPhysiNode;
......
source/common/src/tmsg.c
浏览文件 @ 096c292d
......@@ -539,7 +539,7 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
const char *pFunc = taosArrayGet(pReq->pFuncs, i);
if (tEncodeCStr(&encoder, pFunc) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
......@@ -621,7 +621,7 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
if (NULL == pReq->pFuncs) return -1;
}
for (int32_t i = 0; i < numOfFuncs; ++i) {
char* pFunc = NULL;
char *pFunc = NULL;
if (tDecodeCStrAlloc(&decoder, &pFunc) < 0) return -1;
if (taosArrayPush(pReq->pFuncs, pFunc) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
......@@ -783,6 +783,8 @@ int32_t tSerializeSMCreateSmaReq(void *buf, int32_t bufLen, SMCreateSmaReq *pReq
if (tEncodeI64(&encoder, pReq->interval) < 0) return -1;
if (tEncodeI64(&encoder, pReq->offset) < 0) return -1;
if (tEncodeI64(&encoder, pReq->sliding) < 0) return -1;
if (tEncodeI64(&encoder, pReq->watermark) < 0) return -1;
if (tEncodeI64(&encoder, pReq->maxDelay) < 0) return -1;
if (tEncodeI32(&encoder, pReq->exprLen) < 0) return -1;
if (tEncodeI32(&encoder, pReq->tagsFilterLen) < 0) return -1;
if (tEncodeI32(&encoder, pReq->sqlLen) < 0) return -1;
......@@ -821,6 +823,8 @@ int32_t tDeserializeSMCreateSmaReq(void *buf, int32_t bufLen, SMCreateSmaReq *pR
if (tDecodeI64(&decoder, &pReq->interval) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->offset) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->sliding) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->watermark) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->maxDelay) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->exprLen) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->tagsFilterLen) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->sqlLen) < 0) return -1;
......@@ -1868,7 +1872,7 @@ int32_t tSerializeSTableCfgRsp(void *buf, int32_t bufLen, STableCfgRsp *pRsp) {
if (tEncodeI64(&encoder, pRsp->watermark1) < 0) return -1;
if (tEncodeI64(&encoder, pRsp->watermark2) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->ttl) < 0) return -1;
int32_t numOfFuncs = taosArrayGetSize(pRsp->pFuncs);
if (tEncodeI32(&encoder, numOfFuncs) < 0) return -1;
for (int32_t i = 0; i < numOfFuncs; ++i) {
......@@ -1876,19 +1880,19 @@ int32_t tSerializeSTableCfgRsp(void *buf, int32_t bufLen, STableCfgRsp *pRsp) {
if (tEncodeCStr(&encoder, pFunc) < 0) return -1;
}
if (tEncodeI32(&encoder, pRsp->commentLen) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->commentLen) < 0) return -1;
if (pRsp->commentLen > 0) {
if (tEncodeCStr(&encoder, pRsp->pComment) < 0) return -1;
}
for (int32_t i = 0; i < pRsp->numOfColumns + pRsp->numOfTags; ++i) {
SSchema *pSchema = &pRsp->pSchemas[i];
if (tEncodeSSchema(&encoder, pSchema) < 0) return -1;
}
if (tEncodeI32(&encoder, pRsp->tagsLen) < 0) return -1;
if (tEncodeBinary(&encoder, pRsp->pTags, pRsp->tagsLen) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->tagsLen) < 0) return -1;
if (tEncodeBinary(&encoder, pRsp->pTags, pRsp->tagsLen) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
......@@ -1945,7 +1949,7 @@ int32_t tDeserializeSTableCfgRsp(void *buf, int32_t bufLen, STableCfgRsp *pRsp)
}
if (tDecodeI32(&decoder, &pRsp->tagsLen) < 0) return -1;
if (tDecodeBinaryAlloc(&decoder, (void**)&pRsp->pTags, NULL) < 0) return -1;
if (tDecodeBinaryAlloc(&decoder, (void **)&pRsp->pTags, NULL) < 0) return -1;
tEndDecode(&decoder);
......@@ -1961,7 +1965,7 @@ void tFreeSTableCfgRsp(STableCfgRsp *pRsp) {
taosMemoryFreeClear(pRsp->pComment);
taosMemoryFreeClear(pRsp->pSchemas);
taosMemoryFreeClear(pRsp->pTags);
taosArrayDestroy(pRsp->pFuncs);
}
......
source/libs/executor/src/executorimpl.c
浏览文件 @ 096c292d
......@@ -3581,6 +3581,10 @@ static SArray* setRowTsColumnOutputInfo(SqlFunctionCtx* pCtx, int32_t numOfCols)
return pList;
}
static int64_t getLimit(SNode* pLimit) { return NULL == pLimit ? -1 : ((SLimitNode*)pLimit)->limit; }
static int64_t getOffset(SNode* pLimit) { return NULL == pLimit ? -1 : ((SLimitNode*)pLimit)->offset; }
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode,
SExecTaskInfo* pTaskInfo) {
SProjectOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SProjectOperatorInfo));
......@@ -3593,8 +3597,8 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys
SExprInfo* pExprInfo = createExprInfo(pProjPhyNode->pProjections, NULL, &numOfCols);
SSDataBlock* pResBlock = createResDataBlock(pProjPhyNode->node.pOutputDataBlockDesc);
SLimit limit = {.limit = pProjPhyNode->limit, .offset = pProjPhyNode->offset};
SLimit slimit = {.limit = pProjPhyNode->slimit, .offset = pProjPhyNode->soffset};
SLimit limit = {.limit = getLimit(pProjPhyNode->node.pLimit), .offset = getOffset(pProjPhyNode->node.pLimit)};
SLimit slimit = {.limit = getLimit(pProjPhyNode->node.pSlimit), .offset = getOffset(pProjPhyNode->node.pSlimit)};
pInfo->limit = limit;
pInfo->slimit = slimit;
......
source/libs/function/src/builtins.c
浏览文件 @ 096c292d
......@@ -1857,7 +1857,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "last_row",
.type = FUNCTION_TYPE_LAST_ROW,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
.translateFunc = translateLastRow,
.getEnvFunc = getMinmaxFuncEnv,
.initFunc = minmaxFunctionSetup,
......@@ -2024,7 +2024,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "diff",
.type = FUNCTION_TYPE_DIFF,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC,
.translateFunc = translateDiff,
.getEnvFunc = getDiffFuncEnv,
.initFunc = diffFunctionSetup,
......@@ -2034,7 +2034,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "statecount",
.type = FUNCTION_TYPE_STATE_COUNT,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC,
.translateFunc = translateStateCount,
.getEnvFunc = getStateFuncEnv,
.initFunc = functionSetup,
......@@ -2044,7 +2044,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "stateduration",
.type = FUNCTION_TYPE_STATE_DURATION,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC,
.translateFunc = translateStateDuration,
.getEnvFunc = getStateFuncEnv,
.initFunc = functionSetup,
......@@ -2054,7 +2054,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "csum",
.type = FUNCTION_TYPE_CSUM,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC,
.translateFunc = translateCsum,
.getEnvFunc = getCsumFuncEnv,
.initFunc = functionSetup,
......@@ -2064,7 +2064,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "mavg",
.type = FUNCTION_TYPE_MAVG,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC,
.translateFunc = translateMavg,
.getEnvFunc = getMavgFuncEnv,
.initFunc = mavgFunctionSetup,
......
source/libs/nodes/src/nodesCodeFuncs.c
浏览文件 @ 096c292d
......@@ -491,6 +491,8 @@ static int32_t jsonToTableMeta(const SJson* pJson, void* pObj) {
static const char* jkLogicPlanTargets = "Targets";
static const char* jkLogicPlanConditions = "Conditions";
static const char* jkLogicPlanChildren = "Children";
static const char* jkLogicPlanLimit = "Limit";
static const char* jkLogicPlanSlimit = "SLimit";
static int32_t logicPlanNodeToJson(const void* pObj, SJson* pJson) {
const SLogicNode* pNode = (const SLogicNode*)pObj;
......@@ -502,6 +504,12 @@ static int32_t logicPlanNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkLogicPlanChildren, pNode->pChildren);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkLogicPlanLimit, nodeToJson, pNode->pLimit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkLogicPlanSlimit, nodeToJson, pNode->pSlimit);
}
return code;
}
......@@ -516,6 +524,12 @@ static int32_t jsonToLogicPlanNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkLogicPlanChildren, &pNode->pChildren);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkLogicPlanLimit, &pNode->pLimit);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkLogicPlanSlimit, &pNode->pSlimit);
}
return code;
}
......@@ -581,10 +595,6 @@ static int32_t jsonToLogicScanNode(const SJson* pJson, void* pObj) {
}
static const char* jkProjectLogicPlanProjections = "Projections";
static const char* jkProjectLogicPlanLimit = "Limit";
static const char* jkProjectLogicPlanOffset = "Offset";
static const char* jkProjectLogicPlanSlimit = "SLimit";
static const char* jkProjectLogicPlanSoffset = "SOffset";
static int32_t logicProjectNodeToJson(const void* pObj, SJson* pJson) {
const SProjectLogicNode* pNode = (const SProjectLogicNode*)pObj;
......@@ -593,18 +603,6 @@ static int32_t logicProjectNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkProjectLogicPlanProjections, pNode->pProjections);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkProjectLogicPlanLimit, pNode->limit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkProjectLogicPlanOffset, pNode->offset);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkProjectLogicPlanSlimit, pNode->slimit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkProjectLogicPlanSoffset, pNode->soffset);
}
return code;
}
......@@ -616,18 +614,6 @@ static int32_t jsonToLogicProjectNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkProjectLogicPlanProjections, &pNode->pProjections);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkProjectLogicPlanLimit, &pNode->limit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkProjectLogicPlanOffset, &pNode->offset);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkProjectLogicPlanSlimit, &pNode->slimit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkProjectLogicPlanSoffset, &pNode->soffset);
}
return code;
}
......@@ -1283,6 +1269,8 @@ static int32_t logicJoinNodeToJson(const void* pObj, SJson* pJson) {
static const char* jkPhysiPlanOutputDataBlockDesc = "OutputDataBlockDesc";
static const char* jkPhysiPlanConditions = "Conditions";
static const char* jkPhysiPlanChildren = "Children";
static const char* jkPhysiPlanLimit = "Limit";
static const char* jkPhysiPlanSlimit = "SLimit";
static int32_t physicPlanNodeToJson(const void* pObj, SJson* pJson) {
const SPhysiNode* pNode = (const SPhysiNode*)pObj;
......@@ -1294,6 +1282,12 @@ static int32_t physicPlanNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkPhysiPlanChildren, pNode->pChildren);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkPhysiPlanLimit, nodeToJson, pNode->pLimit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkPhysiPlanSlimit, nodeToJson, pNode->pSlimit);
}
return code;
}
......@@ -1308,6 +1302,12 @@ static int32_t jsonToPhysicPlanNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkPhysiPlanChildren, &pNode->pChildren);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkPhysiPlanLimit, &pNode->pLimit);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkPhysiPlanSlimit, &pNode->pSlimit);
}
return code;
}
......@@ -1590,10 +1590,6 @@ static int32_t jsonToPhysiSysTableScanNode(const SJson* pJson, void* pObj) {
}
static const char* jkProjectPhysiPlanProjections = "Projections";
static const char* jkProjectPhysiPlanLimit = "Limit";
static const char* jkProjectPhysiPlanOffset = "Offset";
static const char* jkProjectPhysiPlanSlimit = "SLimit";
static const char* jkProjectPhysiPlanSoffset = "SOffset";
static int32_t physiProjectNodeToJson(const void* pObj, SJson* pJson) {
const SProjectPhysiNode* pNode = (const SProjectPhysiNode*)pObj;
......@@ -1602,18 +1598,6 @@ static int32_t physiProjectNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkProjectPhysiPlanProjections, pNode->pProjections);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkProjectPhysiPlanLimit, pNode->limit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkProjectPhysiPlanOffset, pNode->offset);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkProjectPhysiPlanSlimit, pNode->slimit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkProjectPhysiPlanSoffset, pNode->soffset);
}
return code;
}
......@@ -1625,18 +1609,6 @@ static int32_t jsonToPhysiProjectNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkProjectPhysiPlanProjections, &pNode->pProjections);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkProjectPhysiPlanLimit, &pNode->limit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkProjectPhysiPlanOffset, &pNode->offset);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkProjectPhysiPlanSlimit, &pNode->slimit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkProjectPhysiPlanSoffset, &pNode->soffset);
}
return code;
}
......@@ -3165,6 +3137,31 @@ static int32_t jsonToOrderByExprNode(const SJson* pJson, void* pObj) {
return code;
}
static const char* jkLimitLimit = "Limit";
static const char* jkLimitOffset = "Offset";
static int32_t limitNodeToJson(const void* pObj, SJson* pJson) {
const SLimitNode* pNode = (const SLimitNode*)pObj;
int32_t code = tjsonAddIntegerToObject(pJson, jkLimitLimit, pNode->limit);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkLimitOffset, pNode->offset);
}
return code;
}
static int32_t jsonToLimitNode(const SJson* pJson, void* pObj) {
SLimitNode* pNode = (SLimitNode*)pObj;
int32_t code = tjsonGetBigIntValue(pJson, jkLimitLimit, &pNode->limit);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkLimitOffset, &pNode->offset);
}
return code;
}
static const char* jkStateWindowCol = "StateWindowCol";
static const char* jkStateWindowExpr = "StateWindowExpr";
......@@ -4031,7 +4028,7 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
case QUERY_NODE_ORDER_BY_EXPR:
return orderByExprNodeToJson(pObj, pJson);
case QUERY_NODE_LIMIT:
break;
return limitNodeToJson(pObj, pJson);
case QUERY_NODE_STATE_WINDOW:
return stateWindowNodeToJson(pObj, pJson);
case QUERY_NODE_SESSION_WINDOW:
......@@ -4191,6 +4188,8 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
return jsonToTempTableNode(pJson, pObj);
case QUERY_NODE_ORDER_BY_EXPR:
return jsonToOrderByExprNode(pJson, pObj);
case QUERY_NODE_LIMIT:
return jsonToLimitNode(pJson, pObj);
case QUERY_NODE_STATE_WINDOW:
return jsonToStateWindowNode(pJson, pObj);
case QUERY_NODE_SESSION_WINDOW:
......
source/libs/nodes/src/nodesUtilFuncs.c
浏览文件 @ 096c292d
......@@ -210,6 +210,7 @@ SNode* nodesMakeNode(ENodeType type) {
case QUERY_NODE_SHOW_VARIABLES_STMT:
case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
return makeNode(type, sizeof(SShowStmt));
case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
return makeNode(type, sizeof(SShowDnodeVariablesStmt));
......@@ -371,6 +372,8 @@ static void destroyScanPhysiNode(SScanPhysiNode* pNode) {
static void destroyDataSinkNode(SDataSinkNode* pNode) { nodesDestroyNode((SNode*)pNode->pInputDataBlockDesc); }
static void destroyExprNode(SExprNode* pExpr) { taosArrayDestroy(pExpr->pAssociation); }
void nodesDestroyNode(SNode* pNode) {
if (NULL == pNode) {
return;
......@@ -378,9 +381,11 @@ void nodesDestroyNode(SNode* pNode) {
switch (nodeType(pNode)) {
case QUERY_NODE_COLUMN: // pProjectRef is weak reference, no need to release
destroyExprNode((SExprNode*)pNode);
break;
case QUERY_NODE_VALUE: {
SValueNode* pValue = (SValueNode*)pNode;
destroyExprNode((SExprNode*)pNode);
taosMemoryFreeClear(pValue->literal);
if (IS_VAR_DATA_TYPE(pValue->node.resType.type)) {
taosMemoryFreeClear(pValue->datum.p);
......@@ -389,14 +394,17 @@ void nodesDestroyNode(SNode* pNode) {
}
case QUERY_NODE_OPERATOR: {
SOperatorNode* pOp = (SOperatorNode*)pNode;
destroyExprNode((SExprNode*)pNode);
nodesDestroyNode(pOp->pLeft);
nodesDestroyNode(pOp->pRight);
break;
}
case QUERY_NODE_LOGIC_CONDITION:
destroyExprNode((SExprNode*)pNode);
nodesDestroyList(((SLogicConditionNode*)pNode)->pParameterList);
break;
case QUERY_NODE_FUNCTION:
destroyExprNode((SExprNode*)pNode);
nodesDestroyList(((SFunctionNode*)pNode)->pParameterList);
break;
case QUERY_NODE_REAL_TABLE: {
......@@ -644,7 +652,8 @@ void nodesDestroyNode(SNode* pNode) {
case QUERY_NODE_SHOW_SCORES_STMT:
case QUERY_NODE_SHOW_VARIABLES_STMT:
case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
case QUERY_NODE_SHOW_TRANSACTIONS_STMT: {
case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT: {
SShowStmt* pStmt = (SShowStmt*)pNode;
nodesDestroyNode(pStmt->pDbName);
nodesDestroyNode(pStmt->pTbName);
......@@ -831,6 +840,7 @@ void nodesDestroyNode(SNode* pNode) {
destroyPhysiNode((SPhysiNode*)pPhyNode);
nodesDestroyList(pPhyNode->pExprs);
nodesDestroyList(pPhyNode->pSortKeys);
nodesDestroyList(pPhyNode->pTargets);
break;
}
case QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL:
......@@ -1094,6 +1104,16 @@ SNode* nodesListGetNode(SNodeList* pList, int32_t index) {
return NULL;
}
SListCell* nodesListGetCell(SNodeList* pList, int32_t index) {
SNode* node;
FOREACH(node, pList) {
if (0 == index--) {
return cell;
}
}
return NULL;
}
void nodesDestroyList(SNodeList* pList) {
if (NULL == pList) {
return;
......
source/libs/parser/inc/parAst.h
浏览文件 @ 096c292d
......@@ -170,7 +170,8 @@ SNode* createDropDnodeStmt(SAstCreateContext* pCxt, const SToken* pDnode);
SNode* createAlterDnodeStmt(SAstCreateContext* pCxt, const SToken* pDnode, const SToken* pConfig, const SToken* pValue);
SNode* createCreateIndexStmt(SAstCreateContext* pCxt, EIndexType type, bool ignoreExists, SToken* pIndexName,
SToken* pTableName, SNodeList* pCols, SNode* pOptions);
SNode* createIndexOption(SAstCreateContext* pCxt, SNodeList* pFuncs, SNode* pInterval, SNode* pOffset, SNode* pSliding);
SNode* createIndexOption(SAstCreateContext* pCxt, SNodeList* pFuncs, SNode* pInterval, SNode* pOffset, SNode* pSliding,
SNode* pStreamOptions);
SNode* createDropIndexStmt(SAstCreateContext* pCxt, bool ignoreNotExists, SToken* pIndexName);
SNode* createCreateComponentNodeStmt(SAstCreateContext* pCxt, ENodeType type, const SToken* pDnodeId);
SNode* createDropComponentNodeStmt(SAstCreateContext* pCxt, ENodeType type, const SToken* pDnodeId);
......
source/libs/parser/inc/sql.y
浏览文件 @ 096c292d
......@@ -388,6 +388,8 @@ cmd ::= SHOW SNODES.
cmd ::= SHOW CLUSTER. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_CLUSTER_STMT); }
cmd ::= SHOW TRANSACTIONS. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_TRANSACTIONS_STMT); }
cmd ::= SHOW TABLE DISTRIBUTED full_table_name(A). { pCxt->pRootNode = createShowTableDistributedStmt(pCxt, A); }
cmd ::= SHOW CONSUMERS. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_CONSUMERS_STMT); }
cmd ::= SHOW SUBSCRIPTIONS. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT); }
db_name_cond_opt(A) ::= . { A = createDefaultDatabaseCondValue(pCxt); }
db_name_cond_opt(A) ::= db_name(B) NK_DOT. { A = createValueNode(pCxt, TSDB_DATA_TYPE_BINARY, &B); }
......@@ -407,11 +409,11 @@ cmd ::= CREATE SMA INDEX not_exists_opt(D)
// index_name(A) ON table_name(B) NK_LP col_name_list(C) NK_RP. { pCxt->pRootNode = createCreateIndexStmt(pCxt, INDEX_TYPE_FULLTEXT, D, &A, &B, C, NULL); }
cmd ::= DROP INDEX exists_opt(B) index_name(A). { pCxt->pRootNode = createDropIndexStmt(pCxt, B, &A); }
index_options(A) ::= . { A = NULL; }
index_options(A) ::= FUNCTION NK_LP func_list(B) NK_RP INTERVAL
NK_LP duration_literal(C) NK_RP sliding_opt(D). { A = createIndexOption(pCxt, B, releaseRawExprNode(pCxt, C), NULL, D); }
NK_LP duration_literal(C) NK_RP sliding_opt(D) sma_stream_opt(E). { A = createIndexOption(pCxt, B, releaseRawExprNode(pCxt, C), NULL, D, E); }
index_options(A) ::= FUNCTION NK_LP func_list(B) NK_RP INTERVAL
NK_LP duration_literal(C) NK_COMMA duration_literal(D) NK_RP sliding_opt(E). { A = createIndexOption(pCxt, B, releaseRawExprNode(pCxt, C), releaseRawExprNode(pCxt, D), E); }
NK_LP duration_literal(C) NK_COMMA duration_literal(D) NK_RP sliding_opt(E)
sma_stream_opt(F). { A = createIndexOption(pCxt, B, releaseRawExprNode(pCxt, C), releaseRawExprNode(pCxt, D), E, F); }
%type func_list { SNodeList* }
%destructor func_list { nodesDestroyList($$); }
......@@ -420,6 +422,10 @@ func_list(A) ::= func_list(B) NK_COMMA func(C).
func(A) ::= function_name(B) NK_LP expression_list(C) NK_RP. { A = createFunctionNode(pCxt, &B, C); }
sma_stream_opt(A) ::= . { A = createStreamOptions(pCxt); }
sma_stream_opt(A) ::= stream_options(B) WATERMARK duration_literal(C). { ((SStreamOptions*)B)->pWatermark = releaseRawExprNode(pCxt, C); A = B; }
sma_stream_opt(A) ::= stream_options(B) MAX_DELAY duration_literal(C). { ((SStreamOptions*)B)->pDelay = releaseRawExprNode(pCxt, C); A = B; }
/************************************************ create/drop topic ***************************************************/
cmd ::= CREATE TOPIC not_exists_opt(A) topic_name(B) AS query_expression(C). { pCxt->pRootNode = createCreateTopicStmtUseQuery(pCxt, A, &B, C); }
cmd ::= CREATE TOPIC not_exists_opt(A) topic_name(B) AS DATABASE db_name(C). { pCxt->pRootNode = createCreateTopicStmtUseDb(pCxt, A, &B, &C, false); }
......
source/libs/parser/src/parAstCreater.c
浏览文件 @ 096c292d
......@@ -1317,8 +1317,8 @@ SNode* createCreateIndexStmt(SAstCreateContext* pCxt, EIndexType type, bool igno
return (SNode*)pStmt;
}
SNode* createIndexOption(SAstCreateContext* pCxt, SNodeList* pFuncs, SNode* pInterval, SNode* pOffset,
SNode* pSliding) {
SNode* createIndexOption(SAstCreateContext* pCxt, SNodeList* pFuncs, SNode* pInterval, SNode* pOffset, SNode* pSliding,
SNode* pStreamOptions) {
CHECK_PARSER_STATUS(pCxt);
SIndexOptions* pOptions = (SIndexOptions*)nodesMakeNode(QUERY_NODE_INDEX_OPTIONS);
CHECK_OUT_OF_MEM(pOptions);
......@@ -1326,6 +1326,7 @@ SNode* createIndexOption(SAstCreateContext* pCxt, SNodeList* pFuncs, SNode* pInt
pOptions->pInterval = pInterval;
pOptions->pOffset = pOffset;
pOptions->pSliding = pSliding;
pOptions->pStreamOptions = pStreamOptions;
return (SNode*)pOptions;
}
......
source/libs/parser/src/parAstParser.c
浏览文件 @ 096c292d
......@@ -380,6 +380,11 @@ static int32_t collectMetaKeyFromShowTopics(SCollectMetaKeyCxt* pCxt, SShowStmt*
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowConsumers(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_PERFORMANCE_SCHEMA_DB, TSDB_PERFS_TABLE_CONSUMERS,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowConnections(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_PERFORMANCE_SCHEMA_DB, TSDB_PERFS_TABLE_CONNECTIONS,
pCxt->pMetaCache);
......@@ -456,6 +461,11 @@ static int32_t collectMetaKeyFromShowBlockDist(SCollectMetaKeyCxt* pCxt, SShowTa
return code;
}
static int32_t collectMetaKeyFromShowSubscriptions(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_PERFORMANCE_SCHEMA_DB, TSDB_PERFS_TABLE_SUBSCRIPTIONS,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromQuery(SCollectMetaKeyCxt* pCxt, SNode* pStmt) {
pCxt->pStmt = pStmt;
switch (nodeType(pStmt)) {
......@@ -517,6 +527,8 @@ static int32_t collectMetaKeyFromQuery(SCollectMetaKeyCxt* pCxt, SNode* pStmt) {
return collectMetaKeyFromShowVgroups(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_TOPICS_STMT:
return collectMetaKeyFromShowTopics(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_CONSUMERS_STMT:
return collectMetaKeyFromShowConsumers(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_CONNECTIONS_STMT:
return collectMetaKeyFromShowConnections(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_QUERIES_STMT:
......@@ -538,6 +550,8 @@ static int32_t collectMetaKeyFromQuery(SCollectMetaKeyCxt* pCxt, SNode* pStmt) {
return collectMetaKeyFromDelete(pCxt, (SDeleteStmt*)pStmt);
case QUERY_NODE_SHOW_TABLE_DISTRIBUTED_STMT:
return collectMetaKeyFromShowBlockDist(pCxt, (SShowTableDistributedStmt*)pStmt);
case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
return collectMetaKeyFromShowSubscriptions(pCxt, (SShowStmt*)pStmt);
default:
break;
}
......
source/libs/parser/src/parCalcConst.c
浏览文件 @ 096c292d
......@@ -175,17 +175,23 @@ static int32_t calcConstProject(SNode* pProject, SNode** pNew) {
}
}
char aliasName[TSDB_COL_NAME_LEN] = {0};
strcpy(aliasName, ((SExprNode*)pProject)->aliasName);
int32_t code = scalarCalculateConstants(pProject, pNew);
if (TSDB_CODE_SUCCESS == code && QUERY_NODE_VALUE == nodeType(*pNew) && NULL != pAssociation) {
strcpy(((SExprNode*)*pNew)->aliasName, aliasName);
int32_t size = taosArrayGetSize(pAssociation);
for (int32_t i = 0; i < size; ++i) {
SNode** pCol = taosArrayGetP(pAssociation, i);
nodesDestroyNode(*pCol);
*pCol = nodesCloneNode(*pNew);
if (NULL == *pCol) {
return TSDB_CODE_OUT_OF_MEMORY;
code = TSDB_CODE_OUT_OF_MEMORY;
break;
}
}
}
taosArrayDestroy(pAssociation);
return code;
}
......@@ -196,6 +202,19 @@ static bool isUselessCol(bool hasSelectValFunc, SExprNode* pProj) {
return NULL == ((SExprNode*)pProj)->pAssociation;
}
static SNode* createConstantValue() {
SValueNode* pVal = (SValueNode*)nodesMakeNode(QUERY_NODE_VALUE);
if (NULL == pVal) {
return NULL;
}
pVal->node.resType.type = TSDB_DATA_TYPE_INT;
pVal->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_INT].bytes;
const int32_t val = 1;
nodesSetValueNodeValue(pVal, (void*)&val);
pVal->translate = true;
return (SNode*)pVal;
}
static int32_t calcConstProjections(SCalcConstContext* pCxt, SSelectStmt* pSelect, bool subquery) {
SNode* pProj = NULL;
WHERE_EACH(pProj, pSelect->pProjectionList) {
......@@ -212,6 +231,9 @@ static int32_t calcConstProjections(SCalcConstContext* pCxt, SSelectStmt* pSelec
}
WHERE_NEXT;
}
if (0 == LIST_LENGTH(pSelect->pProjectionList)) {
return nodesListStrictAppend(pSelect->pProjectionList, createConstantValue());
}
return TSDB_CODE_SUCCESS;
}
......
source/libs/parser/src/parTokenizer.c
浏览文件 @ 096c292d
......@@ -63,6 +63,7 @@ static SKeyword keywordTable[] = {
{"CONNECTION", TK_CONNECTION},
{"CONNECTIONS", TK_CONNECTIONS},
{"CONSUMER", TK_CONSUMER},
{"CONSUMERS", TK_CONSUMERS},
{"COUNT", TK_COUNT},
{"CREATE", TK_CREATE},
{"CONTAINS", TK_CONTAINS},
......@@ -188,6 +189,7 @@ static SKeyword keywordTable[] = {
{"STREAM", TK_STREAM},
{"STREAMS", TK_STREAMS},
{"STRICT", TK_STRICT},
{"SUBSCRIPTIONS", TK_SUBSCRIPTIONS},
{"SYNCDB", TK_SYNCDB},
{"SYSINFO", TK_SYSINFO},
{"TABLE", TK_TABLE},
......
source/libs/parser/src/parTranslater.c
浏览文件 @ 096c292d
......@@ -517,8 +517,9 @@ static int32_t createColumnsByTable(STranslateContext* pCxt, const STableNode* p
if (NULL == pCol) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_OUT_OF_MEMORY);
}
setColumnInfoByExpr(pTable, (SExprNode*)pNode, &pCol);
nodesListAppend(pList, (SNode*)pCol);
SListCell* pCell = nodesListGetCell(pList, LIST_LENGTH(pList) - 1);
setColumnInfoByExpr(pTable, (SExprNode*)pNode, (SColumnNode**)&pCell->pNode);
}
}
return TSDB_CODE_SUCCESS;
......@@ -707,7 +708,11 @@ static EDealRes translateColumn(STranslateContext* pCxt, SColumnNode** pCol) {
res = translateColumnUseAlias(pCxt, pCol, &found);
}
if (DEAL_RES_ERROR != res && !found) {
res = translateColumnWithoutPrefix(pCxt, pCol);
if (NULL != pCxt->pCurrSetOperator) {
res = generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_COLUMN, (*pCol)->colName);
} else {
res = translateColumnWithoutPrefix(pCxt, pCol);
}
}
}
return res;
......@@ -2656,6 +2661,13 @@ static int32_t translateSetOperOrderBy(STranslateContext* pCxt, SSetOperator* pS
return code;
}
static int32_t checkSetOperLimit(STranslateContext* pCxt, SLimitNode* pLimit) {
if ((NULL != pLimit && pLimit->offset < 0)) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_OFFSET_LESS_ZERO);
}
return TSDB_CODE_SUCCESS;
}
static int32_t translateSetOperator(STranslateContext* pCxt, SSetOperator* pSetOperator) {
int32_t code = translateQuery(pCxt, pSetOperator->pLeft);
if (TSDB_CODE_SUCCESS == code) {
......@@ -2671,6 +2683,9 @@ static int32_t translateSetOperator(STranslateContext* pCxt, SSetOperator* pSetO
if (TSDB_CODE_SUCCESS == code) {
code = translateSetOperOrderBy(pCxt, pSetOperator);
}
if (TSDB_CODE_SUCCESS == code) {
code = checkSetOperLimit(pCxt, (SLimitNode*)pSetOperator->pLimit);
}
return code;
}
......@@ -3566,9 +3581,7 @@ static int32_t buildRollupFuncs(SNodeList* pFuncs, SArray** pArray) {
}
*pArray = taosArrayInit(LIST_LENGTH(pFuncs), TSDB_FUNC_NAME_LEN);
SNode* pNode;
FOREACH(pNode, pFuncs) {
taosArrayPush(*pArray, ((SFunctionNode*)pNode)->functionName);
}
FOREACH(pNode, pFuncs) { taosArrayPush(*pArray, ((SFunctionNode*)pNode)->functionName); }
return TSDB_CODE_SUCCESS;
}
......@@ -3867,6 +3880,11 @@ static int32_t buildCreateSmaReq(STranslateContext* pCxt, SCreateIndexStmt* pStm
(NULL != pStmt->pOptions->pSliding ? ((SValueNode*)pStmt->pOptions->pSliding)->datum.i : pReq->interval);
pReq->slidingUnit =
(NULL != pStmt->pOptions->pSliding ? ((SValueNode*)pStmt->pOptions->pSliding)->unit : pReq->intervalUnit);
if (NULL != pStmt->pOptions->pStreamOptions) {
SStreamOptions* pStreamOpt = (SStreamOptions*)pStmt->pOptions->pStreamOptions;
pReq->maxDelay = (NULL != pStreamOpt->pDelay ? ((SValueNode*)pStreamOpt->pDelay)->datum.i : 0);
pReq->watermark = (NULL != pStreamOpt->pWatermark ? ((SValueNode*)pStreamOpt->pWatermark)->datum.i : 0);
}
int32_t code = getSmaIndexDstVgId(pCxt, pStmt->tableName, &pReq->dstVgId);
if (TSDB_CODE_SUCCESS == code) {
......@@ -3888,6 +3906,18 @@ static int32_t translateCreateSmaIndex(STranslateContext* pCxt, SCreateIndexStmt
return pCxt->errCode;
}
if (NULL != pStmt->pOptions->pStreamOptions) {
SStreamOptions* pStreamOpt = (SStreamOptions*)pStmt->pOptions->pStreamOptions;
if (NULL != pStreamOpt->pWatermark &&
(DEAL_RES_ERROR == translateValue(pCxt, (SValueNode*)pStreamOpt->pWatermark))) {
return pCxt->errCode;
}
if (NULL != pStreamOpt->pDelay && (DEAL_RES_ERROR == translateValue(pCxt, (SValueNode*)pStreamOpt->pDelay))) {
return pCxt->errCode;
}
}
SMCreateSmaReq createSmaReq = {0};
int32_t code = buildCreateSmaReq(pCxt, pStmt, &createSmaReq);
if (TSDB_CODE_SUCCESS == code) {
......@@ -4629,7 +4659,6 @@ static int32_t extractShowLocalVariablesResultSchema(int32_t* numOfCols, SSchema
return TSDB_CODE_SUCCESS;
}
int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema) {
if (NULL == pRoot) {
return TSDB_CODE_SUCCESS;
......@@ -4683,6 +4712,8 @@ static const char* getSysDbName(ENodeType type) {
case QUERY_NODE_SHOW_STREAMS_STMT:
case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
case QUERY_NODE_SHOW_APPS_STMT:
case QUERY_NODE_SHOW_CONSUMERS_STMT:
case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
return TSDB_PERFORMANCE_SCHEMA_DB;
default:
break;
......@@ -4738,6 +4769,10 @@ static const char* getSysTableName(ENodeType type) {
return TSDB_PERFS_TABLE_APPS;
case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
return TSDB_INS_TABLE_DNODE_VARIABLES;
case QUERY_NODE_SHOW_CONSUMERS_STMT:
return TSDB_PERFS_TABLE_CONSUMERS;
case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
return TSDB_PERFS_TABLE_SUBSCRIPTIONS;
default:
break;
}
......@@ -5876,6 +5911,8 @@ static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
case QUERY_NODE_SHOW_VARIABLES_STMT:
case QUERY_NODE_SHOW_APPS_STMT:
case QUERY_NODE_SHOW_CONSUMERS_STMT:
case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
code = rewriteShow(pCxt, pQuery);
break;
case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
......
source/libs/parser/src/parUtil.c
浏览文件 @ 096c292d
......@@ -897,7 +897,7 @@ STableCfg* tableCfgDup(STableCfg* pCfg) {
if (pNew->pFuncs) {
pNew->pFuncs = taosArrayDup(pNew->pFuncs);
}
int32_t schemaSize = (pCfg->numOfColumns + pCfg->numOfTags) * sizeof(SSchema);
SSchema* pSchema = taosMemoryMalloc(schemaSize);
......@@ -912,7 +912,7 @@ int32_t getTableCfgFromCache(SParseMetaCache* pMetaCache, const SName* pName, ST
char fullName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pName, fullName);
STableCfg* pCfg = NULL;
int32_t code = getMetaDataFromHash(fullName, strlen(fullName), pMetaCache->pTableCfg, (void**)&pCfg);
int32_t code = getMetaDataFromHash(fullName, strlen(fullName), pMetaCache->pTableCfg, (void**)&pCfg);
if (TSDB_CODE_SUCCESS == code) {
*pOutput = tableCfgDup(pCfg);
if (NULL == *pOutput) {
......@@ -929,10 +929,10 @@ int32_t reserveDnodeRequiredInCache(SParseMetaCache* pMetaCache) {
int32_t getDnodeListFromCache(SParseMetaCache* pMetaCache, SArray** pDnodes) {
SMetaRes* pRes = taosArrayGet(pMetaCache->pDnodes, 0);
if (pRes->code) {
if (TSDB_CODE_SUCCESS != pRes->code) {
return pRes->code;
}
*pDnodes = taosArrayDup((SArray*)pRes->pRes);
if (NULL == *pDnodes) {
return TSDB_CODE_OUT_OF_MEMORY;
......
source/libs/parser/src/sql.c
浏览文件 @ 096c292d
此差异已折叠。
source/libs/parser/test/mockCatalog.cpp
浏览文件 @ 096c292d
......@@ -25,95 +25,125 @@
#ifdef WINDOWS
#define TD_USE_WINSOCK
#endif
#include "mockCatalog.h"
#include "systable.h"
namespace {
void generateInformationSchema(MockCatalogService* mcs) {
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "dnodes", TSDB_SYSTEM_TABLE, 1)
.addColumn("id", TSDB_DATA_TYPE_INT);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_DNODES, TSDB_SYSTEM_TABLE, 1)
.addColumn("id", TSDB_DATA_TYPE_INT);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "mnodes", TSDB_SYSTEM_TABLE, 1)
.addColumn("id", TSDB_DATA_TYPE_INT);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_MNODES, TSDB_SYSTEM_TABLE, 1)
.addColumn("id", TSDB_DATA_TYPE_INT);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "modules", TSDB_SYSTEM_TABLE, 1)
.addColumn("id", TSDB_DATA_TYPE_INT);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_MODULES, TSDB_SYSTEM_TABLE, 1)
.addColumn("id", TSDB_DATA_TYPE_INT);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "qnodes", TSDB_SYSTEM_TABLE, 1)
.addColumn("id", TSDB_DATA_TYPE_INT);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_QNODES, TSDB_SYSTEM_TABLE, 1)
.addColumn("id", TSDB_DATA_TYPE_INT);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "user_databases", TSDB_SYSTEM_TABLE, 1)
.addColumn("name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_DATABASES, TSDB_SYSTEM_TABLE, 1)
.addColumn("name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "user_functions", TSDB_SYSTEM_TABLE, 1)
.addColumn("name", TSDB_DATA_TYPE_BINARY, TSDB_FUNC_NAME_LEN);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_FUNCTIONS, TSDB_SYSTEM_TABLE, 1)
.addColumn("name", TSDB_DATA_TYPE_BINARY, TSDB_FUNC_NAME_LEN);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "user_indexes", TSDB_SYSTEM_TABLE, 2)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
.addColumn("table_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_INDEXES, TSDB_SYSTEM_TABLE, 2)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
.addColumn("table_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "user_stables", TSDB_SYSTEM_TABLE, 2)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
.addColumn("stable_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_STABLES, TSDB_SYSTEM_TABLE, 2)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
.addColumn("stable_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "user_tables", TSDB_SYSTEM_TABLE, 2)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
.addColumn("table_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_TABLES, TSDB_SYSTEM_TABLE, 2)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
.addColumn("table_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN);
builder.done();
}
{
ITableBuilder& builder =
mcs->createTableBuilder("information_schema", "user_table_distributed", TSDB_SYSTEM_TABLE, 1)
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_TABLE_DISTRIBUTED, TSDB_SYSTEM_TABLE, 1)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "user_users", TSDB_SYSTEM_TABLE, 1)
.addColumn("user_name", TSDB_DATA_TYPE_BINARY, TSDB_USER_LEN);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_USERS, TSDB_SYSTEM_TABLE, 1)
.addColumn("user_name", TSDB_DATA_TYPE_BINARY, TSDB_USER_LEN);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "vgroups", TSDB_SYSTEM_TABLE, 1)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_VGROUPS, TSDB_SYSTEM_TABLE, 1)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "configs", TSDB_SYSTEM_TABLE, 1)
.addColumn("name", TSDB_DATA_TYPE_BINARY, TSDB_CONFIG_OPTION_LEN);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_CONFIGS, TSDB_SYSTEM_TABLE, 1)
.addColumn("name", TSDB_DATA_TYPE_BINARY, TSDB_CONFIG_OPTION_LEN);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("information_schema", "dnode_variables", TSDB_SYSTEM_TABLE, 1)
.addColumn("dnode_id", TSDB_DATA_TYPE_INT);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_DNODE_VARIABLES, TSDB_SYSTEM_TABLE, 1)
.addColumn("dnode_id", TSDB_DATA_TYPE_INT);
builder.done();
}
}
void generatePerformanceSchema(MockCatalogService* mcs) {
{
ITableBuilder& builder = mcs->createTableBuilder("performance_schema", "trans", TSDB_SYSTEM_TABLE, 1)
.addColumn("id", TSDB_DATA_TYPE_INT);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_PERFORMANCE_SCHEMA_DB, TSDB_PERFS_TABLE_TRANS, TSDB_SYSTEM_TABLE, 1)
.addColumn("id", TSDB_DATA_TYPE_INT);
builder.done();
}
{
ITableBuilder& builder = mcs->createTableBuilder("performance_schema", "streams", TSDB_SYSTEM_TABLE, 1)
.addColumn("stream_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN);
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_PERFORMANCE_SCHEMA_DB, TSDB_PERFS_TABLE_STREAMS, TSDB_SYSTEM_TABLE, 1)
.addColumn("stream_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN);
builder.done();
}
{
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_PERFORMANCE_SCHEMA_DB, TSDB_PERFS_TABLE_CONSUMERS, TSDB_SYSTEM_TABLE, 1)
.addColumn("stream_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN);
builder.done();
}
{
ITableBuilder& builder =
mcs->createTableBuilder(TSDB_PERFORMANCE_SCHEMA_DB, TSDB_PERFS_TABLE_SUBSCRIPTIONS, TSDB_SYSTEM_TABLE, 1)
.addColumn("stream_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN);
builder.done();
}
}
......@@ -261,7 +291,8 @@ int32_t __catalogGetDnodeList(SCatalog* pCatalog, SRequestConnInfo* pConn, SArra
return g_mockCatalogService->catalogGetDnodeList(pDnodeList);
}
int32_t __catalogRefreshGetTableCfg(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableCfg** pCfg) {
int32_t __catalogRefreshGetTableCfg(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName,
STableCfg** pCfg) {
*pCfg = (STableCfg*)taosMemoryCalloc(1, sizeof(STableCfg));
return 0;
}
......
source/libs/parser/test/mockCatalogService.cpp
浏览文件 @ 096c292d
......@@ -166,13 +166,10 @@ class MockCatalogServiceImpl {
}
int32_t catalogGetDnodeList(SArray** pDnodes) const {
SMetaRes res = {0};
res.pRes = taosArrayInit(dnode_.size(), sizeof(SEpSet));
*pDnodes = taosArrayInit(dnode_.size(), sizeof(SEpSet));
for (const auto& dnode : dnode_) {
taosArrayPush((SArray*)res.pRes, &dnode.second);
taosArrayPush(*pDnodes, &dnode.second);
}
*pDnodes = taosArrayInit(1, sizeof(SMetaRes));
taosArrayPush(*pDnodes, &res);
return TSDB_CODE_SUCCESS;
}
......@@ -200,7 +197,7 @@ class MockCatalogServiceImpl {
code = getAllTableIndex(pCatalogReq->pTableIndex, &pMetaData->pTableIndex);
}
if (TSDB_CODE_SUCCESS == code && pCatalogReq->dNodeRequired) {
code = catalogGetDnodeList(&pMetaData->pDnodeList);
code = getAllDnodeList(&pMetaData->pDnodeList);
}
if (TSDB_CODE_SUCCESS == code) {
code = getAllTableCfg(pCatalogReq->pTableCfg, &pMetaData->pTableCfg);
......@@ -565,6 +562,14 @@ class MockCatalogServiceImpl {
return TSDB_CODE_SUCCESS;
}
int32_t getAllDnodeList(SArray** pDnodes) const {
SMetaRes res = {0};
catalogGetDnodeList((SArray**)&res.pRes);
*pDnodes = taosArrayInit(1, sizeof(SMetaRes));
taosArrayPush(*pDnodes, &res);
return TSDB_CODE_SUCCESS;
}
uint64_t id_;
std::unique_ptr<TableBuilder> builder_;
DbMetaCache meta_;
......
source/libs/parser/test/parSelectTest.cpp
浏览文件 @ 096c292d
......@@ -144,15 +144,17 @@ TEST_F(ParserSelectTest, IndefiniteRowsFunc) {
TEST_F(ParserSelectTest, IndefiniteRowsFuncSemanticCheck) {
useDb("root", "test");
run("SELECT DIFF(c1), c2 FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC, PARSER_STAGE_TRANSLATE);
run("SELECT DIFF(c1), c2 FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
run("SELECT DIFF(c1), tbname FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC, PARSER_STAGE_TRANSLATE);
run("SELECT DIFF(c1), tbname FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
run("SELECT DIFF(c1), count(*) FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC, PARSER_STAGE_TRANSLATE);
run("SELECT DIFF(c1), count(*) FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
run("SELECT DIFF(c1), CSUM(c1) FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC, PARSER_STAGE_TRANSLATE);
run("SELECT DIFF(c1), CSUM(c1) FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
// run("SELECT DIFF(c1) FROM t1 INTERVAL(10s)");
run("SELECT CSUM(c3) FROM t1 STATE_WINDOW(c1)", TSDB_CODE_PAR_WINDOW_NOT_ALLOWED_FUNC);
run("SELECT DIFF(c1) FROM t1 INTERVAL(10s)", TSDB_CODE_PAR_WINDOW_NOT_ALLOWED_FUNC);
}
TEST_F(ParserSelectTest, useDefinedFunc) {
......@@ -384,6 +386,12 @@ TEST_F(ParserSelectTest, setOperator) {
run("SELECT c1, c2 FROM t1 UNION ALL SELECT c1 as a, c2 as b FROM t1 ORDER BY c1");
}
TEST_F(ParserSelectTest, setOperatorSemanticCheck) {
useDb("root", "test");
run("SELECT c1, c2 FROM t1 UNION ALL SELECT c1, c2 FROM t1 ORDER BY ts", TSDB_CODE_PAR_INVALID_COLUMN);
}
TEST_F(ParserSelectTest, informationSchema) {
useDb("root", "test");
......
source/libs/parser/test/parShowToUse.cpp
浏览文件 @ 096c292d
......@@ -25,6 +25,15 @@ class ParserShowToUseTest : public ParserDdlTest {};
// todo SHOW apps
// todo SHOW connections
TEST_F(ParserShowToUseTest, showConsumers) {
useDb("root", "test");
setCheckDdlFunc(
[&](const SQuery* pQuery, ParserStage stage) { ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_SELECT_STMT); });
run("SHOW CONSUMERS");
}
TEST_F(ParserShowToUseTest, showCreateDatabase) {
useDb("root", "test");
......@@ -143,6 +152,15 @@ TEST_F(ParserShowToUseTest, showStreams) {
run("SHOW streams");
}
TEST_F(ParserShowToUseTest, showSubscriptions) {
useDb("root", "test");
setCheckDdlFunc(
[&](const SQuery* pQuery, ParserStage stage) { ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_SELECT_STMT); });
run("SHOW SUBSCRIPTIONS");
}
TEST_F(ParserShowToUseTest, showTransactions) {
useDb("root", "test");
......
source/libs/planner/src/planLogicCreater.c
浏览文件 @ 096c292d
......@@ -19,6 +19,7 @@
typedef struct SLogicPlanContext {
SPlanContext* pPlanCxt;
SLogicNode* pCurrRoot;
} SLogicPlanContext;
typedef int32_t (*FCreateLogicNode)(SLogicPlanContext*, void*, SLogicNode**);
......@@ -140,6 +141,7 @@ static int32_t createRootLogicNode(SLogicPlanContext* pCxt, void* pStmt, uint8_t
if (TSDB_CODE_SUCCESS == code && NULL != pNode) {
pNode->precision = precision;
code = pushLogicNode(pCxt, pRoot, pNode);
pCxt->pCurrRoot = pNode;
}
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode((SNode*)pNode);
......@@ -152,12 +154,16 @@ static int32_t createSelectRootLogicNode(SLogicPlanContext* pCxt, SSelectStmt* p
return createRootLogicNode(pCxt, pSelect, pSelect->precision, (FCreateLogicNode)func, pRoot);
}
static EScanType getScanType(SLogicPlanContext* pCxt, SNodeList* pScanPseudoCols, SNodeList* pScanCols,
int8_t tableType) {
static EScanType getScanType(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SNodeList* pScanPseudoCols,
SNodeList* pScanCols, int8_t tableType) {
if (pCxt->pPlanCxt->topicQuery || pCxt->pPlanCxt->streamQuery) {
return SCAN_TYPE_STREAM;
}
if (pSelect->hasLastRowFunc) {
return SCAN_TYPE_LAST_ROW;
}
if (NULL == pScanCols) {
// select count(*) from t
return NULL == pScanPseudoCols
......@@ -273,7 +279,7 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
code = rewriteExprsForSelect(pScan->pScanPseudoCols, pSelect, SQL_CLAUSE_FROM);
}
pScan->scanType = getScanType(pCxt, pScan->pScanPseudoCols, pScan->pScanCols, pScan->tableType);
pScan->scanType = getScanType(pCxt, pSelect, pScan->pScanPseudoCols, pScan->pScanCols, pScan->tableType);
if (TSDB_CODE_SUCCESS == code) {
code = addPrimaryKeyCol(pScan->tableId, &pScan->pScanCols);
......@@ -319,36 +325,6 @@ static int32_t createColumnByLastRow(SNodeList* pFuncs, SNodeList** pOutput) {
return code;
}
static int32_t createLastRowScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SRealTableNode* pRealTable,
SLogicNode** pLogicNode) {
SScanLogicNode* pScan = NULL;
int32_t code = makeScanLogicNode(pCxt, pRealTable, false, (SLogicNode**)&pScan);
SNodeList* pFuncs = NULL;
if (TSDB_CODE_SUCCESS == code) {
pScan->scanType = SCAN_TYPE_LAST_ROW;
code = nodesCollectFuncs(pSelect, SQL_CLAUSE_FROM, fmIsLastRowFunc, &pFuncs);
}
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pFuncs, pSelect, SQL_CLAUSE_FROM);
}
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByLastRow(pFuncs, &pScan->pScanCols);
}
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByRewriteExprs(pScan->pScanCols, &pScan->node.pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pScan;
} else {
nodesDestroyNode((SNode*)pScan);
}
nodesDestroyList(pFuncs);
return code;
}
static int32_t createSubqueryLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, STempTableNode* pTable,
SLogicNode** pLogicNode) {
return createQueryLogicNode(pCxt, pTable->pSubquery, pLogicNode);
......@@ -420,11 +396,7 @@ static int32_t doCreateLogicNodeByTable(SLogicPlanContext* pCxt, SSelectStmt* pS
SLogicNode** pLogicNode) {
switch (nodeType(pTable)) {
case QUERY_NODE_REAL_TABLE:
if (pSelect->hasLastRowFunc) {
return createLastRowScanLogicNode(pCxt, pSelect, (SRealTableNode*)pTable, pLogicNode);
} else {
return createScanLogicNode(pCxt, pSelect, (SRealTableNode*)pTable, pLogicNode);
}
return createScanLogicNode(pCxt, pSelect, (SRealTableNode*)pTable, pLogicNode);
case QUERY_NODE_TEMP_TABLE:
return createSubqueryLogicNode(pCxt, pSelect, (STempTableNode*)pTable, pLogicNode);
case QUERY_NODE_JOIN_TABLE:
......@@ -447,6 +419,7 @@ static int32_t createLogicNodeByTable(SLogicPlanContext* pCxt, SSelectStmt* pSel
}
pNode->precision = pSelect->precision;
*pLogicNode = pNode;
pCxt->pCurrRoot = pNode;
}
return code;
}
......@@ -728,6 +701,10 @@ static int32_t createFillLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
}
int32_t code = nodesCollectColumns(pSelect, SQL_CLAUSE_WINDOW, NULL, COLLECT_COL_TYPE_ALL, &pFill->node.pTargets);
if (TSDB_CODE_SUCCESS == code && NULL == pFill->node.pTargets) {
code = nodesListMakeStrictAppend(&pFill->node.pTargets,
nodesCloneNode(nodesListGetNode(pCxt->pCurrRoot->pTargets, 0)));
}
pFill->mode = pFillNode->mode;
pFill->timeRange = pFillNode->timeRange;
......@@ -757,6 +734,10 @@ static int32_t createSortLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
}
int32_t code = nodesCollectColumns(pSelect, SQL_CLAUSE_ORDER_BY, NULL, COLLECT_COL_TYPE_ALL, &pSort->node.pTargets);
if (TSDB_CODE_SUCCESS == code && NULL == pSort->node.pTargets) {
code = nodesListMakeStrictAppend(&pSort->node.pTargets,
nodesCloneNode(nodesListGetNode(pCxt->pCurrRoot->pTargets, 0)));
}
if (TSDB_CODE_SUCCESS == code) {
pSort->pSortKeys = nodesCloneList(pSelect->pOrderByList);
......@@ -799,21 +780,8 @@ static int32_t createProjectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSel
return TSDB_CODE_OUT_OF_MEMORY;
}
if (NULL != pSelect->pLimit) {
pProject->limit = pSelect->pLimit->limit;
pProject->offset = pSelect->pLimit->offset;
} else {
pProject->limit = -1;
pProject->offset = -1;
}
if (NULL != pSelect->pSlimit) {
pProject->slimit = ((SLimitNode*)pSelect->pSlimit)->limit;
pProject->soffset = ((SLimitNode*)pSelect->pSlimit)->offset;
} else {
pProject->slimit = -1;
pProject->soffset = -1;
}
pProject->node.pLimit = (SNode*)pSelect->pLimit;
pProject->node.pSlimit = (SNode*)pSelect->pSlimit;
int32_t code = TSDB_CODE_SUCCESS;
......@@ -848,6 +816,9 @@ static int32_t createPartitionLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pS
int32_t code =
nodesCollectColumns(pSelect, SQL_CLAUSE_PARTITION_BY, NULL, COLLECT_COL_TYPE_ALL, &pPartition->node.pTargets);
if (TSDB_CODE_SUCCESS == code && NULL == pPartition->node.pTargets) {
code = nodesListMakeStrictAppend(&pPartition->node.pTargets, nodesListGetNode(pCxt->pCurrRoot->pTargets, 0));
}
if (TSDB_CODE_SUCCESS == code) {
pPartition->pPartitionKeys = nodesCloneList(pSelect->pPartitionByList);
......@@ -969,6 +940,8 @@ static int32_t createSetOpSortLogicNode(SLogicPlanContext* pCxt, SSetOperator* p
return TSDB_CODE_OUT_OF_MEMORY;
}
pSort->node.pLimit = pSetOperator->pLimit;
int32_t code = TSDB_CODE_SUCCESS;
pSort->node.pTargets = nodesCloneList(pSetOperator->pProjectionList);
......@@ -999,12 +972,8 @@ static int32_t createSetOpProjectLogicNode(SLogicPlanContext* pCxt, SSetOperator
return TSDB_CODE_OUT_OF_MEMORY;
}
if (NULL != pSetOperator->pLimit) {
pProject->limit = ((SLimitNode*)pSetOperator->pLimit)->limit;
pProject->offset = ((SLimitNode*)pSetOperator->pLimit)->offset;
} else {
pProject->limit = -1;
pProject->offset = -1;
if (NULL == pSetOperator->pOrderByList) {
pProject->node.pLimit = pSetOperator->pLimit;
}
int32_t code = TSDB_CODE_SUCCESS;
......@@ -1034,6 +1003,10 @@ static int32_t createSetOpAggLogicNode(SLogicPlanContext* pCxt, SSetOperator* pS
return TSDB_CODE_OUT_OF_MEMORY;
}
if (NULL == pSetOperator->pOrderByList) {
pAgg->node.pLimit = pSetOperator->pLimit;
}
int32_t code = TSDB_CODE_SUCCESS;
pAgg->pGroupKeys = nodesCloneList(pSetOperator->pProjectionList);
if (NULL == pAgg->pGroupKeys) {
......
source/libs/planner/src/planOptimizer.c
浏览文件 @ 096c292d
......@@ -1117,8 +1117,7 @@ static bool eliminateProjOptMayBeOptimized(SLogicNode* pNode) {
}
SProjectLogicNode* pProjectNode = (SProjectLogicNode*)pNode;
if (-1 != pProjectNode->limit || -1 != pProjectNode->slimit || -1 != pProjectNode->offset ||
-1 != pProjectNode->soffset) {
if (NULL != pProjectNode->node.pLimit || NULL != pProjectNode->node.pSlimit) {
return false;
}
......
source/libs/planner/src/planPhysiCreater.c
浏览文件 @ 096c292d
......@@ -348,6 +348,9 @@ static SPhysiNode* makePhysiNode(SPhysiPlanContext* pCxt, SLogicNode* pLogicNode
return NULL;
}
pPhysiNode->pLimit = pLogicNode->pLimit;
pPhysiNode->pSlimit = pLogicNode->pSlimit;
int32_t code = createDataBlockDesc(pCxt, pLogicNode->pTargets, &pPhysiNode->pOutputDataBlockDesc);
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode((SNode*)pPhysiNode);
......@@ -921,11 +924,6 @@ static int32_t createProjectPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChild
return TSDB_CODE_OUT_OF_MEMORY;
}
pProject->limit = pProjectLogicNode->limit;
pProject->offset = pProjectLogicNode->offset;
pProject->slimit = pProjectLogicNode->slimit;
pProject->soffset = pProjectLogicNode->soffset;
int32_t code = TSDB_CODE_SUCCESS;
if (0 == LIST_LENGTH(pChildren)) {
pProject->pProjections = nodesCloneList(pProjectLogicNode->pProjections);
......@@ -1199,6 +1197,9 @@ static int32_t createSortPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
nodesDestroyNode((SNode*)pSort);
}
nodesDestroyList(pPrecalcExprs);
nodesDestroyList(pSortKeys);
return code;
}
......
source/libs/planner/src/planSpliter.c
浏览文件 @ 096c292d
......@@ -1040,6 +1040,7 @@ static int32_t unAllSplCreateExchangeNode(SSplitContext* pCxt, SLogicSubplan* pS
if (NULL == pExchange->node.pTargets) {
return TSDB_CODE_OUT_OF_MEMORY;
}
TSWAP(pExchange->node.pLimit, pProject->node.pLimit);
pSubplan->subplanType = SUBPLAN_TYPE_MERGE;
......
source/libs/planner/test/planBasicTest.cpp
浏览文件 @ 096c292d
......@@ -37,13 +37,6 @@ TEST_F(PlanBasicTest, whereClause) {
run("SELECT * FROM t1 WHERE ts > TIMESTAMP '2022-04-01 00:00:00' and ts < TIMESTAMP '2022-04-30 23:59:59'");
}
TEST_F(PlanBasicTest, joinClause) {
useDb("root", "test");
run("SELECT t1.c1, t2.c2 FROM st1s1 t1, st1s2 t2 WHERE t1.ts = t2.ts");
run("SELECT t1.c1, t2.c2 FROM st1s1 t1 JOIN st1s2 t2 ON t1.ts = t2.ts");
}
TEST_F(PlanBasicTest, func) {
useDb("root", "test");
......
source/libs/planner/test/planOrderByTest.cpp
浏览文件 @ 096c292d
......@@ -31,6 +31,8 @@ TEST_F(PlanOrderByTest, basic) {
run("SELECT c1 AS a FROM t1 ORDER BY a");
run("SELECT c1 + 10 AS a FROM t1 ORDER BY a");
run("SELECT 1 FROM t1 ORDER BY c1");
}
TEST_F(PlanOrderByTest, expr) {
......
source/libs/planner/test/planSetOpTest.cpp
浏览文件 @ 096c292d
......@@ -54,6 +54,14 @@ TEST_F(PlanSetOpTest, unionAllWithOrderBy) {
run("SELECT c1, c2 FROM t1 WHERE c1 > 10 UNION ALL SELECT c1, c2 FROM t1 WHERE c1 > 20 ORDER BY 1");
}
TEST_F(PlanSetOpTest, unionAllWithLimit) {
useDb("root", "test");
run("SELECT c1, c2 FROM t1 WHERE c1 > 10 UNION ALL SELECT c1, c2 FROM t1 WHERE c1 > 20 LIMIT 10, 20");
run("SELECT c1, c2 FROM t1 WHERE c1 > 10 UNION ALL SELECT c1, c2 FROM t1 WHERE c1 > 20 ORDER BY 1 LIMIT 10, 20");
}
TEST_F(PlanSetOpTest, union) {
useDb("root", "test");
......@@ -65,6 +73,14 @@ TEST_F(PlanSetOpTest, union) {
"UNION SELECT c1, c2 FROM t1 WHERE c1 > 30");
}
TEST_F(PlanSetOpTest, unionWithLimit) {
useDb("root", "test");
run("SELECT c1, c2 FROM t1 WHERE c1 > 10 UNION SELECT c1, c2 FROM t1 WHERE c1 > 20 LIMIT 10, 20");
run("SELECT c1, c2 FROM t1 WHERE c1 > 10 UNION SELECT c1, c2 FROM t1 WHERE c1 > 20 ORDER BY 1 LIMIT 10, 20");
}
TEST_F(PlanSetOpTest, unionContainJoin) {
useDb("root", "test");
......
source/libs/planner/test/planSubqueryTest.cpp
浏览文件 @ 096c292d
......@@ -33,7 +33,9 @@ TEST_F(PlanSubqeuryTest, basic) {
run("SELECT NOW() FROM (SELECT * FROM t1)");
// run("SELECT NOW() FROM (SELECT * FROM t1) ORDER BY ts");
run("SELECT * FROM (SELECT NOW() FROM t1)");
run("SELECT NOW() FROM (SELECT * FROM t1) ORDER BY ts");
}
TEST_F(PlanSubqeuryTest, doubleGroupBy) {
......
source/libs/planner/test/planTestMain.cpp
浏览文件 @ 096c292d
......@@ -16,9 +16,11 @@
#include <string>
#include <gtest/gtest.h>
#include "functionMgt.h"
#include "getopt.h"
#include "mockCatalog.h"
#include "parser.h"
#include "planTestUtil.h"
class PlannerEnv : public testing::Environment {
......@@ -30,7 +32,12 @@ class PlannerEnv : public testing::Environment {
initLog(TD_TMP_DIR_PATH "td");
}
virtual void TearDown() { destroyMetaDataEnv(); }
virtual void TearDown() {
destroyMetaDataEnv();
qCleanupKeywordsTable();
fmFuncMgtDestroy();
taosCloseLog();
}
PlannerEnv() {}
virtual ~PlannerEnv() {}
......
source/libs/planner/test/planTestUtil.cpp
浏览文件 @ 096c292d
......@@ -102,12 +102,15 @@ class PlannerTestBaseImpl {
try {
SQuery* pQuery = nullptr;
doParseSql(sql, &pQuery);
unique_ptr<SQuery, void (*)(SQuery*)> query(pQuery, qDestroyQuery);
SPlanContext cxt = {0};
setPlanContext(pQuery, &cxt);
SLogicSubplan* pLogicSubplan = nullptr;
doCreateLogicPlan(&cxt, &pLogicSubplan);
unique_ptr<SLogicSubplan, void (*)(SLogicSubplan*)> logicSubplan(pLogicSubplan,
(void (*)(SLogicSubplan*))nodesDestroyNode);
doOptimizeLogicPlan(&cxt, pLogicSubplan);
......@@ -115,9 +118,12 @@ class PlannerTestBaseImpl {
SQueryLogicPlan* pLogicPlan = nullptr;
doScaleOutLogicPlan(&cxt, pLogicSubplan, &pLogicPlan);
unique_ptr<SQueryLogicPlan, void (*)(SQueryLogicPlan*)> logicPlan(pLogicPlan,
(void (*)(SQueryLogicPlan*))nodesDestroyNode);
SQueryPlan* pPlan = nullptr;
doCreatePhysiPlan(&cxt, pLogicPlan, &pPlan);
unique_ptr<SQueryPlan, void (*)(SQueryPlan*)> plan(pPlan, (void (*)(SQueryPlan*))nodesDestroyNode);
dump(g_dumpModule);
} catch (...) {
......@@ -345,8 +351,9 @@ class PlannerTestBaseImpl {
}
void doCreatePhysiPlan(SPlanContext* pCxt, SQueryLogicPlan* pLogicPlan, SQueryPlan** pPlan) {
SArray* pExecNodeList = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SQueryNodeAddr));
DO_WITH_THROW(createPhysiPlan, pCxt, pLogicPlan, pPlan, pExecNodeList);
unique_ptr<SArray, void (*)(SArray*)> execNodeList((SArray*)taosArrayInit(TARRAY_MIN_SIZE, sizeof(SQueryNodeAddr)),
(void (*)(SArray*))taosArrayDestroy);
DO_WITH_THROW(createPhysiPlan, pCxt, pLogicPlan, pPlan, execNodeList.get());
res_.physiPlan_ = toString((SNode*)(*pPlan));
SNode* pNode;
FOREACH(pNode, (*pPlan)->pSubplans) {
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或