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

feat: support database auto compact parameters

This commit is contained in:
Hongze Cheng 2024-11-14 17:01:15 +08:00
parent 1f51f480af
commit be743733a5
12 changed files with 791 additions and 391 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
include/common/tmsg.h
View File

@ -422,7 +422,7 @@ typedef enum ENodeType {
// physical plan node // physical plan node
QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN = 1100, QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN = 1100,
QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN,
QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN, // INACTIVE QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN, // INACTIVE
QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN,
QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN,
QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN, QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN,
@ -436,7 +436,7 @@ typedef enum ENodeType {
QUERY_NODE_PHYSICAL_PLAN_SORT, QUERY_NODE_PHYSICAL_PLAN_SORT,
QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT, QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT,
QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL, QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL, // INACTIVE QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL, // INACTIVE
QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL, QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL, QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL, QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL,
@ -677,7 +677,7 @@ typedef struct {
int32_t tsSlowLogThreshold; int32_t tsSlowLogThreshold;
int32_t tsSlowLogMaxLen; int32_t tsSlowLogMaxLen;
int32_t tsSlowLogScope; int32_t tsSlowLogScope;
int32_t tsSlowLogThresholdTest; //Obsolete int32_t tsSlowLogThresholdTest; // Obsolete
char tsSlowLogExceptDb[TSDB_DB_NAME_LEN]; char tsSlowLogExceptDb[TSDB_DB_NAME_LEN];
} SMonitorParas; } SMonitorParas;
@ -986,7 +986,6 @@ typedef struct SEpSet {
SEp eps[TSDB_MAX_REPLICA]; SEp eps[TSDB_MAX_REPLICA];
} SEpSet; } SEpSet;
int32_t tEncodeSEpSet(SEncoder* pEncoder, const SEpSet* pEp); int32_t tEncodeSEpSet(SEncoder* pEncoder, const SEpSet* pEp);
int32_t tDecodeSEpSet(SDecoder* pDecoder, SEpSet* pEp); int32_t tDecodeSEpSet(SDecoder* pDecoder, SEpSet* pEp);
int32_t taosEncodeSEpSet(void** buf, const SEpSet* pEp); int32_t taosEncodeSEpSet(void** buf, const SEpSet* pEp);
@ -1343,6 +1342,11 @@ typedef struct {
int8_t withArbitrator; int8_t withArbitrator;
int8_t encryptAlgorithm; int8_t encryptAlgorithm;
char dnodeListStr[TSDB_DNODE_LIST_LEN]; char dnodeListStr[TSDB_DNODE_LIST_LEN];
// 1. add auto-compact parameters
int32_t compactInterval;
int32_t compactStartTime;
int32_t compactEndTime;
int32_t compactTimeOffset;
} SCreateDbReq; } SCreateDbReq;
int32_t tSerializeSCreateDbReq(void* buf, int32_t bufLen, SCreateDbReq* pReq); int32_t tSerializeSCreateDbReq(void* buf, int32_t bufLen, SCreateDbReq* pReq);
@ -1374,6 +1378,11 @@ typedef struct {
int32_t sqlLen; int32_t sqlLen;
char* sql; char* sql;
int8_t withArbitrator; int8_t withArbitrator;
// 1. add auto-compact parameters
int32_t compactInterval;
int32_t compactStartTime;
int32_t compactEndTime;
int32_t compactTimeOffset;
} SAlterDbReq; } SAlterDbReq;
int32_t tSerializeSAlterDbReq(void* buf, int32_t bufLen, SAlterDbReq* pReq); int32_t tSerializeSAlterDbReq(void* buf, int32_t bufLen, SAlterDbReq* pReq);
@ -4147,20 +4156,20 @@ typedef struct {
SArray* blockTbName; SArray* blockTbName;
SArray* blockSchema; SArray* blockSchema;
union{ union {
struct{ struct {
int64_t sleepTime; int64_t sleepTime;
}; };
struct{ struct {
int32_t createTableNum; int32_t createTableNum;
SArray* createTableLen; SArray* createTableLen;
SArray* createTableReq; SArray* createTableReq;
}; };
}; };
} SMqDataRsp; } SMqDataRsp;
int32_t tEncodeMqDataRsp(SEncoder *pEncoder, const SMqDataRsp *pObj); int32_t tEncodeMqDataRsp(SEncoder* pEncoder, const SMqDataRsp* pObj);
int32_t tDecodeMqDataRsp(SDecoder* pDecoder, SMqDataRsp* pRsp); int32_t tDecodeMqDataRsp(SDecoder* pDecoder, SMqDataRsp* pRsp);
void tDeleteMqDataRsp(SMqDataRsp* pRsp); void tDeleteMqDataRsp(SMqDataRsp* pRsp);

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

@ -109,6 +109,14 @@ typedef struct SDatabaseOptions {
SValueNode* s3KeepLocalStr; SValueNode* s3KeepLocalStr;
int8_t s3Compact; int8_t s3Compact;
int8_t withArbitrator; int8_t withArbitrator;
// for auto-compact
SValueNode* pCompactIntervalNode;
int32_t compactInterval;
SNodeList* pCompactTimeRangeList;
int32_t compactStartTime;
int32_t compactEndTime;
SValueNode* pCompactTimeOffsetNode;
int32_t compactTimeOffset;
} SDatabaseOptions; } SDatabaseOptions;
typedef struct SCreateDatabaseStmt { typedef struct SCreateDatabaseStmt {
@ -348,7 +356,7 @@ typedef struct SShowStmt {
SNode* pTbName; // SValueNode SNode* pTbName; // SValueNode
EOperatorType tableCondType; EOperatorType tableCondType;
EShowKind showKind; // show databases: user/system, show tables: normal/child, others NULL EShowKind showKind; // show databases: user/system, show tables: normal/child, others NULL
bool withFull; // for show users full; bool withFull; // for show users full;
} SShowStmt; } SShowStmt;
typedef struct SShowCreateDatabaseStmt { typedef struct SShowCreateDatabaseStmt {
@ -658,7 +666,7 @@ typedef struct SCreateTSMAStmt {
bool ignoreExists; bool ignoreExists;
char tsmaName[TSDB_TABLE_NAME_LEN]; char tsmaName[TSDB_TABLE_NAME_LEN];
char dbName[TSDB_DB_NAME_LEN]; char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN]; // base tb name or base tsma name char tableName[TSDB_TABLE_NAME_LEN]; // base tb name or base tsma name
char originalTbName[TSDB_TABLE_NAME_LEN]; char originalTbName[TSDB_TABLE_NAME_LEN];
STSMAOptions* pOptions; STSMAOptions* pOptions;
SNode* pPrevQuery; SNode* pPrevQuery;
@ -667,10 +675,10 @@ typedef struct SCreateTSMAStmt {
} SCreateTSMAStmt; } SCreateTSMAStmt;
typedef struct SDropTSMAStmt { typedef struct SDropTSMAStmt {
ENodeType type; ENodeType type;
bool ignoreNotExists; bool ignoreNotExists;
char dbName[TSDB_DB_NAME_LEN]; char dbName[TSDB_DB_NAME_LEN];
char tsmaName[TSDB_TABLE_NAME_LEN]; char tsmaName[TSDB_TABLE_NAME_LEN];
} SDropTSMAStmt; } SDropTSMAStmt;
#ifdef __cplusplus #ifdef __cplusplus

54
include/util/tdef.h
View File

@ -190,45 +190,22 @@ typedef enum EOperatorType {
} EOperatorType; } EOperatorType;
static const EOperatorType OPERATOR_ARRAY[] = { static const EOperatorType OPERATOR_ARRAY[] = {
OP_TYPE_ADD, OP_TYPE_ADD, OP_TYPE_SUB, OP_TYPE_MULTI, OP_TYPE_DIV, OP_TYPE_REM,
OP_TYPE_SUB,
OP_TYPE_MULTI,
OP_TYPE_DIV,
OP_TYPE_REM,
OP_TYPE_MINUS,
OP_TYPE_BIT_AND,
OP_TYPE_BIT_OR,
OP_TYPE_GREATER_THAN, OP_TYPE_MINUS,
OP_TYPE_GREATER_EQUAL,
OP_TYPE_LOWER_THAN,
OP_TYPE_LOWER_EQUAL,
OP_TYPE_EQUAL,
OP_TYPE_NOT_EQUAL,
OP_TYPE_IN,
OP_TYPE_NOT_IN,
OP_TYPE_LIKE,
OP_TYPE_NOT_LIKE,
OP_TYPE_MATCH,
OP_TYPE_NMATCH,
OP_TYPE_IS_NULL,
OP_TYPE_IS_NOT_NULL,
OP_TYPE_IS_TRUE,
OP_TYPE_IS_FALSE,
OP_TYPE_IS_UNKNOWN,
OP_TYPE_IS_NOT_TRUE,
OP_TYPE_IS_NOT_FALSE,
OP_TYPE_IS_NOT_UNKNOWN,
//OP_TYPE_COMPARE_MAX_VALUE,
OP_TYPE_JSON_GET_VALUE, OP_TYPE_BIT_AND, OP_TYPE_BIT_OR,
OP_TYPE_JSON_CONTAINS,
OP_TYPE_ASSIGN OP_TYPE_GREATER_THAN, OP_TYPE_GREATER_EQUAL, OP_TYPE_LOWER_THAN, OP_TYPE_LOWER_EQUAL, OP_TYPE_EQUAL,
}; OP_TYPE_NOT_EQUAL, OP_TYPE_IN, OP_TYPE_NOT_IN, OP_TYPE_LIKE, OP_TYPE_NOT_LIKE, OP_TYPE_MATCH, OP_TYPE_NMATCH,
OP_TYPE_IS_NULL, OP_TYPE_IS_NOT_NULL, OP_TYPE_IS_TRUE, OP_TYPE_IS_FALSE, OP_TYPE_IS_UNKNOWN, OP_TYPE_IS_NOT_TRUE,
OP_TYPE_IS_NOT_FALSE, OP_TYPE_IS_NOT_UNKNOWN,
// OP_TYPE_COMPARE_MAX_VALUE,
OP_TYPE_JSON_GET_VALUE, OP_TYPE_JSON_CONTAINS,
OP_TYPE_ASSIGN};
#define OP_TYPE_CALC_MAX OP_TYPE_BIT_OR #define OP_TYPE_CALC_MAX OP_TYPE_BIT_OR
@ -528,6 +505,11 @@ typedef enum ELogicConditionType {
#define TSDB_MIN_TABLE_TTL 0 #define TSDB_MIN_TABLE_TTL 0
#define TSDB_DEFAULT_TABLE_TTL 0 #define TSDB_DEFAULT_TABLE_TTL 0
#define TSDB_DEFAULT_COMPACT_INTERVAL 0
#define TSDB_DEFAULT_COMPACT_START_TIME 0
#define TSDB_DEFAULT_COMPACT_END_TIME 0
#define TSDB_DEFAULT_COMPACT_TIME_OFFSET 0
#define TSDB_MIN_EXPLAIN_RATIO 0 #define TSDB_MIN_EXPLAIN_RATIO 0
#define TSDB_MAX_EXPLAIN_RATIO 1 #define TSDB_MAX_EXPLAIN_RATIO 1
#define TSDB_DEFAULT_EXPLAIN_RATIO 0.001 #define TSDB_DEFAULT_EXPLAIN_RATIO 0.001

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

@ -76,7 +76,7 @@ static int32_t tSerializeSMonitorParas(SEncoder *encoder, const SMonitorParas *p
TAOS_CHECK_RETURN(tEncodeI32(encoder, pMonitorParas->tsSlowLogScope)); TAOS_CHECK_RETURN(tEncodeI32(encoder, pMonitorParas->tsSlowLogScope));
TAOS_CHECK_RETURN(tEncodeI32(encoder, pMonitorParas->tsSlowLogMaxLen)); TAOS_CHECK_RETURN(tEncodeI32(encoder, pMonitorParas->tsSlowLogMaxLen));
TAOS_CHECK_RETURN(tEncodeI32(encoder, pMonitorParas->tsSlowLogThreshold)); TAOS_CHECK_RETURN(tEncodeI32(encoder, pMonitorParas->tsSlowLogThreshold));
TAOS_CHECK_RETURN(tEncodeI32(encoder, pMonitorParas->tsSlowLogThresholdTest)); //Obsolete TAOS_CHECK_RETURN(tEncodeI32(encoder, pMonitorParas->tsSlowLogThresholdTest)); // Obsolete
TAOS_CHECK_RETURN(tEncodeCStr(encoder, pMonitorParas->tsSlowLogExceptDb)); TAOS_CHECK_RETURN(tEncodeCStr(encoder, pMonitorParas->tsSlowLogExceptDb));
return 0; return 0;
} }
@ -87,7 +87,7 @@ static int32_t tDeserializeSMonitorParas(SDecoder *decoder, SMonitorParas *pMoni
TAOS_CHECK_RETURN(tDecodeI32(decoder, &pMonitorParas->tsSlowLogScope)); TAOS_CHECK_RETURN(tDecodeI32(decoder, &pMonitorParas->tsSlowLogScope));
TAOS_CHECK_RETURN(tDecodeI32(decoder, &pMonitorParas->tsSlowLogMaxLen)); TAOS_CHECK_RETURN(tDecodeI32(decoder, &pMonitorParas->tsSlowLogMaxLen));
TAOS_CHECK_RETURN(tDecodeI32(decoder, &pMonitorParas->tsSlowLogThreshold)); TAOS_CHECK_RETURN(tDecodeI32(decoder, &pMonitorParas->tsSlowLogThreshold));
TAOS_CHECK_RETURN(tDecodeI32(decoder, &pMonitorParas->tsSlowLogThresholdTest)); //Obsolete TAOS_CHECK_RETURN(tDecodeI32(decoder, &pMonitorParas->tsSlowLogThresholdTest)); // Obsolete
TAOS_CHECK_RETURN(tDecodeCStrTo(decoder, pMonitorParas->tsSlowLogExceptDb)); TAOS_CHECK_RETURN(tDecodeCStrTo(decoder, pMonitorParas->tsSlowLogExceptDb));
return 0; return 0;
} }
@ -2166,9 +2166,9 @@ int32_t tSerializeRetrieveAnalAlgoRsp(void *buf, int32_t bufLen, SRetrieveAnalAl
int32_t numOfAlgos = 0; int32_t numOfAlgos = 0;
void *pIter = taosHashIterate(pRsp->hash, NULL); void *pIter = taosHashIterate(pRsp->hash, NULL);
while (pIter != NULL) { while (pIter != NULL) {
SAnalyticsUrl *pUrl = pIter; SAnalyticsUrl *pUrl = pIter;
size_t nameLen = 0; size_t nameLen = 0;
const char *name = taosHashGetKey(pIter, &nameLen); const char *name = taosHashGetKey(pIter, &nameLen);
if (nameLen > 0 && nameLen <= TSDB_ANAL_ALGO_KEY_LEN && pUrl->urlLen > 0) { if (nameLen > 0 && nameLen <= TSDB_ANAL_ALGO_KEY_LEN && pUrl->urlLen > 0) {
numOfAlgos++; numOfAlgos++;
} }
@ -2181,9 +2181,9 @@ int32_t tSerializeRetrieveAnalAlgoRsp(void *buf, int32_t bufLen, SRetrieveAnalAl
pIter = taosHashIterate(pRsp->hash, NULL); pIter = taosHashIterate(pRsp->hash, NULL);
while (pIter != NULL) { while (pIter != NULL) {
SAnalyticsUrl *pUrl = pIter; SAnalyticsUrl *pUrl = pIter;
size_t nameLen = 0; size_t nameLen = 0;
const char *name = taosHashGetKey(pIter, &nameLen); const char *name = taosHashGetKey(pIter, &nameLen);
if (nameLen > 0 && pUrl->urlLen > 0) { if (nameLen > 0 && pUrl->urlLen > 0) {
TAOS_CHECK_EXIT(tEncodeI32(&encoder, nameLen)); TAOS_CHECK_EXIT(tEncodeI32(&encoder, nameLen));
TAOS_CHECK_EXIT(tEncodeBinary(&encoder, (const uint8_t *)name, nameLen)); TAOS_CHECK_EXIT(tEncodeBinary(&encoder, (const uint8_t *)name, nameLen));
@ -2221,10 +2221,10 @@ int32_t tDeserializeRetrieveAnalAlgoRsp(void *buf, int32_t bufLen, SRetrieveAnal
int32_t lino; int32_t lino;
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
int32_t numOfAlgos = 0; int32_t numOfAlgos = 0;
int32_t nameLen; int32_t nameLen;
int32_t type; int32_t type;
char name[TSDB_ANAL_ALGO_KEY_LEN]; char name[TSDB_ANAL_ALGO_KEY_LEN];
SAnalyticsUrl url = {0}; SAnalyticsUrl url = {0};
TAOS_CHECK_EXIT(tStartDecode(&decoder)); TAOS_CHECK_EXIT(tStartDecode(&decoder));
@ -3937,6 +3937,12 @@ int32_t tSerializeSCreateDbReq(void *buf, int32_t bufLen, SCreateDbReq *pReq) {
TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->s3Compact)); TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->s3Compact));
TAOS_CHECK_EXIT(tEncodeCStr(&encoder, pReq->dnodeListStr)); TAOS_CHECK_EXIT(tEncodeCStr(&encoder, pReq->dnodeListStr));
// auto-compact parameters
TAOS_CHECK_EXIT(tEncodeI32v(&encoder, pReq->compactInterval));
TAOS_CHECK_EXIT(tEncodeI32v(&encoder, pReq->compactStartTime));
TAOS_CHECK_EXIT(tEncodeI32v(&encoder, pReq->compactEndTime));
TAOS_CHECK_EXIT(tEncodeI32v(&encoder, pReq->compactTimeOffset));
tEndEncode(&encoder); tEndEncode(&encoder);
_exit: _exit:
@ -4028,6 +4034,18 @@ int32_t tDeserializeSCreateDbReq(void *buf, int32_t bufLen, SCreateDbReq *pReq)
TAOS_CHECK_EXIT(tDecodeCStrTo(&decoder, pReq->dnodeListStr)); TAOS_CHECK_EXIT(tDecodeCStrTo(&decoder, pReq->dnodeListStr));
} }
if (!tDecodeIsEnd(&decoder)) {
TAOS_CHECK_EXIT(tDecodeI32v(&decoder, &pReq->compactInterval));
TAOS_CHECK_EXIT(tDecodeI32v(&decoder, &pReq->compactStartTime));
TAOS_CHECK_EXIT(tDecodeI32v(&decoder, &pReq->compactEndTime));
TAOS_CHECK_EXIT(tDecodeI32v(&decoder, &pReq->compactTimeOffset));
} else {
pReq->compactInterval = 0;
pReq->compactStartTime = 0;
pReq->compactEndTime = 0;
pReq->compactTimeOffset = 0;
}
tEndDecode(&decoder); tEndDecode(&decoder);
_exit: _exit:
@ -4078,6 +4096,11 @@ int32_t tSerializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
ENCODESQL(); ENCODESQL();
TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->withArbitrator)); TAOS_CHECK_EXIT(tEncodeI8(&encoder, pReq->withArbitrator));
// auto compact config
TAOS_CHECK_EXIT(tEncodeI32v(&encoder, pReq->compactInterval));
TAOS_CHECK_EXIT(tEncodeI32v(&encoder, pReq->compactStartTime));
TAOS_CHECK_EXIT(tEncodeI32v(&encoder, pReq->compactEndTime));
TAOS_CHECK_EXIT(tEncodeI32v(&encoder, pReq->compactTimeOffset));
tEndEncode(&encoder); tEndEncode(&encoder);
_exit: _exit:
@ -4145,6 +4168,19 @@ int32_t tDeserializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
if (!tDecodeIsEnd(&decoder)) { if (!tDecodeIsEnd(&decoder)) {
TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->withArbitrator)); TAOS_CHECK_EXIT(tDecodeI8(&decoder, &pReq->withArbitrator));
} }
// auto compact config
if (!tDecodeIsEnd(&decoder)) {
TAOS_CHECK_EXIT(tDecodeI32v(&decoder, &pReq->compactInterval));
TAOS_CHECK_EXIT(tDecodeI32v(&decoder, &pReq->compactStartTime));
TAOS_CHECK_EXIT(tDecodeI32v(&decoder, &pReq->compactEndTime));
TAOS_CHECK_EXIT(tDecodeI32v(&decoder, &pReq->compactTimeOffset));
} else {
pReq->compactInterval = 0;
pReq->compactStartTime = 0;
pReq->compactEndTime = 0;
pReq->compactTimeOffset = 0;
}
tEndDecode(&decoder); tEndDecode(&decoder);
_exit: _exit:

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

@ -106,8 +106,8 @@ typedef enum {
TRN_CONFLICT_GLOBAL = 1, TRN_CONFLICT_GLOBAL = 1,
TRN_CONFLICT_DB = 2, TRN_CONFLICT_DB = 2,
TRN_CONFLICT_DB_INSIDE = 3, TRN_CONFLICT_DB_INSIDE = 3,
// TRN_CONFLICT_TOPIC = 4, // TRN_CONFLICT_TOPIC = 4,
// TRN_CONFLICT_TOPIC_INSIDE = 5, // TRN_CONFLICT_TOPIC_INSIDE = 5,
TRN_CONFLICT_ARBGROUP = 6, TRN_CONFLICT_ARBGROUP = 6,
TRN_CONFLICT_TSMA = 7, TRN_CONFLICT_TSMA = 7,
} ETrnConflct; } ETrnConflct;
@ -423,6 +423,10 @@ typedef struct {
int8_t s3Compact; int8_t s3Compact;
int8_t withArbitrator; int8_t withArbitrator;
int8_t encryptAlgorithm; int8_t encryptAlgorithm;
int32_t compactInterval;
int32_t compactStartTime;
int32_t compactEndTime;
int32_t compactTimeOffset;
} SDbCfg; } SDbCfg;
typedef struct { typedef struct {
@ -649,12 +653,12 @@ typedef struct {
int32_t maxPollIntervalMs; int32_t maxPollIntervalMs;
} SMqConsumerObj; } SMqConsumerObj;
int32_t tNewSMqConsumerObj(int64_t consumerId, char *cgroup, int8_t updateType, int32_t tNewSMqConsumerObj(int64_t consumerId, char* cgroup, int8_t updateType, char* topic, SCMSubscribeReq* subscribe,
char *topic, SCMSubscribeReq *subscribe, SMqConsumerObj** ppConsumer); SMqConsumerObj** ppConsumer);
void tClearSMqConsumerObj(SMqConsumerObj* pConsumer); void tClearSMqConsumerObj(SMqConsumerObj* pConsumer);
void tDeleteSMqConsumerObj(SMqConsumerObj* pConsumer); void tDeleteSMqConsumerObj(SMqConsumerObj* pConsumer);
int32_t tEncodeSMqConsumerObj(void** buf, const SMqConsumerObj* pConsumer); int32_t tEncodeSMqConsumerObj(void** buf, const SMqConsumerObj* pConsumer);
void* tDecodeSMqConsumerObj(const void* buf, SMqConsumerObj* pConsumer, int8_t sver); void* tDecodeSMqConsumerObj(const void* buf, SMqConsumerObj* pConsumer, int8_t sver);
typedef struct { typedef struct {
int32_t vgId; int32_t vgId;
@ -693,11 +697,11 @@ typedef struct {
char* qmsg; // SubPlanToString char* qmsg; // SubPlanToString
} SMqSubscribeObj; } SMqSubscribeObj;
int32_t tNewSubscribeObj(const char *key, SMqSubscribeObj **ppSub); int32_t tNewSubscribeObj(const char* key, SMqSubscribeObj** ppSub);
int32_t tCloneSubscribeObj(const SMqSubscribeObj* pSub, SMqSubscribeObj **ppSub); int32_t tCloneSubscribeObj(const SMqSubscribeObj* pSub, SMqSubscribeObj** ppSub);
void tDeleteSubscribeObj(SMqSubscribeObj* pSub); void tDeleteSubscribeObj(SMqSubscribeObj* pSub);
int32_t tEncodeSubscribeObj(void** buf, const SMqSubscribeObj* pSub); int32_t tEncodeSubscribeObj(void** buf, const SMqSubscribeObj* pSub);
void* tDecodeSubscribeObj(const void* buf, SMqSubscribeObj* pSub, int8_t sver); void* tDecodeSubscribeObj(const void* buf, SMqSubscribeObj* pSub, int8_t sver);
// typedef struct { // typedef struct {
// int32_t epoch; // int32_t epoch;

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

@ -16,6 +16,7 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "mndDb.h" #include "mndDb.h"
#include "audit.h" #include "audit.h"
#include "command.h"
#include "mndArbGroup.h" #include "mndArbGroup.h"
#include "mndCluster.h" #include "mndCluster.h"
#include "mndDnode.h" #include "mndDnode.h"
@ -34,7 +35,6 @@
#include "systable.h" #include "systable.h"
#include "thttp.h" #include "thttp.h"
#include "tjson.h" #include "tjson.h"
#include "command.h"
#define DB_VER_NUMBER 1 #define DB_VER_NUMBER 1
#define DB_RESERVE_SIZE 27 #define DB_RESERVE_SIZE 27
@ -795,6 +795,10 @@ static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate,
.tsdbPageSize = pCreate->tsdbPageSize, .tsdbPageSize = pCreate->tsdbPageSize,
.withArbitrator = pCreate->withArbitrator, .withArbitrator = pCreate->withArbitrator,
.encryptAlgorithm = pCreate->encryptAlgorithm, .encryptAlgorithm = pCreate->encryptAlgorithm,
.compactInterval = pCreate->compactInterval,
.compactStartTime = pCreate->compactStartTime,
.compactEndTime = pCreate->compactEndTime,
.compactTimeOffset = pCreate->compactTimeOffset,
}; };
dbObj.cfg.numOfRetensions = pCreate->numOfRetensions; dbObj.cfg.numOfRetensions = pCreate->numOfRetensions;
@ -1138,6 +1142,32 @@ static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
code = 0; code = 0;
} }
if (pAlter->compactInterval != TSDB_DEFAULT_COMPACT_INTERVAL && pAlter->compactInterval != pDb->cfg.compactInterval) {
pDb->cfg.compactInterval = pAlter->compactInterval;
pDb->vgVersion++;
code = 0;
}
if (pAlter->compactStartTime != TSDB_DEFAULT_COMPACT_START_TIME &&
pAlter->compactStartTime != pDb->cfg.compactStartTime) {
pDb->cfg.compactStartTime = pAlter->compactStartTime;
pDb->vgVersion++;
code = 0;
}
if (pAlter->compactEndTime != TSDB_DEFAULT_COMPACT_END_TIME && pAlter->compactEndTime != pDb->cfg.compactEndTime) {
pDb->cfg.compactEndTime = pAlter->compactEndTime;
pDb->vgVersion++;
code = 0;
}
if (pAlter->compactTimeOffset != TSDB_DEFAULT_COMPACT_TIME_OFFSET &&
pAlter->compactTimeOffset != pDb->cfg.compactTimeOffset) {
pDb->cfg.compactTimeOffset = pAlter->compactTimeOffset;
pDb->vgVersion++;
code = 0;
}
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -1560,8 +1590,8 @@ static int32_t mndBuildDropVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *
static int32_t mndSetDropDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) { static int32_t mndSetDropDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
int32_t code = 0; int32_t code = 0;
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL; void *pIter = NULL;
while (1) { while (1) {
SVgObj *pVgroup = NULL; SVgObj *pVgroup = NULL;
@ -1941,9 +1971,9 @@ int32_t mndValidateDbInfo(SMnode *pMnode, SDbCacheInfo *pDbs, int32_t numOfDbs,
continue; continue;
} else { } else {
mTrace("db:%s, valid dbinfo, vgVersion:%d cfgVersion:%d stateTs:%" PRId64 mTrace("db:%s, valid dbinfo, vgVersion:%d cfgVersion:%d stateTs:%" PRId64
" numOfTables:%d, changed to vgVersion:%d cfgVersion:%d stateTs:%" PRId64 " numOfTables:%d", " numOfTables:%d, changed to vgVersion:%d cfgVersion:%d stateTs:%" PRId64 " numOfTables:%d",
pDbCacheInfo->dbFName, pDbCacheInfo->vgVersion, pDbCacheInfo->cfgVersion, pDbCacheInfo->stateTs, pDbCacheInfo->dbFName, pDbCacheInfo->vgVersion, pDbCacheInfo->cfgVersion, pDbCacheInfo->stateTs,
pDbCacheInfo->numOfTable, pDb->vgVersion, pDb->cfgVersion, pDb->stateTs, numOfTable); pDbCacheInfo->numOfTable, pDb->vgVersion, pDb->cfgVersion, pDb->stateTs, numOfTable);
} }
if (pDbCacheInfo->cfgVersion < pDb->cfgVersion) { if (pDbCacheInfo->cfgVersion < pDb->cfgVersion) {
@ -1955,7 +1985,7 @@ int32_t mndValidateDbInfo(SMnode *pMnode, SDbCacheInfo *pDbs, int32_t numOfDbs,
rsp.pTsmaRsp = taosMemoryCalloc(1, sizeof(STableTSMAInfoRsp)); rsp.pTsmaRsp = taosMemoryCalloc(1, sizeof(STableTSMAInfoRsp));
if (rsp.pTsmaRsp) rsp.pTsmaRsp->pTsmas = taosArrayInit(4, POINTER_BYTES); if (rsp.pTsmaRsp) rsp.pTsmaRsp->pTsmas = taosArrayInit(4, POINTER_BYTES);
if (rsp.pTsmaRsp && rsp.pTsmaRsp->pTsmas) { if (rsp.pTsmaRsp && rsp.pTsmaRsp->pTsmas) {
bool exist = false; bool exist = false;
int32_t code = mndGetDbTsmas(pMnode, 0, pDb->uid, rsp.pTsmaRsp, &exist); int32_t code = mndGetDbTsmas(pMnode, 0, pDb->uid, rsp.pTsmaRsp, &exist);
if (TSDB_CODE_SUCCESS != code) { if (TSDB_CODE_SUCCESS != code) {
mndReleaseDb(pMnode, pDb); mndReleaseDb(pMnode, pDb);
@ -2386,7 +2416,8 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)strictVstr, false), &lino, _OVER); TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)strictVstr, false), &lino, _OVER);
char durationVstr[128] = {0}; char durationVstr[128] = {0};
int32_t len = formatDurationOrKeep(&durationVstr[VARSTR_HEADER_SIZE], sizeof(durationVstr) - VARSTR_HEADER_SIZE, pDb->cfg.daysPerFile); int32_t len = formatDurationOrKeep(&durationVstr[VARSTR_HEADER_SIZE], sizeof(durationVstr) - VARSTR_HEADER_SIZE,
pDb->cfg.daysPerFile);
varDataSetLen(durationVstr, len); varDataSetLen(durationVstr, len);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
@ -2402,9 +2433,9 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
int32_t lenKeep2 = formatDurationOrKeep(keep2Str, sizeof(keep2Str), pDb->cfg.daysToKeep2); int32_t lenKeep2 = formatDurationOrKeep(keep2Str, sizeof(keep2Str), pDb->cfg.daysToKeep2);
if (pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep1 || pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep2) { if (pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep1 || pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep2) {
len = sprintf(&keepVstr[VARSTR_HEADER_SIZE], "%s,%s,%s", keep1Str, keep2Str, keep0Str); len = sprintf(&keepVstr[VARSTR_HEADER_SIZE], "%s,%s,%s", keep1Str, keep2Str, keep0Str);
} else { } else {
len = sprintf(&keepVstr[VARSTR_HEADER_SIZE], "%s,%s,%s", keep0Str, keep1Str, keep2Str); len = sprintf(&keepVstr[VARSTR_HEADER_SIZE], "%s,%s,%s", keep0Str, keep1Str, keep2Str);
} }
varDataSetLen(keepVstr, len); varDataSetLen(keepVstr, len);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

88
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -1248,7 +1248,7 @@ static const char* jkStreamOption_destHasPrimaryKey = "StreamOptionDestHasPrimar
static int32_t streamNodeOptionToJson(const void* pObj, SJson* pJson) { static int32_t streamNodeOptionToJson(const void* pObj, SJson* pJson) {
const SStreamNodeOption* pNode = (const SStreamNodeOption*)pObj; const SStreamNodeOption* pNode = (const SStreamNodeOption*)pObj;
int32_t code = tjsonAddIntegerToObject(pJson, jkStreamOption_triggerType, pNode->triggerType); int32_t code = tjsonAddIntegerToObject(pJson, jkStreamOption_triggerType, pNode->triggerType);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkStreamOption_watermark, pNode->watermark); code = tjsonAddIntegerToObject(pJson, jkStreamOption_watermark, pNode->watermark);
@ -1270,7 +1270,7 @@ static int32_t streamNodeOptionToJson(const void* pObj, SJson* pJson) {
static int32_t jsonToStreamNodeOption(const SJson* pJson, void* pObj) { static int32_t jsonToStreamNodeOption(const SJson* pJson, void* pObj) {
SStreamNodeOption* pNode = (SStreamNodeOption*)pObj; SStreamNodeOption* pNode = (SStreamNodeOption*)pObj;
int32_t code = tjsonGetTinyIntValue(pJson, jkStreamOption_triggerType, &pNode->triggerType); int32_t code = tjsonGetTinyIntValue(pJson, jkStreamOption_triggerType, &pNode->triggerType);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkStreamOption_watermark, &pNode->watermark); code = tjsonGetBigIntValue(pJson, jkStreamOption_watermark, &pNode->watermark);
@ -1333,7 +1333,8 @@ static int32_t logicInterpFuncNodeToJson(const void* pObj, SJson* pJson) {
code = tjsonAddObject(pJson, jkInterpFuncLogicPlanTimeSeries, nodeToJson, pNode->pTimeSeries); code = tjsonAddObject(pJson, jkInterpFuncLogicPlanTimeSeries, nodeToJson, pNode->pTimeSeries);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkInterpFuncLogicPlanStreamNodeOption, streamNodeOptionToJson, &pNode->streamNodeOption); code =
tjsonAddObject(pJson, jkInterpFuncLogicPlanStreamNodeOption, streamNodeOptionToJson, &pNode->streamNodeOption);
} }
return code; return code;
@ -1371,7 +1372,8 @@ static int32_t jsonToLogicInterpFuncNode(const SJson* pJson, void* pObj) {
code = jsonToNodeObject(pJson, jkInterpFuncLogicPlanTimeSeries, &pNode->pTimeSeries); code = jsonToNodeObject(pJson, jkInterpFuncLogicPlanTimeSeries, &pNode->pTimeSeries);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonToObject(pJson, jkInterpFuncLogicPlanStreamNodeOption, jsonToStreamNodeOption, &pNode->streamNodeOption); code =
tjsonToObject(pJson, jkInterpFuncLogicPlanStreamNodeOption, jsonToStreamNodeOption, &pNode->streamNodeOption);
} }
return code; return code;
@ -2363,7 +2365,7 @@ static int32_t physiMergeJoinNodeToJson(const void* pObj, SJson* pJson) {
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanAsofOp, pNode->asofOpType); code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanAsofOp, pNode->asofOpType);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkJoinPhysiPlanLeftPrimExpr, nodeToJson, pNode->leftPrimExpr); code = tjsonAddObject(pJson, jkJoinPhysiPlanLeftPrimExpr, nodeToJson, pNode->leftPrimExpr);
} }
@ -2372,10 +2374,10 @@ static int32_t physiMergeJoinNodeToJson(const void* pObj, SJson* pJson) {
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanLeftPrimSlotId, pNode->leftPrimSlotId); code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanLeftPrimSlotId, pNode->leftPrimSlotId);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanRightPrimSlotId, pNode->rightPrimSlotId); code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanRightPrimSlotId, pNode->rightPrimSlotId);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkJoinPhysiPlanLeftEqCols, pNode->pEqLeft); code = nodeListToJson(pJson, jkJoinPhysiPlanLeftEqCols, pNode->pEqLeft);
} }
@ -2477,7 +2479,7 @@ static int32_t jsonToPhysiMergeJoinNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBoolValue(pJson, jkJoinPhysiPlanGroupJoin, &pNode->grpJoin); code = tjsonGetBoolValue(pJson, jkJoinPhysiPlanGroupJoin, &pNode->grpJoin);
} }
return code; return code;
} }
@ -2490,7 +2492,7 @@ static int32_t physiHashJoinNodeToJson(const void* pObj, SJson* pJson) {
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanSubType, pNode->subType); code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanSubType, pNode->subType);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkJoinPhysiPlanOnLeftCols, pNode->pOnLeft); code = nodeListToJson(pJson, jkJoinPhysiPlanOnLeftCols, pNode->pOnLeft);
} }
@ -2505,10 +2507,10 @@ static int32_t physiHashJoinNodeToJson(const void* pObj, SJson* pJson) {
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanLeftPrimSlotId, pNode->leftPrimSlotId); code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanLeftPrimSlotId, pNode->leftPrimSlotId);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanRightPrimSlotId, pNode->rightPrimSlotId); code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanRightPrimSlotId, pNode->rightPrimSlotId);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkJoinPhysiPlanOnConditions, nodeToJson, pNode->pFullOnCond); code = tjsonAddObject(pJson, jkJoinPhysiPlanOnConditions, nodeToJson, pNode->pFullOnCond);
} }
@ -2555,7 +2557,7 @@ static int32_t jsonToPhysiHashJoinNode(const SJson* pJson, void* pObj) {
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
tjsonGetNumberValue(pJson, jkJoinPhysiPlanSubType, pNode->subType, code); tjsonGetNumberValue(pJson, jkJoinPhysiPlanSubType, pNode->subType, code);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkJoinPhysiPlanOnLeftCols, &pNode->pOnLeft); code = jsonToNodeList(pJson, jkJoinPhysiPlanOnLeftCols, &pNode->pOnLeft);
} }
@ -2573,7 +2575,7 @@ static int32_t jsonToPhysiHashJoinNode(const SJson* pJson, void* pObj) {
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
tjsonGetNumberValue(pJson, jkJoinPhysiPlanRightPrimSlotId, pNode->rightPrimSlotId, code); tjsonGetNumberValue(pJson, jkJoinPhysiPlanRightPrimSlotId, pNode->rightPrimSlotId, code);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkJoinPhysiPlanOnConditions, &pNode->pFullOnCond); code = jsonToNodeObject(pJson, jkJoinPhysiPlanOnConditions, &pNode->pFullOnCond);
} }
@ -2600,7 +2602,7 @@ static int32_t jsonToPhysiHashJoinNode(const SJson* pJson, void* pObj) {
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
tjsonGetNumberValue(pJson, jkJoinPhysiPlanTimeRangeTarget, pNode->timeRangeTarget, code); tjsonGetNumberValue(pJson, jkJoinPhysiPlanTimeRangeTarget, pNode->timeRangeTarget, code);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkJoinPhysiPlanTimeRangeSKey, &pNode->timeRange.skey); code = tjsonGetBigIntValue(pJson, jkJoinPhysiPlanTimeRangeSKey, &pNode->timeRange.skey);
} }
@ -3350,7 +3352,8 @@ static int32_t physiInterpFuncNodeToJson(const void* pObj, SJson* pJson) {
code = tjsonAddObject(pJson, jkInterpFuncPhysiPlanTimeSeries, nodeToJson, pNode->pTimeSeries); code = tjsonAddObject(pJson, jkInterpFuncPhysiPlanTimeSeries, nodeToJson, pNode->pTimeSeries);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkInterpFuncPhysiPlanStreamNodeOption, streamNodeOptionToJson, &pNode->streamNodeOption); code =
tjsonAddObject(pJson, jkInterpFuncPhysiPlanStreamNodeOption, streamNodeOptionToJson, &pNode->streamNodeOption);
} }
return code; return code;
@ -3391,7 +3394,8 @@ static int32_t jsonToPhysiInterpFuncNode(const SJson* pJson, void* pObj) {
code = jsonToNodeObject(pJson, jkInterpFuncPhysiPlanTimeSeries, &pNode->pTimeSeries); code = jsonToNodeObject(pJson, jkInterpFuncPhysiPlanTimeSeries, &pNode->pTimeSeries);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonToObject(pJson, jkInterpFuncPhysiPlanStreamNodeOption, jsonToStreamNodeOption, &pNode->streamNodeOption); code =
tjsonToObject(pJson, jkInterpFuncPhysiPlanStreamNodeOption, jsonToStreamNodeOption, &pNode->streamNodeOption);
} }
return code; return code;
@ -4089,7 +4093,7 @@ static int32_t columnNodeToJson(const void* pObj, SJson* pJson) {
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddBoolToObject(pJson, jkColumnIsPk, pNode->isPk); code = tjsonAddBoolToObject(pJson, jkColumnIsPk, pNode->isPk);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkColumnNumOfPKs, pNode->numOfPKs); code = tjsonAddIntegerToObject(pJson, jkColumnNumOfPKs, pNode->numOfPKs);
} }
@ -4138,10 +4142,10 @@ static int32_t jsonToColumnNode(const SJson* pJson, void* pObj) {
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBoolValue(pJson, jkColumnIsPk, &pNode->isPk); code = tjsonGetBoolValue(pJson, jkColumnIsPk, &pNode->isPk);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetSmallIntValue(pJson, jkColumnNumOfPKs, &pNode->numOfPKs); code = tjsonGetSmallIntValue(pJson, jkColumnNumOfPKs, &pNode->numOfPKs);
} }
return code; return code;
} }
@ -4184,7 +4188,7 @@ static int32_t datumToJson(const void* pObj, SJson* pJson) {
case TSDB_DATA_TYPE_NCHAR: { case TSDB_DATA_TYPE_NCHAR: {
// cJSON only support utf-8 encoding. Convert memory content to hex string. // cJSON only support utf-8 encoding. Convert memory content to hex string.
int32_t bufSize = varDataLen(pNode->datum.p) * 2 + 1; int32_t bufSize = varDataLen(pNode->datum.p) * 2 + 1;
char* buf = taosMemoryCalloc(bufSize, sizeof(char)); char* buf = taosMemoryCalloc(bufSize, sizeof(char));
if (!buf) return terrno; if (!buf) return terrno;
code = taosHexEncode(varDataVal(pNode->datum.p), buf, varDataLen(pNode->datum.p), bufSize); code = taosHexEncode(varDataVal(pNode->datum.p), buf, varDataLen(pNode->datum.p), bufSize);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
@ -4551,7 +4555,7 @@ static int32_t jsonToFunctionNode(const SJson* pJson, void* pObj) {
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetIntValue(pJson, jkFunctionPkBytes, &pNode->pkBytes); code = tjsonGetIntValue(pJson, jkFunctionPkBytes, &pNode->pkBytes);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBoolValue(pJson, jkFunctionIsMergeFunc, &pNode->hasOriginalFunc); code = tjsonGetBoolValue(pJson, jkFunctionIsMergeFunc, &pNode->hasOriginalFunc);
} }
@ -5255,7 +5259,7 @@ static int32_t jsonToColumnDefNode(const SJson* pJson, void* pObj) {
code = tjsonGetBoolValue(pJson, jkColumnDefSma, &pNode->sma); code = tjsonGetBoolValue(pJson, jkColumnDefSma, &pNode->sma);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkColumnDefOptions, (SNode**)&pNode->pOptions); code = jsonToNodeObject(pJson, jkColumnDefOptions, (SNode**)&pNode->pOptions);
} }
return code; return code;
} }
@ -5315,7 +5319,7 @@ static int32_t jsonToDownstreamSourceNode(const SJson* pJson, void* pObj) {
static const char* jkWindowOffsetStartOffset = "StartOffset"; static const char* jkWindowOffsetStartOffset = "StartOffset";
static const char* jkWindowOffsetEndOffset = "EndOffset"; static const char* jkWindowOffsetEndOffset = "EndOffset";
static int32_t windowOffsetNodeToJson(const void* pObj, SJson* pJson) { static int32_t windowOffsetNodeToJson(const void* pObj, SJson* pJson) {
const SWindowOffsetNode* pNode = (const SWindowOffsetNode*)pObj; const SWindowOffsetNode* pNode = (const SWindowOffsetNode*)pObj;
int32_t code = tjsonAddObject(pJson, jkWindowOffsetStartOffset, nodeToJson, pNode->pStartOffset); int32_t code = tjsonAddObject(pJson, jkWindowOffsetStartOffset, nodeToJson, pNode->pStartOffset);
@ -5335,7 +5339,6 @@ static int32_t jsonToWindowOffsetNode(const SJson* pJson, void* pObj) {
return code; return code;
} }
static const char* jkDatabaseOptionsBuffer = "Buffer"; static const char* jkDatabaseOptionsBuffer = "Buffer";
static const char* jkDatabaseOptionsCacheModel = "CacheModel"; static const char* jkDatabaseOptionsCacheModel = "CacheModel";
static const char* jkDatabaseOptionsCompressionLevel = "CompressionLevel"; static const char* jkDatabaseOptionsCompressionLevel = "CompressionLevel";
@ -5359,6 +5362,9 @@ static const char* jkDatabaseOptionsS3ChunkSize = "S3ChunkSize";
static const char* jkDatabaseOptionsS3KeepLocalNode = "S3KeepLocalNode"; static const char* jkDatabaseOptionsS3KeepLocalNode = "S3KeepLocalNode";
static const char* jkDatabaseOptionsS3KeepLocal = "S3KeepLocal"; static const char* jkDatabaseOptionsS3KeepLocal = "S3KeepLocal";
static const char* jkDatabaseOptionsS3Compact = "S3Compact"; static const char* jkDatabaseOptionsS3Compact = "S3Compact";
static const char* jkDatabaseOptionsCompactIntervalNode = "compactIntervalNode";
static const char* jkDatabaseOptionsCompactTimeRange = "compactTimeRange";
static const char* jkDatabaseOptionsCompactTimeOffsetNode = "compactTimeOffsetNode";
static int32_t databaseOptionsToJson(const void* pObj, SJson* pJson) { static int32_t databaseOptionsToJson(const void* pObj, SJson* pJson) {
const SDatabaseOptions* pNode = (const SDatabaseOptions*)pObj; const SDatabaseOptions* pNode = (const SDatabaseOptions*)pObj;
@ -5430,6 +5436,15 @@ static int32_t databaseOptionsToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsS3Compact, pNode->s3Compact); code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsS3Compact, pNode->s3Compact);
} }
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkDatabaseOptionsCompactIntervalNode, nodeToJson, pNode->pCompactIntervalNode);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkDatabaseOptionsCompactTimeRange, pNode->pCompactTimeRangeList);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkDatabaseOptionsCompactTimeOffsetNode, nodeToJson, pNode->pCompactTimeOffsetNode);
}
return code; return code;
} }
@ -5504,6 +5519,15 @@ static int32_t jsonToDatabaseOptions(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkDatabaseOptionsS3Compact, &pNode->s3Compact); code = tjsonGetTinyIntValue(pJson, jkDatabaseOptionsS3Compact, &pNode->s3Compact);
} }
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkDatabaseOptionsCompactIntervalNode, (SNode**)&pNode->pCompactIntervalNode);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkDatabaseOptionsCompactTimeRange, &pNode->pCompactTimeRangeList);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkDatabaseOptionsCompactTimeOffsetNode, (SNode**)&pNode->pCompactTimeOffsetNode);
}
return code; return code;
} }
@ -6747,7 +6771,7 @@ static int32_t dropQnodeStmtToJson(const void* pObj, SJson* pJson) { return drop
static int32_t jsonToDropQnodeStmt(const SJson* pJson, void* pObj) { return jsonToDropComponentNodeStmt(pJson, pObj); } static int32_t jsonToDropQnodeStmt(const SJson* pJson, void* pObj) { return jsonToDropComponentNodeStmt(pJson, pObj); }
static const char* jkCreateAnodeStmtUrl = "Url"; static const char* jkCreateAnodeStmtUrl = "Url";
static const char* jkUpdateDropANodeStmtId = "AnodeId"; static const char* jkUpdateDropANodeStmtId = "AnodeId";
static int32_t createAnodeStmtToJson(const void* pObj, SJson* pJson) { static int32_t createAnodeStmtToJson(const void* pObj, SJson* pJson) {
const SCreateAnodeStmt* pNode = (const SCreateAnodeStmt*)pObj; const SCreateAnodeStmt* pNode = (const SCreateAnodeStmt*)pObj;
@ -7741,7 +7765,7 @@ static const char* jkTSMAOptionTsPrecision = "Precision";
static int32_t tsmaOptionToJson(const void* pObj, SJson* pJson) { static int32_t tsmaOptionToJson(const void* pObj, SJson* pJson) {
const STSMAOptions* pNode = (const STSMAOptions*)pObj; const STSMAOptions* pNode = (const STSMAOptions*)pObj;
int32_t code = nodeListToJson(pJson, jkTSMAOptionFuncs, pNode->pFuncs); int32_t code = nodeListToJson(pJson, jkTSMAOptionFuncs, pNode->pFuncs);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkTSMAOptionInterval, nodeToJson, pNode->pInterval); code = tjsonAddObject(pJson, jkTSMAOptionInterval, nodeToJson, pNode->pInterval);
} }
@ -7753,7 +7777,7 @@ static int32_t tsmaOptionToJson(const void* pObj, SJson* pJson) {
static int32_t jsonToTSMAOption(const SJson* pJson, void* pObj) { static int32_t jsonToTSMAOption(const SJson* pJson, void* pObj) {
STSMAOptions* pNode = (STSMAOptions*)pObj; STSMAOptions* pNode = (STSMAOptions*)pObj;
int32_t code = jsonToNodeList(pJson, jkTSMAOptionFuncs, &pNode->pFuncs); int32_t code = jsonToNodeList(pJson, jkTSMAOptionFuncs, &pNode->pFuncs);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkTSMAOptionInterval, &pNode->pInterval); code = jsonToNodeObject(pJson, jkTSMAOptionInterval, &pNode->pInterval);
} }
@ -7771,7 +7795,7 @@ static const char* jkCreateTSMAStmtpOptions = "Options";
static int32_t createTSMAStmtToJson(const void* pObj, SJson* pJson) { static int32_t createTSMAStmtToJson(const void* pObj, SJson* pJson) {
const SCreateTSMAStmt* pNode = (const SCreateTSMAStmt*)pObj; const SCreateTSMAStmt* pNode = (const SCreateTSMAStmt*)pObj;
int32_t code = tjsonAddBoolToObject(pJson, jkCreateTSMAStmtIgnoreExists, pNode->ignoreExists); int32_t code = tjsonAddBoolToObject(pJson, jkCreateTSMAStmtIgnoreExists, pNode->ignoreExists);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddStringToObject(pJson, jkCreateTSMAStmtTsmaName, pNode->tsmaName); code = tjsonAddStringToObject(pJson, jkCreateTSMAStmtTsmaName, pNode->tsmaName);
} }
@ -7789,7 +7813,7 @@ static int32_t createTSMAStmtToJson(const void* pObj, SJson* pJson) {
static int32_t jsonToCreateTSMAStmt(const SJson* pJson, void* pObj) { static int32_t jsonToCreateTSMAStmt(const SJson* pJson, void* pObj) {
SCreateTSMAStmt* pNode = (SCreateTSMAStmt*)pObj; SCreateTSMAStmt* pNode = (SCreateTSMAStmt*)pObj;
int32_t code = tjsonGetBoolValue(pJson, jkCreateTSMAStmtIgnoreExists, &pNode->ignoreExists); int32_t code = tjsonGetBoolValue(pJson, jkCreateTSMAStmtIgnoreExists, &pNode->ignoreExists);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetStringValue(pJson, jkCreateTSMAStmtTsmaName, pNode->tsmaName); code = tjsonGetStringValue(pJson, jkCreateTSMAStmtTsmaName, pNode->tsmaName);
} }
@ -7811,7 +7835,7 @@ static const char* jkDropTSMAStmtTsmaName = "TSMAName";
static int32_t dropTSMAStmtToJson(const void* pObj, SJson* pJson) { static int32_t dropTSMAStmtToJson(const void* pObj, SJson* pJson) {
const SDropTSMAStmt* pNode = (const SDropTSMAStmt*)pObj; const SDropTSMAStmt* pNode = (const SDropTSMAStmt*)pObj;
int32_t code = tjsonAddBoolToObject(pJson, jkDropTSMAStmtIgnoreNotExists, pNode->ignoreNotExists); int32_t code = tjsonAddBoolToObject(pJson, jkDropTSMAStmtIgnoreNotExists, pNode->ignoreNotExists);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddStringToObject(pJson, jkDropTSMAStmtDbName, pNode->dbName); code = tjsonAddStringToObject(pJson, jkDropTSMAStmtDbName, pNode->dbName);
} }
@ -7823,7 +7847,7 @@ static int32_t dropTSMAStmtToJson(const void* pObj, SJson* pJson) {
static int32_t jsonToDropTSMAStmt(const SJson* pJson, void* pObj) { static int32_t jsonToDropTSMAStmt(const SJson* pJson, void* pObj) {
SDropTSMAStmt* pNode = (SDropTSMAStmt*)pObj; SDropTSMAStmt* pNode = (SDropTSMAStmt*)pObj;
int32_t code = tjsonGetBoolValue(pJson, jkDropTSMAStmtIgnoreNotExists, &pNode->ignoreNotExists); int32_t code = tjsonGetBoolValue(pJson, jkDropTSMAStmtIgnoreNotExists, &pNode->ignoreNotExists);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetStringValue(pJson, jkDropTSMAStmtDbName, pNode->dbName); code = tjsonGetStringValue(pJson, jkDropTSMAStmtDbName, pNode->dbName);
} }
@ -8361,7 +8385,7 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
case QUERY_NODE_BALANCE_VGROUP_LEADER_STMT: case QUERY_NODE_BALANCE_VGROUP_LEADER_STMT:
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
case QUERY_NODE_BALANCE_VGROUP_LEADER_DATABASE_STMT: case QUERY_NODE_BALANCE_VGROUP_LEADER_DATABASE_STMT:
return TSDB_CODE_SUCCESS; // SBalanceVgroupLeaderStmt has no fields to deserialize. return TSDB_CODE_SUCCESS; // SBalanceVgroupLeaderStmt has no fields to deserialize.
case QUERY_NODE_MERGE_VGROUP_STMT: case QUERY_NODE_MERGE_VGROUP_STMT:
return jsonToMergeVgroupStmt(pJson, pObj); return jsonToMergeVgroupStmt(pJson, pObj);
case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT: case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT:

669
source/libs/nodes/src/nodesUtilFuncs.c
View File

File diff suppressed because it is too large Load Diff

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

@ -70,6 +70,9 @@ typedef enum EDatabaseOptionType {
DB_OPTION_KEEP_TIME_OFFSET, DB_OPTION_KEEP_TIME_OFFSET,
DB_OPTION_ENCRYPT_ALGORITHM, DB_OPTION_ENCRYPT_ALGORITHM,
DB_OPTION_DNODES, DB_OPTION_DNODES,
DB_OPTION_COMPACT_INTERVAL,
DB_OPTION_COMPACT_TIME_RANGE,
DB_OPTION_COMPACT_TIME_OFFSET,
} EDatabaseOptionType; } EDatabaseOptionType;
typedef enum ETableOptionType { typedef enum ETableOptionType {

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

@ -288,6 +288,9 @@ db_options(A) ::= db_options(B) S3_COMPACT NK_INTEGER(C).
db_options(A) ::= db_options(B) KEEP_TIME_OFFSET NK_INTEGER(C). { A = setDatabaseOption(pCxt, B, DB_OPTION_KEEP_TIME_OFFSET, &C); } db_options(A) ::= db_options(B) KEEP_TIME_OFFSET NK_INTEGER(C). { A = setDatabaseOption(pCxt, B, DB_OPTION_KEEP_TIME_OFFSET, &C); }
db_options(A) ::= db_options(B) ENCRYPT_ALGORITHM NK_STRING(C). { A = setDatabaseOption(pCxt, B, DB_OPTION_ENCRYPT_ALGORITHM, &C); } db_options(A) ::= db_options(B) ENCRYPT_ALGORITHM NK_STRING(C). { A = setDatabaseOption(pCxt, B, DB_OPTION_ENCRYPT_ALGORITHM, &C); }
db_options(A) ::= db_options(B) DNODES NK_STRING(C). { A = setDatabaseOption(pCxt, B, DB_OPTION_DNODES, &C); } db_options(A) ::= db_options(B) DNODES NK_STRING(C). { A = setDatabaseOption(pCxt, B, DB_OPTION_DNODES, &C); }
db_options(A) ::= db_options(B) COMPACT_INTERVAL NK_VARIABLE(C). { A = setDatabaseOption(pCxt, B, DB_OPTION_COMPACT_INTERVAL, &C); }
db_options(A) ::= db_options(B) COMPACT_TIME_RANGE variable_list(C). { A = setDatabaseOption(pCxt, B, DB_OPTION_COMPACT_TIME_RANGE, &C); }
db_options(A) ::= db_options(B) COMPACT_TIME_OFFSET NK_VARIABLE(C). { A = setDatabaseOption(pCxt, B, DB_OPTION_COMPACT_TIME_OFFSET, &C); }
alter_db_options(A) ::= alter_db_option(B). { A = createAlterDatabaseOptions(pCxt); A = setAlterDatabaseOption(pCxt, A, &B); } alter_db_options(A) ::= alter_db_option(B). { A = createAlterDatabaseOptions(pCxt); A = setAlterDatabaseOption(pCxt, A, &B); }
alter_db_options(A) ::= alter_db_options(B) alter_db_option(C). { A = setAlterDatabaseOption(pCxt, B, &C); } alter_db_options(A) ::= alter_db_options(B) alter_db_option(C). { A = setAlterDatabaseOption(pCxt, B, &C); }
@ -323,6 +326,9 @@ alter_db_option(A) ::= S3_KEEPLOCAL NK_VARIABLE(B).
alter_db_option(A) ::= S3_COMPACT NK_INTEGER(B). { A.type = DB_OPTION_S3_COMPACT, A.val = B; } alter_db_option(A) ::= S3_COMPACT NK_INTEGER(B). { A.type = DB_OPTION_S3_COMPACT, A.val = B; }
alter_db_option(A) ::= KEEP_TIME_OFFSET NK_INTEGER(B). { A.type = DB_OPTION_KEEP_TIME_OFFSET; A.val = B; } alter_db_option(A) ::= KEEP_TIME_OFFSET NK_INTEGER(B). { A.type = DB_OPTION_KEEP_TIME_OFFSET; A.val = B; }
alter_db_option(A) ::= ENCRYPT_ALGORITHM NK_STRING(B). { A.type = DB_OPTION_ENCRYPT_ALGORITHM; A.val = B; } alter_db_option(A) ::= ENCRYPT_ALGORITHM NK_STRING(B). { A.type = DB_OPTION_ENCRYPT_ALGORITHM; A.val = B; }
alter_db_option(A) ::= COMPACT_INTERVAL NK_VARIABLE(B). { A.type = DB_OPTION_COMPACT_INTERVAL; A.val = B; }
alter_db_option(A) ::= COMPACT_TIME_RANGE variable_list(B). { A.type = DB_OPTION_COMPACT_TIME_RANGE; A.pList = B; }
alter_db_option(A) ::= COMPACT_TIME_OFFSET NK_VARIABLE(B). { A.type = DB_OPTION_COMPACT_TIME_OFFSET; A.val = B; }
%type integer_list { SNodeList* } %type integer_list { SNodeList* }
%destructor integer_list { nodesDestroyList($$); } %destructor integer_list { nodesDestroyList($$); }

19
source/libs/parser/src/parAstCreater.c
View File

@ -1800,6 +1800,10 @@ SNode* createDefaultDatabaseOptions(SAstCreateContext* pCxt) {
pOptions->withArbitrator = TSDB_DEFAULT_DB_WITH_ARBITRATOR; pOptions->withArbitrator = TSDB_DEFAULT_DB_WITH_ARBITRATOR;
pOptions->encryptAlgorithm = TSDB_DEFAULT_ENCRYPT_ALGO; pOptions->encryptAlgorithm = TSDB_DEFAULT_ENCRYPT_ALGO;
pOptions->dnodeListStr[0] = 0; pOptions->dnodeListStr[0] = 0;
pOptions->compactInterval = TSDB_DEFAULT_COMPACT_INTERVAL;
pOptions->compactStartTime = TSDB_DEFAULT_COMPACT_START_TIME;
pOptions->compactEndTime = TSDB_DEFAULT_COMPACT_END_TIME;
pOptions->compactTimeOffset = TSDB_DEFAULT_COMPACT_TIME_OFFSET;
return (SNode*)pOptions; return (SNode*)pOptions;
_err: _err:
return NULL; return NULL;
@ -1844,6 +1848,10 @@ SNode* createAlterDatabaseOptions(SAstCreateContext* pCxt) {
pOptions->withArbitrator = -1; pOptions->withArbitrator = -1;
pOptions->encryptAlgorithm = -1; pOptions->encryptAlgorithm = -1;
pOptions->dnodeListStr[0] = 0; pOptions->dnodeListStr[0] = 0;
pOptions->compactInterval = TSDB_DEFAULT_COMPACT_INTERVAL;
pOptions->compactStartTime = TSDB_DEFAULT_COMPACT_START_TIME;
pOptions->compactEndTime = TSDB_DEFAULT_COMPACT_END_TIME;
pOptions->compactTimeOffset = TSDB_DEFAULT_COMPACT_TIME_OFFSET;
return (SNode*)pOptions; return (SNode*)pOptions;
_err: _err:
return NULL; return NULL;
@ -1991,6 +1999,17 @@ static SNode* setDatabaseOptionImpl(SAstCreateContext* pCxt, SNode* pOptions, ED
} else { } else {
COPY_STRING_FORM_STR_TOKEN(pDbOptions->dnodeListStr, (SToken*)pVal); COPY_STRING_FORM_STR_TOKEN(pDbOptions->dnodeListStr, (SToken*)pVal);
} }
break;
case DB_OPTION_COMPACT_INTERVAL:
pDbOptions->pCompactIntervalNode = (SValueNode*)createDurationValueNode(pCxt, (SToken*)pVal);
break;
case DB_OPTION_COMPACT_TIME_RANGE:
pDbOptions->pCompactTimeRangeList = pVal;
break;
case DB_OPTION_COMPACT_TIME_OFFSET:
pDbOptions->pCompactTimeOffsetNode = (SValueNode*)createDurationValueNode(pCxt, (SToken*)pVal);
;
break;
default: default:
break; break;
} }

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

@ -1853,7 +1853,7 @@ static bool clauseSupportAlias(ESqlClause clause) {
return SQL_CLAUSE_GROUP_BY == clause || SQL_CLAUSE_PARTITION_BY == clause || SQL_CLAUSE_ORDER_BY == clause; return SQL_CLAUSE_GROUP_BY == clause || SQL_CLAUSE_PARTITION_BY == clause || SQL_CLAUSE_ORDER_BY == clause;
} }
static EDealRes translateColumnInGroupByClause(STranslateContext* pCxt, SColumnNode** pCol, bool *translateAsAlias) { static EDealRes translateColumnInGroupByClause(STranslateContext* pCxt, SColumnNode** pCol, bool* translateAsAlias) {
*translateAsAlias = false; *translateAsAlias = false;
// count(*)/first(*)/last(*) and so on // count(*)/first(*)/last(*) and so on
if (0 == strcmp((*pCol)->colName, "*")) { if (0 == strcmp((*pCol)->colName, "*")) {
@ -1862,7 +1862,7 @@ static EDealRes translateColumnInGroupByClause(STranslateContext* pCxt, SColumnN
if (pCxt->pParseCxt->biMode) { if (pCxt->pParseCxt->biMode) {
SNode** ppNode = (SNode**)pCol; SNode** ppNode = (SNode**)pCol;
bool ret; bool ret;
pCxt->errCode = biRewriteToTbnameFunc(pCxt, ppNode, &ret); pCxt->errCode = biRewriteToTbnameFunc(pCxt, ppNode, &ret);
if (TSDB_CODE_SUCCESS != pCxt->errCode) return DEAL_RES_ERROR; if (TSDB_CODE_SUCCESS != pCxt->errCode) return DEAL_RES_ERROR;
if (ret) { if (ret) {
@ -1876,9 +1876,8 @@ static EDealRes translateColumnInGroupByClause(STranslateContext* pCxt, SColumnN
} else { } else {
bool found = false; bool found = false;
res = translateColumnWithoutPrefix(pCxt, pCol); res = translateColumnWithoutPrefix(pCxt, pCol);
if (!(*pCol)->node.asParam && if (!(*pCol)->node.asParam && res != DEAL_RES_CONTINUE && res != DEAL_RES_END &&
res != DEAL_RES_CONTINUE && pCxt->errCode != TSDB_CODE_PAR_AMBIGUOUS_COLUMN) {
res != DEAL_RES_END && pCxt->errCode != TSDB_CODE_PAR_AMBIGUOUS_COLUMN) {
res = translateColumnUseAlias(pCxt, pCol, &found); res = translateColumnUseAlias(pCxt, pCol, &found);
*translateAsAlias = true; *translateAsAlias = true;
} }
@ -3311,26 +3310,28 @@ static int32_t selectCommonType(SDataType* commonType, const SDataType* newType)
} else { } else {
resultType = gDisplyTypes[type2][type1]; resultType = gDisplyTypes[type2][type1];
} }
if (resultType == -1) { if (resultType == -1) {
return TSDB_CODE_SCALAR_CONVERT_ERROR; return TSDB_CODE_SCALAR_CONVERT_ERROR;
} }
if (commonType->type == newType->type) { if (commonType->type == newType->type) {
commonType->bytes = TMAX(commonType->bytes, newType->bytes); commonType->bytes = TMAX(commonType->bytes, newType->bytes);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
if ((resultType == TSDB_DATA_TYPE_VARCHAR) && (IS_MATHABLE_TYPE(commonType->type) || IS_MATHABLE_TYPE(newType->type))) { if ((resultType == TSDB_DATA_TYPE_VARCHAR) &&
(IS_MATHABLE_TYPE(commonType->type) || IS_MATHABLE_TYPE(newType->type))) {
commonType->bytes = TMAX(TMAX(commonType->bytes, newType->bytes), QUERY_NUMBER_MAX_DISPLAY_LEN); commonType->bytes = TMAX(TMAX(commonType->bytes, newType->bytes), QUERY_NUMBER_MAX_DISPLAY_LEN);
} else if ((resultType == TSDB_DATA_TYPE_NCHAR) && (IS_MATHABLE_TYPE(commonType->type) || IS_MATHABLE_TYPE(newType->type))) { } else if ((resultType == TSDB_DATA_TYPE_NCHAR) &&
(IS_MATHABLE_TYPE(commonType->type) || IS_MATHABLE_TYPE(newType->type))) {
commonType->bytes = TMAX(TMAX(commonType->bytes, newType->bytes), QUERY_NUMBER_MAX_DISPLAY_LEN * TSDB_NCHAR_SIZE); commonType->bytes = TMAX(TMAX(commonType->bytes, newType->bytes), QUERY_NUMBER_MAX_DISPLAY_LEN * TSDB_NCHAR_SIZE);
} else { } else {
commonType->bytes = TMAX(TMAX(commonType->bytes, newType->bytes), TYPE_BYTES[resultType]); commonType->bytes = TMAX(TMAX(commonType->bytes, newType->bytes), TYPE_BYTES[resultType]);
} }
commonType->type = resultType; commonType->type = resultType;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -5480,7 +5481,7 @@ static EDealRes translateGroupPartitionByImpl(SNode** pNode, void* pContext) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
STranslateContext* pTransCxt = pCxt->pTranslateCxt; STranslateContext* pTransCxt = pCxt->pTranslateCxt;
if (QUERY_NODE_VALUE == nodeType(*pNode)) { if (QUERY_NODE_VALUE == nodeType(*pNode)) {
SValueNode* pVal = (SValueNode*) *pNode; SValueNode* pVal = (SValueNode*)*pNode;
if (DEAL_RES_ERROR == translateValue(pTransCxt, pVal)) { if (DEAL_RES_ERROR == translateValue(pTransCxt, pVal)) {
return DEAL_RES_CONTINUE; return DEAL_RES_CONTINUE;
} }
@ -5528,8 +5529,7 @@ static int32_t translateGroupByList(STranslateContext* pCxt, SSelectStmt* pSelec
if (NULL == pSelect->pGroupByList) { if (NULL == pSelect->pGroupByList) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SReplaceGroupByAliasCxt cxt = { SReplaceGroupByAliasCxt cxt = {.pTranslateCxt = pCxt, .pProjectionList = pSelect->pProjectionList};
.pTranslateCxt = pCxt, .pProjectionList = pSelect->pProjectionList};
nodesRewriteExprsPostOrder(pSelect->pGroupByList, translateGroupPartitionByImpl, &cxt); nodesRewriteExprsPostOrder(pSelect->pGroupByList, translateGroupPartitionByImpl, &cxt);
return pCxt->errCode; return pCxt->errCode;
@ -5540,8 +5540,7 @@ static int32_t translatePartitionByList(STranslateContext* pCxt, SSelectStmt* pS
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SReplaceGroupByAliasCxt cxt = { SReplaceGroupByAliasCxt cxt = {.pTranslateCxt = pCxt, .pProjectionList = pSelect->pProjectionList};
.pTranslateCxt = pCxt, .pProjectionList = pSelect->pProjectionList};
nodesRewriteExprsPostOrder(pSelect->pPartitionByList, translateGroupPartitionByImpl, &cxt); nodesRewriteExprsPostOrder(pSelect->pPartitionByList, translateGroupPartitionByImpl, &cxt);
return pCxt->errCode; return pCxt->errCode;
@ -7559,6 +7558,12 @@ static int32_t buildCreateDbReq(STranslateContext* pCxt, SCreateDatabaseStmt* pS
pReq->encryptAlgorithm = pStmt->pOptions->encryptAlgorithm; pReq->encryptAlgorithm = pStmt->pOptions->encryptAlgorithm;
tstrncpy(pReq->dnodeListStr, pStmt->pOptions->dnodeListStr, TSDB_DNODE_LIST_LEN); tstrncpy(pReq->dnodeListStr, pStmt->pOptions->dnodeListStr, TSDB_DNODE_LIST_LEN);
// auto-compact options
pReq->compactInterval = pStmt->pOptions->compactInterval;
pReq->compactStartTime = pStmt->pOptions->compactStartTime;
pReq->compactEndTime = pStmt->pOptions->compactEndTime;
pReq->compactTimeOffset = pStmt->pOptions->compactTimeOffset;
return buildCreateDbRetentions(pStmt->pOptions->pRetentions, pReq); return buildCreateDbRetentions(pStmt->pOptions->pRetentions, pReq);
} }
@ -7922,6 +7927,79 @@ static int32_t checkOptionsDependency(STranslateContext* pCxt, const char* pDbNa
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t checkDbCompactIntervalOption(STranslateContext* pCxt, SDatabaseOptions* pOptions) {
if (NULL == pOptions->pCompactIntervalNode) return TSDB_CODE_SUCCESS;
if (DEAL_RES_ERROR == translateValue(pCxt, pOptions->pCompactIntervalNode)) {
return pCxt->errCode;
}
if (TIME_UNIT_MINUTE != pOptions->pCompactIntervalNode->unit &&
TIME_UNIT_HOUR != pOptions->pCompactIntervalNode->unit) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
"Invalid option compact_interval unit: %c, only %c, %c allowed",
pOptions->pCompactIntervalNode->unit, TIME_UNIT_MINUTE, TIME_UNIT_HOUR);
}
pOptions->compactInterval = getBigintFromValueNode(pOptions->pCompactIntervalNode);
// TODO: check the semantic of compact_interval
return TSDB_CODE_SUCCESS;
}
static int32_t checkDbCompactTimeRangeOption(STranslateContext* pCxt, SDatabaseOptions* pOptions) {
if (NULL == pOptions->pCompactTimeRangeList) {
return TSDB_CODE_SUCCESS;
}
if (LIST_LENGTH(pOptions->pCompactTimeRangeList) != 2) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
"Invalid option compact_time_range, should have 2 values");
}
SValueNode* pStart = (SValueNode*)nodesListGetNode(pOptions->pCompactTimeRangeList, 0);
SValueNode* pEnd = (SValueNode*)nodesListGetNode(pOptions->pCompactTimeRangeList, 1);
if (DEAL_RES_ERROR == translateValue(pCxt, pStart)) {
return pCxt->errCode;
}
if (DEAL_RES_ERROR == translateValue(pCxt, pEnd)) {
return pCxt->errCode;
}
if (TIME_UNIT_MINUTE != pStart->unit && TIME_UNIT_HOUR != pStart->unit && TIME_UNIT_DAY != pStart->unit) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
"Invalid option compact_time_range start unit: %c, only %c, %c, %c allowed",
pStart->unit, TIME_UNIT_MINUTE, TIME_UNIT_HOUR, TIME_UNIT_DAY);
}
if (TIME_UNIT_MINUTE != pEnd->unit && TIME_UNIT_HOUR != pEnd->unit && TIME_UNIT_DAY != pEnd->unit) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
"Invalid option compact_time_range end unit: %c, only %c, %c, %c allowed",
pEnd->unit, TIME_UNIT_MINUTE, TIME_UNIT_HOUR, TIME_UNIT_DAY);
}
pOptions->compactStartTime = getBigintFromValueNode(pStart);
pOptions->compactEndTime = getBigintFromValueNode(pEnd);
// TODO: check the semantic of compact_time_range
return TSDB_CODE_SUCCESS;
}
static int32_t checkDbCompactTimeOffsetOption(STranslateContext* pCxt, SDatabaseOptions* pOptions) {
if (NULL == pOptions->pCompactTimeOffsetNode) {
return TSDB_CODE_SUCCESS;
}
if (DEAL_RES_ERROR == translateValue(pCxt, pOptions->pCompactTimeOffsetNode)) {
return pCxt->errCode;
}
if (TIME_UNIT_MINUTE != pOptions->pCompactTimeOffsetNode->unit &&
TIME_UNIT_HOUR != pOptions->pCompactTimeOffsetNode->unit) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
"Invalid option compact_time_offset unit: %c, only %c, %c allowed",
pOptions->pCompactTimeOffsetNode->unit, TIME_UNIT_MINUTE, TIME_UNIT_HOUR);
}
pOptions->compactTimeOffset = getBigintFromValueNode(pOptions->pCompactTimeOffsetNode);
// TODO: check the semantic of compact_time_offset
return TSDB_CODE_SUCCESS;
}
static int32_t checkDatabaseOptions(STranslateContext* pCxt, const char* pDbName, SDatabaseOptions* pOptions) { static int32_t checkDatabaseOptions(STranslateContext* pCxt, const char* pDbName, SDatabaseOptions* pOptions) {
int32_t code = int32_t code =
checkDbRangeOption(pCxt, "buffer", pOptions->buffer, TSDB_MIN_BUFFER_PER_VNODE, TSDB_MAX_BUFFER_PER_VNODE); checkDbRangeOption(pCxt, "buffer", pOptions->buffer, TSDB_MIN_BUFFER_PER_VNODE, TSDB_MAX_BUFFER_PER_VNODE);
@ -8038,6 +8116,15 @@ static int32_t checkDatabaseOptions(STranslateContext* pCxt, const char* pDbName
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = checkDbRangeOption(pCxt, "s3_compact", pOptions->s3Compact, TSDB_MIN_S3_COMPACT, TSDB_MAX_S3_COMPACT); code = checkDbRangeOption(pCxt, "s3_compact", pOptions->s3Compact, TSDB_MIN_S3_COMPACT, TSDB_MAX_S3_COMPACT);
} }
if (TSDB_CODE_SUCCESS == code) {
code = checkDbCompactIntervalOption(pCxt, pOptions);
}
if (TSDB_CODE_SUCCESS == code) {
code = checkDbCompactTimeRangeOption(pCxt, pOptions);
}
if (TSDB_CODE_SUCCESS == code) {
code = checkDbCompactTimeOffsetOption(pCxt, pOptions);
}
return code; return code;
} }
@ -8274,6 +8361,10 @@ static int32_t buildAlterDbReq(STranslateContext* pCxt, SAlterDatabaseStmt* pStm
pReq->s3KeepLocal = pStmt->pOptions->s3KeepLocal; pReq->s3KeepLocal = pStmt->pOptions->s3KeepLocal;
pReq->s3Compact = pStmt->pOptions->s3Compact; pReq->s3Compact = pStmt->pOptions->s3Compact;
pReq->withArbitrator = pStmt->pOptions->withArbitrator; pReq->withArbitrator = pStmt->pOptions->withArbitrator;
pReq->compactInterval = pStmt->pOptions->compactInterval;
pReq->compactStartTime = pStmt->pOptions->compactStartTime;
pReq->compactEndTime = pStmt->pOptions->compactEndTime;
pReq->compactTimeOffset = pStmt->pOptions->compactTimeOffset;
return code; return code;
} }
@ -10601,7 +10692,8 @@ static void getSourceDatabase(SNode* pStmt, int32_t acctId, char* pDbFName) {
(void)tNameGetFullDbName(&name, pDbFName); (void)tNameGetFullDbName(&name, pDbFName);
} }
static void getStreamQueryFirstProjectAliasName(SHashObj* pUserAliasSet, char* aliasName, int32_t len, char* defaultName[]) { static void getStreamQueryFirstProjectAliasName(SHashObj* pUserAliasSet, char* aliasName, int32_t len,
char* defaultName[]) {
for (int32_t i = 0; defaultName[i] != NULL; i++) { for (int32_t i = 0; defaultName[i] != NULL; i++) {
if (NULL == taosHashGet(pUserAliasSet, defaultName[i], strlen(defaultName[i]))) { if (NULL == taosHashGet(pUserAliasSet, defaultName[i], strlen(defaultName[i]))) {
snprintf(aliasName, len, "%s", defaultName[i]); snprintf(aliasName, len, "%s", defaultName[i]);
@ -10627,8 +10719,8 @@ static int32_t setColumnDefNodePrimaryKey(SColumnDefNode* pNode, bool isPk) {
return code; return code;
} }
static int32_t addIrowTsToCreateStreamQueryImpl(STranslateContext* pCxt, SSelectStmt* pSelect, static int32_t addIrowTsToCreateStreamQueryImpl(STranslateContext* pCxt, SSelectStmt* pSelect, SHashObj* pUserAliasSet,
SHashObj* pUserAliasSet, SNodeList* pCols, SCMCreateStreamReq* pReq) { SNodeList* pCols, SCMCreateStreamReq* pReq) {
SNode* pProj = nodesListGetNode(pSelect->pProjectionList, 0); SNode* pProj = nodesListGetNode(pSelect->pProjectionList, 0);
if (!pSelect->hasInterpFunc || if (!pSelect->hasInterpFunc ||
(QUERY_NODE_FUNCTION == nodeType(pProj) && 0 == strcmp("_irowts", ((SFunctionNode*)pProj)->functionName))) { (QUERY_NODE_FUNCTION == nodeType(pProj) && 0 == strcmp("_irowts", ((SFunctionNode*)pProj)->functionName))) {
@ -10675,7 +10767,7 @@ static int32_t addWstartTsToCreateStreamQueryImpl(STranslateContext* pCxt, SSele
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SFunctionNode* pFunc = NULL; SFunctionNode* pFunc = NULL;
int32_t code = nodesMakeNode(QUERY_NODE_FUNCTION, (SNode**)&pFunc); int32_t code = nodesMakeNode(QUERY_NODE_FUNCTION, (SNode**)&pFunc);
if (NULL == pFunc) { if (NULL == pFunc) {
return code; return code;
} }
@ -10707,7 +10799,7 @@ static int32_t addWstartTsToCreateStreamQueryImpl(STranslateContext* pCxt, SSele
} }
static int32_t addTsKeyToCreateStreamQuery(STranslateContext* pCxt, SNode* pStmt, SNodeList* pCols, static int32_t addTsKeyToCreateStreamQuery(STranslateContext* pCxt, SNode* pStmt, SNodeList* pCols,
SCMCreateStreamReq* pReq) { SCMCreateStreamReq* pReq) {
SSelectStmt* pSelect = (SSelectStmt*)pStmt; SSelectStmt* pSelect = (SSelectStmt*)pStmt;
SHashObj* pUserAliasSet = NULL; SHashObj* pUserAliasSet = NULL;
int32_t code = checkProjectAlias(pCxt, pSelect->pProjectionList, &pUserAliasSet); int32_t code = checkProjectAlias(pCxt, pSelect->pProjectionList, &pUserAliasSet);
@ -11070,21 +11162,18 @@ static int32_t checkStreamQuery(STranslateContext* pCxt, SCreateStreamStmt* pStm
if (pStmt->pOptions->triggerType == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) { if (pStmt->pOptions->triggerType == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) {
if (pStmt->pOptions->fillHistory) { if (pStmt->pOptions->fillHistory) {
return generateSyntaxErrMsgExt( return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY,
&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "When trigger was force window close, Stream unsupported Fill history");
"When trigger was force window close, Stream unsupported Fill history");
} }
if (pStmt->pOptions->ignoreExpired != 1) { if (pStmt->pOptions->ignoreExpired != 1) {
return generateSyntaxErrMsgExt( return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY,
&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "When trigger was force window close, Stream must not set ignore expired 0");
"When trigger was force window close, Stream must not set ignore expired 0");
} }
if (pStmt->pOptions->ignoreUpdate != 1) { if (pStmt->pOptions->ignoreUpdate != 1) {
return generateSyntaxErrMsgExt( return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY,
&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "When trigger was force window close, Stream must not set ignore update 0");
"When trigger was force window close, Stream must not set ignore update 0");
} }
if (pSelect->pWindow != NULL && QUERY_NODE_INTERVAL_WINDOW == nodeType(pSelect->pWindow)) { if (pSelect->pWindow != NULL && QUERY_NODE_INTERVAL_WINDOW == nodeType(pSelect->pWindow)) {