xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #11660 from taosdata/feature/3.0_wxy 

feat: sql command 'create table ... rollup ...'
This commit is contained in:
Xiaoyu Wang 2022-04-20 14:21:50 +08:00 committed by GitHub
parent a44725f178 5dea946bd4
commit d0b9b90781
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 3201 additions and 2824 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/common/tmsg.h
View File

@ -280,10 +280,14 @@ typedef struct {
int32_t numOfTags;
int32_t numOfSmas;
int32_t commentLen;
int32_t ast1Len;
int32_t ast2Len;
SArray* pColumns; // array of SField
SArray* pTags; // array of SField
SArray* pSmas; // array of SField
char* comment;
char* pAst1;
char* pAst2;
} SMCreateStbReq;
int32_t tSerializeSMCreateStbReq(void* buf, int32_t bufLen, SMCreateStbReq* pReq);
@ -609,6 +613,8 @@ typedef struct {
int8_t cacheLastRow;
int8_t streamMode;
int8_t singleSTable;
int32_t numOfRetensions;
SArray* pRetensions;
} SDbCfgRsp;
int32_t tSerializeSDbCfgRsp(void* buf, int32_t bufLen, const SDbCfgRsp* pRsp);

299
include/common/ttokendef.h
View File

@ -86,155 +86,156 @@
#define TK_SINGLE_STABLE 68
#define TK_STREAM_MODE 69
#define TK_RETENTIONS 70
#define TK_NK_COMMA 71
#define TK_NK_COLON 72
#define TK_TABLE 73
#define TK_NK_LP 74
#define TK_NK_RP 75
#define TK_STABLE 76
#define TK_ADD 77
#define TK_COLUMN 78
#define TK_MODIFY 79
#define TK_RENAME 80
#define TK_TAG 81
#define TK_SET 82
#define TK_NK_EQ 83
#define TK_USING 84
#define TK_TAGS 85
#define TK_NK_DOT 86
#define TK_COMMENT 87
#define TK_BOOL 88
#define TK_TINYINT 89
#define TK_SMALLINT 90
#define TK_INT 91
#define TK_INTEGER 92
#define TK_BIGINT 93
#define TK_FLOAT 94
#define TK_DOUBLE 95
#define TK_BINARY 96
#define TK_TIMESTAMP 97
#define TK_NCHAR 98
#define TK_UNSIGNED 99
#define TK_JSON 100
#define TK_VARCHAR 101
#define TK_MEDIUMBLOB 102
#define TK_BLOB 103
#define TK_VARBINARY 104
#define TK_DECIMAL 105
#define TK_SMA 106
#define TK_ROLLUP 107
#define TK_FILE_FACTOR 108
#define TK_NK_FLOAT 109
#define TK_DELAY 110
#define TK_SHOW 111
#define TK_DATABASES 112
#define TK_TABLES 113
#define TK_STABLES 114
#define TK_MNODES 115
#define TK_MODULES 116
#define TK_QNODES 117
#define TK_FUNCTIONS 118
#define TK_INDEXES 119
#define TK_FROM 120
#define TK_ACCOUNTS 121
#define TK_APPS 122
#define TK_CONNECTIONS 123
#define TK_LICENCE 124
#define TK_GRANTS 125
#define TK_QUERIES 126
#define TK_SCORES 127
#define TK_TOPICS 128
#define TK_VARIABLES 129
#define TK_BNODES 130
#define TK_SNODES 131
#define TK_LIKE 132
#define TK_INDEX 133
#define TK_FULLTEXT 134
#define TK_FUNCTION 135
#define TK_INTERVAL 136
#define TK_TOPIC 137
#define TK_AS 138
#define TK_DESC 139
#define TK_DESCRIBE 140
#define TK_RESET 141
#define TK_QUERY 142
#define TK_EXPLAIN 143
#define TK_ANALYZE 144
#define TK_VERBOSE 145
#define TK_NK_BOOL 146
#define TK_RATIO 147
#define TK_COMPACT 148
#define TK_VNODES 149
#define TK_IN 150
#define TK_OUTPUTTYPE 151
#define TK_AGGREGATE 152
#define TK_BUFSIZE 153
#define TK_STREAM 154
#define TK_INTO 155
#define TK_TRIGGER 156
#define TK_AT_ONCE 157
#define TK_WINDOW_CLOSE 158
#define TK_WATERMARK 159
#define TK_KILL 160
#define TK_CONNECTION 161
#define TK_MERGE 162
#define TK_VGROUP 163
#define TK_REDISTRIBUTE 164
#define TK_SPLIT 165
#define TK_SYNCDB 166
#define TK_NULL 167
#define TK_NK_QUESTION 168
#define TK_NK_ARROW 169
#define TK_ROWTS 170
#define TK_TBNAME 171
#define TK_QSTARTTS 172
#define TK_QENDTS 173
#define TK_WSTARTTS 174
#define TK_WENDTS 175
#define TK_WDURATION 176
#define TK_CAST 177
#define TK_NOW 178
#define TK_TODAY 179
#define TK_TIMEZONE 180
#define TK_COUNT 181
#define TK_FIRST 182
#define TK_LAST 183
#define TK_LAST_ROW 184
#define TK_BETWEEN 185
#define TK_IS 186
#define TK_NK_LT 187
#define TK_NK_GT 188
#define TK_NK_LE 189
#define TK_NK_GE 190
#define TK_NK_NE 191
#define TK_MATCH 192
#define TK_NMATCH 193
#define TK_CONTAINS 194
#define TK_JOIN 195
#define TK_INNER 196
#define TK_SELECT 197
#define TK_DISTINCT 198
#define TK_WHERE 199
#define TK_PARTITION 200
#define TK_BY 201
#define TK_SESSION 202
#define TK_STATE_WINDOW 203
#define TK_SLIDING 204
#define TK_FILL 205
#define TK_VALUE 206
#define TK_NONE 207
#define TK_PREV 208
#define TK_LINEAR 209
#define TK_NEXT 210
#define TK_GROUP 211
#define TK_HAVING 212
#define TK_ORDER 213
#define TK_SLIMIT 214
#define TK_SOFFSET 215
#define TK_LIMIT 216
#define TK_OFFSET 217
#define TK_ASC 218
#define TK_NULLS 219
#define TK_STRICT 71
#define TK_NK_COMMA 72
#define TK_NK_COLON 73
#define TK_TABLE 74
#define TK_NK_LP 75
#define TK_NK_RP 76
#define TK_STABLE 77
#define TK_ADD 78
#define TK_COLUMN 79
#define TK_MODIFY 80
#define TK_RENAME 81
#define TK_TAG 82
#define TK_SET 83
#define TK_NK_EQ 84
#define TK_USING 85
#define TK_TAGS 86
#define TK_NK_DOT 87
#define TK_COMMENT 88
#define TK_BOOL 89
#define TK_TINYINT 90
#define TK_SMALLINT 91
#define TK_INT 92
#define TK_INTEGER 93
#define TK_BIGINT 94
#define TK_FLOAT 95
#define TK_DOUBLE 96
#define TK_BINARY 97
#define TK_TIMESTAMP 98
#define TK_NCHAR 99
#define TK_UNSIGNED 100
#define TK_JSON 101
#define TK_VARCHAR 102
#define TK_MEDIUMBLOB 103
#define TK_BLOB 104
#define TK_VARBINARY 105
#define TK_DECIMAL 106
#define TK_SMA 107
#define TK_ROLLUP 108
#define TK_FILE_FACTOR 109
#define TK_NK_FLOAT 110
#define TK_DELAY 111
#define TK_SHOW 112
#define TK_DATABASES 113
#define TK_TABLES 114
#define TK_STABLES 115
#define TK_MNODES 116
#define TK_MODULES 117
#define TK_QNODES 118
#define TK_FUNCTIONS 119
#define TK_INDEXES 120
#define TK_FROM 121
#define TK_ACCOUNTS 122
#define TK_APPS 123
#define TK_CONNECTIONS 124
#define TK_LICENCE 125
#define TK_GRANTS 126
#define TK_QUERIES 127
#define TK_SCORES 128
#define TK_TOPICS 129
#define TK_VARIABLES 130
#define TK_BNODES 131
#define TK_SNODES 132
#define TK_LIKE 133
#define TK_INDEX 134
#define TK_FULLTEXT 135
#define TK_FUNCTION 136
#define TK_INTERVAL 137
#define TK_TOPIC 138
#define TK_AS 139
#define TK_DESC 140
#define TK_DESCRIBE 141
#define TK_RESET 142
#define TK_QUERY 143
#define TK_EXPLAIN 144
#define TK_ANALYZE 145
#define TK_VERBOSE 146
#define TK_NK_BOOL 147
#define TK_RATIO 148
#define TK_COMPACT 149
#define TK_VNODES 150
#define TK_IN 151
#define TK_OUTPUTTYPE 152
#define TK_AGGREGATE 153
#define TK_BUFSIZE 154
#define TK_STREAM 155
#define TK_INTO 156
#define TK_TRIGGER 157
#define TK_AT_ONCE 158
#define TK_WINDOW_CLOSE 159
#define TK_WATERMARK 160
#define TK_KILL 161
#define TK_CONNECTION 162
#define TK_MERGE 163
#define TK_VGROUP 164
#define TK_REDISTRIBUTE 165
#define TK_SPLIT 166
#define TK_SYNCDB 167
#define TK_NULL 168
#define TK_NK_QUESTION 169
#define TK_NK_ARROW 170
#define TK_ROWTS 171
#define TK_TBNAME 172
#define TK_QSTARTTS 173
#define TK_QENDTS 174
#define TK_WSTARTTS 175
#define TK_WENDTS 176
#define TK_WDURATION 177
#define TK_CAST 178
#define TK_NOW 179
#define TK_TODAY 180
#define TK_TIMEZONE 181
#define TK_COUNT 182
#define TK_FIRST 183
#define TK_LAST 184
#define TK_LAST_ROW 185
#define TK_BETWEEN 186
#define TK_IS 187
#define TK_NK_LT 188
#define TK_NK_GT 189
#define TK_NK_LE 190
#define TK_NK_GE 191
#define TK_NK_NE 192
#define TK_MATCH 193
#define TK_NMATCH 194
#define TK_CONTAINS 195
#define TK_JOIN 196
#define TK_INNER 197
#define TK_SELECT 198
#define TK_DISTINCT 199
#define TK_WHERE 200
#define TK_PARTITION 201
#define TK_BY 202
#define TK_SESSION 203
#define TK_STATE_WINDOW 204
#define TK_SLIDING 205
#define TK_FILL 206
#define TK_VALUE 207
#define TK_NONE 208
#define TK_PREV 209
#define TK_LINEAR 210
#define TK_NEXT 211
#define TK_GROUP 212
#define TK_HAVING 213
#define TK_ORDER 214
#define TK_SLIMIT 215
#define TK_SOFFSET 216
#define TK_LIMIT 217
#define TK_OFFSET 218
#define TK_ASC 219
#define TK_NULLS 220
#define TK_NK_SPACE 300
#define TK_NK_COMMENT 301

1
include/libs/nodes/cmdnodes.h
View File

@ -47,6 +47,7 @@ typedef struct SDatabaseOptions {
SValueNode* pNumOfVgroups;
SValueNode* pSingleStable;
SValueNode* pStreamMode;
SValueNode* pStrict;
SNodeList* pRetentions;
} SDatabaseOptions;

8
include/util/tdef.h
View File

@ -390,10 +390,14 @@ typedef enum ELogicConditionType {
#define TSDB_MAX_DB_SINGLE_STABLE_OPTION 1
#define TSDB_DEFAULT_DB_SINGLE_STABLE_OPTION 0
#define TSDB_MIN_DB_STREAM_MODE_OPTION 0
#define TSDB_MAX_DB_STREAM_MODE_OPTION 1
#define TSDB_DB_STREAM_MODE_OPTION_OFF 0
#define TSDB_DB_STREAM_MODE_OPTION_ON 1
#define TSDB_DEFAULT_DB_STREAM_MODE_OPTION 0
#define TSDB_DB_STRICT_OPTION_OFF 0
#define TSDB_DB_STRICT_OPTION_ON 1
#define TSDB_DEFAULT_DB_STRICT_OPTION 0
#define TSDB_MAX_JOIN_TABLE_NUM 10
#define TSDB_MAX_UNION_CLAUSE 5

49
source/common/src/tmsg.c
View File

@ -611,6 +611,8 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
if (tEncodeI32(&encoder, pReq->numOfTags) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numOfSmas) < 0) return -1;
if (tEncodeI32(&encoder, pReq->commentLen) < 0) return -1;
if (tEncodeI32(&encoder, pReq->ast1Len) < 0) return -1;
if (tEncodeI32(&encoder, pReq->ast2Len) < 0) return -1;
for (int32_t i = 0; i < pReq->numOfColumns; ++i) {
SField *pField = taosArrayGet(pReq->pColumns, i);
@ -636,6 +638,12 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
if (pReq->commentLen > 0) {
if (tEncodeBinary(&encoder, pReq->comment, pReq->commentLen) < 0) return -1;
}
if (pReq->ast1Len > 0) {
if (tEncodeBinary(&encoder, pReq->pAst1, pReq->ast1Len) < 0) return -1;
}
if (pReq->ast2Len > 0) {
if (tEncodeBinary(&encoder, pReq->pAst2, pReq->ast2Len) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -658,6 +666,8 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
if (tDecodeI32(&decoder, &pReq->numOfTags) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numOfSmas) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->commentLen) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->ast1Len) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->ast2Len) < 0) return -1;
pReq->pColumns = taosArrayInit(pReq->numOfColumns, sizeof(SField));
pReq->pTags = taosArrayInit(pReq->numOfTags, sizeof(SField));
@ -706,6 +716,18 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
if (tDecodeCStrTo(&decoder, pReq->comment) < 0) return -1;
}
if (pReq->ast1Len > 0) {
pReq->pAst1 = taosMemoryMalloc(pReq->ast1Len);
if (pReq->pAst1 == NULL) return -1;
if (tDecodeCStrTo(&decoder, pReq->pAst1) < 0) return -1;
}
if (pReq->ast2Len > 0) {
pReq->pAst2 = taosMemoryMalloc(pReq->ast2Len);
if (pReq->pAst2 == NULL) return -1;
if (tDecodeCStrTo(&decoder, pReq->pAst2) < 0) return -1;
}
tEndDecode(&decoder);
tCoderClear(&decoder);
@ -717,6 +739,8 @@ void tFreeSMCreateStbReq(SMCreateStbReq *pReq) {
taosArrayDestroy(pReq->pTags);
taosArrayDestroy(pReq->pSmas);
taosMemoryFreeClear(pReq->comment);
taosMemoryFreeClear(pReq->pAst1);
taosMemoryFreeClear(pReq->pAst2);
pReq->pColumns = NULL;
pReq->pTags = NULL;
pReq->pSmas = NULL;
@ -2211,6 +2235,14 @@ int32_t tSerializeSDbCfgRsp(void *buf, int32_t bufLen, const SDbCfgRsp *pRsp) {
if (tEncodeI8(&encoder, pRsp->update) < 0) return -1;
if (tEncodeI8(&encoder, pRsp->cacheLastRow) < 0) return -1;
if (tEncodeI8(&encoder, pRsp->streamMode) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->numOfRetensions) < 0) return -1;
for (int32_t i = 0; i < pRsp->numOfRetensions; ++i) {
SRetention *pRetension = taosArrayGet(pRsp->pRetensions, i);
if (tEncodeI32(&encoder, pRetension->freq) < 0) return -1;
if (tEncodeI32(&encoder, pRetension->keep) < 0) return -1;
if (tEncodeI8(&encoder, pRetension->freqUnit) < 0) return -1;
if (tEncodeI8(&encoder, pRetension->keepUnit) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -2242,7 +2274,24 @@ int32_t tDeserializeSDbCfgRsp(void *buf, int32_t bufLen, SDbCfgRsp *pRsp) {
if (tDecodeI8(&decoder, &pRsp->update) < 0) return -1;
if (tDecodeI8(&decoder, &pRsp->cacheLastRow) < 0) return -1;
if (tDecodeI8(&decoder, &pRsp->streamMode) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->numOfRetensions) < 0) return -1;
pRsp->pRetensions = taosArrayInit(pRsp->numOfRetensions, sizeof(SRetention));
if (pRsp->pRetensions == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
for (int32_t i = 0; i < pRsp->numOfRetensions; ++i) {
SRetention rentension = {0};
if (tDecodeI32(&decoder, &rentension.freq) < 0) return -1;
if (tDecodeI32(&decoder, &rentension.keep) < 0) return -1;
if (tDecodeI8(&decoder, &rentension.freqUnit) < 0) return -1;
if (tDecodeI8(&decoder, &rentension.keepUnit) < 0) return -1;
if (taosArrayPush(pRsp->pRetensions, &rentension) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
}
tEndDecode(&decoder);
tCoderClear(&decoder);

4
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -355,10 +355,14 @@ typedef struct {
int32_t numOfTags;
int32_t numOfSmas;
int32_t commentLen;
int32_t ast1Len;
int32_t ast2Len;
SSchema* pColumns;
SSchema* pTags;
SSchema* pSmas;
char* comment;
char* pAst1;
char* pAst2;
SRWLatch lock;
} SStbObj;

46
source/dnode/mnode/impl/src/mndDb.c
View File

@ -762,27 +762,29 @@ static int32_t mndProcessGetDbCfgReq(SNodeMsg *pReq) {
goto GET_DB_CFG_OVER;
}
cfgRsp.numOfVgroups = pDb->cfg.numOfVgroups;
cfgRsp.cacheBlockSize = pDb->cfg.cacheBlockSize;
cfgRsp.totalBlocks = pDb->cfg.totalBlocks;
cfgRsp.daysPerFile = pDb->cfg.daysPerFile;
cfgRsp.daysToKeep0 = pDb->cfg.daysToKeep0;
cfgRsp.daysToKeep1 = pDb->cfg.daysToKeep1;
cfgRsp.daysToKeep2 = pDb->cfg.daysToKeep2;
cfgRsp.minRows = pDb->cfg.minRows;
cfgRsp.maxRows = pDb->cfg.maxRows;
cfgRsp.commitTime = pDb->cfg.commitTime;
cfgRsp.fsyncPeriod = pDb->cfg.fsyncPeriod;
cfgRsp.ttl = pDb->cfg.ttl;
cfgRsp.walLevel = pDb->cfg.walLevel;
cfgRsp.precision = pDb->cfg.precision;
cfgRsp.compression = pDb->cfg.compression;
cfgRsp.replications = pDb->cfg.replications;
cfgRsp.quorum = pDb->cfg.quorum;
cfgRsp.update = pDb->cfg.update;
cfgRsp.cacheLastRow = pDb->cfg.cacheLastRow;
cfgRsp.streamMode = pDb->cfg.streamMode;
cfgRsp.singleSTable = pDb->cfg.singleSTable;
cfgRsp.numOfVgroups = pDb->cfg.numOfVgroups;
cfgRsp.cacheBlockSize = pDb->cfg.cacheBlockSize;
cfgRsp.totalBlocks = pDb->cfg.totalBlocks;
cfgRsp.daysPerFile = pDb->cfg.daysPerFile;
cfgRsp.daysToKeep0 = pDb->cfg.daysToKeep0;
cfgRsp.daysToKeep1 = pDb->cfg.daysToKeep1;
cfgRsp.daysToKeep2 = pDb->cfg.daysToKeep2;
cfgRsp.minRows = pDb->cfg.minRows;
cfgRsp.maxRows = pDb->cfg.maxRows;
cfgRsp.commitTime = pDb->cfg.commitTime;
cfgRsp.fsyncPeriod = pDb->cfg.fsyncPeriod;
cfgRsp.ttl = pDb->cfg.ttl;
cfgRsp.walLevel = pDb->cfg.walLevel;
cfgRsp.precision = pDb->cfg.precision;
cfgRsp.compression = pDb->cfg.compression;
cfgRsp.replications = pDb->cfg.replications;
cfgRsp.quorum = pDb->cfg.quorum;
cfgRsp.update = pDb->cfg.update;
cfgRsp.cacheLastRow = pDb->cfg.cacheLastRow;
cfgRsp.streamMode = pDb->cfg.streamMode;
cfgRsp.singleSTable = pDb->cfg.singleSTable;
cfgRsp.numOfRetensions = pDb->cfg.numOfRetensions;
cfgRsp.pRetensions = pDb->cfg.pRetensions;
int32_t contLen = tSerializeSDbCfgRsp(NULL, 0, &cfgRsp);
void *pRsp = rpcMallocCont(contLen);
@ -797,6 +799,8 @@ static int32_t mndProcessGetDbCfgReq(SNodeMsg *pReq) {
pReq->pRsp = pRsp;
pReq->rspLen = contLen;
code = 0;
GET_DB_CFG_OVER:
if (code != 0 && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {

74
source/dnode/mnode/impl/src/mndStb.c
View File

@ -72,7 +72,7 @@ SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
int32_t size = sizeof(SStbObj) + (pStb->numOfColumns + pStb->numOfTags + pStb->numOfSmas) * sizeof(SSchema) +
TSDB_STB_RESERVE_SIZE;
+ pStb->commentLen + pStb->ast1Len + pStb->ast2Len + TSDB_STB_RESERVE_SIZE;
SSdbRaw *pRaw = sdbAllocRaw(SDB_STB, TSDB_STB_VER_NUMBER, size);
if (pRaw == NULL) goto _OVER;
@ -93,6 +93,8 @@ SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
SDB_SET_INT32(pRaw, dataPos, pStb->numOfTags, _OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->numOfSmas, _OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->commentLen, _OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->ast1Len, _OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->ast2Len, _OVER)
for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
SSchema *pSchema = &pStb->pColumns[i];
@ -121,6 +123,12 @@ SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
if (pStb->commentLen > 0) {
SDB_SET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen, _OVER)
}
if (pStb->ast1Len > 0) {
SDB_SET_BINARY(pRaw, dataPos, pStb->pAst1, pStb->ast1Len, _OVER)
}
if (pStb->ast2Len > 0) {
SDB_SET_BINARY(pRaw, dataPos, pStb->pAst2, pStb->ast2Len, _OVER)
}
SDB_SET_RESERVE(pRaw, dataPos, TSDB_STB_RESERVE_SIZE, _OVER)
SDB_SET_DATALEN(pRaw, dataPos, _OVER)
@ -173,6 +181,8 @@ static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT32(pRaw, dataPos, &pStb->numOfTags, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->numOfSmas, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->commentLen, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->ast1Len, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->ast2Len, _OVER)
pStb->pColumns = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchema));
pStb->pTags = taosMemoryCalloc(pStb->numOfTags, sizeof(SSchema));
@ -210,6 +220,16 @@ static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {
if (pStb->comment == NULL) goto _OVER;
SDB_GET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen, _OVER)
}
if (pStb->ast1Len > 0) {
pStb->pAst1 = taosMemoryCalloc(pStb->ast1Len, 1);
if (pStb->pAst1 == NULL) goto _OVER;
SDB_GET_BINARY(pRaw, dataPos, pStb->pAst1, pStb->ast1Len, _OVER)
}
if (pStb->ast2Len > 0) {
pStb->pAst2 = taosMemoryCalloc(pStb->ast2Len, 1);
if (pStb->pAst2 == NULL) goto _OVER;
SDB_GET_BINARY(pRaw, dataPos, pStb->pAst2, pStb->ast2Len, _OVER)
}
SDB_GET_RESERVE(pRaw, dataPos, TSDB_STB_RESERVE_SIZE, _OVER)
terrno = 0;
@ -238,6 +258,8 @@ static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb) {
taosMemoryFreeClear(pStb->pColumns);
taosMemoryFreeClear(pStb->pTags);
taosMemoryFreeClear(pStb->comment);
taosMemoryFreeClear(pStb->pAst1);
taosMemoryFreeClear(pStb->pAst2);
return 0;
}
@ -294,6 +316,30 @@ static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {
}
}
if (pOld->ast1Len < pNew->ast1Len) {
void *pAst1 = taosMemoryMalloc(pNew->ast1Len);
if (pAst1 != NULL) {
taosMemoryFree(pOld->pAst1);
pOld->pAst1 = pAst1;
} else {
terrno = TSDB_CODE_OUT_OF_MEMORY;
mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());
taosWUnLockLatch(&pOld->lock);
}
}
if (pOld->ast2Len < pNew->ast2Len) {
void *pAst2 = taosMemoryMalloc(pNew->ast2Len);
if (pAst2 != NULL) {
taosMemoryFree(pOld->pAst2);
pOld->pAst2 = pAst2;
} else {
terrno = TSDB_CODE_OUT_OF_MEMORY;
mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());
taosWUnLockLatch(&pOld->lock);
}
}
pOld->updateTime = pNew->updateTime;
pOld->version = pNew->version;
pOld->nextColId = pNew->nextColId;
@ -304,6 +350,12 @@ static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {
if (pNew->commentLen != 0) {
memcpy(pOld->comment, pNew->comment, TSDB_STB_COMMENT_LEN);
}
if (pNew->ast1Len != 0) {
memcpy(pOld->pAst1, pNew->pAst1, pNew->ast1Len);
}
if (pNew->ast2Len != 0) {
memcpy(pOld->pAst2, pNew->pAst2, pNew->ast2Len);
}
taosWUnLockLatch(&pOld->lock);
return 0;
}
@ -646,6 +698,26 @@ static int32_t mndCreateStb(SMnode *pMnode, SNodeMsg *pReq, SMCreateStbReq *pCre
memcpy(stbObj.comment, pCreate->comment, stbObj.commentLen);
}
stbObj.ast1Len = pCreate->ast1Len;
if (stbObj.ast1Len > 0) {
stbObj.pAst1 = taosMemoryCalloc(stbObj.ast1Len, 1);
if (stbObj.pAst1 == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
memcpy(stbObj.pAst1, pCreate->pAst1, stbObj.ast1Len);
}
stbObj.ast2Len = pCreate->ast2Len;
if (stbObj.ast2Len > 0) {
stbObj.pAst2 = taosMemoryCalloc(stbObj.ast2Len, 1);
if (stbObj.pAst2 == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
memcpy(stbObj.pAst2, pCreate->pAst2, stbObj.ast2Len);
}
stbObj.pColumns = taosMemoryMalloc(stbObj.numOfColumns * sizeof(SSchema));
stbObj.pTags = taosMemoryMalloc(stbObj.numOfTags * sizeof(SSchema));
stbObj.pSmas = taosMemoryMalloc(stbObj.numOfSmas * sizeof(SSchema));

1
source/libs/parser/inc/parAst.h
View File

@ -53,6 +53,7 @@ typedef enum EDatabaseOptionType {
DB_OPTION_VGROUPS,
DB_OPTION_SINGLE_STABLE,
DB_OPTION_STREAM_MODE,
DB_OPTION_STRICT,
DB_OPTION_RETENTIONS
} EDatabaseOptionType;

4
source/libs/parser/inc/sql.y
View File

@ -161,6 +161,7 @@ db_options(A) ::= db_options(B) VGROUPS NK_INTEGER(C).
db_options(A) ::= db_options(B) SINGLE_STABLE NK_INTEGER(C). { ((SDatabaseOptions*)B)->pSingleStable = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
db_options(A) ::= db_options(B) STREAM_MODE NK_INTEGER(C). { ((SDatabaseOptions*)B)->pStreamMode = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
db_options(A) ::= db_options(B) RETENTIONS retention_list(C). { ((SDatabaseOptions*)B)->pRetentions = C; A = B; }
db_options(A) ::= db_options(B) STRICT NK_INTEGER(C). { ((SDatabaseOptions*)B)->pStrict = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
alter_db_options(A) ::= alter_db_option(B). { A = createDatabaseOptions(pCxt); A = setDatabaseAlterOption(pCxt, A, &B); }
alter_db_options(A) ::= alter_db_options(B) alter_db_option(C). { A = setDatabaseAlterOption(pCxt, B, &C); }
@ -175,6 +176,7 @@ alter_db_option(A) ::= WAL NK_INTEGER(B).
alter_db_option(A) ::= QUORUM NK_INTEGER(B). { A.type = DB_OPTION_QUORUM; A.pVal = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &B); }
alter_db_option(A) ::= CACHELAST NK_INTEGER(B). { A.type = DB_OPTION_CACHELAST; A.pVal = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &B); }
alter_db_option(A) ::= REPLICA NK_INTEGER(B). { A.type = DB_OPTION_REPLICA; A.pVal = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &B); }
alter_db_option(A) ::= STRICT NK_INTEGER(B). { A.type = DB_OPTION_STRICT; A.pVal = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &B); }
%type integer_list { SNodeList* }
%destructor integer_list { nodesDestroyList($$); }
@ -359,7 +361,7 @@ from_db_opt(A) ::= FROM db_name(B).
%type func_name_list { SNodeList* }
%destructor func_name_list { nodesDestroyList($$); }
func_name_list(A) ::= func_name(B). { A = createNodeList(pCxt, B); }
func_name_list(A) ::= func_name_list(B) NK_COMMA col_name(C). { A = addNodeToList(pCxt, B, C); }
func_name_list(A) ::= func_name_list(B) NK_COMMA func_name(C). { A = addNodeToList(pCxt, B, C); }
func_name(A) ::= function_name(B). { A = createFunctionNode(pCxt, &B, NULL); }

1
source/libs/parser/src/parTokenizer.c
View File

@ -168,6 +168,7 @@ static SKeyword keywordTable[] = {
{"STREAM", TK_STREAM},
{"STREAMS", TK_STREAMS},
{"STREAM_MODE", TK_STREAM_MODE},
{"STRICT", TK_STRICT},
{"SYNCDB", TK_SYNCDB},
{"TABLE", TK_TABLE},
{"TABLES", TK_TABLES},

484
source/libs/parser/src/parTranslater.c
View File

@ -198,6 +198,56 @@ static int32_t getDBVgVersion(STranslateContext* pCxt, const char* pDbFName, int
return code;
}
static int32_t getDBCfg(STranslateContext* pCxt, const char* pDbName, SDbCfgInfo* pInfo) {
SParseContext* pParCxt = pCxt->pParseCxt;
SName name;
tNameSetDbName(&name, pCxt->pParseCxt->acctId, pDbName, strlen(pDbName));
char dbFname[TSDB_DB_FNAME_LEN] = {0};
tNameGetFullDbName(&name, dbFname);
int32_t code = collectUseDatabaseImpl(dbFname, pCxt->pDbs);
if (TSDB_CODE_SUCCESS == code) {
code = catalogGetDBCfg(pParCxt->pCatalog, pParCxt->pTransporter, &pParCxt->mgmtEpSet, dbFname, pInfo);
}
if (TSDB_CODE_SUCCESS != code) {
parserError("catalogGetDBCfg error, code:%s, dbFName:%s", tstrerror(code), dbFname);
}
return code;
}
static int32_t initTranslateContext(SParseContext* pParseCxt, STranslateContext* pCxt) {
pCxt->pParseCxt = pParseCxt;
pCxt->errCode = TSDB_CODE_SUCCESS;
pCxt->msgBuf.buf = pParseCxt->pMsg;
pCxt->msgBuf.len = pParseCxt->msgLen;
pCxt->pNsLevel = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
pCxt->currLevel = 0;
pCxt->currClause = 0;
pCxt->pDbs = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
pCxt->pTables = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
if (NULL == pCxt->pNsLevel || NULL == pCxt->pDbs || NULL == pCxt->pTables) {
return TSDB_CODE_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
static void destroyTranslateContext(STranslateContext* pCxt) {
if (NULL != pCxt->pNsLevel) {
size_t size = taosArrayGetSize(pCxt->pNsLevel);
for (size_t i = 0; i < size; ++i) {
taosArrayDestroy(taosArrayGetP(pCxt->pNsLevel, i));
}
taosArrayDestroy(pCxt->pNsLevel);
}
if (NULL != pCxt->pCmdMsg) {
taosMemoryFreeClear(pCxt->pCmdMsg->pMsg);
taosMemoryFreeClear(pCxt->pCmdMsg);
}
taosHashCleanup(pCxt->pDbs);
taosHashCleanup(pCxt->pTables);
}
static bool belongTable(const char* currentDb, const SColumnNode* pCol, const STableNode* pTable) {
int cmp = 0;
if ('\0' != pCol->dbName[0]) {
@ -749,15 +799,18 @@ static int32_t translateTable(STranslateContext* pCxt, SNode* pTable) {
case QUERY_NODE_REAL_TABLE: {
SRealTableNode* pRealTable = (SRealTableNode*)pTable;
pRealTable->ratio = (NULL != pCxt->pExplainOpt ? pCxt->pExplainOpt->ratio : 1.0);
SName name;
code = getTableMetaImpl(
pCxt, toName(pCxt->pParseCxt->acctId, pRealTable->table.dbName, pRealTable->table.tableName, &name),
&(pRealTable->pMeta));
if (TSDB_CODE_SUCCESS != code) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_TABLE_NOT_EXIST, pRealTable->table.tableName);
// The SRealTableNode created through ROLLUP already has STableMeta.
if (NULL == pRealTable->pMeta) {
SName name;
code = getTableMetaImpl(
pCxt, toName(pCxt->pParseCxt->acctId, pRealTable->table.dbName, pRealTable->table.tableName, &name),
&(pRealTable->pMeta));
if (TSDB_CODE_SUCCESS != code) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_TABLE_NOT_EXIST, pRealTable->table.tableName);
}
code = setTableVgroupList(pCxt, &name, pRealTable);
}
pRealTable->table.precision = pRealTable->pMeta->tableInfo.precision;
code = setTableVgroupList(pCxt, &name, pRealTable);
if (TSDB_CODE_SUCCESS == code) {
code = addNamespace(pCxt, pRealTable);
}
@ -1365,6 +1418,7 @@ static int32_t buildCreateDbReq(STranslateContext* pCxt, SCreateDatabaseStmt* pS
pReq->streamMode = GET_OPTION_VAL(pStmt->pOptions->pStreamMode, TSDB_DEFAULT_DB_STREAM_MODE_OPTION);
pReq->ttl = GET_OPTION_VAL(pStmt->pOptions->pTtl, TSDB_DEFAULT_DB_TTL_OPTION);
pReq->singleSTable = GET_OPTION_VAL(pStmt->pOptions->pSingleStable, TSDB_DEFAULT_DB_SINGLE_STABLE_OPTION);
// pReq->strict = GET_OPTION_VAL(pStmt->pOptions->pStrict, TSDB_DEFAULT_DB_SINGLE_STABLE_OPTION);
return buildCreateDbRetentions(pStmt->pOptions->pRetentions, pReq);
}
@ -1573,12 +1627,16 @@ static int32_t checkDatabaseOptions(STranslateContext* pCxt, SDatabaseOptions* p
TSDB_MAX_DB_SINGLE_STABLE_OPTION);
}
if (TSDB_CODE_SUCCESS == code) {
code = checkDbEnumOption(pCxt, "streamMode", pOptions->pStreamMode, TSDB_MIN_DB_STREAM_MODE_OPTION,
TSDB_MAX_DB_STREAM_MODE_OPTION);
code = checkDbEnumOption(pCxt, "streamMode", pOptions->pStreamMode, TSDB_DB_STREAM_MODE_OPTION_OFF,
TSDB_DB_STREAM_MODE_OPTION_ON);
}
if (TSDB_CODE_SUCCESS == code) {
code = checkDbRetentionsOption(pCxt, pOptions->pRetentions);
}
if (TSDB_CODE_SUCCESS == code) {
code = checkDbEnumOption(pCxt, "strict", pOptions->pStrict, TSDB_DB_STRICT_OPTION_OFF,
TSDB_DB_STRICT_OPTION_ON);
}
return code;
}
@ -1789,35 +1847,259 @@ static int32_t getAggregationMethod(SNodeList* pFuncs) {
return ((SFunctionNode*)nodesListGetNode(pFuncs, 0))->funcId;
}
static int32_t translateCreateSuperTable(STranslateContext* pCxt, SCreateTableStmt* pStmt) {
int32_t code = checkCreateTable(pCxt, pStmt);
if (TSDB_CODE_SUCCESS != code) {
return code;
static void toSchema(const SColumnDefNode* pCol, col_id_t colId, SSchema* pSchema) {
int8_t flags = 0;
if (pCol->sma) {
flags |= SCHEMA_SMA_ON;
}
pSchema->colId = colId;
pSchema->type = pCol->dataType.type;
pSchema->bytes = calcTypeBytes(pCol->dataType);
pSchema->flags = flags;
strcpy(pSchema->name, pCol->colName);
}
typedef struct SSampleAstInfo {
const char* pDbName;
const char* pTableName;
SNodeList* pFuncs;
SNode* pInterval;
SNode* pOffset;
SNode* pSliding;
STableMeta* pRollupTableMeta;
} SSampleAstInfo;
static int32_t buildSampleAst(STranslateContext* pCxt, SSampleAstInfo* pInfo, char** pAst, int32_t* pLen) {
SSelectStmt* pSelect = nodesMakeNode(QUERY_NODE_SELECT_STMT);
if (NULL == pSelect) {
return TSDB_CODE_OUT_OF_MEMORY;
}
sprintf(pSelect->stmtName, "%p", pSelect);
SRealTableNode* pTable = nodesMakeNode(QUERY_NODE_REAL_TABLE);
if (NULL == pTable) {
nodesDestroyNode(pSelect);
return TSDB_CODE_OUT_OF_MEMORY;
}
strcpy(pTable->table.dbName, pInfo->pDbName);
strcpy(pTable->table.tableName, pInfo->pTableName);
TSWAP(pTable->pMeta, pInfo->pRollupTableMeta, STableMeta*);
pSelect->pFromTable = (SNode*)pTable;
TSWAP(pSelect->pProjectionList, pInfo->pFuncs, SNodeList*);
SFunctionNode* pFunc = nodesMakeNode(QUERY_NODE_FUNCTION);
if (NULL == pSelect->pProjectionList || NULL == pFunc) {
nodesDestroyNode(pSelect);
return TSDB_CODE_OUT_OF_MEMORY;
}
strcpy(pFunc->functionName, "_wstartts");
nodesListPushFront(pSelect->pProjectionList, pFunc);
SNode* pProject = NULL;
FOREACH(pProject, pSelect->pProjectionList) { sprintf(((SExprNode*)pProject)->aliasName, "#%p", pProject); }
SIntervalWindowNode* pInterval = nodesMakeNode(QUERY_NODE_INTERVAL_WINDOW);
if (NULL == pInterval) {
nodesDestroyNode(pSelect);
return TSDB_CODE_OUT_OF_MEMORY;
}
pSelect->pWindow = (SNode*)pInterval;
TSWAP(pInterval->pInterval, pInfo->pInterval, SNode*);
TSWAP(pInterval->pOffset, pInfo->pOffset, SNode*);
TSWAP(pInterval->pSliding, pInfo->pSliding, SNode*);
pInterval->pCol = nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pInterval->pCol) {
nodesDestroyNode(pSelect);
return TSDB_CODE_OUT_OF_MEMORY;
}
((SColumnNode*)pInterval->pCol)->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
strcpy(((SColumnNode*)pInterval->pCol)->colName, PK_TS_COL_INTERNAL_NAME);
int32_t code = translateQuery(pCxt, (SNode*)pSelect);
if (TSDB_CODE_SUCCESS == code) {
code = nodesNodeToString(pSelect, false, pAst, pLen);
}
nodesDestroyNode(pSelect);
return code;
}
static void clearSampleAstInfo(SSampleAstInfo* pInfo) {
nodesDestroyList(pInfo->pFuncs);
nodesDestroyNode(pInfo->pInterval);
nodesDestroyNode(pInfo->pOffset);
nodesDestroyNode(pInfo->pSliding);
}
static SNode* makeIntervalVal(SRetention* pRetension, int8_t precision) {
SValueNode* pVal = nodesMakeNode(QUERY_NODE_VALUE);
if (NULL == pVal) {
return NULL;
}
int64_t timeVal = convertTimeFromPrecisionToUnit(pRetension->freq, precision, pRetension->freqUnit);
char buf[20] = {0};
int32_t len = snprintf(buf, sizeof(buf), "%"PRId64"%c", timeVal, pRetension->freqUnit);
pVal->literal = strndup(buf, len);
if (NULL == pVal->literal) {
nodesDestroyNode(pVal);
return NULL;
}
pVal->isDuration = true;
pVal->node.resType.type = TSDB_DATA_TYPE_BIGINT;
pVal->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes;
pVal->node.resType.precision = precision;
return (SNode*)pVal;
}
static SNode* createColumnFromDef(SColumnDefNode* pDef) {
SColumnNode* pCol = nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {
return NULL;
}
strcpy(pCol->colName, pDef->colName);
return (SNode*)pCol;
}
static SNode* createRollupFunc(SNode* pSrcFunc, SColumnDefNode* pColDef) {
SFunctionNode* pFunc = nodesCloneNode(pSrcFunc);
if (NULL == pFunc) {
return NULL;
}
if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pFunc->pParameterList, createColumnFromDef(pColDef))) {
nodesDestroyNode(pFunc);
return NULL;
}
return (SNode*)pFunc;
}
static SNodeList* createRollupFuncs(SCreateTableStmt* pStmt) {
SNodeList* pFuncs = nodesMakeList();
if (NULL == pFuncs) {
return NULL;
}
SMCreateStbReq createReq = {0};
createReq.igExists = pStmt->ignoreExists;
createReq.aggregationMethod = getAggregationMethod(pStmt->pOptions->pFuncs);
createReq.xFilesFactor = GET_OPTION_VAL(pStmt->pOptions->pFilesFactor, TSDB_DEFAULT_DB_FILE_FACTOR);
createReq.delay = GET_OPTION_VAL(pStmt->pOptions->pDelay, TSDB_DEFAULT_DB_DELAY);
columnDefNodeToField(pStmt->pCols, &createReq.pColumns);
columnDefNodeToField(pStmt->pTags, &createReq.pTags);
createReq.numOfColumns = LIST_LENGTH(pStmt->pCols);
createReq.numOfTags = LIST_LENGTH(pStmt->pTags);
SNode* pFunc = NULL;
FOREACH(pFunc, pStmt->pOptions->pFuncs) {
SNode* pCol = NULL;
bool primaryKey = true;
FOREACH(pCol, pStmt->pCols) {
if (primaryKey) {
primaryKey = false;
continue;
}
if (TSDB_CODE_SUCCESS != nodesListStrictAppend(pFuncs, createRollupFunc(pFunc, (SColumnDefNode*)pCol))) {
nodesDestroyList(pFuncs);
return NULL;
}
}
}
return pFuncs;
}
static STableMeta* createRollupTableMeta(SCreateTableStmt* pStmt, int8_t precision) {
int32_t numOfField = LIST_LENGTH(pStmt->pCols) + LIST_LENGTH(pStmt->pTags);
STableMeta* pMeta = taosMemoryCalloc(1, sizeof(STableMeta) + numOfField * sizeof(SSchema));
if (NULL == pMeta) {
return NULL;
}
pMeta->tableType = TSDB_SUPER_TABLE;
pMeta->tableInfo.numOfTags = LIST_LENGTH(pStmt->pTags);
pMeta->tableInfo.precision = precision;
pMeta->tableInfo.numOfColumns = LIST_LENGTH(pStmt->pCols);
int32_t index = 0;
SNode* pCol = NULL;
FOREACH(pCol, pStmt->pCols) {
toSchema((SColumnDefNode*)pCol, index + 1, pMeta->schema + index);
++index;
}
SNode* pTag = NULL;
FOREACH(pTag, pStmt->pTags) {
toSchema((SColumnDefNode*)pTag, index + 1, pMeta->schema + index);
++index;
}
return pMeta;
}
static int32_t buildSampleAstInfoByTable(STranslateContext* pCxt,
SCreateTableStmt* pStmt, SRetention* pRetension, int8_t precision, SSampleAstInfo* pInfo) {
pInfo->pDbName = pStmt->dbName;
pInfo->pTableName = pStmt->tableName;
pInfo->pFuncs = createRollupFuncs(pStmt);
pInfo->pInterval = makeIntervalVal(pRetension, precision);
pInfo->pRollupTableMeta = createRollupTableMeta(pStmt, precision);
if (NULL == pInfo->pFuncs || NULL == pInfo->pInterval || NULL == pInfo->pRollupTableMeta) {
return TSDB_CODE_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
static int32_t getRollupAst(STranslateContext* pCxt,
SCreateTableStmt* pStmt, SRetention* pRetension, int8_t precision, char** pAst, int32_t* pLen) {
SSampleAstInfo info = {0};
int32_t code = buildSampleAstInfoByTable(pCxt, pStmt, pRetension, precision, &info);
if (TSDB_CODE_SUCCESS == code) {
code = buildSampleAst(pCxt, &info, pAst, pLen);
}
clearSampleAstInfo(&info);
return code;
}
static int32_t buildRollupAst(STranslateContext* pCxt, SCreateTableStmt* pStmt, SMCreateStbReq* pReq) {
SDbCfgInfo dbCfg = {0};
int32_t code = getDBCfg(pCxt, pStmt->dbName, &dbCfg);
int32_t num = taosArrayGetSize(dbCfg.pRetensions);
if (TSDB_CODE_SUCCESS != code || num < 2) {
return code;
}
for (int32_t i = 1; i < num; ++i) {
SRetention *pRetension = taosArrayGet(dbCfg.pRetensions, i);
STranslateContext cxt = {0};
initTranslateContext(pCxt->pParseCxt, &cxt);
code = getRollupAst(&cxt, pStmt, pRetension, dbCfg.precision,
1 == i ? &pReq->pAst1 : &pReq->pAst2, 1 == i ? &pReq->ast1Len : &pReq->ast2Len);
destroyTranslateContext(&cxt);
if (TSDB_CODE_SUCCESS != code) {
break;
}
}
return code;
}
static int32_t buildCreateStbReq(STranslateContext* pCxt, SCreateTableStmt* pStmt, SMCreateStbReq* pReq) {
pReq->igExists = pStmt->ignoreExists;
pReq->aggregationMethod = getAggregationMethod(pStmt->pOptions->pFuncs);
pReq->xFilesFactor = GET_OPTION_VAL(pStmt->pOptions->pFilesFactor, TSDB_DEFAULT_DB_FILE_FACTOR);
pReq->delay = GET_OPTION_VAL(pStmt->pOptions->pDelay, TSDB_DEFAULT_DB_DELAY);
columnDefNodeToField(pStmt->pCols, &pReq->pColumns);
columnDefNodeToField(pStmt->pTags, &pReq->pTags);
pReq->numOfColumns = LIST_LENGTH(pStmt->pCols);
pReq->numOfTags = LIST_LENGTH(pStmt->pTags);
if (NULL == pStmt->pOptions->pSma) {
columnDefNodeToField(pStmt->pCols, &createReq.pSmas);
createReq.numOfSmas = createReq.numOfColumns;
columnDefNodeToField(pStmt->pCols, &pReq->pSmas);
pReq->numOfSmas = pReq->numOfColumns;
} else {
columnNodeToField(pStmt->pOptions->pSma, &createReq.pSmas);
createReq.numOfSmas = LIST_LENGTH(pStmt->pOptions->pSma);
columnNodeToField(pStmt->pOptions->pSma, &pReq->pSmas);
pReq->numOfSmas = LIST_LENGTH(pStmt->pOptions->pSma);
}
SName tableName = {.type = TSDB_TABLE_NAME_T, .acctId = pCxt->pParseCxt->acctId};
strcpy(tableName.dbname, pStmt->dbName);
strcpy(tableName.tname, pStmt->tableName);
tNameExtractFullName(&tableName, createReq.name);
tNameExtractFullName(&tableName, pReq->name);
code = buildCmdMsg(pCxt, TDMT_MND_CREATE_STB, (FSerializeFunc)tSerializeSMCreateStbReq, &createReq);
return buildRollupAst(pCxt, pStmt, pReq);
}
static int32_t translateCreateSuperTable(STranslateContext* pCxt, SCreateTableStmt* pStmt) {
SMCreateStbReq createReq = {0};
int32_t code = checkCreateTable(pCxt, pStmt);
if (TSDB_CODE_SUCCESS == code) {
code = buildCreateStbReq(pCxt, pStmt, &createReq);
}
if (TSDB_CODE_SUCCESS == code) {
code = buildCmdMsg(pCxt, TDMT_MND_CREATE_STB, (FSerializeFunc)tSerializeSMCreateStbReq, &createReq);
}
tFreeSMCreateStbReq(&createReq);
return code;
}
@ -1983,20 +2265,20 @@ static int32_t translateAlterDnode(STranslateContext* pCxt, SAlterDnodeStmt* pSt
}
static int32_t nodeTypeToShowType(ENodeType nt) {
// switch (nt) {
// case QUERY_NODE_SHOW_CONNECTIONS_STMT:
// return TSDB_MGMT_TABLE_CONNS;
// case QUERY_NODE_SHOW_LICENCE_STMT:
// return TSDB_MGMT_TABLE_GRANTS;
// case QUERY_NODE_SHOW_QUERIES_STMT:
// return TSDB_MGMT_TABLE_QUERIES;
// case QUERY_NODE_SHOW_TOPICS_STMT:
// return 0; // todo
// case QUERY_NODE_SHOW_VARIABLE_STMT:
// return TSDB_MGMT_TABLE_VARIABLES;
// default:
// break;
// }
switch (nt) {
case QUERY_NODE_SHOW_CONNECTIONS_STMT:
return TSDB_MGMT_TABLE_CONNS;
case QUERY_NODE_SHOW_LICENCE_STMT:
return TSDB_MGMT_TABLE_GRANTS;
case QUERY_NODE_SHOW_QUERIES_STMT:
return TSDB_MGMT_TABLE_QUERIES;
case QUERY_NODE_SHOW_TOPICS_STMT:
return 0; // todo
case QUERY_NODE_SHOW_VARIABLE_STMT:
return 0; // todo
default:
break;
}
return 0;
}
@ -2027,57 +2309,28 @@ static int32_t getSmaIndexExpr(STranslateContext* pCxt, SCreateIndexStmt* pStmt,
return nodesListToString(pStmt->pOptions->pFuncs, false, pExpr, pLen);
}
static int32_t getSmaIndexBuildAst(STranslateContext* pCxt, SCreateIndexStmt* pStmt, char** pAst, int32_t* pLen) {
SSelectStmt* pSelect = nodesMakeNode(QUERY_NODE_SELECT_STMT);
if (NULL == pSelect) {
static int32_t buildSampleAstInfoByIndex(STranslateContext* pCxt, SCreateIndexStmt* pStmt, SSampleAstInfo* pInfo) {
pInfo->pDbName = pCxt->pParseCxt->db;
pInfo->pTableName = pStmt->tableName;
pInfo->pFuncs = nodesCloneList(pStmt->pOptions->pFuncs);
pInfo->pInterval = nodesCloneNode(pStmt->pOptions->pInterval);
pInfo->pOffset = nodesCloneNode(pStmt->pOptions->pOffset);
pInfo->pSliding = nodesCloneNode(pStmt->pOptions->pSliding);
if (NULL == pInfo->pFuncs || NULL == pInfo->pInterval ||
(NULL != pStmt->pOptions->pOffset && NULL == pInfo->pOffset) ||
(NULL != pStmt->pOptions->pSliding && NULL == pInfo->pSliding)) {
return TSDB_CODE_OUT_OF_MEMORY;
}
sprintf(pSelect->stmtName, "%p", pSelect);
return TSDB_CODE_SUCCESS;
}
SRealTableNode* pTable = nodesMakeNode(QUERY_NODE_REAL_TABLE);
if (NULL == pTable) {
nodesDestroyNode(pSelect);
return TSDB_CODE_OUT_OF_MEMORY;
}
strcpy(pTable->table.dbName, pCxt->pParseCxt->db);
strcpy(pTable->table.tableName, pStmt->tableName);
pSelect->pFromTable = (SNode*)pTable;
pSelect->pProjectionList = nodesCloneList(pStmt->pOptions->pFuncs);
SFunctionNode* pFunc = nodesMakeNode(QUERY_NODE_FUNCTION);
if (NULL == pSelect->pProjectionList || NULL == pFunc) {
nodesDestroyNode(pSelect);
return TSDB_CODE_OUT_OF_MEMORY;
}
strcpy(pFunc->functionName, "_wstartts");
nodesListPushFront(pSelect->pProjectionList, pFunc);
SNode* pProject = NULL;
FOREACH(pProject, pSelect->pProjectionList) { sprintf(((SExprNode*)pProject)->aliasName, "#sma_%p", pProject); }
SIntervalWindowNode* pInterval = nodesMakeNode(QUERY_NODE_INTERVAL_WINDOW);
if (NULL == pInterval) {
nodesDestroyNode(pSelect);
return TSDB_CODE_OUT_OF_MEMORY;
}
pSelect->pWindow = (SNode*)pInterval;
pInterval->pCol = nodesMakeNode(QUERY_NODE_COLUMN);
pInterval->pInterval = nodesCloneNode(pStmt->pOptions->pInterval);
pInterval->pOffset = nodesCloneNode(pStmt->pOptions->pOffset);
pInterval->pSliding = nodesCloneNode(pStmt->pOptions->pSliding);
if (NULL == pInterval->pCol || NULL == pInterval->pInterval ||
(NULL != pStmt->pOptions->pOffset && NULL == pInterval->pOffset) ||
(NULL != pStmt->pOptions->pSliding && NULL == pInterval->pSliding)) {
nodesDestroyNode(pSelect);
return TSDB_CODE_OUT_OF_MEMORY;
}
((SColumnNode*)pInterval->pCol)->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
strcpy(((SColumnNode*)pInterval->pCol)->colName, PK_TS_COL_INTERNAL_NAME);
int32_t code = translateQuery(pCxt, (SNode*)pSelect);
static int32_t getSmaIndexAst(STranslateContext* pCxt, SCreateIndexStmt* pStmt, char** pAst, int32_t* pLen) {
SSampleAstInfo info = {0};
int32_t code = buildSampleAstInfoByIndex(pCxt, pStmt, &info);
if (TSDB_CODE_SUCCESS == code) {
code = nodesNodeToString(pSelect, false, pAst, pLen);
code = buildSampleAst(pCxt, &info, pAst, pLen);
}
nodesDestroyNode(pSelect);
clearSampleAstInfo(&info);
return code;
}
@ -2106,7 +2359,7 @@ static int32_t buildCreateSmaReq(STranslateContext* pCxt, SCreateIndexStmt* pStm
code = getSmaIndexExpr(pCxt, pStmt, &pReq->expr, &pReq->exprLen);
}
if (TSDB_CODE_SUCCESS == code) {
code = getSmaIndexBuildAst(pCxt, pStmt, &pReq->ast, &pReq->astLen);
code = getSmaIndexAst(pCxt, pStmt, &pReq->ast, &pReq->astLen);
}
return code;
@ -2129,10 +2382,12 @@ static int32_t translateCreateSmaIndex(STranslateContext* pCxt, SCreateIndexStmt
tFreeSMCreateSmaReq(&createSmaReq);
return code;
}
static int32_t buildCreateFullTextReq(STranslateContext* pCxt, SCreateIndexStmt* pStmt, SMCreateFullTextReq* pReq) {
// impl later
return 0;
}
static int32_t translateCreateFullTextIndex(STranslateContext* pCxt, SCreateIndexStmt* pStmt) {
SMCreateFullTextReq createFTReq = {0};
int32_t code = buildCreateFullTextReq(pCxt, pStmt, &createFTReq);
@ -2535,24 +2790,6 @@ int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pS
return TSDB_CODE_FAILED;
}
static void destroyTranslateContext(STranslateContext* pCxt) {
if (NULL != pCxt->pNsLevel) {
size_t size = taosArrayGetSize(pCxt->pNsLevel);
for (size_t i = 0; i < size; ++i) {
taosArrayDestroy(taosArrayGetP(pCxt->pNsLevel, i));
}
taosArrayDestroy(pCxt->pNsLevel);
}
if (NULL != pCxt->pCmdMsg) {
taosMemoryFreeClear(pCxt->pCmdMsg->pMsg);
taosMemoryFreeClear(pCxt->pCmdMsg);
}
taosHashCleanup(pCxt->pDbs);
taosHashCleanup(pCxt->pTables);
}
static const char* getSysDbName(ENodeType type) {
switch (type) {
case QUERY_NODE_SHOW_DATABASES_STMT:
@ -2613,8 +2850,9 @@ static const char* getSysTableName(ENodeType type) {
case QUERY_NODE_SHOW_LICENCE_STMT:
return TSDB_INS_TABLE_LICENCES;
case QUERY_NODE_SHOW_CONNECTIONS_STMT:
return TSDB_PERFS_TABLE_CONNECTIONS;
case QUERY_NODE_SHOW_QUERIES_STMT:
// todo
return TSDB_PERFS_TABLE_QUERIES;
default:
break;
}
@ -2739,18 +2977,6 @@ typedef struct SVgroupTablesBatch {
char dbName[TSDB_DB_NAME_LEN];
} SVgroupTablesBatch;
static void toSchemaEx(const SColumnDefNode* pCol, col_id_t colId, SSchema* pSchema) {
int8_t flags = 0;
if (pCol->sma) {
flags |= SCHEMA_SMA_ON;
}
pSchema->colId = colId;
pSchema->type = pCol->dataType.type;
pSchema->bytes = calcTypeBytes(pCol->dataType);
pSchema->flags = flags;
strcpy(pSchema->name, pCol->colName);
}
static void destroyCreateTbReq(SVCreateTbReq* pReq) {
taosMemoryFreeClear(pReq->name);
taosMemoryFreeClear(pReq->ntbCfg.pSchema);
@ -2798,7 +3024,7 @@ static int32_t buildNormalTableBatchReq(int32_t acctId, const SCreateTableStmt*
SNode* pCol;
col_id_t index = 0;
FOREACH(pCol, pStmt->pCols) {
toSchemaEx((SColumnDefNode*)pCol, index + 1, req.ntbCfg.pSchema + index);
toSchema((SColumnDefNode*)pCol, index + 1, req.ntbCfg.pSchema + index);
++index;
}
if (TSDB_CODE_SUCCESS != buildSmaParam(pStmt->pOptions, &req)) {
@ -3224,19 +3450,11 @@ static int32_t setQuery(STranslateContext* pCxt, SQuery* pQuery) {
}
int32_t translate(SParseContext* pParseCxt, SQuery* pQuery) {
STranslateContext cxt = {
.pParseCxt = pParseCxt,
.errCode = TSDB_CODE_SUCCESS,
.msgBuf = {.buf = pParseCxt->pMsg, .len = pParseCxt->msgLen},
.pNsLevel = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES),
.currLevel = 0,
.currClause = 0,
.pDbs = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK),
.pTables = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK)};
if (NULL == cxt.pNsLevel) {
return TSDB_CODE_OUT_OF_MEMORY;
STranslateContext cxt = {0};
int32_t code = initTranslateContext(pParseCxt, &cxt);
if (TSDB_CODE_SUCCESS == code) {
code = fmFuncMgtInit();
}
int32_t code = fmFuncMgtInit();
if (TSDB_CODE_SUCCESS == code) {
code = rewriteQuery(&cxt, pQuery);
}

5048
source/libs/parser/src/sql.c
View File

File diff suppressed because it is too large Load Diff