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

feat: vgroup manage command

include/common/tmsg.h

@@ -1315,6 +1315,31 @@ typedef struct {
 int32_t tSerializeSKillTransReq(void* buf, int32_t bufLen, SKillTransReq* pReq);
 int32_t tDeserializeSKillTransReq(void* buf, int32_t bufLen, SKillTransReq* pReq);
 
+typedef struct {
+  int32_t useless;  // useless
+} SBalanceVgroupReq;
+
+int32_t tSerializeSBalanceVgroupReq(void* buf, int32_t bufLen, SBalanceVgroupReq* pReq);
+int32_t tDeserializeSBalanceVgroupReq(void* buf, int32_t bufLen, SBalanceVgroupReq* pReq);
+
+typedef struct {
+  int32_t vgId1;
+  int32_t vgId2;
+} SMergeVgroupReq;
+
+int32_t tSerializeSMergeVgroupReq(void* buf, int32_t bufLen, SMergeVgroupReq* pReq);
+int32_t tDeserializeSMergeVgroupReq(void* buf, int32_t bufLen, SMergeVgroupReq* pReq);
+
+typedef struct {
+  int32_t vgId;
+  int32_t dnodeId1;
+  int32_t dnodeId2;
+  int32_t dnodeId3;
+} SRedistributeVgroupReq;
+
+int32_t tSerializeSRedistributeVgroupReq(void* buf, int32_t bufLen, SRedistributeVgroupReq* pReq);
+int32_t tDeserializeSRedistributeVgroupReq(void* buf, int32_t bufLen, SRedistributeVgroupReq* pReq);
+
 typedef struct {
   char user[TSDB_USER_LEN];
   char spi;

include/common/tmsgdef.h

@@ -153,6 +153,9 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_GRANT, "grant", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_AUTH, "auth", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_APPLY_MSG, "mnode-apply", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_BALANCE_VGROUP, "balance-vgroup", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_MERGE_VGROUP, "merge-vgroup", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_REDISTRIBUTE_VGROUP, "redistribute-vgroup", NULL, NULL)
 
   TD_NEW_MSG_SEG(TDMT_VND_MSG)
   TD_DEF_MSG_TYPE(TDMT_VND_SUBMIT, "submit", SSubmitReq, SSubmitRsp)

include/common/ttokendef.h

@@ -188,10 +188,10 @@
 #define TK_KILL          170
 #define TK_CONNECTION    171
 #define TK_TRANSACTION   172
-#define TK_MERGE         173
+#define TK_BALANCE       173
 #define TK_VGROUP        174
-#define TK_REDISTRIBUTE  175
-#define TK_SPLIT         176
+#define TK_MERGE         175
+#define TK_REDISTRIBUTE  176
 #define TK_SYNCDB        177
 #define TK_DELETE        178
 #define TK_NULL          179

include/libs/nodes/cmdnodes.h

@@ -28,6 +28,14 @@ extern "C" {
 #define DESCRIBE_RESULT_TYPE_LEN  (20 + VARSTR_HEADER_SIZE)
 #define DESCRIBE_RESULT_NOTE_LEN  (8 + VARSTR_HEADER_SIZE)
 
+#define PRIVILEGE_TYPE_MASK(n) (1 << n)
+
+#define PRIVILEGE_TYPE_ALL   PRIVILEGE_TYPE_MASK(0)
+#define PRIVILEGE_TYPE_READ  PRIVILEGE_TYPE_MASK(1)
+#define PRIVILEGE_TYPE_WRITE PRIVILEGE_TYPE_MASK(2)
+
+#define PRIVILEGE_TYPE_TEST_MASK(val, mask) (((val) & (mask)) != 0)
+
 typedef struct SDatabaseOptions {
   ENodeType   type;
   int32_t     buffer;
@@ -316,14 +324,6 @@ typedef struct SDropFunctionStmt {
   bool      ignoreNotExists;
 } SDropFunctionStmt;
 
-#define PRIVILEGE_TYPE_MASK(n) (1 << n)
-
-#define PRIVILEGE_TYPE_ALL   PRIVILEGE_TYPE_MASK(0)
-#define PRIVILEGE_TYPE_READ  PRIVILEGE_TYPE_MASK(1)
-#define PRIVILEGE_TYPE_WRITE PRIVILEGE_TYPE_MASK(2)
-
-#define PRIVILEGE_TYPE_TEST_MASK(val, mask) (((val) & (mask)) != 0)
-
 typedef struct SGrantStmt {
   ENodeType type;
   char      userName[TSDB_USER_LEN];
@@ -333,6 +333,30 @@ typedef struct SGrantStmt {
 
 typedef SGrantStmt SRevokeStmt;
 
+typedef struct SBalanceVgroupStmt {
+  ENodeType type;
+} SBalanceVgroupStmt;
+
+typedef struct SMergeVgroupStmt {
+  ENodeType type;
+  int32_t   vgId1;
+  int32_t   vgId2;
+} SMergeVgroupStmt;
+
+typedef struct SRedistributeVgroupStmt {
+  ENodeType  type;
+  int32_t    vgId;
+  int32_t    dnodeId1;
+  int32_t    dnodeId2;
+  int32_t    dnodeId3;
+  SNodeList* pDnodes;
+} SRedistributeVgroupStmt;
+
+typedef struct SSplitVgroupStmt {
+  ENodeType type;
+  int32_t   vgId;
+} SSplitVgroupStmt;
+
 #ifdef __cplusplus
 }
 #endif

include/libs/nodes/nodes.h

@@ -140,6 +140,7 @@ typedef enum ENodeType {
   QUERY_NODE_DROP_FUNCTION_STMT,
   QUERY_NODE_CREATE_STREAM_STMT,
   QUERY_NODE_DROP_STREAM_STMT,
+  QUERY_NODE_BALANCE_VGROUP_STMT,
   QUERY_NODE_MERGE_VGROUP_STMT,
   QUERY_NODE_REDISTRIBUTE_VGROUP_STMT,
   QUERY_NODE_SPLIT_VGROUP_STMT,

include/util/taoserror.h

@@ -653,6 +653,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_VALUE_TOO_LONG            TAOS_DEF_ERROR_CODE(0, 0x2653)
 #define TSDB_CODE_PAR_INTERNAL_ERROR            TAOS_DEF_ERROR_CODE(0, 0x2654)
 #define TSDB_CODE_PAR_INVALID_DELETE_WHERE      TAOS_DEF_ERROR_CODE(0, 0x2655)
+#define TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG   TAOS_DEF_ERROR_CODE(0, 0x2656)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

source/common/src/tmsg.c

@@ -3258,6 +3258,89 @@ int32_t tDeserializeSKillTransReq(void *buf, int32_t bufLen, SKillTransReq *pReq
   return 0;
 }
 
+int32_t tSerializeSBalanceVgroupReq(void *buf, int32_t bufLen, SBalanceVgroupReq *pReq) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->useless) < 0) return -1;
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSBalanceVgroupReq(void *buf, int32_t bufLen, SBalanceVgroupReq *pReq) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->useless) < 0) return -1;
+  tEndDecode(&decoder);
+
+  tDecoderClear(&decoder);
+  return 0;
+}
+
+int32_t tSerializeSMergeVgroupReq(void *buf, int32_t bufLen, SMergeVgroupReq *pReq) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->vgId1) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->vgId2) < 0) return -1;
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSMergeVgroupReq(void *buf, int32_t bufLen, SMergeVgroupReq *pReq) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->vgId1) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->vgId2) < 0) return -1;
+  tEndDecode(&decoder);
+
+  tDecoderClear(&decoder);
+  return 0;
+}
+
+int32_t tSerializeSRedistributeVgroupReq(void *buf, int32_t bufLen, SRedistributeVgroupReq *pReq) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->vgId) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->dnodeId1) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->dnodeId2) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->dnodeId3) < 0) return -1;
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSRedistributeVgroupReq(void *buf, int32_t bufLen, SRedistributeVgroupReq *pReq) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->vgId) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->dnodeId1) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->dnodeId2) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->dnodeId3) < 0) return -1;
+  tEndDecode(&decoder);
+
+  tDecoderClear(&decoder);
+  return 0;
+}
+
 int32_t tSerializeSDCreateMnodeReq(void *buf, int32_t bufLen, SDCreateMnodeReq *pReq) {
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, bufLen);

source/libs/nodes/src/nodesCloneFuncs.c

@@ -449,6 +449,34 @@ static SNode* logicSubplanCopy(const SLogicSubplan* pSrc, SLogicSubplan* pDst) {
   return (SNode*)pDst;
 }
 
+static SNode* physiNodeCopy(const SPhysiNode* pSrc, SPhysiNode* pDst) {
+  CLONE_NODE_FIELD(pOutputDataBlockDesc);
+  CLONE_NODE_FIELD(pConditions);
+  CLONE_NODE_LIST_FIELD(pChildren);
+  return (SNode*)pDst;
+}
+
+static SNode* physiWindowCopy(const SWinodwPhysiNode* pSrc, SWinodwPhysiNode* pDst) {
+  COPY_BASE_OBJECT_FIELD(node, physiNodeCopy);
+  CLONE_NODE_LIST_FIELD(pExprs);
+  CLONE_NODE_LIST_FIELD(pFuncs);
+  CLONE_NODE_FIELD(pTspk);
+  COPY_SCALAR_FIELD(triggerType);
+  COPY_SCALAR_FIELD(watermark);
+  COPY_SCALAR_FIELD(filesFactor);
+  return (SNode*)pDst;
+}
+
+static SNode* physiIntervalCopy(const SIntervalPhysiNode* pSrc, SIntervalPhysiNode* pDst) {
+  COPY_BASE_OBJECT_FIELD(window, physiWindowCopy);
+  COPY_SCALAR_FIELD(interval);
+  COPY_SCALAR_FIELD(offset);
+  COPY_SCALAR_FIELD(sliding);
+  COPY_SCALAR_FIELD(intervalUnit);
+  COPY_SCALAR_FIELD(slidingUnit);
+  return (SNode*)pDst;
+}
+
 static SNode* dataBlockDescCopy(const SDataBlockDescNode* pSrc, SDataBlockDescNode* pDst) {
   COPY_SCALAR_FIELD(dataBlockId);
   CLONE_NODE_LIST_FIELD(pSlots);
@@ -575,6 +603,12 @@ SNodeptr nodesCloneNode(const SNodeptr pNode) {
       return logicIndefRowsFuncCopy((const SIndefRowsFuncLogicNode*)pNode, (SIndefRowsFuncLogicNode*)pDst);
     case QUERY_NODE_LOGIC_SUBPLAN:
       return logicSubplanCopy((const SLogicSubplan*)pNode, (SLogicSubplan*)pDst);
+    case QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL:
+    case QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL:
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL:
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL:
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:
+      return physiIntervalCopy((const SIntervalPhysiNode*)pNode, (SIntervalPhysiNode*)pDst);
     default:
       break;
   }

source/libs/nodes/src/nodesUtilFuncs.c

@@ -164,9 +164,14 @@ SNodeptr nodesMakeNode(ENodeType type) {
       return makeNode(type, sizeof(SCreateStreamStmt));
     case QUERY_NODE_DROP_STREAM_STMT:
       return makeNode(type, sizeof(SDropStreamStmt));
+    case QUERY_NODE_BALANCE_VGROUP_STMT:
+      return makeNode(type, sizeof(SBalanceVgroupStmt));
     case QUERY_NODE_MERGE_VGROUP_STMT:
+      return makeNode(type, sizeof(SMergeVgroupStmt));
     case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT:
+      return makeNode(type, sizeof(SRedistributeVgroupStmt));
     case QUERY_NODE_SPLIT_VGROUP_STMT:
+      return makeNode(type, sizeof(SSplitVgroupStmt));
     case QUERY_NODE_SYNCDB_STMT:
       break;
     case QUERY_NODE_GRANT_STMT:

source/libs/parser/inc/parAst.h

@@ -187,6 +187,7 @@ SNode* createCreateStreamStmt(SAstCreateContext* pCxt, bool ignoreExists, const
                               SNode* pOptions, SNode* pQuery);
 SNode* createDropStreamStmt(SAstCreateContext* pCxt, bool ignoreNotExists, const SToken* pStreamName);
 SNode* createKillStmt(SAstCreateContext* pCxt, ENodeType type, const SToken* pId);
+SNode* createBalanceVgroupStmt(SAstCreateContext* pCxt);
 SNode* createMergeVgroupStmt(SAstCreateContext* pCxt, const SToken* pVgId1, const SToken* pVgId2);
 SNode* createRedistributeVgroupStmt(SAstCreateContext* pCxt, const SToken* pVgId, SNodeList* pDnodes);
 SNode* createSplitVgroupStmt(SAstCreateContext* pCxt, const SToken* pVgId);

source/libs/parser/inc/sql.y

@@ -469,9 +469,10 @@ cmd ::= KILL QUERY NK_INTEGER(A).
 cmd ::= KILL TRANSACTION NK_INTEGER(A).                                           { pCxt->pRootNode = createKillStmt(pCxt, QUERY_NODE_KILL_TRANSACTION_STMT, &A); }
 
 /************************************************ merge/redistribute/ vgroup ******************************************/
+cmd ::= BALANCE VGROUP.                                                           { pCxt->pRootNode = createBalanceVgroupStmt(pCxt); }
 cmd ::= MERGE VGROUP NK_INTEGER(A) NK_INTEGER(B).                                 { pCxt->pRootNode = createMergeVgroupStmt(pCxt, &A, &B); }
 cmd ::= REDISTRIBUTE VGROUP NK_INTEGER(A) dnode_list(B).                          { pCxt->pRootNode = createRedistributeVgroupStmt(pCxt, &A, B); }
-cmd ::= SPLIT VGROUP NK_INTEGER(A).                                               { pCxt->pRootNode = createSplitVgroupStmt(pCxt, &A); }
+//cmd ::= SPLIT VGROUP NK_INTEGER(A).                                               { pCxt->pRootNode = createSplitVgroupStmt(pCxt, &A); }
 
 %type dnode_list                                                                  { SNodeList* }
 %destructor dnode_list                                                            { nodesDestroyList($$); }

source/libs/parser/src/parAstCreater.c

@@ -1435,25 +1435,37 @@ SNode* createKillStmt(SAstCreateContext* pCxt, ENodeType type, const SToken* pId
   return (SNode*)pStmt;
 }
 
+SNode* createBalanceVgroupStmt(SAstCreateContext* pCxt) {
+  CHECK_PARSER_STATUS(pCxt);
+  SBalanceVgroupStmt* pStmt = nodesMakeNode(QUERY_NODE_BALANCE_VGROUP_STMT);
+  CHECK_OUT_OF_MEM(pStmt);
+  return (SNode*)pStmt;
+}
+
 SNode* createMergeVgroupStmt(SAstCreateContext* pCxt, const SToken* pVgId1, const SToken* pVgId2) {
   CHECK_PARSER_STATUS(pCxt);
-  SNode* pStmt = nodesMakeNode(QUERY_NODE_MERGE_VGROUP_STMT);
+  SMergeVgroupStmt* pStmt = nodesMakeNode(QUERY_NODE_MERGE_VGROUP_STMT);
   CHECK_OUT_OF_MEM(pStmt);
-  return pStmt;
+  pStmt->vgId1 = taosStr2Int32(pVgId1->z, NULL, 10);
+  pStmt->vgId2 = taosStr2Int32(pVgId2->z, NULL, 10);
+  return (SNode*)pStmt;
 }
 
 SNode* createRedistributeVgroupStmt(SAstCreateContext* pCxt, const SToken* pVgId, SNodeList* pDnodes) {
   CHECK_PARSER_STATUS(pCxt);
-  SNode* pStmt = nodesMakeNode(QUERY_NODE_REDISTRIBUTE_VGROUP_STMT);
+  SRedistributeVgroupStmt* pStmt = nodesMakeNode(QUERY_NODE_REDISTRIBUTE_VGROUP_STMT);
   CHECK_OUT_OF_MEM(pStmt);
-  return pStmt;
+  pStmt->vgId = taosStr2Int32(pVgId->z, NULL, 10);
+  pStmt->pDnodes = pDnodes;
+  return (SNode*)pStmt;
 }
 
 SNode* createSplitVgroupStmt(SAstCreateContext* pCxt, const SToken* pVgId) {
   CHECK_PARSER_STATUS(pCxt);
-  SNode* pStmt = nodesMakeNode(QUERY_NODE_SPLIT_VGROUP_STMT);
+  SSplitVgroupStmt* pStmt = nodesMakeNode(QUERY_NODE_SPLIT_VGROUP_STMT);
   CHECK_OUT_OF_MEM(pStmt);
-  return pStmt;
+  pStmt->vgId = taosStr2Int32(pVgId->z, NULL, 10);
+  return (SNode*)pStmt;
 }
 
 SNode* createSyncdbStmt(SAstCreateContext* pCxt, const SToken* pDbName) {

source/libs/parser/src/parTokenizer.c

@@ -41,6 +41,7 @@ static SKeyword keywordTable[] = {
     {"AS",            TK_AS},
     {"ASC",           TK_ASC},
     {"AT_ONCE",       TK_AT_ONCE},
+    {"BALANCE",       TK_BALANCE},
     {"BETWEEN",       TK_BETWEEN},
     {"BINARY",        TK_BINARY},
     {"BIGINT",        TK_BIGINT},
@@ -116,6 +117,7 @@ static SKeyword keywordTable[] = {
     {"LOCAL",         TK_LOCAL},
     {"MATCH",         TK_MATCH},
     {"MAXROWS",       TK_MAXROWS},
+    {"MERGE",         TK_MERGE},
     {"MINROWS",       TK_MINROWS},
     {"MINUS",         TK_MINUS},
     {"MNODE",         TK_MNODE},
@@ -151,6 +153,7 @@ static SKeyword keywordTable[] = {
     {"QUERY",         TK_QUERY},
     {"RATIO",         TK_RATIO},
     {"READ",          TK_READ},
+    {"REDISTRIBUTE",  TK_REDISTRIBUTE},
     {"RENAME",        TK_RENAME},
     {"REPLICA",       TK_REPLICA},
     {"RESET",         TK_RESET},
@@ -171,6 +174,7 @@ static SKeyword keywordTable[] = {
     {"SNODE",         TK_SNODE},
     {"SNODES",        TK_SNODES},
     {"SOFFSET",       TK_SOFFSET},
+    // {"SPLIT",         TK_SPLIT},
     {"STABLE",        TK_STABLE},
     {"STABLES",       TK_STABLES},
     {"STATE",         TK_STATE},
@@ -208,6 +212,7 @@ static SKeyword keywordTable[] = {
     {"VARCHAR",       TK_VARCHAR},
     {"VARIABLES",     TK_VARIABLES},
     {"VERBOSE",       TK_VERBOSE},
+    {"VGROUP",        TK_VGROUP},
     {"VGROUPS",       TK_VGROUPS},
     {"VNODES",        TK_VNODES},
     {"WAL",           TK_WAL},

source/libs/parser/src/parTranslater.c

@@ -847,9 +847,9 @@ static EDealRes translateJsonOperator(STranslateContext* pCxt, SOperatorNode* pO
   if (TSDB_DATA_TYPE_JSON != ldt.type || TSDB_DATA_TYPE_BINARY != rdt.type) {
     return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pRight))->aliasName);
   }
-  if(pOp->opType == OP_TYPE_JSON_GET_VALUE){
+  if (pOp->opType == OP_TYPE_JSON_GET_VALUE) {
     pOp->node.resType.type = TSDB_DATA_TYPE_JSON;
-  }else if(pOp->opType == OP_TYPE_JSON_CONTAINS){
+  } else if (pOp->opType == OP_TYPE_JSON_CONTAINS) {
     pOp->node.resType.type = TSDB_DATA_TYPE_BOOL;
   }
   pOp->node.resType.bytes = tDataTypes[pOp->node.resType.type].bytes;
@@ -3611,6 +3611,55 @@ static int32_t translateRevoke(STranslateContext* pCxt, SRevokeStmt* pStmt) {
   return buildCmdMsg(pCxt, TDMT_MND_ALTER_USER, (FSerializeFunc)tSerializeSAlterUserReq, &req);
 }
 
+static int32_t translateBalanceVgroup(STranslateContext* pCxt, SBalanceVgroupStmt* pStmt) {
+  SBalanceVgroupReq req = {0};
+  return buildCmdMsg(pCxt, TDMT_MND_BALANCE_VGROUP, (FSerializeFunc)tSerializeSBalanceVgroupReq, &req);
+}
+
+static int32_t translateMergeVgroup(STranslateContext* pCxt, SMergeVgroupStmt* pStmt) {
+  SMergeVgroupReq req = {.vgId1 = pStmt->vgId1, .vgId2 = pStmt->vgId2};
+  return buildCmdMsg(pCxt, TDMT_MND_MERGE_VGROUP, (FSerializeFunc)tSerializeSMergeVgroupReq, &req);
+}
+
+static int32_t checkDnodeIds(STranslateContext* pCxt, SRedistributeVgroupStmt* pStmt) {
+  int32_t numOfDnodes = LIST_LENGTH(pStmt->pDnodes);
+  if (numOfDnodes > 3 || numOfDnodes < 1) {
+    return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG);
+  }
+
+  SNode* pNode = NULL;
+  FOREACH(pNode, pStmt->pDnodes) {
+    SValueNode* pVal = (SValueNode*)pNode;
+    if (DEAL_RES_ERROR == translateValue(pCxt, pVal)) {
+      return pCxt->errCode;
+    }
+  }
+
+  pStmt->dnodeId1 = getBigintFromValueNode((SValueNode*)nodesListGetNode(pStmt->pDnodes, 0));
+  pStmt->dnodeId2 = -1;
+  pStmt->dnodeId3 = -1;
+  if (numOfDnodes > 1) {
+    pStmt->dnodeId2 = getBigintFromValueNode((SValueNode*)nodesListGetNode(pStmt->pDnodes, 1));
+  }
+  if (numOfDnodes > 2) {
+    pStmt->dnodeId3 = getBigintFromValueNode((SValueNode*)nodesListGetNode(pStmt->pDnodes, 2));
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateRedistributeVgroup(STranslateContext* pCxt, SRedistributeVgroupStmt* pStmt) {
+  SRedistributeVgroupReq req = {.vgId = pStmt->vgId};
+  int32_t                code = checkDnodeIds(pCxt, pStmt);
+  if (TSDB_CODE_SUCCESS == code) {
+    req.dnodeId1 = pStmt->dnodeId1;
+    req.dnodeId2 = pStmt->dnodeId2;
+    req.dnodeId3 = pStmt->dnodeId3;
+    code = buildCmdMsg(pCxt, TDMT_MND_REDISTRIBUTE_VGROUP, (FSerializeFunc)tSerializeSRedistributeVgroupReq, &req);
+  }
+  return code;
+}
+
 static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode) {
   int32_t code = TSDB_CODE_SUCCESS;
   switch (nodeType(pNode)) {
@@ -3733,6 +3782,15 @@ static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode) {
     case QUERY_NODE_REVOKE_STMT:
       code = translateRevoke(pCxt, (SRevokeStmt*)pNode);
       break;
+    case QUERY_NODE_BALANCE_VGROUP_STMT:
+      code = translateBalanceVgroup(pCxt, (SBalanceVgroupStmt*)pNode);
+      break;
+    case QUERY_NODE_MERGE_VGROUP_STMT:
+      code = translateMergeVgroup(pCxt, (SMergeVgroupStmt*)pNode);
+      break;
+    case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT:
+      code = translateRedistributeVgroup(pCxt, (SRedistributeVgroupStmt*)pNode);
+      break;
     default:
       break;
   }

source/libs/parser/src/parUtil.c

@@ -180,6 +180,8 @@ static char* getSyntaxErrFormat(int32_t errCode) {
       return "Value too long for column/tag: %s";
     case TSDB_CODE_PAR_INVALID_DELETE_WHERE:
       return "The DELETE statement must have a definite time window range";
+    case TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG:
+      return "The REDISTRIBUTE VGROUP statement only support 1 to 3 dnodes";
     case TSDB_CODE_OUT_OF_MEMORY:
       return "Out of memory";
     default:

source/libs/parser/test/parExplainToSyncdbTest.cpp

@@ -19,7 +19,7 @@ using namespace std;
 
 namespace ParserTest {
 
-class ParserExplainToSyncdbTest : public ParserTestBase {};
+class ParserExplainToSyncdbTest : public ParserDdlTest {};
 
 TEST_F(ParserExplainToSyncdbTest, explain) {
   useDb("root", "test");
@@ -43,8 +43,62 @@ TEST_F(ParserExplainToSyncdbTest, grant) {
 // todo kill connection
 // todo kill query
 // todo kill stream
-// todo merge vgroup
-// todo redistribute vgroup
+
+TEST_F(ParserExplainToSyncdbTest, mergeVgroup) {
+  useDb("root", "test");
+
+  SMergeVgroupReq expect = {0};
+
+  auto setMergeVgroupReqFunc = [&](int32_t vgId1, int32_t vgId2) {
+    expect.vgId1 = vgId1;
+    expect.vgId2 = vgId2;
+  };
+
+  setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
+    ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_MERGE_VGROUP_STMT);
+    ASSERT_EQ(pQuery->pCmdMsg->msgType, TDMT_MND_MERGE_VGROUP);
+    SMergeVgroupReq req = {0};
+    ASSERT_EQ(tDeserializeSMergeVgroupReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req), TSDB_CODE_SUCCESS);
+    ASSERT_EQ(req.vgId1, expect.vgId1);
+    ASSERT_EQ(req.vgId2, expect.vgId2);
+  });
+
+  setMergeVgroupReqFunc(1, 2);
+  run("MERGE VGROUP 1 2");
+}
+
+TEST_F(ParserExplainToSyncdbTest, redistributeVgroup) {
+  useDb("root", "test");
+
+  SRedistributeVgroupReq expect = {0};
+
+  auto setRedistributeVgroupReqFunc = [&](int32_t vgId, int32_t dnodeId1, int32_t dnodeId2 = -1,
+                                          int32_t dnodeId3 = -1) {
+    expect.vgId = vgId;
+    expect.dnodeId1 = dnodeId1;
+    expect.dnodeId2 = dnodeId2;
+    expect.dnodeId3 = dnodeId3;
+  };
+
+  setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
+    ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_REDISTRIBUTE_VGROUP_STMT);
+    ASSERT_EQ(pQuery->pCmdMsg->msgType, TDMT_MND_REDISTRIBUTE_VGROUP);
+    SRedistributeVgroupReq req = {0};
+    ASSERT_EQ(tDeserializeSRedistributeVgroupReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req),
+              TSDB_CODE_SUCCESS);
+    ASSERT_EQ(req.vgId, expect.vgId);
+    ASSERT_EQ(req.dnodeId1, expect.dnodeId1);
+    ASSERT_EQ(req.dnodeId2, expect.dnodeId2);
+    ASSERT_EQ(req.dnodeId3, expect.dnodeId3);
+  });
+
+  setRedistributeVgroupReqFunc(3, 1);
+  run("REDISTRIBUTE VGROUP 3 DNODE 1");
+
+  setRedistributeVgroupReqFunc(5, 10, 20, 30);
+  run("REDISTRIBUTE VGROUP 5 DNODE 10 DNODE 20 DNODE 30");
+}
+
 // todo reset query cache
 
 TEST_F(ParserExplainToSyncdbTest, revoke) {

source/libs/parser/test/parInitialATest.cpp

@@ -305,6 +305,19 @@ TEST_F(ParserInitialATest, alterUser) {
   run("ALTER user wxy privilege 'write'");
 }
 
+TEST_F(ParserInitialATest, balanceVgroup) {
+  useDb("root", "test");
+
+  setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
+    ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_BALANCE_VGROUP_STMT);
+    ASSERT_EQ(pQuery->pCmdMsg->msgType, TDMT_MND_BALANCE_VGROUP);
+    SBalanceVgroupReq req = {0};
+    ASSERT_EQ(tDeserializeSBalanceVgroupReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req), TSDB_CODE_SUCCESS);
+  });
+
+  run("BALANCE VGROUP");
+}
+
 TEST_F(ParserInitialATest, bug001) {
   useDb("root", "test");
 

source/util/src/terror.c

@@ -451,9 +451,89 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SCH_TIMEOUT_ERROR,            "Task timeout")
 TAOS_DEFINE_ERROR(TSDB_CODE_QW_MSG_ERROR,                 "Invalid msg order")
 
 // parser
-TAOS_DEFINE_ERROR(TSDB_CODE_PAR_TABLE_NOT_EXIST,          "Table does not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_PAR_PERMISSION_DENIED,         "Permission denied")
 TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INTERNAL_ERROR,            "Parser internal error")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_SYNTAX_ERROR,              "syntax error near")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INCOMPLETE_SQL,            "Incomplete SQL statement")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_COLUMN,            "Invalid column name")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_TABLE_NOT_EXIST,           "Table does not exist")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_AMBIGUOUS_COLUMN,          "Column ambiguously defined")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_WRONG_VALUE_TYPE,          "Invalid value type")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION,  "There mustn't be aggregation")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_WRONG_NUMBER_OF_SELECT,    "ORDER BY item must be the number of a SELECT-list expression")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION,   "Not a GROUP BY expression")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_NOT_SELECTED_EXPRESSION,   "Not SELECTed expression")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_NOT_SINGLE_GROUP,          "Not a single-group group function")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_TAGS_NOT_MATCHED,          "Tags number not matched")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_TAG_NAME,          "Invalid tag name")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_NAME_OR_PASSWD_TOO_LONG,   "Name or password too long")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_PASSWD_EMPTY,              "Password can not be empty")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_PORT,              "Port should be an integer that is less than 65535 and greater than 0")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_ENDPOINT,          "Endpoint should be in the format of 'fqdn:port'")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_EXPRIE_STATEMENT,          "This statement is no longer supported")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INTER_VALUE_TOO_SMALL,     "Interval too small")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_DB_NOT_SPECIFIED,          "Database not specified")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_IDENTIFIER_NAME,   "Invalid identifier name")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_CORRESPONDING_STABLE_ERR,  "Corresponding super table not in this db")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_RANGE_OPTION,      "Invalid option")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_STR_OPTION,        "Invalid option")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_ENUM_OPTION,       "Invalid option")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_KEEP_NUM,          "Invalid number of keep options")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_KEEP_ORDER,        "Invalid keep value, should be keep0 <= keep1 <= keep2")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_KEEP_VALUE,        "Invalid option keep")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_COMMENT_OPTION,    "Invalid option comment")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_F_RANGE_OPTION,    "Invalid option")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_ROLLUP_OPTION,     "Invalid option rollup: only one function is allowed")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_RETENTIONS_OPTION, "Invalid option retentions")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_GROUPBY_WINDOW_COEXIST,    "GROUP BY and WINDOW-clause can't be used together")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_OPTION_UNIT,       "Invalid option unit: only m, h, d allowed")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_KEEP_UNIT,         "Invalid option keep unit: only m, h, d allowed")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_AGG_FUNC_NESTING,          "Aggregate functions do not support nesting")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_STATE_WIN_TYPE,    "Only support STATE_WINDOW on integer/bool/varchar column")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_STATE_WIN_COL,     "Not support STATE_WINDOW on tag column")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_STATE_WIN_TABLE,   "STATE_WINDOW not support for super table query")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INTER_SESSION_GAP,         "SESSION gap should be fixed time window, and greater than 0")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INTER_SESSION_COL,         "Only support SESSION on primary timestamp column")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INTER_OFFSET_NEGATIVE,     "Interval offset cannot be negative")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INTER_OFFSET_UNIT,         "Cannot use 'year' as offset when interval is 'month'")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INTER_OFFSET_TOO_BIG,      "Interval offset should be shorter than interval")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INTER_SLIDING_UNIT,        "Does not support sliding when interval is natural month/year")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INTER_SLIDING_TOO_BIG,     "sliding value no larger than the interval value")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INTER_SLIDING_TOO_SMALL,   "sliding value can not less than 1% of interval value")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_ONLY_ONE_JSON_TAG,         "Only one tag if there is a json tag")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INCORRECT_NUM_OF_COL,      "Query block has incorrect number of result columns")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INCORRECT_TIMESTAMP_VAL,   "Incorrect TIMESTAMP value")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_DAYS_VALUE,        "Invalid days value, should be keep2 >= keep1 >= keep0 >= days")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_OFFSET_LESS_ZERO,          "soffset/offset can not be less than 0")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_SLIMIT_LEAK_PARTITION_BY,  "slimit/soffset only available for PARTITION BY query")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_TOPIC_QUERY,        "Invalid topic query")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_DROP_STABLE,        "Cannot drop super table in batch")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_FILL_TIME_RANGE,    "Start(end) time of query range required or time range too large")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_DUPLICATED_COLUMN,          "Duplicated column names")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_TAGS_LENGTH,        "Tags length exceeds max length")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_ROW_LENGTH,         "Row length exceeds max length")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_COLUMNS_NUM,        "Illegal number of columns")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_TOO_MANY_COLUMNS,           "Too many columns")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_FIRST_COLUMN,       "First column must be timestamp")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN,     "Invalid binary/nchar column length")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_TAGS_NUM,           "Invalid number of tag columns")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_INTERNAL_PK,        "Invalid _c0 or _rowts expression")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_TIMELINE_FUNC,      "Invalid timeline function")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_PASSWD,             "Invalid password")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_ALTER_TABLE,        "Invalid alter table statement")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_CANNOT_DROP_PRIMARY_KEY,    "Primary timestamp column cannot be dropped")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_MODIFY_COL,         "Only binary/nchar column length could be modified")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_TBNAME,             "Invalid tbname pseudo column")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_FUNCTION_NAME,      "Invalid function name")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_COMMENT_TOO_LONG,           "Comment too long")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_NOT_ALLOWED_FUNC,           "Some functions are allowed only in the SELECT list of a query. "
+                                                            "And, cannot be mixed with other non scalar functions or columns.")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_NOT_ALLOWED_WIN_QUERY,      "Window query not supported, since the result of subquery not include valid timestamp column")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_DROP_COL,           "No columns can be dropped")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_COL_JSON,           "Only tag can be json type")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_VALUE_TOO_LONG,             "Value too long for column/tag")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_DELETE_WHERE,       "The DELETE statement must have a definite time window range")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG,    "The REDISTRIBUTE VGROUP statement only support 1 to 3 dnodes")
 
 //planner
 TAOS_DEFINE_ERROR(TSDB_CODE_PLAN_INTERNAL_ERROR,          "Planner internal error")