Merge branch '3.0' into test/jcy

include/common/ttokendef.h

@@ -67,101 +67,101 @@
 #define TK_IF            49
 #define TK_NOT           50
 #define TK_EXISTS        51
-#define TK_BLOCKS                          52
-#define TK_CACHE                           53
-#define TK_CACHELAST                       54
-#define TK_COMP                            55
-#define TK_DAYS                            56
-#define TK_NK_VARIABLE                     57
-#define TK_FSYNC                           58
-#define TK_MAXROWS                         59
-#define TK_MINROWS                         60
-#define TK_KEEP                            61
-#define TK_PRECISION                       62
-#define TK_QUORUM                          63
+#define TK_BUFFER        52
+#define TK_CACHELAST     53
+#define TK_COMP          54
+#define TK_DAYS          55
+#define TK_NK_VARIABLE   56
+#define TK_FSYNC         57
+#define TK_MAXROWS       58
+#define TK_MINROWS       59
+#define TK_KEEP          60
+#define TK_PAGES         61
+#define TK_PAGESIZE      62
+#define TK_PRECISION     63
 #define TK_REPLICA       64
-#define TK_TTL                             65
+#define TK_STRICT        65
 #define TK_WAL           66
 #define TK_VGROUPS       67
 #define TK_SINGLE_STABLE 68
-#define TK_STREAM_MODE                     69
-#define TK_RETENTIONS                      70
-#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_RETENTIONS    69
+#define TK_NK_COMMA      70
+#define TK_NK_COLON      71
+#define TK_TABLE         72
+#define TK_NK_LP         73
+#define TK_NK_RP         74
+#define TK_STABLE        75
+#define TK_ADD           76
+#define TK_COLUMN        77
+#define TK_MODIFY        78
+#define TK_RENAME        79
+#define TK_TAG           80
+#define TK_SET           81
+#define TK_NK_EQ         82
+#define TK_USING         83
+#define TK_TAGS          84
+#define TK_NK_DOT        85
+#define TK_COMMENT       86
+#define TK_BOOL          87
+#define TK_TINYINT       88
+#define TK_SMALLINT      89
+#define TK_INT           90
+#define TK_INTEGER       91
+#define TK_BIGINT        92
+#define TK_FLOAT         93
+#define TK_DOUBLE        94
+#define TK_BINARY        95
+#define TK_TIMESTAMP     96
+#define TK_NCHAR         97
+#define TK_UNSIGNED      98
+#define TK_JSON          99
+#define TK_VARCHAR       100
+#define TK_MEDIUMBLOB    101
+#define TK_BLOB          102
+#define TK_VARBINARY     103
+#define TK_DECIMAL       104
+#define TK_DELAY         105
+#define TK_FILE_FACTOR   106
+#define TK_NK_FLOAT      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_CLUSTER                        133
-#define TK_LIKE                           134
-#define TK_INDEX                          135
-#define TK_FULLTEXT                       136
-#define TK_FUNCTION                       137
-#define TK_INTERVAL                       138
-#define TK_TOPIC                          139
-#define TK_AS                             140
-#define TK_WITH                           141
-#define TK_SCHEMA                         142
-#define TK_DESC                           143
-#define TK_DESCRIBE                       144
-#define TK_RESET                          145
-#define TK_QUERY                          146
+#define TK_TTL           109
+#define TK_SMA           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_CLUSTER       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_WITH          140
+#define TK_SCHEMA        141
+#define TK_DESC          142
+#define TK_DESCRIBE      143
+#define TK_RESET         144
+#define TK_QUERY         145
+#define TK_CACHE         146
 #define TK_EXPLAIN       147
 #define TK_ANALYZE       148
 #define TK_VERBOSE       149

include/libs/nodes/cmdnodes.h

@@ -30,24 +30,25 @@ extern "C" {
 
 typedef struct SDatabaseOptions {
   ENodeType   type;
-  SValueNode* pNumOfBlocks;
-  SValueNode* pCacheBlockSize;
-  SValueNode* pCachelast;
-  SValueNode* pCompressionLevel;
+  int32_t     buffer;
+  int8_t      cachelast;
+  int8_t      compressionLevel;
+  int32_t     daysPerFile;
   SValueNode* pDaysPerFile;
-  SValueNode* pFsyncPeriod;
-  SValueNode* pMaxRowsPerBlock;
-  SValueNode* pMinRowsPerBlock;
+  int32_t     fsyncPeriod;
+  int32_t     maxRowsPerBlock;
+  int32_t     minRowsPerBlock;
   SNodeList*  pKeep;
-  SValueNode* pPrecision;
-  SValueNode* pQuorum;
-  SValueNode* pReplica;
-  SValueNode* pTtl;
-  SValueNode* pWalLevel;
-  SValueNode* pNumOfVgroups;
-  SValueNode* pSingleStable;
-  SValueNode* pStreamMode;
-  SValueNode* pStrict;
+  int32_t     keep[3];
+  int32_t     pages;
+  int32_t     pagesize;
+  char        precisionStr[3];
+  int8_t      precision;
+  int8_t      replica;
+  int8_t      strict;
+  int8_t      walLevel;
+  int32_t     numOfVgroups;
+  int8_t      singleStable;
   SNodeList*  pRetentions;
 } SDatabaseOptions;
 
@@ -77,13 +78,12 @@ typedef struct SAlterDatabaseStmt {
 
 typedef struct STableOptions {
   ENodeType  type;
-  SNodeList* pKeep;
-  SValueNode* pTtl;
-  SValueNode* pComments;
+  char       comment[TSDB_STB_COMMENT_LEN];
+  int32_t    delay;
+  float      filesFactor;
+  SNodeList* pRollupFuncs;
+  int32_t    ttl;
   SNodeList* pSma;
-  SNodeList* pFuncs;
-  SValueNode* pFilesFactor;
-  SValueNode* pDelay;
 } STableOptions;
 
 typedef struct SColumnDefNode {
@@ -201,10 +201,7 @@ typedef struct SShowCreatStmt {
   char      tableName[TSDB_TABLE_NAME_LEN];
 } SShowCreatStmt;
 
-typedef enum EIndexType {
-  INDEX_TYPE_SMA = 1,
-  INDEX_TYPE_FULLTEXT
-} EIndexType;
+typedef enum EIndexType { INDEX_TYPE_SMA = 1, INDEX_TYPE_FULLTEXT } EIndexType;
 
 typedef struct SIndexOptions {
   ENodeType  type;

include/util/taoserror.h

@@ -586,7 +586,6 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_RANGE_OPTION      TAOS_DEF_ERROR_CODE(0, 0x2619)
 #define TSDB_CODE_PAR_INVALID_STR_OPTION        TAOS_DEF_ERROR_CODE(0, 0x261A)
 #define TSDB_CODE_PAR_INVALID_ENUM_OPTION       TAOS_DEF_ERROR_CODE(0, 0x261B)
-#define TSDB_CODE_PAR_INVALID_TTL_OPTION        TAOS_DEF_ERROR_CODE(0, 0x261C)
 #define TSDB_CODE_PAR_INVALID_KEEP_NUM          TAOS_DEF_ERROR_CODE(0, 0x261D)
 #define TSDB_CODE_PAR_INVALID_KEEP_ORDER        TAOS_DEF_ERROR_CODE(0, 0x261E)
 #define TSDB_CODE_PAR_INVALID_KEEP_VALUE        TAOS_DEF_ERROR_CODE(0, 0x261F)

include/util/tdef.h

@@ -350,8 +350,6 @@ typedef enum ELogicConditionType {
 #define TSDB_MIN_FSYNC_PERIOD           0
 #define TSDB_MAX_FSYNC_PERIOD           180000  // millisecond
 #define TSDB_DEFAULT_FSYNC_PERIOD       3000    // three second
-#define TSDB_MIN_DB_TTL               1
-#define TSDB_DEFAULT_DB_TTL           1
 #define TSDB_MIN_WAL_LEVEL              1
 #define TSDB_MAX_WAL_LEVEL              2
 #define TSDB_DEFAULT_WAL_LEVEL          1
@@ -379,13 +377,23 @@ typedef enum ELogicConditionType {
 #define TSDB_DB_SINGLE_STABLE_ON        0
 #define TSDB_DB_SINGLE_STABLE_OFF       1
 #define TSDB_DEFAULT_DB_SINGLE_STABLE   0
+#define TSDB_MIN_BUFFER_PER_VNODE       3  // unit MB
+#define TSDB_DEFAULT_BUFFER_PER_VNODE   96
+#define TSDB_MIN_PAGES_PER_VNODE        64
+#define TSDB_DEFAULT_PAGES_PER_VNODE    256
+#define TSDB_MIN_PAGESIZE_PER_VNODE     1  // unit KB
+#define TSDB_MAX_PAGESIZE_PER_VNODE     16384
+#define TSDB_DEFAULT_PAGESIZE_PER_VNODE 4
+
+#define TSDB_MIN_ROLLUP_FILE_FACTOR     0
+#define TSDB_MAX_ROLLUP_FILE_FACTOR     1
+#define TSDB_DEFAULT_ROLLUP_FILE_FACTOR 0.1
+#define TSDB_MIN_ROLLUP_DELAY           1
+#define TSDB_MAX_ROLLUP_DELAY           10
+#define TSDB_DEFAULT_ROLLUP_DELAY       2
+#define TSDB_MIN_TABLE_TTL              0
+#define TSDB_DEFAULT_TABLE_TTL          0
 
-#define TSDB_MIN_DB_FILE_FACTOR     0
-#define TSDB_MAX_DB_FILE_FACTOR     1
-#define TSDB_DEFAULT_DB_FILE_FACTOR 0.1
-#define TSDB_MIN_DB_DELAY           1
-#define TSDB_MAX_DB_DELAY           10
-#define TSDB_DEFAULT_DB_DELAY       2
 #define TSDB_MIN_EXPLAIN_RATIO     0
 #define TSDB_MAX_EXPLAIN_RATIO     1
 #define TSDB_DEFAULT_EXPLAIN_RATIO 0.001

source/client/inc/clientInt.h

@@ -188,6 +188,7 @@ typedef struct SRequestSendRecvBody {
 
 typedef struct {
   int8_t         resType;
+  int32_t        code;
   char           topic[TSDB_TOPIC_FNAME_LEN];
   int32_t        vgId;
   SSchemaWrapper schema;
@@ -311,7 +312,6 @@ SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code
 int32_t      getQueryPlan(SRequestObj* pRequest, SQuery* pQuery, SArray** pNodeList);
 int32_t      scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList);
 
-
 #ifdef __cplusplus
 }
 #endif

source/client/src/clientMain.c

@@ -110,16 +110,23 @@ int taos_errno(TAOS_RES *tres) {
     return terrno;
   }
 
+  if (TD_RES_TMQ(tres)) {
+    return 0;
+  }
+
   return ((SRequestObj *)tres)->code;
 }
 
 const char *taos_errstr(TAOS_RES *res) {
-  SRequestObj *pRequest = (SRequestObj *)res;
-
-  if (pRequest == NULL) {
+  if (res == NULL) {
     return (const char *)tstrerror(terrno);
   }
 
+  if (TD_RES_TMQ(res)) {
+    return "success";
+  }
+
+  SRequestObj *pRequest = (SRequestObj *)res;
   if (NULL != pRequest->msgBuf && (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR)) {
     return pRequest->msgBuf;
   } else {
@@ -632,9 +639,7 @@ int taos_stmt_set_tbname(TAOS_STMT *stmt, const char *name) {
   return stmtSetTbName(stmt, name);
 }
 
-int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name) {
-  return taos_stmt_set_tbname(stmt, name);
-}
+int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name) { return taos_stmt_set_tbname(stmt, name); }
 
 int taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
   if (stmt == NULL || bind == NULL) {
@@ -750,9 +755,7 @@ TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt) {
   return stmtUseResult(stmt);
 }
 
-char *taos_stmt_errstr(TAOS_STMT *stmt) {
-  return (char *)stmtErrstr(stmt);
-}
+char *taos_stmt_errstr(TAOS_STMT *stmt) { return (char *)stmtErrstr(stmt); }
 
 int taos_stmt_affected_rows(TAOS_STMT *stmt) {
   if (stmt == NULL) {

source/dnode/mgmt/mgmt_vnode/src/vmHandle.c

@@ -107,7 +107,7 @@ static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
 
   pCfg->vgId = pCreate->vgId;
   strcpy(pCfg->dbname, pCreate->db);
-  pCfg->szBuf = pCreate->cacheBlockSize * 1024 * 1024;
+  // pCfg->szBuf = pCreate->cacheBlockSize * 1024 * 1024;
   pCfg->streamMode = pCreate->streamMode;
   pCfg->isWeak = true;
   pCfg->tsdbCfg.days = 10;

source/dnode/mnode/impl/inc/mndDef.h

@@ -514,12 +514,12 @@ void*    tDecodeSMqVgEp(const void* buf, SMqVgEp* pVgEp);
 typedef struct {
   int64_t consumerId;  // -1 for unassigned
   SArray* vgs;         // SArray<SMqVgEp*>
-} SMqConsumerEpInSub;
+} SMqConsumerEp;
 
-SMqConsumerEpInSub* tCloneSMqConsumerEpInSub(const SMqConsumerEpInSub* pEpInSub);
-void                tDeleteSMqConsumerEpInSub(SMqConsumerEpInSub* pEpInSub);
-int32_t             tEncodeSMqConsumerEpInSub(void** buf, const SMqConsumerEpInSub* pEpInSub);
-void*               tDecodeSMqConsumerEpInSub(const void* buf, SMqConsumerEpInSub* pEpInSub);
+SMqConsumerEp* tCloneSMqConsumerEp(const SMqConsumerEp* pEp);
+void           tDeleteSMqConsumerEp(SMqConsumerEp* pEp);
+int32_t        tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pEp);
+void*          tDecodeSMqConsumerEp(const void* buf, SMqConsumerEp* pEp);
 
 typedef struct {
   char      key[TSDB_SUBSCRIBE_KEY_LEN];
@@ -529,9 +529,8 @@ typedef struct {
   int8_t    withTbName;
   int8_t    withSchema;
   int8_t    withTag;
-  SHashObj* consumerHash;  // consumerId -> SMqConsumerEpInSub
-  // TODO put -1 into unassignVgs
-  // SArray*   unassignedVgs;
+  SHashObj* consumerHash;   // consumerId -> SMqConsumerEp
+  SArray*   unassignedVgs;  // SArray<SMqVgEp*>
 } SMqSubscribeObj;
 
 SMqSubscribeObj* tNewSubscribeObj(const char key[TSDB_SUBSCRIBE_KEY_LEN]);
@@ -542,7 +541,7 @@ void*            tDecodeSubscribeObj(const void* buf, SMqSubscribeObj* pSub);
 
 typedef struct {
   int32_t epoch;
-  SArray* consumers;  // SArray<SMqConsumerEpInSub*>
+  SArray* consumers;  // SArray<SMqConsumerEp*>
 } SMqSubActionLogEntry;
 
 SMqSubActionLogEntry* tCloneSMqSubActionLogEntry(SMqSubActionLogEntry* pEntry);

source/dnode/mnode/impl/src/mndConsumer.c

@@ -302,8 +302,8 @@ static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) {
       mndReleaseTopic(pMnode, pTopic);
 
       // 2.2 iterate all vg assigned to the consumer of that topic
-      SMqConsumerEpInSub *pEpInSub = taosHashGet(pSub->consumerHash, &consumerId, sizeof(int64_t));
-      int32_t             vgNum = taosArrayGetSize(pEpInSub->vgs);
+      SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &consumerId, sizeof(int64_t));
+      int32_t        vgNum = taosArrayGetSize(pConsumerEp->vgs);
 
       topicEp.vgs = taosArrayInit(vgNum, sizeof(SMqSubVgEp));
       if (topicEp.vgs == NULL) {
@@ -313,7 +313,7 @@ static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) {
       }
 
       for (int32_t j = 0; j < vgNum; j++) {
-        SMqVgEp *pVgEp = taosArrayGetP(pEpInSub->vgs, j);
+        SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j);
         char     offsetKey[TSDB_PARTITION_KEY_LEN];
         mndMakePartitionKey(offsetKey, pConsumer->cgroup, topic, pVgEp->vgId);
         // 2.2.1 build vg ep

source/dnode/mnode/impl/src/mndDb.c

@@ -268,8 +268,10 @@ static int32_t mndCheckDbName(const char *dbName, SUserObj *pUser) {
 
 static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
   if (pCfg->numOfVgroups < TSDB_MIN_VNODES_PER_DB || pCfg->numOfVgroups > TSDB_MAX_VNODES_PER_DB) return -1;
+  /*
   if (pCfg->cacheBlockSize < TSDB_MIN_CACHE_BLOCK_SIZE || pCfg->cacheBlockSize > TSDB_MAX_CACHE_BLOCK_SIZE) return -1;
   if (pCfg->totalBlocks < TSDB_MIN_TOTAL_BLOCKS || pCfg->totalBlocks > TSDB_MAX_TOTAL_BLOCKS) return -1;
+  */
   if (pCfg->daysPerFile < TSDB_MIN_DAYS_PER_FILE || pCfg->daysPerFile > TSDB_MAX_DAYS_PER_FILE) return -1;
   if (pCfg->daysToKeep0 < TSDB_MIN_KEEP || pCfg->daysToKeep0 > TSDB_MAX_KEEP) return -1;
   if (pCfg->daysToKeep1 < TSDB_MIN_KEEP || pCfg->daysToKeep1 > TSDB_MAX_KEEP) return -1;
@@ -282,7 +284,7 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
   if (pCfg->minRows > pCfg->maxRows) return -1;
   if (pCfg->commitTime < TSDB_MIN_COMMIT_TIME || pCfg->commitTime > TSDB_MAX_COMMIT_TIME) return -1;
   if (pCfg->fsyncPeriod < TSDB_MIN_FSYNC_PERIOD || pCfg->fsyncPeriod > TSDB_MAX_FSYNC_PERIOD) return -1;
-  if (pCfg->ttl < TSDB_MIN_DB_TTL) return -1;
+  // if (pCfg->ttl < TSDB_MIN_TABLE_TTL) return -1;
   if (pCfg->walLevel < TSDB_MIN_WAL_LEVEL || pCfg->walLevel > TSDB_MAX_WAL_LEVEL) return -1;
   if (pCfg->precision < TSDB_MIN_PRECISION && pCfg->precision > TSDB_MAX_PRECISION) return -1;
   if (pCfg->compression < TSDB_MIN_COMP_LEVEL || pCfg->compression > TSDB_MAX_COMP_LEVEL) return -1;
@@ -310,7 +312,7 @@ static void mndSetDefaultDbCfg(SDbCfg *pCfg) {
   if (pCfg->maxRows < 0) pCfg->maxRows = TSDB_DEFAULT_MAXROWS_FBLOCK;
   if (pCfg->commitTime < 0) pCfg->commitTime = TSDB_DEFAULT_COMMIT_TIME;
   if (pCfg->fsyncPeriod < 0) pCfg->fsyncPeriod = TSDB_DEFAULT_FSYNC_PERIOD;
-  if (pCfg->ttl < 0) pCfg->ttl = TSDB_DEFAULT_DB_TTL;
+  if (pCfg->ttl < 0) pCfg->ttl = TSDB_DEFAULT_TABLE_TTL;
   if (pCfg->walLevel < 0) pCfg->walLevel = TSDB_DEFAULT_WAL_LEVEL;
   if (pCfg->precision < 0) pCfg->precision = TSDB_DEFAULT_PRECISION;
   if (pCfg->compression < 0) pCfg->compression = TSDB_DEFAULT_COMP_LEVEL;

source/dnode/mnode/impl/src/mndDef.c

@@ -211,42 +211,47 @@ void *tDecodeSMqVgEp(const void *buf, SMqVgEp *pVgEp) {
   return (void *)buf;
 }
 
-SMqConsumerEpInSub *tCloneSMqConsumerEpInSub(const SMqConsumerEpInSub *pEpInSub) {
-  SMqConsumerEpInSub *pEpInSubNew = taosMemoryMalloc(sizeof(SMqConsumerEpInSub));
-  if (pEpInSubNew == NULL) return NULL;
-  pEpInSubNew->consumerId = pEpInSub->consumerId;
-  pEpInSubNew->vgs = taosArrayDeepCopy(pEpInSub->vgs, (FCopy)tCloneSMqVgEp);
-  return pEpInSubNew;
+SMqConsumerEp *tCloneSMqConsumerEp(const SMqConsumerEp *pConsumerEpOld) {
+  SMqConsumerEp *pConsumerEpNew = taosMemoryMalloc(sizeof(SMqConsumerEp));
+  if (pConsumerEpNew == NULL) return NULL;
+  pConsumerEpNew->consumerId = pConsumerEpOld->consumerId;
+  pConsumerEpNew->vgs = taosArrayDeepCopy(pConsumerEpOld->vgs, (FCopy)tCloneSMqVgEp);
+  return pConsumerEpNew;
 }
 
-void tDeleteSMqConsumerEpInSub(SMqConsumerEpInSub *pEpInSub) {
-  taosArrayDestroyEx(pEpInSub->vgs, (FDelete)tDeleteSMqVgEp);
+void tDeleteSMqConsumerEp(SMqConsumerEp *pConsumerEp) {
+  //
+  taosArrayDestroyP(pConsumerEp->vgs, (FDelete)tDeleteSMqVgEp);
 }
 
-int32_t tEncodeSMqConsumerEpInSub(void **buf, const SMqConsumerEpInSub *pEpInSub) {
+int32_t tEncodeSMqConsumerEp(void **buf, const SMqConsumerEp *pConsumerEp) {
   int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, pEpInSub->consumerId);
-  int32_t sz = taosArrayGetSize(pEpInSub->vgs);
+  tlen += taosEncodeFixedI64(buf, pConsumerEp->consumerId);
+  tlen += taosEncodeArray(buf, pConsumerEp->vgs, (FEncode)tEncodeSMqVgEp);
+#if 0
+  int32_t sz = taosArrayGetSize(pConsumerEp->vgs);
   tlen += taosEncodeFixedI32(buf, sz);
   for (int32_t i = 0; i < sz; i++) {
-    SMqVgEp *pVgEp = taosArrayGetP(pEpInSub->vgs, i);
+    SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, i);
     tlen += tEncodeSMqVgEp(buf, pVgEp);
   }
-  /*tlen += taosEncodeArray(buf, pEpInSub->vgs, (FEncode)tEncodeSMqVgEp);*/
+#endif
   return tlen;
 }
 
-void *tDecodeSMqConsumerEpInSub(const void *buf, SMqConsumerEpInSub *pEpInSub) {
-  buf = taosDecodeFixedI64(buf, &pEpInSub->consumerId);
-  /*buf = taosDecodeArray(buf, &pEpInSub->vgs, (FDecode)tDecodeSMqVgEp, sizeof(SMqSubVgEp));*/
+void *tDecodeSMqConsumerEp(const void *buf, SMqConsumerEp *pConsumerEp) {
+  buf = taosDecodeFixedI64(buf, &pConsumerEp->consumerId);
+  buf = taosDecodeArray(buf, &pConsumerEp->vgs, (FDecode)tDecodeSMqVgEp, sizeof(SMqSubVgEp));
+#if 0
   int32_t sz;
   buf = taosDecodeFixedI32(buf, &sz);
-  pEpInSub->vgs = taosArrayInit(sz, sizeof(void *));
+  pConsumerEp->vgs = taosArrayInit(sz, sizeof(void *));
   for (int32_t i = 0; i < sz; i++) {
     SMqVgEp *pVgEp = taosMemoryMalloc(sizeof(SMqVgEp));
     buf = tDecodeSMqVgEp(buf, pVgEp);
-    taosArrayPush(pEpInSub->vgs, &pVgEp);
+    taosArrayPush(pConsumerEp->vgs, &pVgEp);
   }
+#endif
 
   return (void *)buf;
 }
@@ -258,13 +263,11 @@ SMqSubscribeObj *tNewSubscribeObj(const char key[TSDB_SUBSCRIBE_KEY_LEN]) {
   taosInitRWLatch(&pSubNew->lock);
   pSubNew->vgNum = 0;
   pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
-  // TODO set free fp
-  SMqConsumerEpInSub epInSub = {
-      .consumerId = -1,
-      .vgs = taosArrayInit(0, sizeof(void *)),
-  };
-  int64_t unexistKey = -1;
-  taosHashPut(pSubNew->consumerHash, &unexistKey, sizeof(int64_t), &epInSub, sizeof(SMqConsumerEpInSub));
+  // TODO set hash free fp
+  /*taosHashSetFreeFp(pSubNew->consumerHash, tDeleteSMqConsumerEp);*/
+
+  pSubNew->unassignedVgs = taosArrayInit(0, sizeof(void *));
+
   return pSubNew;
 }
 
@@ -281,25 +284,27 @@ SMqSubscribeObj *tCloneSubscribeObj(const SMqSubscribeObj *pSub) {
 
   pSubNew->vgNum = pSub->vgNum;
   pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
-  /*taosHashSetFreeFp(pSubNew->consumerHash, taosArrayDestroy);*/
+  // TODO set hash free fp
+  /*taosHashSetFreeFp(pSubNew->consumerHash, tDeleteSMqConsumerEp);*/
   void          *pIter = NULL;
-  SMqConsumerEpInSub *pEpInSub = NULL;
+  SMqConsumerEp *pConsumerEp = NULL;
   while (1) {
     pIter = taosHashIterate(pSub->consumerHash, pIter);
     if (pIter == NULL) break;
-    pEpInSub = (SMqConsumerEpInSub *)pIter;
-    SMqConsumerEpInSub newEp = {
-        .consumerId = pEpInSub->consumerId,
-        .vgs = taosArrayDeepCopy(pEpInSub->vgs, (FCopy)tCloneSMqVgEp),
+    pConsumerEp = (SMqConsumerEp *)pIter;
+    SMqConsumerEp newEp = {
+        .consumerId = pConsumerEp->consumerId,
+        .vgs = taosArrayDeepCopy(pConsumerEp->vgs, (FCopy)tCloneSMqVgEp),
     };
-    taosHashPut(pSubNew->consumerHash, &newEp.consumerId, sizeof(int64_t), &newEp, sizeof(SMqConsumerEpInSub));
+    taosHashPut(pSubNew->consumerHash, &newEp.consumerId, sizeof(int64_t), &newEp, sizeof(SMqConsumerEp));
   }
+  pSubNew->unassignedVgs = taosArrayDeepCopy(pSub->unassignedVgs, (FCopy)tCloneSMqVgEp);
   return pSubNew;
 }
 
 void tDeleteSubscribeObj(SMqSubscribeObj *pSub) {
-  /*taosArrayDestroyEx(pSub->consumerEps, (FDelete)tDeleteSMqConsumerEpInSub);*/
   taosHashCleanup(pSub->consumerHash);
+  taosArrayDestroyP(pSub->unassignedVgs, (FDelete)tDeleteSMqVgEp);
 }
 
 int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {
@@ -319,12 +324,12 @@ int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {
   while (1) {
     pIter = taosHashIterate(pSub->consumerHash, pIter);
     if (pIter == NULL) break;
-    SMqConsumerEpInSub *pEpInSub = (SMqConsumerEpInSub *)pIter;
-    tlen += tEncodeSMqConsumerEpInSub(buf, pEpInSub);
+    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
+    tlen += tEncodeSMqConsumerEp(buf, pConsumerEp);
     cnt++;
   }
   ASSERT(cnt == sz);
-  /*tlen += taosEncodeArray(buf, pSub->consumerEps, (FEncode)tEncodeSMqConsumerEpInSub);*/
+  tlen += taosEncodeArray(buf, pSub->unassignedVgs, (FEncode)tEncodeSMqVgEp);
   return tlen;
 }
 
@@ -342,13 +347,12 @@ void *tDecodeSubscribeObj(const void *buf, SMqSubscribeObj *pSub) {
 
   pSub->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
   for (int32_t i = 0; i < sz; i++) {
-    /*SMqConsumerEpInSub* pEpInSub = taosMemoryMalloc(sizeof(SMqConsumerEpInSub));*/
-    SMqConsumerEpInSub epInSub = {0};
-    buf = tDecodeSMqConsumerEpInSub(buf, &epInSub);
-    taosHashPut(pSub->consumerHash, &epInSub.consumerId, sizeof(int64_t), &epInSub, sizeof(SMqConsumerEpInSub));
+    SMqConsumerEp consumerEp = {0};
+    buf = tDecodeSMqConsumerEp(buf, &consumerEp);
+    taosHashPut(pSub->consumerHash, &consumerEp.consumerId, sizeof(int64_t), &consumerEp, sizeof(SMqConsumerEp));
   }
 
-  /*buf = taosDecodeArray(buf, &pSub->consumerEps, (FDecode)tDecodeSMqConsumerEpInSub, sizeof(SMqConsumerEpInSub));*/
+  buf = taosDecodeArray(buf, &pSub->unassignedVgs, (FDecode)tDecodeSMqVgEp, sizeof(SMqVgEp));
   return (void *)buf;
 }
 
@@ -356,12 +360,12 @@ SMqSubActionLogEntry *tCloneSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {
   SMqSubActionLogEntry *pEntryNew = taosMemoryMalloc(sizeof(SMqSubActionLogEntry));
   if (pEntryNew == NULL) return NULL;
   pEntryNew->epoch = pEntry->epoch;
-  pEntryNew->consumers = taosArrayDeepCopy(pEntry->consumers, (FCopy)tCloneSMqConsumerEpInSub);
+  pEntryNew->consumers = taosArrayDeepCopy(pEntry->consumers, (FCopy)tCloneSMqConsumerEp);
   return pEntryNew;
 }
 
 void tDeleteSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {
-  taosArrayDestroyEx(pEntry->consumers, (FDelete)tDeleteSMqConsumerEpInSub);
+  taosArrayDestroyEx(pEntry->consumers, (FDelete)tDeleteSMqConsumerEp);
 }
 
 int32_t tEncodeSMqSubActionLogEntry(void **buf, const SMqSubActionLogEntry *pEntry) {
@@ -381,12 +385,12 @@ SMqSubActionLogObj *tCloneSMqSubActionLogObj(SMqSubActionLogObj *pLog) {
   SMqSubActionLogObj *pLogNew = taosMemoryMalloc(sizeof(SMqSubActionLogObj));
   if (pLogNew == NULL) return pLogNew;
   memcpy(pLogNew->key, pLog->key, TSDB_SUBSCRIBE_KEY_LEN);
-  pLogNew->logs = taosArrayDeepCopy(pLog->logs, (FCopy)tCloneSMqConsumerEpInSub);
+  pLogNew->logs = taosArrayDeepCopy(pLog->logs, (FCopy)tCloneSMqConsumerEp);
   return pLogNew;
 }
 
 void tDeleteSMqSubActionLogObj(SMqSubActionLogObj *pLog) {
-  taosArrayDestroyEx(pLog->logs, (FDelete)tDeleteSMqConsumerEpInSub);
+  taosArrayDestroyEx(pLog->logs, (FDelete)tDeleteSMqConsumerEp);
 }
 
 int32_t tEncodeSMqSubActionLogObj(void **buf, const SMqSubActionLogObj *pLog) {

source/dnode/mnode/impl/src/mndScheduler.c

@@ -504,11 +504,8 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     plan = nodesListGetNode(inner->pNodeList, 0);
   }
 
-  int64_t             unexistKey = -1;
-  SMqConsumerEpInSub* pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t));
-  ASSERT(pEpInSub);
-
-  ASSERT(taosHashGetSize(pSub->consumerHash) == 1);
+  ASSERT(pSub->unassignedVgs);
+  ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
 
   void* pIter = NULL;
   while (1) {
@@ -524,7 +521,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     SMqVgEp* pVgEp = taosMemoryMalloc(sizeof(SMqVgEp));
     pVgEp->epSet = mndGetVgroupEpset(pMnode, pVgroup);
     pVgEp->vgId = pVgroup->vgId;
-    taosArrayPush(pEpInSub->vgs, &pVgEp);
+    taosArrayPush(pSub->unassignedVgs, &pVgEp);
 
     mDebug("init subscription %s, assign vg: %d", pSub->key, pVgEp->vgId);
 
@@ -543,17 +540,11 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     } else {
       pVgEp->qmsg = strdup("");
     }
-
-    ASSERT(taosHashGetSize(pSub->consumerHash) == 1);
-
-    /*taosArrayPush(pSub->unassignedVg, &consumerEp);*/
   }
 
-  pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t));
-
-  ASSERT(pEpInSub->vgs->size > 0);
+  ASSERT(pSub->unassignedVgs->size > 0);
 
-  ASSERT(taosHashGetSize(pSub->consumerHash) == 1);
+  ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
 
   qDestroyQueryPlan(pPlan);
 

source/dnode/mnode/impl/src/mndSubscribe.c

@@ -85,7 +85,8 @@ static SMqSubscribeObj *mndCreateSub(SMnode *pMnode, const SMqTopicObj *pTopic,
   pSub->withSchema = pTopic->withSchema;
   pSub->withTag = pTopic->withTag;
 
-  ASSERT(taosHashGetSize(pSub->consumerHash) == 1);
+  ASSERT(pSub->unassignedVgs->size == 0);
+  ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
 
   if (mndSchedInitSubEp(pMnode, pTopic, pSub) < 0) {
     tDeleteSubscribeObj(pSub);
@@ -93,7 +94,8 @@ static SMqSubscribeObj *mndCreateSub(SMnode *pMnode, const SMqTopicObj *pTopic,
     return NULL;
   }
 
-  ASSERT(taosHashGetSize(pSub->consumerHash) == 1);
+  ASSERT(pSub->unassignedVgs->size > 0);
+  ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
 
   return pSub;
 }
@@ -185,7 +187,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
   if (pInput->pTopic != NULL) {
     // create subscribe
     pOutput->pSub = mndCreateSub(pMnode, pInput->pTopic, pInput->pRebInfo->key);
-    ASSERT(taosHashGetSize(pOutput->pSub->consumerHash) == 1);
+    ASSERT(taosHashGetSize(pOutput->pSub->consumerHash) == 0);
   } else {
     pOutput->pSub = tCloneSubscribeObj(pInput->pOldSub);
   }
@@ -196,21 +198,20 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
   // 1. build temporary hash(vgId -> SMqRebOutputVg) to store modified vg
   SHashObj *pHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
 
-  ASSERT(taosHashGetSize(pOutput->pSub->consumerHash) > 0);
   // 2. check and get actual removed consumers, put their vg into hash
   int32_t removedNum = taosArrayGetSize(pInput->pRebInfo->removedConsumers);
   int32_t actualRemoved = 0;
   for (int32_t i = 0; i < removedNum; i++) {
     int64_t consumerId = *(int64_t *)taosArrayGet(pInput->pRebInfo->removedConsumers, i);
     ASSERT(consumerId > 0);
-    SMqConsumerEpInSub *pEpInSub = taosHashGet(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t));
-    ASSERT(pEpInSub);
-    if (pEpInSub) {
-      ASSERT(consumerId == pEpInSub->consumerId);
+    SMqConsumerEp *pConsumerEp = taosHashGet(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t));
+    ASSERT(pConsumerEp);
+    if (pConsumerEp) {
+      ASSERT(consumerId == pConsumerEp->consumerId);
       actualRemoved++;
-      int32_t consumerVgNum = taosArrayGetSize(pEpInSub->vgs);
+      int32_t consumerVgNum = taosArrayGetSize(pConsumerEp->vgs);
       for (int32_t j = 0; j < consumerVgNum; j++) {
-        SMqVgEp       *pVgEp = taosArrayGetP(pEpInSub->vgs, j);
+        SMqVgEp       *pVgEp = taosArrayGetP(pConsumerEp->vgs, j);
         SMqRebOutputVg outputVg = {
             .oldConsumerId = consumerId,
             .newConsumerId = -1,
@@ -224,16 +225,12 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
     }
   }
   ASSERT(removedNum == actualRemoved);
-  ASSERT(taosHashGetSize(pOutput->pSub->consumerHash) > 0);
 
   // if previously no consumer, there are vgs not assigned
   {
-    int64_t             unexistKey = -1;
-    SMqConsumerEpInSub *pEpInSub = taosHashGet(pOutput->pSub->consumerHash, &unexistKey, sizeof(int64_t));
-    ASSERT(pEpInSub);
-    int32_t consumerVgNum = taosArrayGetSize(pEpInSub->vgs);
+    int32_t consumerVgNum = taosArrayGetSize(pOutput->pSub->unassignedVgs);
     for (int32_t i = 0; i < consumerVgNum; i++) {
-      SMqVgEp       *pVgEp = *(SMqVgEp **)taosArrayPop(pEpInSub->vgs);
+      SMqVgEp       *pVgEp = *(SMqVgEp **)taosArrayPop(pOutput->pSub->unassignedVgs);
       SMqRebOutputVg rebOutput = {
           .oldConsumerId = -1,
           .newConsumerId = -1,
@@ -246,7 +243,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
   // 3. calc vg number of each consumer
   int32_t oldSz = 0;
   if (pInput->pOldSub) {
-    oldSz = taosHashGetSize(pInput->pOldSub->consumerHash) - 1;
+    oldSz = taosHashGetSize(pInput->pOldSub->consumerHash);
   }
   int32_t afterRebConsumerNum =
       oldSz + taosArrayGetSize(pInput->pRebInfo->newConsumers) - taosArrayGetSize(pInput->pRebInfo->removedConsumers);
@@ -264,23 +261,22 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
   while (1) {
     pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
     if (pIter == NULL) break;
-    SMqConsumerEpInSub *pEpInSub = (SMqConsumerEpInSub *)pIter;
-    if (pEpInSub->consumerId == -1) continue;
-    ASSERT(pEpInSub->consumerId > 0);
-    int32_t consumerVgNum = taosArrayGetSize(pEpInSub->vgs);
+    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
+    ASSERT(pConsumerEp->consumerId > 0);
+    int32_t consumerVgNum = taosArrayGetSize(pConsumerEp->vgs);
     // all old consumers still existing are touched
     // TODO optimize: touch only consumer whose vgs changed
-    taosArrayPush(pOutput->touchedConsumers, &pEpInSub->consumerId);
+    taosArrayPush(pOutput->touchedConsumers, &pConsumerEp->consumerId);
     if (consumerVgNum > minVgCnt) {
       if (imbCnt < imbConsumerNum) {
         if (consumerVgNum == minVgCnt + 1) {
           continue;
         } else {
           // pop until equal minVg + 1
-          while (taosArrayGetSize(pEpInSub->vgs) > minVgCnt + 1) {
-            SMqVgEp       *pVgEp = *(SMqVgEp **)taosArrayPop(pEpInSub->vgs);
+          while (taosArrayGetSize(pConsumerEp->vgs) > minVgCnt + 1) {
+            SMqVgEp       *pVgEp = *(SMqVgEp **)taosArrayPop(pConsumerEp->vgs);
             SMqRebOutputVg outputVg = {
-                .oldConsumerId = pEpInSub->consumerId,
+                .oldConsumerId = pConsumerEp->consumerId,
                 .newConsumerId = -1,
                 .pVgEp = pVgEp,
             };
@@ -290,10 +286,10 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
         }
       } else {
         // pop until equal minVg
-        while (taosArrayGetSize(pEpInSub->vgs) > minVgCnt) {
-          SMqVgEp       *pVgEp = *(SMqVgEp **)taosArrayPop(pEpInSub->vgs);
+        while (taosArrayGetSize(pConsumerEp->vgs) > minVgCnt) {
+          SMqVgEp       *pVgEp = *(SMqVgEp **)taosArrayPop(pConsumerEp->vgs);
           SMqRebOutputVg outputVg = {
-              .oldConsumerId = pEpInSub->consumerId,
+              .oldConsumerId = pConsumerEp->consumerId,
               .newConsumerId = -1,
               .pVgEp = pVgEp,
           };
@@ -309,12 +305,11 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
     for (int32_t i = 0; i < consumerNum; i++) {
       int64_t consumerId = *(int64_t *)taosArrayGet(pInput->pRebInfo->newConsumers, i);
       ASSERT(consumerId > 0);
-      SMqConsumerEpInSub newConsumerEp;
+      SMqConsumerEp newConsumerEp;
       newConsumerEp.consumerId = consumerId;
       newConsumerEp.vgs = taosArrayInit(0, sizeof(void *));
-      taosHashPut(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t), &newConsumerEp,
-                  sizeof(SMqConsumerEpInSub));
-      /*SMqConsumerEpInSub *pTestNew = taosHashGet(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t));*/
+      taosHashPut(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t), &newConsumerEp, sizeof(SMqConsumerEp));
+      /*SMqConsumer* pTestNew = taosHashGet(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t));*/
       /*ASSERT(pTestNew->consumerId == consumerId);*/
       /*ASSERT(pTestNew->vgs == newConsumerEp.vgs);*/
       taosArrayPush(pOutput->newConsumers, &consumerId);
@@ -329,25 +324,24 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
   while (1) {
     pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
     if (pIter == NULL) break;
-    SMqConsumerEpInSub *pEpInSub = (SMqConsumerEpInSub *)pIter;
-    if (pEpInSub->consumerId == -1) continue;
-    ASSERT(pEpInSub->consumerId > 0);
+    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
+    ASSERT(pConsumerEp->consumerId > 0);
 
     // push until equal minVg
-    while (taosArrayGetSize(pEpInSub->vgs) < minVgCnt) {
+    while (taosArrayGetSize(pConsumerEp->vgs) < minVgCnt) {
       // iter hash and find one vg
       pRemovedIter = taosHashIterate(pHash, pRemovedIter);
       ASSERT(pRemovedIter);
       pRebVg = (SMqRebOutputVg *)pRemovedIter;
       // push
-      taosArrayPush(pEpInSub->vgs, &pRebVg->pVgEp);
-      pRebVg->newConsumerId = pEpInSub->consumerId;
+      taosArrayPush(pConsumerEp->vgs, &pRebVg->pVgEp);
+      pRebVg->newConsumerId = pConsumerEp->consumerId;
       taosArrayPush(pOutput->rebVgs, pRebVg);
     }
   }
 
   // 7. handle unassigned vg
-  if (taosHashGetSize(pOutput->pSub->consumerHash) != 1) {
+  if (taosHashGetSize(pOutput->pSub->consumerHash) != 0) {
     // if has consumer, assign all left vg
     while (1) {
       pRemovedIter = taosHashIterate(pHash, pRemovedIter);
@@ -355,20 +349,14 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
       pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
       ASSERT(pIter);
       pRebVg = (SMqRebOutputVg *)pRemovedIter;
-      SMqConsumerEpInSub *pEpInSub = (SMqConsumerEpInSub *)pIter;
-      if (pEpInSub->consumerId == -1) continue;
-      ASSERT(pEpInSub->consumerId > 0);
-      taosArrayPush(pEpInSub->vgs, &pRebVg->pVgEp);
-      pRebVg->newConsumerId = pEpInSub->consumerId;
+      SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
+      ASSERT(pConsumerEp->consumerId > 0);
+      taosArrayPush(pConsumerEp->vgs, &pRebVg->pVgEp);
+      pRebVg->newConsumerId = pConsumerEp->consumerId;
       taosArrayPush(pOutput->rebVgs, pRebVg);
     }
   } else {
     // if all consumer is removed, put all vg into unassigned
-    int64_t             unexistKey = -1;
-    SMqConsumerEpInSub *pEpInSub = taosHashGet(pOutput->pSub->consumerHash, &unexistKey, sizeof(int64_t));
-    ASSERT(pEpInSub);
-    ASSERT(pEpInSub->consumerId == -1);
-
     pIter = NULL;
     SMqRebOutputVg *pRebOutput = NULL;
     while (1) {
@@ -376,7 +364,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
       if (pIter == NULL) break;
       pRebOutput = (SMqRebOutputVg *)pIter;
       ASSERT(pRebOutput->newConsumerId == -1);
-      taosArrayPush(pEpInSub->vgs, &pRebOutput->pVgEp);
+      taosArrayPush(pOutput->pSub->unassignedVgs, &pRebOutput->pVgEp);
       taosArrayPush(pOutput->rebVgs, pRebOutput);
     }
   }
@@ -512,6 +500,7 @@ static int32_t mndProcessRebalanceReq(SNodeMsg *pMsg) {
     // possibly no vg is changed
     /*ASSERT(taosArrayGetSize(rebOutput.rebVgs) != 0);*/
 
+    // TODO replace assert with error check
     ASSERT(mndPersistRebResult(pMnode, pMsg, &rebOutput) == 0);
 
     if (rebInput.pTopic) {
@@ -631,6 +620,10 @@ static int32_t mndSubActionUpdate(SSdb *pSdb, SMqSubscribeObj *pOldSub, SMqSubsc
   pOldSub->consumerHash = pNewSub->consumerHash;
   pNewSub->consumerHash = tmp;
 
+  SArray *tmp1 = pOldSub->unassignedVgs;
+  pOldSub->unassignedVgs = pNewSub->unassignedVgs;
+  pNewSub->unassignedVgs = tmp1;
+
   taosWUnLockLatch(&pOldSub->lock);
   return 0;
 }

source/libs/nodes/src/nodesUtilFuncs.c

@@ -389,21 +389,25 @@ void nodesDestroyNode(SNodeptr pNode) {
     case QUERY_NODE_COLUMN_DEF:         // no pointer field
     case QUERY_NODE_DOWNSTREAM_SOURCE:  // no pointer field
       break;
-    case QUERY_NODE_DATABASE_OPTIONS:
-      nodesDestroyList(((SDatabaseOptions*)pNode)->pRetentions);
+    case QUERY_NODE_DATABASE_OPTIONS: {
+      SDatabaseOptions* pOptions = (SDatabaseOptions*)pNode;
+      nodesDestroyNode(pOptions->pDaysPerFile);
+      nodesDestroyList(pOptions->pKeep);
+      nodesDestroyList(pOptions->pRetentions);
       break;
+    }
     case QUERY_NODE_TABLE_OPTIONS: {
-      STableOptions* pStmt = (STableOptions*)pNode;
-      nodesDestroyList(pStmt->pSma);
-      nodesDestroyList(pStmt->pFuncs);
+      STableOptions* pOptions = (STableOptions*)pNode;
+      nodesDestroyList(pOptions->pSma);
+      nodesDestroyList(pOptions->pRollupFuncs);
       break;
     }
     case QUERY_NODE_INDEX_OPTIONS: {
-      SIndexOptions* pStmt = (SIndexOptions*)pNode;
-      nodesDestroyList(pStmt->pFuncs);
-      nodesDestroyNode(pStmt->pInterval);
-      nodesDestroyNode(pStmt->pOffset);
-      nodesDestroyNode(pStmt->pSliding);
+      SIndexOptions* pOptions = (SIndexOptions*)pNode;
+      nodesDestroyList(pOptions->pFuncs);
+      nodesDestroyNode(pOptions->pInterval);
+      nodesDestroyNode(pOptions->pOffset);
+      nodesDestroyNode(pOptions->pSliding);
       break;
     }
     case QUERY_NODE_SET_OPERATOR: {

source/libs/parser/inc/parAst.h

@@ -36,8 +36,7 @@ typedef struct SAstCreateContext {
 } SAstCreateContext;
 
 typedef enum EDatabaseOptionType {
-  DB_OPTION_BLOCKS = 1,
-  DB_OPTION_CACHE,
+  DB_OPTION_BUFFER = 1,
   DB_OPTION_CACHELAST,
   DB_OPTION_COMP,
   DB_OPTION_DAYS,
@@ -45,30 +44,29 @@ typedef enum EDatabaseOptionType {
   DB_OPTION_MAXROWS,
   DB_OPTION_MINROWS,
   DB_OPTION_KEEP,
+  DB_OPTION_PAGES,
+  DB_OPTION_PAGESIZE,
   DB_OPTION_PRECISION,
-  DB_OPTION_QUORUM,
   DB_OPTION_REPLICA,
-  DB_OPTION_TTL,
+  DB_OPTION_STRICT,
   DB_OPTION_WAL,
   DB_OPTION_VGROUPS,
   DB_OPTION_SINGLE_STABLE,
-  DB_OPTION_STREAM_MODE,
-  DB_OPTION_STRICT,
   DB_OPTION_RETENTIONS
 } EDatabaseOptionType;
 
 typedef enum ETableOptionType {
-  TABLE_OPTION_KEEP = 1,
-  TABLE_OPTION_TTL,
-  TABLE_OPTION_COMMENT,
-  TABLE_OPTION_SMA,
+  TABLE_OPTION_COMMENT = 1,
+  TABLE_OPTION_DELAY,
   TABLE_OPTION_FILE_FACTOR,
-  TABLE_OPTION_DELAY
+  TABLE_OPTION_ROLLUP,
+  TABLE_OPTION_TTL,
+  TABLE_OPTION_SMA
 } ETableOptionType;
 
 typedef struct SAlterOption {
   int32_t    type;
-  SValueNode* pVal;
+  SToken     val;
   SNodeList* pList;
 } SAlterOption;
 
@@ -121,25 +119,29 @@ SNode* addLimitClause(SAstCreateContext* pCxt, SNode* pStmt, SNode* pLimit);
 SNode* createSelectStmt(SAstCreateContext* pCxt, bool isDistinct, SNodeList* pProjectionList, SNode* pTable);
 SNode* createSetOperator(SAstCreateContext* pCxt, ESetOperatorType type, SNode* pLeft, SNode* pRight);
 
-SNode*    createDatabaseOptions(SAstCreateContext* pCxt);
-SNode*    setDatabaseAlterOption(SAstCreateContext* pCxt, SNode* pOptions, SAlterOption* pAlterOption);
+SDataType createDataType(uint8_t type);
+SDataType createVarLenDataType(uint8_t type, const SToken* pLen);
+
+SNode* createDefaultDatabaseOptions(SAstCreateContext* pCxt);
+SNode* createAlterDatabaseOptions(SAstCreateContext* pCxt);
+SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOptionType type, void* pVal);
+SNode* setAlterDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, SAlterOption* pAlterOption);
 SNode* createCreateDatabaseStmt(SAstCreateContext* pCxt, bool ignoreExists, SToken* pDbName, SNode* pOptions);
 SNode* createDropDatabaseStmt(SAstCreateContext* pCxt, bool ignoreNotExists, SToken* pDbName);
 SNode* createAlterDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName, SNode* pOptions);
-SNode*    createTableOptions(SAstCreateContext* pCxt);
-SNode*    setTableAlterOption(SAstCreateContext* pCxt, SNode* pOptions, SAlterOption* pAlterOption);
+SNode* createDefaultTableOptions(SAstCreateContext* pCxt);
+SNode* createAlterTableOptions(SAstCreateContext* pCxt);
+SNode* setTableOption(SAstCreateContext* pCxt, SNode* pOptions, ETableOptionType type, void* pVal);
 SNode* createColumnDefNode(SAstCreateContext* pCxt, SToken* pColName, SDataType dataType, const SToken* pComment);
-SDataType createDataType(uint8_t type);
-SDataType createVarLenDataType(uint8_t type, const SToken* pLen);
 SNode* createCreateTableStmt(SAstCreateContext* pCxt, bool ignoreExists, SNode* pRealTable, SNodeList* pCols,
                              SNodeList* pTags, SNode* pOptions);
 SNode* createCreateSubTableClause(SAstCreateContext* pCxt, bool ignoreExists, SNode* pRealTable, SNode* pUseRealTable,
-                                  SNodeList* pSpecificTags, SNodeList* pValsOfTags);
+                                  SNodeList* pSpecificTags, SNodeList* pValsOfTags, SNode* pOptions);
 SNode* createCreateMultiTableStmt(SAstCreateContext* pCxt, SNodeList* pSubTables);
 SNode* createDropTableClause(SAstCreateContext* pCxt, bool ignoreNotExists, SNode* pRealTable);
 SNode* createDropTableStmt(SAstCreateContext* pCxt, SNodeList* pTables);
 SNode* createDropSuperTableStmt(SAstCreateContext* pCxt, bool ignoreNotExists, SNode* pRealTable);
-SNode* createAlterTableOption(SAstCreateContext* pCxt, SNode* pRealTable, SNode* pOptions);
+SNode* createAlterTableModifyOptions(SAstCreateContext* pCxt, SNode* pRealTable, SNode* pOptions);
 SNode* createAlterTableAddModifyCol(SAstCreateContext* pCxt, SNode* pRealTable, int8_t alterType,
                                     const SToken* pColName, SDataType dataType);
 SNode* createAlterTableDropCol(SAstCreateContext* pCxt, SNode* pRealTable, int8_t alterType, const SToken* pColName);

source/libs/parser/inc/sql.y

@@ -137,43 +137,41 @@ not_exists_opt(A) ::= .
 exists_opt(A) ::= IF EXISTS.                                                      { A = true; }
 exists_opt(A) ::= .                                                               { A = false; }
 
-db_options(A) ::= .                                                               { A = createDatabaseOptions(pCxt); }
-db_options(A) ::= db_options(B) BLOCKS NK_INTEGER(C).                             { ((SDatabaseOptions*)B)->pNumOfBlocks = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) CACHE NK_INTEGER(C).                              { ((SDatabaseOptions*)B)->pCacheBlockSize = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) CACHELAST NK_INTEGER(C).                          { ((SDatabaseOptions*)B)->pCachelast = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) COMP NK_INTEGER(C).                               { ((SDatabaseOptions*)B)->pCompressionLevel = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) DAYS NK_INTEGER(C).                               { ((SDatabaseOptions*)B)->pDaysPerFile = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) DAYS NK_VARIABLE(C).                              { ((SDatabaseOptions*)B)->pDaysPerFile = (SValueNode*)createDurationValueNode(pCxt, &C); A = B; }
-db_options(A) ::= db_options(B) FSYNC NK_INTEGER(C).                              { ((SDatabaseOptions*)B)->pFsyncPeriod = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) MAXROWS NK_INTEGER(C).                            { ((SDatabaseOptions*)B)->pMaxRowsPerBlock = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) MINROWS NK_INTEGER(C).                            { ((SDatabaseOptions*)B)->pMinRowsPerBlock = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) KEEP integer_list(C).                             { ((SDatabaseOptions*)B)->pKeep = C; A = B; }
-db_options(A) ::= db_options(B) KEEP variable_list(C).                            { ((SDatabaseOptions*)B)->pKeep = C; A = B; }
-db_options(A) ::= db_options(B) PRECISION NK_STRING(C).                           { ((SDatabaseOptions*)B)->pPrecision = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BINARY, &C); A = B; }
-db_options(A) ::= db_options(B) QUORUM NK_INTEGER(C).                             { ((SDatabaseOptions*)B)->pQuorum = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) REPLICA NK_INTEGER(C).                            { ((SDatabaseOptions*)B)->pReplica = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) TTL NK_INTEGER(C).                                { ((SDatabaseOptions*)B)->pTtl = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) WAL NK_INTEGER(C).                                { ((SDatabaseOptions*)B)->pWalLevel = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-db_options(A) ::= db_options(B) VGROUPS NK_INTEGER(C).                            { ((SDatabaseOptions*)B)->pNumOfVgroups = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-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); }
+db_options(A) ::= .                                                               { A = createDefaultDatabaseOptions(pCxt); }
+db_options(A) ::= db_options(B) BUFFER NK_INTEGER(C).                             { A = setDatabaseOption(pCxt, B, DB_OPTION_BUFFER, &C); }
+db_options(A) ::= db_options(B) CACHELAST NK_INTEGER(C).                          { A = setDatabaseOption(pCxt, B, DB_OPTION_CACHELAST, &C); }
+db_options(A) ::= db_options(B) COMP NK_INTEGER(C).                               { A = setDatabaseOption(pCxt, B, DB_OPTION_COMP, &C); }
+db_options(A) ::= db_options(B) DAYS NK_INTEGER(C).                               { A = setDatabaseOption(pCxt, B, DB_OPTION_DAYS, &C); }
+db_options(A) ::= db_options(B) DAYS NK_VARIABLE(C).                              { A = setDatabaseOption(pCxt, B, DB_OPTION_DAYS, &C); }
+db_options(A) ::= db_options(B) FSYNC NK_INTEGER(C).                              { A = setDatabaseOption(pCxt, B, DB_OPTION_FSYNC, &C); }
+db_options(A) ::= db_options(B) MAXROWS NK_INTEGER(C).                            { A = setDatabaseOption(pCxt, B, DB_OPTION_MAXROWS, &C); }
+db_options(A) ::= db_options(B) MINROWS NK_INTEGER(C).                            { A = setDatabaseOption(pCxt, B, DB_OPTION_MINROWS, &C); }
+db_options(A) ::= db_options(B) KEEP integer_list(C).                             { A = setDatabaseOption(pCxt, B, DB_OPTION_KEEP, C); }
+db_options(A) ::= db_options(B) KEEP variable_list(C).                            { A = setDatabaseOption(pCxt, B, DB_OPTION_KEEP, C); }
+db_options(A) ::= db_options(B) PAGES NK_INTEGER(C).                              { A = setDatabaseOption(pCxt, B, DB_OPTION_PAGES, &C); }
+db_options(A) ::= db_options(B) PAGESIZE NK_INTEGER(C).                           { A = setDatabaseOption(pCxt, B, DB_OPTION_PAGESIZE, &C); }
+db_options(A) ::= db_options(B) PRECISION NK_STRING(C).                           { A = setDatabaseOption(pCxt, B, DB_OPTION_PRECISION, &C); }
+db_options(A) ::= db_options(B) REPLICA NK_INTEGER(C).                            { A = setDatabaseOption(pCxt, B, DB_OPTION_REPLICA, &C); }
+db_options(A) ::= db_options(B) STRICT NK_INTEGER(C).                             { A = setDatabaseOption(pCxt, B, DB_OPTION_STRICT, &C); }
+db_options(A) ::= db_options(B) WAL NK_INTEGER(C).                                { A = setDatabaseOption(pCxt, B, DB_OPTION_WAL, &C); }
+db_options(A) ::= db_options(B) VGROUPS NK_INTEGER(C).                            { A = setDatabaseOption(pCxt, B, DB_OPTION_VGROUPS, &C); }
+db_options(A) ::= db_options(B) SINGLE_STABLE NK_INTEGER(C).                      { A = setDatabaseOption(pCxt, B, DB_OPTION_SINGLE_STABLE, &C); }
+db_options(A) ::= db_options(B) RETENTIONS retention_list(C).                     { A = setDatabaseOption(pCxt, B, DB_OPTION_RETENTIONS, C); }
+
+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); }
 
 %type alter_db_option                                                             { SAlterOption }
 %destructor alter_db_option                                                       { }
-alter_db_option(A) ::= BLOCKS NK_INTEGER(B).                                      { A.type = DB_OPTION_BLOCKS; A.pVal = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &B); }
-alter_db_option(A) ::= FSYNC NK_INTEGER(B).                                       { A.type = DB_OPTION_FSYNC; A.pVal = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &B); }
+alter_db_option(A) ::= BUFFER NK_INTEGER(B).                                      { A.type = DB_OPTION_BUFFER; A.val = B; }
+alter_db_option(A) ::= CACHELAST NK_INTEGER(B).                                   { A.type = DB_OPTION_CACHELAST; A.val = B; }
+alter_db_option(A) ::= FSYNC NK_INTEGER(B).                                       { A.type = DB_OPTION_FSYNC; A.val = B; }
 alter_db_option(A) ::= KEEP integer_list(B).                                      { A.type = DB_OPTION_KEEP; A.pList = B; }
 alter_db_option(A) ::= KEEP variable_list(B).                                     { A.type = DB_OPTION_KEEP; A.pList = B; }
-alter_db_option(A) ::= WAL NK_INTEGER(B).                                         { A.type = DB_OPTION_WAL; A.pVal = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &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); }
+alter_db_option(A) ::= PAGES NK_INTEGER(B).                                       { A.type = DB_OPTION_PAGES; A.val = B; }
+alter_db_option(A) ::= REPLICA NK_INTEGER(B).                                     { A.type = DB_OPTION_REPLICA; A.val = B; }
+alter_db_option(A) ::= STRICT NK_INTEGER(B).                                      { A.type = DB_OPTION_STRICT; A.val = B; }
+alter_db_option(A) ::= WAL NK_INTEGER(B).                                         { A.type = DB_OPTION_WAL; A.val = B; }
 
 %type integer_list                                                                { SNodeList* }
 %destructor integer_list                                                          { nodesDestroyList($$); }
@@ -204,7 +202,7 @@ cmd ::= DROP STABLE exists_opt(A) full_table_name(B).
 cmd ::= ALTER TABLE alter_table_clause(A).                                        { pCxt->pRootNode = A; }
 cmd ::= ALTER STABLE alter_table_clause(A).                                       { pCxt->pRootNode = A; }
 
-alter_table_clause(A) ::= full_table_name(B) alter_table_options(C).              { A = createAlterTableOption(pCxt, B, C); }
+alter_table_clause(A) ::= full_table_name(B) alter_table_options(C).              { A = createAlterTableModifyOptions(pCxt, B, C); }
 alter_table_clause(A) ::=
   full_table_name(B) ADD COLUMN column_name(C) type_name(D).                      { A = createAlterTableAddModifyCol(pCxt, B, TSDB_ALTER_TABLE_ADD_COLUMN, &C, D); }
 alter_table_clause(A) ::= full_table_name(B) DROP COLUMN column_name(C).          { A = createAlterTableDropCol(pCxt, B, TSDB_ALTER_TABLE_DROP_COLUMN, &C); }
@@ -229,7 +227,7 @@ multi_create_clause(A) ::= multi_create_clause(B) create_subtable_clause(C).
 
 create_subtable_clause(A) ::=
   not_exists_opt(B) full_table_name(C) USING full_table_name(D)
-  specific_tags_opt(E) TAGS NK_LP literal_list(F) NK_RP.                          { A = createCreateSubTableClause(pCxt, B, C, D, E, F); }
+  specific_tags_opt(E) TAGS NK_LP literal_list(F) NK_RP table_options(G).         { A = createCreateSubTableClause(pCxt, B, C, D, E, F, G); }
 
 %type multi_drop_clause                                                           { SNodeList* }
 %destructor multi_drop_clause                                                     { nodesDestroyList($$); }
@@ -289,25 +287,21 @@ tags_def_opt(A) ::= tags_def(B).
 %destructor tags_def                                                              { nodesDestroyList($$); }
 tags_def(A) ::= TAGS NK_LP column_def_list(B) NK_RP.                              { A = B; }
 
-table_options(A) ::= .                                                            { A = createTableOptions(pCxt); }
-table_options(A) ::= table_options(B) COMMENT NK_STRING(C).                       { ((STableOptions*)B)->pComments = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BINARY, &C); A = B; }
-table_options(A) ::= table_options(B) KEEP integer_list(C).                       { ((STableOptions*)B)->pKeep = C; A = B; }
-table_options(A) ::= table_options(B) KEEP variable_list(C).                      { ((STableOptions*)B)->pKeep = C; A = B; }
-table_options(A) ::= table_options(B) TTL NK_INTEGER(C).                          { ((STableOptions*)B)->pTtl = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
-table_options(A) ::= table_options(B) SMA NK_LP col_name_list(C) NK_RP.           { ((STableOptions*)B)->pSma = C; A = B; }
-table_options(A) ::= table_options(B) ROLLUP NK_LP func_name_list(C) NK_RP.       { ((STableOptions*)B)->pFuncs = C; A = B; }
-table_options(A) ::= table_options(B) FILE_FACTOR NK_FLOAT(C).                    { ((STableOptions*)B)->pFilesFactor = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_DOUBLE, &C); A = B; }
-table_options(A) ::= table_options(B) DELAY NK_INTEGER(C).                        { ((STableOptions*)B)->pDelay = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &C); A = B; }
+table_options(A) ::= .                                                            { A = createDefaultTableOptions(pCxt); }
+table_options(A) ::= table_options(B) COMMENT NK_STRING(C).                       { A = setTableOption(pCxt, B, TABLE_OPTION_COMMENT, &C); }
+table_options(A) ::= table_options(B) DELAY NK_INTEGER(C).                        { A = setTableOption(pCxt, B, TABLE_OPTION_DELAY, &C); }
+table_options(A) ::= table_options(B) FILE_FACTOR NK_FLOAT(C).                    { A = setTableOption(pCxt, B, TABLE_OPTION_FILE_FACTOR, &C); }
+table_options(A) ::= table_options(B) ROLLUP NK_LP func_name_list(C) NK_RP.       { A = setTableOption(pCxt, B, TABLE_OPTION_ROLLUP, C); }
+table_options(A) ::= table_options(B) TTL NK_INTEGER(C).                          { A = setTableOption(pCxt, B, TABLE_OPTION_TTL, &C); }
+table_options(A) ::= table_options(B) SMA NK_LP col_name_list(C) NK_RP.           { A = setTableOption(pCxt, B, TABLE_OPTION_SMA, C); }
 
-alter_table_options(A) ::= alter_table_option(B).                                 { A = createTableOptions(pCxt); A = setTableAlterOption(pCxt, A, &B); }
-alter_table_options(A) ::= alter_table_options(B) alter_table_option(C).          { A = setTableAlterOption(pCxt, B, &C); }
+alter_table_options(A) ::= alter_table_option(B).                                 { A = createAlterTableOptions(pCxt); A = setTableOption(pCxt, A, B.type, &B.val); }
+alter_table_options(A) ::= alter_table_options(B) alter_table_option(C).          { A = setTableOption(pCxt, B, C.type, &C.val); }
 
 %type alter_table_option                                                          { SAlterOption }
 %destructor alter_table_option                                                    { }
-alter_table_option(A) ::= COMMENT NK_STRING(B).                                   { A.type = TABLE_OPTION_COMMENT; A.pVal = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BINARY, &B); }
-alter_table_option(A) ::= KEEP integer_list(B).                                   { A.type = TABLE_OPTION_KEEP; A.pList = B; }
-alter_table_option(A) ::= KEEP variable_list(B).                                  { A.type = TABLE_OPTION_KEEP; A.pList = B; }
-alter_table_option(A) ::= TTL NK_INTEGER(B).                                      { A.type = TABLE_OPTION_TTL; A.pVal = (SValueNode*)createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &B); }
+alter_table_option(A) ::= COMMENT NK_STRING(B).                                   { A.type = TABLE_OPTION_COMMENT; A.val = B; }
+alter_table_option(A) ::= TTL NK_INTEGER(B).                                      { A.type = TABLE_OPTION_TTL; A.val = B; }
 
 %type col_name_list                                                               { SNodeList* }
 %destructor col_name_list                                                         { nodesDestroyList($$); }

source/libs/parser/src/parAstCreater.c

@@ -47,6 +47,12 @@ void initAstCreateContext(SParseContext* pParseCxt, SAstCreateContext* pCxt) {
   pCxt->placeholderNo = 1;
 }
 
+static void copyStringFormStringToken(SToken* pToken, char* pBuf, int32_t len) {
+  if (pToken->n > 2) {
+    strncpy(pBuf, pToken->z + 1, TMIN(pToken->n - 2, len - 1));
+  }
+}
+
 static void trimEscape(SToken* pName) {
   // todo need to deal with `ioo``ii` -> ioo`ii
   if (NULL != pName && pName->n > 1 && '`' == pName->z[0]) {
@@ -604,64 +610,113 @@ SNode* createSetOperator(SAstCreateContext* pCxt, ESetOperatorType type, SNode*
   return (SNode*)setOp;
 }
 
-SNode* createDatabaseOptions(SAstCreateContext* pCxt) {
+SNode* createDefaultDatabaseOptions(SAstCreateContext* pCxt) {
   SDatabaseOptions* pOptions = nodesMakeNode(QUERY_NODE_DATABASE_OPTIONS);
   CHECK_OUT_OF_MEM(pOptions);
+  pOptions->buffer = TSDB_DEFAULT_BUFFER_PER_VNODE;
+  pOptions->cachelast = TSDB_DEFAULT_CACHE_LAST_ROW;
+  pOptions->compressionLevel = TSDB_DEFAULT_COMP_LEVEL;
+  pOptions->daysPerFile = TSDB_DEFAULT_DAYS_PER_FILE;
+  pOptions->fsyncPeriod = TSDB_DEFAULT_FSYNC_PERIOD;
+  pOptions->maxRowsPerBlock = TSDB_DEFAULT_MAXROWS_FBLOCK;
+  pOptions->minRowsPerBlock = TSDB_DEFAULT_MINROWS_FBLOCK;
+  pOptions->keep[0] = TSDB_DEFAULT_KEEP;
+  pOptions->keep[1] = TSDB_DEFAULT_KEEP;
+  pOptions->keep[2] = TSDB_DEFAULT_KEEP;
+  pOptions->pages = TSDB_DEFAULT_PAGES_PER_VNODE;
+  pOptions->pagesize = TSDB_DEFAULT_PAGESIZE_PER_VNODE;
+  pOptions->precision = TSDB_DEFAULT_PRECISION;
+  pOptions->replica = TSDB_DEFAULT_DB_REPLICA;
+  pOptions->strict = TSDB_DEFAULT_DB_STRICT;
+  pOptions->walLevel = TSDB_DEFAULT_WAL_LEVEL;
+  pOptions->numOfVgroups = TSDB_DEFAULT_VN_PER_DB;
+  pOptions->singleStable = TSDB_DEFAULT_DB_SINGLE_STABLE;
   return (SNode*)pOptions;
 }
 
-SNode* setDatabaseAlterOption(SAstCreateContext* pCxt, SNode* pOptions, SAlterOption* pAlterOption) {
-  switch (pAlterOption->type) {
-    case DB_OPTION_BLOCKS:
-      ((SDatabaseOptions*)pOptions)->pNumOfBlocks = pAlterOption->pVal;
-      break;
-    case DB_OPTION_CACHE:
-      ((SDatabaseOptions*)pOptions)->pCacheBlockSize = pAlterOption->pVal;
+SNode* createAlterDatabaseOptions(SAstCreateContext* pCxt) {
+  SDatabaseOptions* pOptions = nodesMakeNode(QUERY_NODE_DATABASE_OPTIONS);
+  CHECK_OUT_OF_MEM(pOptions);
+  pOptions->buffer = -1;
+  pOptions->cachelast = -1;
+  pOptions->compressionLevel = -1;
+  pOptions->daysPerFile = -1;
+  pOptions->fsyncPeriod = -1;
+  pOptions->maxRowsPerBlock = -1;
+  pOptions->minRowsPerBlock = -1;
+  pOptions->keep[0] = -1;
+  pOptions->keep[1] = -1;
+  pOptions->keep[2] = -1;
+  pOptions->pages = -1;
+  pOptions->pagesize = -1;
+  pOptions->precision = -1;
+  pOptions->replica = -1;
+  pOptions->strict = -1;
+  pOptions->walLevel = -1;
+  pOptions->numOfVgroups = -1;
+  pOptions->singleStable = -1;
+  return (SNode*)pOptions;
+}
+
+SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOptionType type, void* pVal) {
+  switch (type) {
+    case DB_OPTION_BUFFER:
+      ((SDatabaseOptions*)pOptions)->buffer = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_CACHELAST:
-      ((SDatabaseOptions*)pOptions)->pCachelast = pAlterOption->pVal;
+      ((SDatabaseOptions*)pOptions)->cachelast = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_COMP:
-      ((SDatabaseOptions*)pOptions)->pCompressionLevel = pAlterOption->pVal;
+      ((SDatabaseOptions*)pOptions)->compressionLevel = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
-    case DB_OPTION_DAYS:
-      ((SDatabaseOptions*)pOptions)->pDaysPerFile = pAlterOption->pVal;
+    case DB_OPTION_DAYS: {
+      SToken* pToken = pVal;
+      if (TK_NK_INTEGER == pToken->type) {
+        ((SDatabaseOptions*)pOptions)->daysPerFile = strtol(pToken->z, NULL, 10);
+      } else {
+        ((SDatabaseOptions*)pOptions)->pDaysPerFile = (SValueNode*)createDurationValueNode(pCxt, pToken);
+      }
       break;
+    }
     case DB_OPTION_FSYNC:
-      ((SDatabaseOptions*)pOptions)->pFsyncPeriod = pAlterOption->pVal;
+      ((SDatabaseOptions*)pOptions)->fsyncPeriod = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_MAXROWS:
-      ((SDatabaseOptions*)pOptions)->pMaxRowsPerBlock = pAlterOption->pVal;
+      ((SDatabaseOptions*)pOptions)->maxRowsPerBlock = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_MINROWS:
-      ((SDatabaseOptions*)pOptions)->pMinRowsPerBlock = pAlterOption->pVal;
+      ((SDatabaseOptions*)pOptions)->minRowsPerBlock = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_KEEP:
-      ((SDatabaseOptions*)pOptions)->pKeep = pAlterOption->pList;
+      ((SDatabaseOptions*)pOptions)->pKeep = pVal;
+      break;
+    case DB_OPTION_PAGES:
+      ((SDatabaseOptions*)pOptions)->pages = strtol(((SToken*)pVal)->z, NULL, 10);
+      break;
+    case DB_OPTION_PAGESIZE:
+      ((SDatabaseOptions*)pOptions)->pagesize = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_PRECISION:
-      ((SDatabaseOptions*)pOptions)->pPrecision = pAlterOption->pVal;
+      copyStringFormStringToken((SToken*)pVal, ((SDatabaseOptions*)pOptions)->precisionStr,
+                                sizeof(((SDatabaseOptions*)pOptions)->precisionStr));
       break;
     case DB_OPTION_REPLICA:
-      ((SDatabaseOptions*)pOptions)->pReplica = pAlterOption->pVal;
+      ((SDatabaseOptions*)pOptions)->replica = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
-    case DB_OPTION_TTL:
-      ((SDatabaseOptions*)pOptions)->pTtl = pAlterOption->pVal;
+    case DB_OPTION_STRICT:
+      ((SDatabaseOptions*)pOptions)->strict = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_WAL:
-      ((SDatabaseOptions*)pOptions)->pWalLevel = pAlterOption->pVal;
+      ((SDatabaseOptions*)pOptions)->walLevel = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_VGROUPS:
-      ((SDatabaseOptions*)pOptions)->pNumOfVgroups = pAlterOption->pVal;
+      ((SDatabaseOptions*)pOptions)->numOfVgroups = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_SINGLE_STABLE:
-      ((SDatabaseOptions*)pOptions)->pSingleStable = pAlterOption->pVal;
-      break;
-    case DB_OPTION_STREAM_MODE:
-      ((SDatabaseOptions*)pOptions)->pStreamMode = pAlterOption->pVal;
+      ((SDatabaseOptions*)pOptions)->singleStable = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_RETENTIONS:
-      ((SDatabaseOptions*)pOptions)->pRetentions = pAlterOption->pList;
+      ((SDatabaseOptions*)pOptions)->pRetentions = pVal;
       break;
     default:
       break;
@@ -669,6 +724,17 @@ SNode* setDatabaseAlterOption(SAstCreateContext* pCxt, SNode* pOptions, SAlterOp
   return pOptions;
 }
 
+SNode* setAlterDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, SAlterOption* pAlterOption) {
+  switch (pAlterOption->type) {
+    case DB_OPTION_KEEP:
+    case DB_OPTION_RETENTIONS:
+      return setDatabaseOption(pCxt, pOptions, pAlterOption->type, pAlterOption->pList);
+    default:
+      break;
+  }
+  return setDatabaseOption(pCxt, pOptions, pAlterOption->type, &pAlterOption->val);
+}
+
 SNode* createCreateDatabaseStmt(SAstCreateContext* pCxt, bool ignoreExists, SToken* pDbName, SNode* pOptions) {
   if (!checkDbName(pCxt, pDbName, false)) {
     return NULL;
@@ -703,31 +769,44 @@ SNode* createAlterDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName, SNode*
   return (SNode*)pStmt;
 }
 
-SNode* createTableOptions(SAstCreateContext* pCxt) {
+SNode* createDefaultTableOptions(SAstCreateContext* pCxt) {
   STableOptions* pOptions = nodesMakeNode(QUERY_NODE_TABLE_OPTIONS);
   CHECK_OUT_OF_MEM(pOptions);
+  pOptions->delay = TSDB_DEFAULT_ROLLUP_DELAY;
+  pOptions->filesFactor = TSDB_DEFAULT_ROLLUP_FILE_FACTOR;
+  pOptions->ttl = TSDB_DEFAULT_TABLE_TTL;
   return (SNode*)pOptions;
 }
 
-SNode* setTableAlterOption(SAstCreateContext* pCxt, SNode* pOptions, SAlterOption* pAlterOption) {
-  switch (pAlterOption->type) {
-    case TABLE_OPTION_KEEP:
-      ((STableOptions*)pOptions)->pKeep = pAlterOption->pList;
-      break;
-    case TABLE_OPTION_TTL:
-      ((STableOptions*)pOptions)->pTtl = pAlterOption->pVal;
-      break;
+SNode* createAlterTableOptions(SAstCreateContext* pCxt) {
+  STableOptions* pOptions = nodesMakeNode(QUERY_NODE_TABLE_OPTIONS);
+  CHECK_OUT_OF_MEM(pOptions);
+  pOptions->delay = -1;
+  pOptions->filesFactor = -1;
+  pOptions->ttl = -1;
+  return (SNode*)pOptions;
+}
+
+SNode* setTableOption(SAstCreateContext* pCxt, SNode* pOptions, ETableOptionType type, void* pVal) {
+  switch (type) {
     case TABLE_OPTION_COMMENT:
-      ((STableOptions*)pOptions)->pComments = pAlterOption->pVal;
+      copyStringFormStringToken((SToken*)pVal, ((STableOptions*)pOptions)->comment,
+                                sizeof(((STableOptions*)pOptions)->comment));
       break;
-    case TABLE_OPTION_SMA:
-      ((STableOptions*)pOptions)->pSma = pAlterOption->pList;
+    case TABLE_OPTION_DELAY:
+      ((STableOptions*)pOptions)->delay = strtol(((SToken*)pVal)->z, NULL, 10);
       break;
     case TABLE_OPTION_FILE_FACTOR:
-      ((STableOptions*)pOptions)->pFilesFactor = pAlterOption->pVal;
+      ((STableOptions*)pOptions)->filesFactor = strtod(((SToken*)pVal)->z, NULL);
       break;
-    case TABLE_OPTION_DELAY:
-      ((STableOptions*)pOptions)->pDelay = pAlterOption->pVal;
+    case TABLE_OPTION_ROLLUP:
+      ((STableOptions*)pOptions)->pRollupFuncs = pVal;
+      break;
+    case TABLE_OPTION_TTL:
+      ((STableOptions*)pOptions)->ttl = strtol(((SToken*)pVal)->z, NULL, 10);
+      break;
+    case TABLE_OPTION_SMA:
+      ((STableOptions*)pOptions)->pSma = pVal;
       break;
     default:
       break;
@@ -778,7 +857,7 @@ SNode* createCreateTableStmt(SAstCreateContext* pCxt, bool ignoreExists, SNode*
 }
 
 SNode* createCreateSubTableClause(SAstCreateContext* pCxt, bool ignoreExists, SNode* pRealTable, SNode* pUseRealTable,
-                                  SNodeList* pSpecificTags, SNodeList* pValsOfTags) {
+                                  SNodeList* pSpecificTags, SNodeList* pValsOfTags, SNode* pOptions) {
   if (NULL == pRealTable) {
     return NULL;
   }
@@ -833,7 +912,7 @@ SNode* createDropSuperTableStmt(SAstCreateContext* pCxt, bool ignoreNotExists, S
   return (SNode*)pStmt;
 }
 
-SNode* createAlterTableOption(SAstCreateContext* pCxt, SNode* pRealTable, SNode* pOptions) {
+SNode* createAlterTableModifyOptions(SAstCreateContext* pCxt, SNode* pRealTable, SNode* pOptions) {
   if (NULL == pRealTable) {
     return NULL;
   }

source/libs/parser/src/parTokenizer.c

@@ -43,7 +43,7 @@ static SKeyword keywordTable[] = {
     {"BETWEEN", TK_BETWEEN},
     {"BINARY", TK_BINARY},
     {"BIGINT", TK_BIGINT},
-    {"BLOCKS", TK_BLOCKS},
+    // {"BLOCKS", TK_BLOCKS},
     {"BNODE", TK_BNODE},
     {"BNODES", TK_BNODES},
     {"BOOL", TK_BOOL},
@@ -142,7 +142,7 @@ static SKeyword keywordTable[] = {
     {"QTIME", TK_QTIME},
     {"QUERIES", TK_QUERIES},
     {"QUERY", TK_QUERY},
-    {"QUORUM", TK_QUORUM},
+    // {"QUORUM", TK_QUORUM},
     {"RATIO", TK_RATIO},
     {"REPLICA", TK_REPLICA},
     {"RESET", TK_RESET},
@@ -169,7 +169,7 @@ static SKeyword keywordTable[] = {
     {"STORAGE", TK_STORAGE},
     {"STREAM", TK_STREAM},
     {"STREAMS", TK_STREAMS},
-    {"STREAM_MODE", TK_STREAM_MODE},
+    // {"STREAM_MODE", TK_STREAM_MODE},
     {"STRICT", TK_STRICT},
     {"SYNCDB", TK_SYNCDB},
     {"TABLE", TK_TABLE},

source/libs/parser/src/parUtil.c

@@ -68,8 +68,6 @@ static char* getSyntaxErrFormat(int32_t errCode) {
       return "Invalid option %s: %s";
     case TSDB_CODE_PAR_INVALID_ENUM_OPTION:
       return "Invalid option %s: %" PRId64 ", only %d, %d allowed";
-    case TSDB_CODE_PAR_INVALID_TTL_OPTION:
-      return "Invalid option ttl: %" PRId64 ", should be greater than or equal to %d";
     case TSDB_CODE_PAR_INVALID_KEEP_NUM:
       return "Invalid number of keep options";
     case TSDB_CODE_PAR_INVALID_KEEP_ORDER:
@@ -89,7 +87,7 @@ static char* getSyntaxErrFormat(int32_t errCode) {
     case TSDB_CODE_PAR_INVALID_OPTION_UNIT:
       return "Invalid option %s unit: %c, only m, h, d allowed";
     case TSDB_CODE_PAR_INVALID_KEEP_UNIT:
-      return "Invalid option keep unit: %c, %c, %c, only m, h, d allowed";
+      return "Invalid option keep unit: %c, only m, h, d allowed";
     case TSDB_CODE_PAR_AGG_FUNC_NESTING:
       return "Aggregate functions do not support nesting";
     case TSDB_CODE_PAR_INVALID_STATE_WIN_TYPE:

source/libs/parser/test/parserTestUtil.h → source/libs/parser/test/parAlterTest.cpp

@@ -13,9 +13,14 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef PARSER_TEST_UTIL_H
-#define PARSER_TEST_UTIL_H
+#include "parTestUtil.h"
 
-extern bool g_isDump;
+using namespace std;
 
-#endif  // PARSER_TEST_UTIL_H
+class ParserAlterTest : public ParserTestBase {};
+
+TEST_F(ParserAlterTest, stmt) {
+  useDb("root", "test");
+
+  run("create database db1");
+}

source/libs/parser/test/parserTestMain.cpp → source/libs/parser/test/parTestMain.cpp

@@ -13,23 +13,22 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include <getopt.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string>
 
-#include <getopt.h>
 #include <gtest/gtest.h>
 
 #ifdef WINDOWS
 #define TD_USE_WINSOCK
 #endif
+
 #include "functionMgt.h"
 #include "mockCatalog.h"
 #include "os.h"
+#include "parTestUtil.h"
 #include "parToken.h"
-#include "parserTestUtil.h"
-
-bool g_isDump = false;
 
 class ParserEnv : public testing::Environment {
  public:

source/libs/parser/test/parTestUtil.cpp

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "parTestUtil.h"
+
+#include <algorithm>
+#include <array>
+
+#include "parInt.h"
+
+using namespace std;
+using namespace testing;
+
+#define DO_WITH_THROW(func, ...)                                                                                   \
+  do {                                                                                                             \
+    int32_t code__ = func(__VA_ARGS__);                                                                            \
+    if (TSDB_CODE_SUCCESS != code__) {                                                                             \
+      throw runtime_error("sql:[" + stmtEnv_.sql_ + "] " #func " code:" + to_string(code__) +                      \
+                          ", strerror:" + string(tstrerror(code__)) + ", msg:" + string(stmtEnv_.msgBuf_.data())); \
+    }                                                                                                              \
+  } while (0);
+
+bool g_isDump = false;
+
+class ParserTestBaseImpl {
+ public:
+  void useDb(const string& acctId, const string& db) {
+    caseEnv_.acctId_ = acctId;
+    caseEnv_.db_ = db;
+  }
+
+  void run(const string& sql) {
+    reset();
+    try {
+      SParseContext cxt = {0};
+      setParseContext(sql, &cxt);
+
+      SQuery* pQuery = nullptr;
+      doParse(&cxt, &pQuery);
+
+      doTranslate(&cxt, pQuery);
+
+      doCalculateConstant(&cxt, pQuery);
+
+      if (g_isDump) {
+        dump();
+      }
+    } catch (...) {
+      dump();
+      throw;
+    }
+  }
+
+ private:
+  struct caseEnv {
+    string acctId_;
+    string db_;
+  };
+
+  struct stmtEnv {
+    string            sql_;
+    array<char, 1024> msgBuf_;
+  };
+
+  struct stmtRes {
+    string parsedAst_;
+    string translatedAst_;
+    string calcConstAst_;
+  };
+
+  void reset() {
+    stmtEnv_.sql_.clear();
+    stmtEnv_.msgBuf_.fill(0);
+
+    res_.parsedAst_.clear();
+    res_.translatedAst_.clear();
+    res_.calcConstAst_.clear();
+  }
+
+  void dump() {
+    cout << "==========================================sql : [" << stmtEnv_.sql_ << "]" << endl;
+    cout << "raw syntax tree : " << endl;
+    cout << res_.parsedAst_ << endl;
+    cout << "translated syntax tree : " << endl;
+    cout << res_.translatedAst_ << endl;
+    cout << "optimized syntax tree : " << endl;
+    cout << res_.calcConstAst_ << endl;
+  }
+
+  void setParseContext(const string& sql, SParseContext* pCxt) {
+    stmtEnv_.sql_ = sql;
+    transform(stmtEnv_.sql_.begin(), stmtEnv_.sql_.end(), stmtEnv_.sql_.begin(), ::tolower);
+
+    pCxt->acctId = atoi(caseEnv_.acctId_.c_str());
+    pCxt->db = caseEnv_.db_.c_str();
+    pCxt->pSql = stmtEnv_.sql_.c_str();
+    pCxt->sqlLen = stmtEnv_.sql_.length();
+    pCxt->pMsg = stmtEnv_.msgBuf_.data();
+    pCxt->msgLen = stmtEnv_.msgBuf_.max_size();
+  }
+
+  void doParse(SParseContext* pCxt, SQuery** pQuery) {
+    DO_WITH_THROW(parse, pCxt, pQuery);
+    res_.parsedAst_ = toString((*pQuery)->pRoot);
+  }
+
+  void doTranslate(SParseContext* pCxt, SQuery* pQuery) {
+    DO_WITH_THROW(translate, pCxt, pQuery);
+    res_.translatedAst_ = toString(pQuery->pRoot);
+  }
+
+  void doCalculateConstant(SParseContext* pCxt, SQuery* pQuery) {
+    DO_WITH_THROW(calculateConstant, pCxt, pQuery);
+    res_.calcConstAst_ = toString(pQuery->pRoot);
+  }
+
+  string toString(const SNode* pRoot) {
+    char*   pStr = NULL;
+    int32_t len = 0;
+    DO_WITH_THROW(nodesNodeToString, pRoot, false, &pStr, &len)
+    string str(pStr);
+    taosMemoryFreeClear(pStr);
+    return str;
+  }
+
+  caseEnv caseEnv_;
+  stmtEnv stmtEnv_;
+  stmtRes res_;
+};
+
+ParserTestBase::ParserTestBase() : impl_(new ParserTestBaseImpl()) {}
+
+ParserTestBase::~ParserTestBase() {}
+
+void ParserTestBase::useDb(const std::string& acctId, const std::string& db) { impl_->useDb(acctId, db); }
+
+void ParserTestBase::run(const std::string& sql) { return impl_->run(sql); }

source/libs/parser/test/parTestUtil.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef PARSER_TEST_UTIL_H
+#define PARSER_TEST_UTIL_H
+
+#include <gtest/gtest.h>
+
+class ParserTestBaseImpl;
+
+class ParserTestBase : public testing::Test {
+ public:
+  ParserTestBase();
+  virtual ~ParserTestBase();
+
+  void useDb(const std::string& acctId, const std::string& db);
+  void run(const std::string& sql);
+
+ private:
+  std::unique_ptr<ParserTestBaseImpl> impl_;
+};
+
+extern bool g_isDump;
+
+#endif  // PARSER_TEST_UTIL_H

source/libs/parser/test/parserAstTest.cpp

@@ -19,7 +19,7 @@
 #include <gtest/gtest.h>
 
 #include "parInt.h"
-#include "parserTestUtil.h"
+#include "parTestUtil.h"
 
 using namespace std;
 using namespace testing;

source/libs/planner/test/planTestUtil.cpp

@@ -14,9 +14,9 @@
  */
 
 #include "planTestUtil.h"
-#include <array>
 
 #include <algorithm>
+#include <array>
 
 #include "cmdnodes.h"
 #include "parser.h"

source/libs/scalar/src/sclvector.c

@@ -1058,36 +1058,36 @@ void vectorMathDivide(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *p
   _getDoubleValue_fn_t getVectorDoubleValueFnRight = getVectorDoubleValueFn(pRightCol->info.type);
 
   double *output = (double *)pOutputCol->pData;
-  if (pLeft->numOfRows == pRight->numOfRows) {  // check for the 0 value
+  if (pLeft->numOfRows == pRight->numOfRows) {
     for (; i < pRight->numOfRows && i >= 0; i += step, output += 1) {
-      if (colDataIsNull_s(pLeft->columnData, i) || colDataIsNull_s(pRight->columnData, i)) {
+      if (colDataIsNull_s(pLeft->columnData, i) || colDataIsNull_s(pRight->columnData, i) || (getVectorDoubleValueFnRight(RIGHT_COL, i) == 0)) { //divide by 0 check
         colDataAppendNULL(pOutputCol, i);
-        continue;  // TODO set null or ignore
+        continue;
       }
       *output = getVectorDoubleValueFnLeft(LEFT_COL, i)
-                 /getVectorDoubleValueFnRight(RIGHT_COL, i);
+                / getVectorDoubleValueFnRight(RIGHT_COL, i);
     }
   } else if (pLeft->numOfRows == 1) {
     if (colDataIsNull_s(pLeftCol, 0)) {  // Set pLeft->numOfRows NULL value
       colDataAppendNNULL(pOutputCol, 0, pRight->numOfRows);
     } else {
       for (; i >= 0 && i < pRight->numOfRows; i += step, output += 1) {
-        if (colDataIsNull_s(pRightCol, i)) {
+        if (colDataIsNull_s(pRightCol, i) || (getVectorDoubleValueFnRight(RIGHT_COL, i) == 0)) { // divide by 0 check
           colDataAppendNULL(pOutputCol, i);
-          continue;  // TODO set null or ignore
+          continue;
         }
         *output = getVectorDoubleValueFnLeft(LEFT_COL, 0)
                   / getVectorDoubleValueFnRight(RIGHT_COL, i);
       }
     }
   } else if (pRight->numOfRows == 1) {
-    if (colDataIsNull_s(pRightCol, 0)) {  // Set pLeft->numOfRows NULL value
+    if (colDataIsNull_s(pRightCol, 0) || (getVectorDoubleValueFnRight(RIGHT_COL, 0) == 0)) {  // Set pLeft->numOfRows NULL value (divde by 0 check)
       colDataAppendNNULL(pOutputCol, 0, pLeft->numOfRows);
     } else {
       for (; i >= 0 && i < pLeft->numOfRows; i += step, output += 1) {
         if (colDataIsNull_s(pLeftCol, i)) {
           colDataAppendNULL(pOutputCol, i);
-          continue;  // TODO set null or ignore
+          continue;
         }
         *output = getVectorDoubleValueFnLeft(LEFT_COL, i)
                   / getVectorDoubleValueFnRight(RIGHT_COL, 0);
@@ -1195,9 +1195,10 @@ void vectorMathMinus(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pO
   for (; i < pLeft->numOfRows && i >= 0; i += step, output += 1) {
     if (colDataIsNull_s(pLeft->columnData, i)) {
       colDataAppendNULL(pOutputCol, i);
-      continue;  // TODO set null or ignore
+      continue;
     }
-    *output = - getVectorDoubleValueFnLeft(LEFT_COL, i);
+    double result = getVectorDoubleValueFnLeft(LEFT_COL, i);
+    *output = (result == 0) ? 0 : -result;
   }
 
   doReleaseVec(pLeftCol,  leftConvert);

tests/script/tsim/db/alter_option.sim

@@ -66,7 +66,7 @@ print ============= create database
 #  | REPLICA value      [1 | 3]
 #  | WAL value          [1 | 2]
 
-sql create database db BLOCKS 7 CACHE 3 CACHELAST 3 COMP 0 DAYS 345600 FSYNC 1000 MAXROWS 8000 MINROWS 10 KEEP 1440000 PRECISION 'ns' REPLICA 1 TTL 7 WAL 2 VGROUPS 6 SINGLE_STABLE 1 STREAM_MODE 1
+sql create database db CACHELAST 3 COMP 0 DAYS 345600 FSYNC 1000 MAXROWS 8000 MINROWS 10 KEEP 1440000 PRECISION 'ns' REPLICA 1 WAL 2 VGROUPS 6 SINGLE_STABLE 1
 sql show databases
 print rows: $rows
 print $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09
@@ -98,12 +98,12 @@ endi
 if $data7_db != 1440000,1440000,1440000 then # keep
   return -1
 endi
-if $data8_db != 3 then  # cache
-  return -1
-endi
-if $data9_db != 7 then  # blocks
-  return -1
-endi
+#if $data8_db != 3 then  # cache
+#  return -1
+#endi
+#if $data9_db != 7 then  # blocks
+#  return -1
+#endi
 if $data10_db != 10 then  # minrows
   return -1
 endi
@@ -250,41 +250,41 @@ sql_error alter database db keep 0
 sql_error alter database db keep -1
 #sql_error alter database db keep 365001
 
-print ============== modify cache
-sql_error alter database db cache 12
-sql_error alter database db cache 1
-sql_error alter database db cache 60
-sql_error alter database db cache 50
-sql_error alter database db cache 20
-sql_error alter database db cache 3
-sql_error alter database db cache 129
-sql_error alter database db cache 300
-sql_error alter database db cache 0
-sql_error alter database db cache -1
+#print ============== modify cache
+#sql_error alter database db cache 12
+#sql_error alter database db cache 1
+#sql_error alter database db cache 60
+#sql_error alter database db cache 50
+#sql_error alter database db cache 20
+#sql_error alter database db cache 3
+#sql_error alter database db cache 129
+#sql_error alter database db cache 300
+#sql_error alter database db cache 0
+#sql_error alter database db cache -1
 
-print ============== modify blocks
-sql alter database db blocks 3
-sql show databases
-print blocks $data9_db
-if $data9_db != 3 then 
-  return -1
-endi
-sql alter database db blocks 11
-sql show databases
-print blocks $data9_db
-if $data9_db != 11 then 
-  return -1
-endi
+#print ============== modify blocks
+#sql alter database db blocks 3
+#sql show databases
+#print blocks $data9_db
+#if $data9_db != 3 then 
+#  return -1
+#endi
+#sql alter database db blocks 11
+#sql show databases
+#print blocks $data9_db
+#if $data9_db != 11 then 
+#  return -1
+#endi
 
-sql alter database db blocks 40
-sql alter database db blocks 30
-sql alter database db blocks 20
-sql alter database db blocks 10
-sql_error alter database db blocks 2
-sql_error alter database db blocks 1
-sql_error alter database db blocks 0
-sql_error alter database db blocks -1
-sql_error alter database db blocks 10001
+#sql alter database db blocks 40
+#sql alter database db blocks 30
+#sql alter database db blocks 20
+#sql alter database db blocks 10
+#sql_error alter database db blocks 2
+#sql_error alter database db blocks 1
+#sql_error alter database db blocks 0
+#sql_error alter database db blocks -1
+#sql_error alter database db blocks 10001
 
 print ============== modify minrows
 sql_error alter database db minrows 8

tests/script/tsim/db/basic6.sim

@@ -15,7 +15,8 @@ $tb = $tbPrefix . $i
 
 print =============== step1
 # quorum  presicion 
-sql create database $db vgroups 8 replica 1 days 2880 keep 3650 cache 32 blocks 12 minrows 80 maxrows 10000 wal 2 fsync 1000 comp 0 cachelast 2 precision 'us'
+#sql create database $db vgroups 8 replica 1 days 2880 keep 3650 cache 32 blocks 12 minrows 80 maxrows 10000 wal 2 fsync 1000 comp 0 cachelast 2 precision 'us'
+sql create database $db vgroups 8 replica 1 days 2880 keep 3650 minrows 80 maxrows 10000 wal 2 fsync 1000 comp 0 cachelast 2 precision 'us'
 sql show databases
 print $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09
 
@@ -40,12 +41,12 @@ endi
 if $data27 != 3650,3650,3650 then
   return -1
 endi
-if $data28 != 32 then
-  return -1
-endi
-if $data29 != 12 then
-  return -1
-endi
+#if $data28 != 32 then
+#  return -1
+#endi
+#if $data29 != 12 then
+#  return -1
+#endi
  
 print =============== step2
 sql_error create database $db

tests/script/tsim/db/create_all_options.sim

@@ -124,12 +124,12 @@ endi
 if $data7_db != 5256000,5256000,5256000 then # keep
   return -1
 endi
-if $data8_db != 16 then  # cache
-  return -1
-endi
-if $data9_db != 6 then  # blocks
-  return -1
-endi
+#if $data8_db != 16 then  # cache
+#  return -1
+#endi
+#if $data9_db != 6 then  # blocks
+#  return -1
+#endi
 if $data10_db != 100 then  # minrows
   return -1
 endi
@@ -153,42 +153,42 @@ if $data16_db != ms then  # precision
 endi
 sql drop database db
 
-print ====> BLOCKS value       [3~1000, default: 6]
-sql create database db BLOCKS 3
-sql show databases
-print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
-if $data9_db != 3 then
-  return -1
-endi
-sql drop database db
+#print ====> BLOCKS value       [3~1000, default: 6]
+#sql create database db BLOCKS 3
+#sql show databases
+#print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
+#if $data9_db != 3 then
+#  return -1
+#endi
+#sql drop database db
 
-sql create database db BLOCKS 1000
-sql show databases
-print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
-if $data9_db != 1000 then
-  return -1
-endi
-sql drop database db
-sql_error create database db BLOCKS 2
-sql_error create database db BLOCKS 0
-sql_error create database db BLOCKS -1
+#sql create database db BLOCKS 1000
+#sql show databases
+#print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
+#if $data9_db != 1000 then
+#  return -1
+#endi
+#sql drop database db
+#sql_error create database db BLOCKS 2
+#sql_error create database db BLOCKS 0
+#sql_error create database db BLOCKS -1
 
-print ====> CACHE value [default: 16]
-sql create database db CACHE 1
-sql show databases
-print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
-if $data8_db != 1 then
-  return -1
-endi
-sql drop database db
+#print ====> CACHE value [default: 16]
+#sql create database db CACHE 1
+#sql show databases
+#print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
+#if $data8_db != 1 then
+#  return -1
+#endi
+#sql drop database db
 
-sql create database db CACHE 128
-sql show databases
-print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
-if $data8_db != 128 then
-  return -1
-endi
-sql drop database db
+#sql create database db CACHE 128
+#sql show databases
+#print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
+#if $data8_db != 128 then
+#  return -1
+#endi
+#sql drop database db
 
 print ====> CACHELAST value [0, 1, 2, 3, default: 0]
 sql create database db CACHELAST 1
@@ -387,24 +387,24 @@ sql_error create database db REPLICA 0
 sql_error create database db REPLICA -1
 sql_error create database db REPLICA 4
 
-print ====> TTL value [1d ~ , default: 1]
-sql create database db TTL 1
-sql show databases
-print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
+#print ====> TTL value [1d ~ , default: 1]
+#sql create database db TTL 1
+#sql show databases
+#print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
 #if $dataXX_db != 1 then
 #  return -1
 #endi
-sql drop database db
+#sql drop database db
 
-sql create database db TTL 10
-sql show databases
-print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
+#sql create database db TTL 10
+#sql show databases
+#print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
 #if $dataXX_db != 10 then
 #  return -1
 #endi
-sql drop database db
-sql_error create database db TTL 0
-sql_error create database db TTL -1
+#sql drop database db
+#sql_error create database db TTL 0
+#sql_error create database db TTL -1
 
 print ====> WAL value [1 | 2, default: 1]
 sql create database db WAL 2
@@ -465,24 +465,24 @@ sql drop database db
 sql_error create database db SINGLE_STABLE 2
 sql_error create database db SINGLE_STABLE -1
 
-print ====> STREAM_MODE [0 | 1, default: ]
-sql create database db STREAM_MODE 1
-sql show databases
-print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
+#print ====> STREAM_MODE [0 | 1, default: ]
+#sql create database db STREAM_MODE 1
+#sql show databases
+#print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
 #if $dataXXX_db != 1 then
 #  return -1
 #endi
-sql drop database db
+#sql drop database db
 
-sql create database db STREAM_MODE 0
-sql show databases
-print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
+#sql create database db STREAM_MODE 0
+#sql show databases
+#print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db
 #if $dataXXX_db != 0 then
 #  return -1
 #endi
-sql drop database db
-sql_error create database db STREAM_MODE 2
-sql_error create database db STREAM_MODE -1
+#sql drop database db
+#sql_error create database db STREAM_MODE 2
+#sql_error create database db STREAM_MODE -1
 
 system sh/exec.sh -n dnode1 -s stop -x SIGINT
 system sh/exec.sh -n dnode2 -s stop -x SIGINT

tests/script/tsim/parser/fourArithmetic-basic.sim

@@ -94,7 +94,7 @@ endi
 if $data01 != 10.000000000 then
  return -1
 endi
-if $data02 != -nan then
+if $data02 != NULL then
  return -1
 endi
 if $data03 != -10.000000000 then

tools/shell/inc/shellInt.h

@@ -62,7 +62,7 @@ typedef struct {
   bool        is_check;
   bool        is_startup;
   bool        is_help;
-  uint16_t    port;
+  int32_t     port;
   int32_t     pktLen;
   int32_t     pktNum;
   int32_t     displayWidth;

tools/shell/src/shellArguments.c

@@ -99,6 +99,7 @@ static int32_t shellParseSingleOpt(int32_t key, char *arg) {
       break;
     case 'P':
       pArgs->port = atoi(arg);
+      if (pArgs->port == 0) pArgs->port = -1;
       break;
     case 'u':
       pArgs->user = arg;
@@ -304,6 +305,11 @@ static int32_t shellCheckArgs() {
     return -1;
   }
 
+  if (pArgs->port < 0 || pArgs->port > 65535) {
+    printf("Invalid port\n");
+    return -1;
+  }
+
   if (pArgs->pktLen < SHELL_MIN_PKG_LEN || pArgs->pktLen > SHELL_MAX_PKG_LEN) {
     printf("Invalid pktLen:%d, range:[%d, %d]\n", pArgs->pktLen, SHELL_MIN_PKG_LEN, SHELL_MAX_PKG_LEN);
     return -1;