Merge pull request #9194 from taosdata/feature/dnode3

Feature/dnode3

include/common/taosmsg.h

@@ -750,31 +750,36 @@ typedef struct {
 } SReplica;
 
 typedef struct {
-  char     db[TSDB_FULL_DB_NAME_LEN];
   int32_t  vgId;
+  int32_t  dnodeId;
+  char     db[TSDB_FULL_DB_NAME_LEN];
+  uint64_t dbUid;
   int32_t  cacheBlockSize;
   int32_t  totalBlocks;
   int32_t  daysPerFile;
   int32_t  daysToKeep0;
   int32_t  daysToKeep1;
   int32_t  daysToKeep2;
-  int32_t  minRowsPerFileBlock;
-  int32_t  maxRowsPerFileBlock;
+  int32_t  minRows;
+  int32_t  maxRows;
+  int32_t  commitTime;
   int32_t  fsyncPeriod;
-  int8_t   reserved[16];
+  int8_t   walLevel;
   int8_t   precision;
   int8_t   compression;
-  int8_t   cacheLastRow;
-  int8_t   update;
-  int8_t   walLevel;
   int8_t   quorum;
+  int8_t   update;
+  int8_t   cacheLastRow;
   int8_t   replica;
   int8_t   selfIndex;
   SReplica replicas[TSDB_MAX_REPLICA];
 } SCreateVnodeMsg, SAlterVnodeMsg;
 
 typedef struct {
-  int32_t vgId;
+  int32_t  vgId;
+  int32_t  dnodeId;
+  char     db[TSDB_FULL_DB_NAME_LEN];
+  uint64_t dbUid;
 } SDropVnodeMsg, SSyncVnodeMsg, SCompactVnodeMsg;
 
 typedef struct {

include/util/taoserror.h

@@ -120,17 +120,18 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_TSC_INVALID_INPUT             TAOS_DEF_ERROR_CODE(0, 0X0224)  //"Invalid tsc input")
 
 // mnode-common
-#define TSDB_CODE_MND_NOT_READY                 TAOS_DEF_ERROR_CODE(0, 0x0300)
-#define TSDB_CODE_MND_MSG_NOT_PROCESSED         TAOS_DEF_ERROR_CODE(0, 0x0301)
-#define TSDB_CODE_MND_ACTION_IN_PROGRESS        TAOS_DEF_ERROR_CODE(0, 0x0302)
-#define TSDB_CODE_MND_ACTION_NEED_REPROCESSED   TAOS_DEF_ERROR_CODE(0, 0x0303)
-#define TSDB_CODE_MND_NO_RIGHTS                 TAOS_DEF_ERROR_CODE(0, 0x0304)
-#define TSDB_CODE_MND_INVALID_OPTIONS           TAOS_DEF_ERROR_CODE(0, 0x0305)
-#define TSDB_CODE_MND_INVALID_CONNECTION        TAOS_DEF_ERROR_CODE(0, 0x0306)
-#define TSDB_CODE_MND_INVALID_MSG_VERSION       TAOS_DEF_ERROR_CODE(0, 0x0307)
-#define TSDB_CODE_MND_INVALID_MSG_LEN           TAOS_DEF_ERROR_CODE(0, 0x0308)
-#define TSDB_CODE_MND_INVALID_MSG_TYPE          TAOS_DEF_ERROR_CODE(0, 0x0309)
-#define TSDB_CODE_MND_TOO_MANY_SHELL_CONNS      TAOS_DEF_ERROR_CODE(0, 0x030A)
+#define TSDB_CODE_MND_APP_ERROR                 TAOS_DEF_ERROR_CODE(0, 0x0300)
+#define TSDB_CODE_MND_NOT_READY                 TAOS_DEF_ERROR_CODE(0, 0x0301)
+#define TSDB_CODE_MND_MSG_NOT_PROCESSED         TAOS_DEF_ERROR_CODE(0, 0x0302)
+#define TSDB_CODE_MND_ACTION_IN_PROGRESS        TAOS_DEF_ERROR_CODE(0, 0x0303)
+#define TSDB_CODE_MND_ACTION_NEED_REPROCESSED   TAOS_DEF_ERROR_CODE(0, 0x0304)
+#define TSDB_CODE_MND_NO_RIGHTS                 TAOS_DEF_ERROR_CODE(0, 0x0305)
+#define TSDB_CODE_MND_INVALID_OPTIONS           TAOS_DEF_ERROR_CODE(0, 0x0306)
+#define TSDB_CODE_MND_INVALID_CONNECTION        TAOS_DEF_ERROR_CODE(0, 0x0307)
+#define TSDB_CODE_MND_INVALID_MSG_VERSION       TAOS_DEF_ERROR_CODE(0, 0x0308)
+#define TSDB_CODE_MND_INVALID_MSG_LEN           TAOS_DEF_ERROR_CODE(0, 0x0309)
+#define TSDB_CODE_MND_INVALID_MSG_TYPE          TAOS_DEF_ERROR_CODE(0, 0x030A)
+#define TSDB_CODE_MND_TOO_MANY_SHELL_CONNS      TAOS_DEF_ERROR_CODE(0, 0x030B)
 
 // mnode-show
 #define TSDB_CODE_MND_INVALID_SHOWOBJ           TAOS_DEF_ERROR_CODE(0, 0x0310)

include/util/tdef.h

@@ -231,7 +231,7 @@ do { \
 #define TSDB_DEFAULT_PAYLOAD_SIZE 5120   // default payload size, greater than PATH_MAX value
 #define TSDB_EXTRA_PAYLOAD_SIZE   128    // extra bytes for auth
 #define TSDB_CQ_SQL_SIZE          1024
-#define TSDB_MIN_VNODES           64
+#define TSDB_MIN_VNODES           16
 #define TSDB_MAX_VNODES           512
 #define TSDB_MIN_VNODES_PER_DB    1
 #define TSDB_MAX_VNODES_PER_DB    4096

source/dnode/mgmt/impl/src/dndVnodes.c

@@ -95,7 +95,7 @@ static int32_t dndProcessCompactVnodeReq(SDnode *pDnode, SRpcMsg *rpcMsg);
 
 static SVnodeObj *dndAcquireVnode(SDnode *pDnode, int32_t vgId) {
   SVnodesMgmt *pMgmt = &pDnode->vmgmt;
-  SVnodeObj *  pVnode = NULL;
+  SVnodeObj   *pVnode = NULL;
   int32_t      refCount = 0;
 
   taosRLockLatch(&pMgmt->latch);
@@ -107,23 +107,23 @@ static SVnodeObj *dndAcquireVnode(SDnode *pDnode, int32_t vgId) {
   }
   taosRUnLockLatch(&pMgmt->latch);
 
-  dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount);
+  if (pVnode != NULL) {
+    dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount);
+  }
+
   return pVnode;
 }
 
 static void dndReleaseVnode(SDnode *pDnode, SVnodeObj *pVnode) {
+  if (pVnode == NULL) return;
+
   SVnodesMgmt *pMgmt = &pDnode->vmgmt;
-  int32_t      refCount = 0;
 
   taosRLockLatch(&pMgmt->latch);
-  if (pVnode != NULL) {
-    refCount = atomic_sub_fetch_32(&pVnode->refCount, 1);
-  }
+  int32_t refCount = atomic_sub_fetch_32(&pVnode->refCount, 1);
   taosRUnLockLatch(&pMgmt->latch);
 
-  if (pVnode != NULL) {
-    dTrace("vgId:%d, release vnode, refCount:%d", pVnode->vgId, refCount);
-  }
+  dTrace("vgId:%d, release vnode, refCount:%d", pVnode->vgId, refCount);
 }
 
 static int32_t dndCreateVnodeWrapper(SDnode *pDnode, int32_t vgId, char *path, SVnode *pImpl) {
@@ -457,7 +457,7 @@ static int32_t dndOpenVnodes(SDnode *pDnode) {
 
   pMgmt->totalVnodes = numOfVnodes;
 
-  int32_t threadNum = tsNumOfCores;
+  int32_t threadNum = pDnode->opt.numOfCores;
   int32_t vnodesPerThread = numOfVnodes / threadNum + 1;
 
   SVnodeThread *threads = calloc(threadNum, sizeof(SVnodeThread));
@@ -525,33 +525,49 @@ static void dndCloseVnodes(SDnode *pDnode) {
 
 static int32_t dndParseCreateVnodeReq(SRpcMsg *rpcMsg, int32_t *vgId, SVnodeCfg *pCfg) {
   SCreateVnodeMsg *pCreate = rpcMsg->pCont;
-  *vgId = htonl(pCreate->vgId);
+  pCreate->vgId = htonl(pCreate->vgId);
+  pCreate->dnodeId = htonl(pCreate->dnodeId);
+  pCreate->dbUid = htobe64(pCreate->dbUid);
+  pCreate->cacheBlockSize = htonl(pCreate->cacheBlockSize);
+  pCreate->totalBlocks = htonl(pCreate->totalBlocks);
+  pCreate->daysPerFile = htonl(pCreate->daysPerFile);
+  pCreate->daysToKeep0 = htonl(pCreate->daysToKeep0);
+  pCreate->daysToKeep1 = htonl(pCreate->daysToKeep1);
+  pCreate->daysToKeep2 = htonl(pCreate->daysToKeep2);
+  pCreate->minRows = htonl(pCreate->minRows);
+  pCreate->maxRows = htonl(pCreate->maxRows);
+  pCreate->commitTime = htonl(pCreate->commitTime);
+  pCreate->fsyncPeriod = htonl(pCreate->fsyncPeriod);
+  for (int r = 0; r < pCreate->replica; ++r) {
+    SReplica *pReplica = &pCreate->replicas[r];
+    pReplica->id = htonl(pReplica->id);
+    pReplica->port = htons(pReplica->port);
+  }
+
+  *vgId = pCreate->vgId;
 
 #if 0
-  tstrncpy(pCfg->db, pCreate->db, TSDB_FULL_DB_NAME_LEN);
-  pCfg->cacheBlockSize = htonl(pCreate->cacheBlockSize);
-  pCfg->totalBlocks = htonl(pCreate->totalBlocks);
-  pCfg->daysPerFile = htonl(pCreate->daysPerFile);
-  pCfg->daysToKeep0 = htonl(pCreate->daysToKeep0);
-  pCfg->daysToKeep1 = htonl(pCreate->daysToKeep1);
-  pCfg->daysToKeep2 = htonl(pCreate->daysToKeep2);
-  pCfg->minRowsPerFileBlock = htonl(pCreate->minRowsPerFileBlock);
-  pCfg->maxRowsPerFileBlock = htonl(pCreate->maxRowsPerFileBlock);
-  pCfg->precision = pCreate->precision;
-  pCfg->compression = pCreate->compression;
-  pCfg->cacheLastRow = pCreate->cacheLastRow;
-  pCfg->update = pCreate->update;
-  pCfg->quorum = pCreate->quorum;
-  pCfg->replica = pCreate->replica;
-  pCfg->walLevel = pCreate->walLevel;
-  pCfg->fsyncPeriod = htonl(pCreate->fsyncPeriod);
-
-  for (int32_t i = 0; i < pCfg->replica; ++i) {
-    pCfg->replicas[i].port = htons(pCreate->replicas[i].port);
-    tstrncpy(pCfg->replicas[i].fqdn, pCreate->replicas[i].fqdn, TSDB_FQDN_LEN);
-  }
+  pCfg->wsize = pCreate->cacheBlockSize;
+  pCfg->ssize = pCreate->cacheBlockSize;
+  pCfg->wsize = pCreate->cacheBlockSize;
+  pCfg->lsize = pCreate->cacheBlockSize;
+  pCfg->isHeapAllocator = true;
+  pCfg->ttl = 4;
+  pCfg->keep = pCreate->daysToKeep0;
+  pCfg->isWeak = true;
+  pCfg->tsdbCfg.keep0 = pCreate->daysToKeep0;
+  pCfg->tsdbCfg.keep1 = pCreate->daysToKeep2;
+  pCfg->tsdbCfg.keep2 = pCreate->daysToKeep0;
+  pCfg->tsdbCfg.lruCacheSize = pCreate->cacheBlockSize;
+  pCfg->metaCfg.lruSize = pCreate->cacheBlockSize;
+  pCfg->walCfg.fsyncPeriod = pCreate->fsyncPeriod;
+  pCfg->walCfg.level = pCreate->walLevel;
+  pCfg->walCfg.retentionPeriod = 10;
+  pCfg->walCfg.retentionSize = 128;
+  pCfg->walCfg.rollPeriod = 128;
+  pCfg->walCfg.segSize = 128;
+  pCfg->walCfg.vgId = pCreate->vgId;
 #endif
-
   return 0;
 }
 
@@ -1016,7 +1032,7 @@ static int32_t dndInitVnodeWriteWorker(SDnode *pDnode) {
   SVnodesMgmt * pMgmt = &pDnode->vmgmt;
   SMWorkerPool *pPool = &pMgmt->writePool;
   pPool->name = "vnode-write";
-  pPool->max = tsNumOfCores;
+  pPool->max = pDnode->opt.numOfCores;
   if (tMWorkerInit(pPool) != 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
@@ -1050,7 +1066,7 @@ static void dndFreeVnodeSyncQueue(SDnode *pDnode, SVnodeObj *pVnode) {
 }
 
 static int32_t dndInitVnodeSyncWorker(SDnode *pDnode) {
-  int32_t maxThreads = tsNumOfCores / 2;
+  int32_t maxThreads = pDnode->opt.numOfCores / 2;
   if (maxThreads < 1) maxThreads = 1;
 
   SVnodesMgmt  *pMgmt = &pDnode->vmgmt;

source/dnode/mgmt/impl/test/CMakeLists.txt

@@ -15,6 +15,6 @@ add_subdirectory(stb)
 # add_subdirectory(telem)
 # add_subdirectory(trans)
 add_subdirectory(user)
-# add_subdirectory(vgroup)
+add_subdirectory(vgroup)
 
 # add_subdirectory(common)

source/dnode/mgmt/impl/test/db/db.cpp

@@ -232,6 +232,7 @@ TEST_F(DndTestDb, 02_Create_Alter_Drop_Db) {
     SRpcMsg* pMsg = pClient->pRsp;
     ASSERT_NE(pMsg, nullptr);
     ASSERT_EQ(pMsg->code, 0);
+    // taosMsleep(1000000);
   }
 
   SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DB, "show databases", 17, NULL);

source/dnode/mgmt/impl/test/vgroup/CMakeLists.txt

+add_executable(dnode_test_vgroup "")
+
+target_sources(dnode_test_vgroup
+    PRIVATE
+    "vgroup.cpp"
+    "../sut/deploy.cpp"
+)
+
+target_link_libraries(
+    dnode_test_vgroup
+    PUBLIC dnode
+    PUBLIC util
+    PUBLIC os
+    PUBLIC gtest_main
+)
+
+target_include_directories(dnode_test_vgroup
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/server/dnode/mgmt"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../../inc"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../sut"
+)
+
+add_test(
+    NAME dnode_test_vgroup
+    COMMAND dnode_test_vgroup
+)

source/dnode/mgmt/impl/test/vgroup/vgroup.cpp

+/**
+ * @file db.cpp
+ * @author slguan (slguan@taosdata.com)
+ * @brief DNODE module vgroup-msg tests
+ * @version 0.1
+ * @date 2021-12-20
+ *
+ * @copyright Copyright (c) 2021
+ *
+ */
+
+#include "deploy.h"
+
+class DndTestVgroup : public ::testing::Test {
+ protected:
+  static SServer* CreateServer(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
+    SServer* pServer = createServer(path, fqdn, port, firstEp);
+    ASSERT(pServer);
+    return pServer;
+  }
+
+  static void SetUpTestSuite() {
+    initLog("/tmp/tdlog");
+
+    const char* fqdn = "localhost";
+    const char* firstEp = "localhost:9150";
+    pServer = CreateServer("/tmp/dnode_test_vgroup", fqdn, 9150, firstEp);
+    pClient = createClient("root", "taosdata", fqdn, 9150);
+    taosMsleep(1100);
+  }
+
+  static void TearDownTestSuite() {
+    stopServer(pServer);
+    dropClient(pClient);
+    pServer = NULL;
+    pClient = NULL;
+  }
+
+  static SServer* pServer;
+  static SClient* pClient;
+  static int32_t  connId;
+
+ public:
+  void SetUp() override {}
+  void TearDown() override {}
+
+  void SendTheCheckShowMetaMsg(int8_t showType, const char* showName, int32_t columns, const char* db) {
+    SShowMsg* pShow = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
+    pShow->type = showType;
+    if (db != NULL) {
+      strcpy(pShow->db, db);
+    }
+    SRpcMsg showRpcMsg = {0};
+    showRpcMsg.pCont = pShow;
+    showRpcMsg.contLen = sizeof(SShowMsg);
+    showRpcMsg.msgType = TSDB_MSG_TYPE_SHOW;
+
+    sendMsg(pClient, &showRpcMsg);
+    ASSERT_NE(pClient->pRsp, nullptr);
+    ASSERT_EQ(pClient->pRsp->code, 0);
+    ASSERT_NE(pClient->pRsp->pCont, nullptr);
+
+    SShowRsp* pShowRsp = (SShowRsp*)pClient->pRsp->pCont;
+    ASSERT_NE(pShowRsp, nullptr);
+    pShowRsp->showId = htonl(pShowRsp->showId);
+    pMeta = &pShowRsp->tableMeta;
+    pMeta->numOfTags = htonl(pMeta->numOfTags);
+    pMeta->numOfColumns = htonl(pMeta->numOfColumns);
+    pMeta->sversion = htonl(pMeta->sversion);
+    pMeta->tversion = htonl(pMeta->tversion);
+    pMeta->tuid = htobe64(pMeta->tuid);
+    pMeta->suid = htobe64(pMeta->suid);
+
+    showId = pShowRsp->showId;
+
+    EXPECT_NE(pShowRsp->showId, 0);
+    EXPECT_STREQ(pMeta->tbFname, showName);
+    EXPECT_EQ(pMeta->numOfTags, 0);
+    EXPECT_EQ(pMeta->numOfColumns, columns);
+    EXPECT_EQ(pMeta->precision, 0);
+    EXPECT_EQ(pMeta->tableType, 0);
+    EXPECT_EQ(pMeta->update, 0);
+    EXPECT_EQ(pMeta->sversion, 0);
+    EXPECT_EQ(pMeta->tversion, 0);
+    EXPECT_EQ(pMeta->tuid, 0);
+    EXPECT_EQ(pMeta->suid, 0);
+  }
+
+  void CheckSchema(int32_t index, int8_t type, int32_t bytes, const char* name) {
+    SSchema* pSchema = &pMeta->pSchema[index];
+    pSchema->bytes = htonl(pSchema->bytes);
+    EXPECT_EQ(pSchema->colId, 0);
+    EXPECT_EQ(pSchema->type, type);
+    EXPECT_EQ(pSchema->bytes, bytes);
+    EXPECT_STREQ(pSchema->name, name);
+  }
+
+  void SendThenCheckShowRetrieveMsg(int32_t rows) {
+    SRetrieveTableMsg* pRetrieve = (SRetrieveTableMsg*)rpcMallocCont(sizeof(SRetrieveTableMsg));
+    pRetrieve->showId = htonl(showId);
+    pRetrieve->free = 0;
+
+    SRpcMsg retrieveRpcMsg = {0};
+    retrieveRpcMsg.pCont = pRetrieve;
+    retrieveRpcMsg.contLen = sizeof(SRetrieveTableMsg);
+    retrieveRpcMsg.msgType = TSDB_MSG_TYPE_SHOW_RETRIEVE;
+
+    sendMsg(pClient, &retrieveRpcMsg);
+
+    ASSERT_NE(pClient->pRsp, nullptr);
+    ASSERT_EQ(pClient->pRsp->code, 0);
+    ASSERT_NE(pClient->pRsp->pCont, nullptr);
+
+    pRetrieveRsp = (SRetrieveTableRsp*)pClient->pRsp->pCont;
+    ASSERT_NE(pRetrieveRsp, nullptr);
+    pRetrieveRsp->numOfRows = htonl(pRetrieveRsp->numOfRows);
+    pRetrieveRsp->useconds = htobe64(pRetrieveRsp->useconds);
+    pRetrieveRsp->compLen = htonl(pRetrieveRsp->compLen);
+
+    EXPECT_EQ(pRetrieveRsp->numOfRows, rows);
+    EXPECT_EQ(pRetrieveRsp->useconds, 0);
+    // EXPECT_EQ(pRetrieveRsp->completed, completed);
+    EXPECT_EQ(pRetrieveRsp->precision, TSDB_TIME_PRECISION_MILLI);
+    EXPECT_EQ(pRetrieveRsp->compressed, 0);
+    EXPECT_EQ(pRetrieveRsp->compLen, 0);
+
+    pData = pRetrieveRsp->data;
+    pos = 0;
+  }
+
+  void CheckInt8(int8_t val) {
+    int8_t data = *((int8_t*)(pData + pos));
+    pos += sizeof(int8_t);
+    EXPECT_EQ(data, val);
+  }
+
+  void CheckInt16(int16_t val) {
+    int16_t data = *((int16_t*)(pData + pos));
+    pos += sizeof(int16_t);
+    EXPECT_EQ(data, val);
+  }
+
+  void CheckInt32(int32_t val) {
+    int32_t data = *((int32_t*)(pData + pos));
+    pos += sizeof(int32_t);
+    EXPECT_EQ(data, val);
+  }
+
+  void CheckInt64(int64_t val) {
+    int64_t data = *((int64_t*)(pData + pos));
+    pos += sizeof(int64_t);
+    EXPECT_EQ(data, val);
+  }
+
+  void CheckTimestamp() {
+    int64_t data = *((int64_t*)(pData + pos));
+    pos += sizeof(int64_t);
+    EXPECT_GT(data, 0);
+  }
+
+  void CheckBinary(const char* val, int32_t len) {
+    pos += sizeof(VarDataLenT);
+    char* data = (char*)(pData + pos);
+    pos += len;
+    EXPECT_STREQ(data, val);
+  }
+
+  int32_t            showId;
+  STableMetaMsg*     pMeta;
+  SRetrieveTableRsp* pRetrieveRsp;
+  char*              pData;
+  int32_t            pos;
+};
+
+SServer* DndTestVgroup::pServer;
+SClient* DndTestVgroup::pClient;
+int32_t  DndTestVgroup::connId;
+
+
+TEST_F(DndTestVgroup, 01_Create_Restart_Drop_Vnode) {
+  {
+    SCreateVnodeMsg* pReq = (SCreateVnodeMsg*)rpcMallocCont(sizeof(SCreateVnodeMsg));
+    pReq->vgId = htonl(2);
+    pReq->dnodeId = htonl(1);
+    strcpy(pReq->db, "1.d1");
+    pReq->dbUid = htobe64(9527);
+    pReq->cacheBlockSize = htonl(16);
+    pReq->totalBlocks = htonl(10);
+    pReq->daysPerFile = htonl(10);
+    pReq->daysToKeep0 = htonl(3650);
+    pReq->daysToKeep1 = htonl(3650);
+    pReq->daysToKeep2 = htonl(3650);
+    pReq->minRows = htonl(100);
+    pReq->minRows = htonl(4096);
+    pReq->commitTime = htonl(3600);
+    pReq->fsyncPeriod = htonl(3000);
+    pReq->walLevel = 1;
+    pReq->precision = 0;
+    pReq->compression = 2;
+    pReq->replica = 1;
+    pReq->quorum = 1;
+    pReq->update = 0;
+    pReq->cacheLastRow = 0;
+    pReq->selfIndex = 0;
+    for (int r = 0; r < pReq->replica; ++r) {
+      SReplica* pReplica = &pReq->replicas[r];
+      pReplica->id = htonl(1);
+      pReplica->port = htons(9150);
+    }
+
+    SRpcMsg rpcMsg = {0};
+    rpcMsg.pCont = pReq;
+    rpcMsg.contLen = sizeof(SCreateVnodeMsg);
+    rpcMsg.msgType = TSDB_MSG_TYPE_CREATE_VNODE_IN;
+
+    sendMsg(pClient, &rpcMsg);
+    SRpcMsg* pMsg = pClient->pRsp;
+    ASSERT_NE(pMsg, nullptr);
+    ASSERT_EQ(pMsg->code, 0);
+    taosMsleep(1000000);
+  }
+
+}
+

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

@@ -30,8 +30,8 @@ void    mndTransDrop(STrans *pTrans);
 int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw);
 int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw);
 int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw);
-int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *, void *pMsg);
-int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *, void *pMsg);
+int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont);
+int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont);
 int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);
 void    mndTransApply(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg, int32_t code);
 char   *mndTransStageStr(ETrnStage stage);

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

@@ -28,7 +28,9 @@ SVgObj  *mndAcquireVgroup(SMnode *pMnode, int32_t vgId);
 void     mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup);
 int32_t  mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups);
 SSdbRaw *mndVgroupActionEncode(SVgObj *pVgroup);
-SSdbRow *mndVgroupActionDecode(SSdbRaw *pRaw);
+
+SCreateVnodeMsg *mndBuildCreateVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup);
+SDropVnodeMsg   *mndBuildDropVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup);
 
 #ifdef __cplusplus
 }

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

@@ -285,10 +285,60 @@ static int32_t mndSetCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVg
 }
 
 static int32_t mndSetRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
+  for (int v = 0; v < pDb->cfg.numOfVgroups; ++v) {
+    SVgObj *pVgroup = pVgroups + v;
+
+    for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
+      SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
+      SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
+      if (pDnode == NULL) {
+        return -1;
+      }
+
+      SEpSet epset = mndGetDnodeEpset(pDnode);
+      mndReleaseDnode(pMnode, pDnode);
+
+      SCreateVnodeMsg *pMsg = mndBuildCreateVnodeMsg(pMnode, pDnode, pDb, pVgroup);
+      if (pMsg == NULL) {
+        return -1;
+      }
+
+      if (mndTransAppendRedoAction(pTrans, &epset, TSDB_MSG_TYPE_ALTER_VNODE_IN, sizeof(SCreateVnodeMsg), pMsg) != 0) {
+        free(pMsg);
+        return -1;
+      }
+    }
+  }
+
   return 0;
 }
 
 static int32_t mndSetUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
+  for (int v = 0; v < pDb->cfg.numOfVgroups; ++v) {
+    SVgObj *pVgroup = pVgroups + v;
+
+    for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
+      SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
+      SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
+      if (pDnode == NULL) {
+        return -1;
+      }
+
+      SEpSet epset = mndGetDnodeEpset(pDnode);
+      mndReleaseDnode(pMnode, pDnode);
+
+      SDropVnodeMsg *pMsg = mndBuildDropVnodeMsg(pMnode, pDnode, pDb, pVgroup);
+      if (pMsg == NULL) {
+        return -1;
+      }
+
+      if (mndTransAppendUndoAction(pTrans, &epset, TSDB_MSG_TYPE_DROP_VNODE_IN, sizeof(SDropVnodeMsg), pMsg) != 0) {
+        free(pMsg);
+        return -1;
+      }
+    }
+  }
+
   return 0;
 }
 
@@ -644,7 +694,7 @@ static int32_t mndProcessUseDbMsg(SMnodeMsg *pMsg) {
     return -1;
   }
 
-  int32_t contLen = sizeof(SUseDbRsp) + pDb->cfg.numOfVgroups * sizeof(SVgroupInfo);
+  int32_t    contLen = sizeof(SUseDbRsp) + pDb->cfg.numOfVgroups * sizeof(SVgroupInfo);
   SUseDbRsp *pRsp = rpcMallocCont(contLen);
   if (pRsp == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;

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

@@ -180,8 +180,12 @@ static int32_t mndDnodeActionUpdate(SSdb *pSdb, SDnodeObj *pOldDnode, SDnodeObj
 }
 
 SDnodeObj *mndAcquireDnode(SMnode *pMnode, int32_t dnodeId) {
-  SSdb *pSdb = pMnode->pSdb;
-  return sdbAcquire(pSdb, SDB_DNODE, &dnodeId);
+  SSdb      *pSdb = pMnode->pSdb;
+  SDnodeObj *pDnode = sdbAcquire(pSdb, SDB_DNODE, &dnodeId);
+  if (pDnode == NULL) {
+    terrno = TSDB_CODE_MND_DNODE_NOT_EXIST;
+  }
+  return pDnode;
 }
 
 void mndReleaseDnode(SMnode *pMnode, SDnodeObj *pDnode) {

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

@@ -21,6 +21,13 @@
 #define TSDB_TRN_ARRAY_SIZE 8
 #define TSDB_TRN_RESERVE_SIZE 64
 
+typedef struct {
+  SEpSet  epSet;
+  int8_t  msgType;
+  int32_t contLen;
+  void   *pCont;
+} STransAction;
+
 static SSdbRaw *mndTransActionEncode(STrans *pTrans);
 static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw);
 static int32_t  mndTransActionInsert(SSdb *pSdb, STrans *pTrans);
@@ -29,9 +36,12 @@ static int32_t  mndTransActionDelete(SSdb *pSdb, STrans *pTrans);
 
 static void    mndTransSetRpcHandle(STrans *pTrans, void *rpcHandle);
 static void    mndTransSendRpcRsp(STrans *pTrans, int32_t code);
-static int32_t mndTransAppendArray(SArray *pArray, SSdbRaw *pRaw);
-static void    mndTransDropArray(SArray *pArray);
-static int32_t mndTransExecuteArray(SMnode *pMnode, SArray *pArray);
+static int32_t mndTransAppendLog(SArray *pArray, SSdbRaw *pRaw);
+static int32_t mndTransAppendAction(SArray *pArray, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont);
+static void    mndTransDropLogs(SArray *pArray);
+static void    mndTransDropActions(SArray *pArray);
+static int32_t mndTransExecuteLogs(SMnode *pMnode, SArray *pArray);
+static int32_t mndTransExecuteActions(SMnode *pMnode, SArray *pArray);
 static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans);
 static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans);
 static int32_t mndTransExecuteCommitLogs(SMnode *pMnode, STrans *pTrans);
@@ -58,7 +68,7 @@ int32_t mndInitTrans(SMnode *pMnode) {
 void mndCleanupTrans(SMnode *pMnode) {}
 
 static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
-  int32_t rawDataLen = 16 * sizeof(int32_t) + TSDB_TRN_RESERVE_SIZE;
+  int32_t rawDataLen = sizeof(STrans) + TSDB_TRN_RESERVE_SIZE;
   int32_t redoLogNum = taosArrayGetSize(pTrans->redoLogs);
   int32_t undoLogNum = taosArrayGetSize(pTrans->undoLogs);
   int32_t commitLogNum = taosArrayGetSize(pTrans->commitLogs);
@@ -80,6 +90,16 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
     rawDataLen += sdbGetRawTotalSize(pTmp);
   }
 
+  for (int32_t i = 0; i < redoActionNum; ++i) {
+    STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
+    rawDataLen += (sizeof(STransAction) + pAction->contLen);
+  }
+
+  for (int32_t i = 0; i < undoActionNum; ++i) {
+    STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
+    rawDataLen += (sizeof(STransAction) + pAction->contLen);
+  }
+
   SSdbRaw *pRaw = sdbAllocRaw(SDB_TRANS, TSDB_TRANS_VER, rawDataLen);
   if (pRaw == NULL) {
     mError("trans:%d, failed to alloc raw since %s", pTrans->id, terrstr());
@@ -116,6 +136,22 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
     SDB_SET_BINARY(pRaw, dataPos, (void *)pTmp, len)
   }
 
+  for (int32_t i = 0; i < redoActionNum; ++i) {
+    STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
+    SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet));
+    SDB_SET_INT8(pRaw, dataPos, pAction->msgType)
+    SDB_SET_INT32(pRaw, dataPos, pAction->contLen)
+    SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen);
+  }
+
+  for (int32_t i = 0; i < undoActionNum; ++i) {
+    STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
+    SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet));
+    SDB_SET_INT8(pRaw, dataPos, pAction->msgType)
+    SDB_SET_INT32(pRaw, dataPos, pAction->contLen)
+    SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pCont, pAction->contLen);
+  }
+
   SDB_SET_RESERVE(pRaw, dataPos, TSDB_TRN_RESERVE_SIZE)
   SDB_SET_DATALEN(pRaw, dataPos);
   mTrace("trans:%d, encode to raw:%p, len:%d", pTrans->id, pRaw, dataPos);
@@ -147,8 +183,8 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
   pTrans->redoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
   pTrans->undoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
   pTrans->commitLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
-  pTrans->redoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
-  pTrans->undoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
+  pTrans->redoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(STransAction));
+  pTrans->undoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(STransAction));
 
   if (pTrans->redoLogs == NULL || pTrans->undoLogs == NULL || pTrans->commitLogs == NULL ||
       pTrans->redoActions == NULL || pTrans->undoActions == NULL) {
@@ -175,42 +211,77 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
   for (int32_t i = 0; i < redoLogNum; ++i) {
     int32_t dataLen = 0;
     SDB_GET_INT32(pRaw, pRow, dataPos, &dataLen)
-
     char *pData = malloc(dataLen);
     SDB_GET_BINARY(pRaw, pRow, dataPos, pData, dataLen);
+
     void *ret = taosArrayPush(pTrans->redoLogs, &pData);
     if (ret == NULL) {
       code = TSDB_CODE_OUT_OF_MEMORY;
       goto TRANS_DECODE_OVER;
-      break;
     }
   }
 
   for (int32_t i = 0; i < undoLogNum; ++i) {
     int32_t dataLen = 0;
     SDB_GET_INT32(pRaw, pRow, dataPos, &dataLen)
-
     char *pData = malloc(dataLen);
     SDB_GET_BINARY(pRaw, pRow, dataPos, pData, dataLen);
+
     void *ret = taosArrayPush(pTrans->undoLogs, &pData);
     if (ret == NULL) {
       code = TSDB_CODE_OUT_OF_MEMORY;
       goto TRANS_DECODE_OVER;
-      break;
     }
   }
 
   for (int32_t i = 0; i < commitLogNum; ++i) {
     int32_t dataLen = 0;
     SDB_GET_INT32(pRaw, pRow, dataPos, &dataLen)
-
     char *pData = malloc(dataLen);
     SDB_GET_BINARY(pRaw, pRow, dataPos, pData, dataLen);
+
     void *ret = taosArrayPush(pTrans->commitLogs, &pData);
     if (ret == NULL) {
       code = TSDB_CODE_OUT_OF_MEMORY;
       goto TRANS_DECODE_OVER;
-      break;
+    }
+  }
+
+  for (int32_t i = 0; i < redoActionNum; ++i) {
+    STransAction action = {0};
+    SDB_GET_BINARY(pRaw, pRow, dataPos, (void *)&action.epSet, sizeof(SEpSet));
+    SDB_GET_INT8(pRaw, pRow, dataPos, &action.msgType)
+    SDB_GET_INT32(pRaw, pRow, dataPos, &action.contLen)
+    action.pCont = malloc(action.contLen);
+    if (action.pCont == NULL) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+      goto TRANS_DECODE_OVER;
+    }
+    SDB_GET_BINARY(pRaw, pRow, dataPos, action.pCont, action.contLen);
+
+    void *ret = taosArrayPush(pTrans->redoActions, &action);
+    if (ret == NULL) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+      goto TRANS_DECODE_OVER;
+    }
+  }
+
+  for (int32_t i = 0; i < undoActionNum; ++i) {
+    STransAction action = {0};
+    SDB_GET_BINARY(pRaw, pRow, dataPos, (void *)&action.epSet, sizeof(SEpSet));
+    SDB_GET_INT8(pRaw, pRow, dataPos, &action.msgType)
+    SDB_GET_INT32(pRaw, pRow, dataPos, &action.contLen)
+    action.pCont = malloc(action.contLen);
+    if (action.pCont == NULL) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+      goto TRANS_DECODE_OVER;
+    }
+    SDB_GET_BINARY(pRaw, pRow, dataPos, action.pCont, action.contLen);
+
+    void *ret = taosArrayPush(pTrans->undoActions, &action);
+    if (ret == NULL) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+      goto TRANS_DECODE_OVER;
     }
   }
 
@@ -237,11 +308,11 @@ static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans) {
 static int32_t mndTransActionDelete(SSdb *pSdb, STrans *pTrans) {
   mTrace("trans:%d, perform delete action, stage:%s", pTrans->id, mndTransStageStr(pTrans->stage));
 
-  mndTransDropArray(pTrans->redoLogs);
-  mndTransDropArray(pTrans->undoLogs);
-  mndTransDropArray(pTrans->commitLogs);
-  mndTransDropArray(pTrans->redoActions);
-  mndTransDropArray(pTrans->undoActions);
+  mndTransDropLogs(pTrans->redoLogs);
+  mndTransDropLogs(pTrans->undoLogs);
+  mndTransDropLogs(pTrans->commitLogs);
+  mndTransDropActions(pTrans->redoActions);
+  mndTransDropActions(pTrans->undoActions);
 
   return 0;
 }
@@ -274,6 +345,8 @@ char *mndTransStageStr(ETrnStage stage) {
       return "rollback";
     case TRN_STAGE_RETRY:
       return "retry";
+    case TRN_STAGE_OVER:
+      return "stop";
     default:
       return "undefined";
   }
@@ -305,8 +378,8 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, void *rpcHandle) {
   pTrans->redoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
   pTrans->undoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
   pTrans->commitLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
-  pTrans->redoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
-  pTrans->undoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
+  pTrans->redoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(STransAction));
+  pTrans->undoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(STransAction));
 
   if (pTrans->redoLogs == NULL || pTrans->undoLogs == NULL || pTrans->commitLogs == NULL ||
       pTrans->redoActions == NULL || pTrans->undoActions == NULL) {
@@ -319,7 +392,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, void *rpcHandle) {
   return pTrans;
 }
 
-static void mndTransDropArray(SArray *pArray) {
+static void mndTransDropLogs(SArray *pArray) {
   for (int32_t i = 0; i < pArray->size; ++i) {
     SSdbRaw *pRaw = taosArrayGetP(pArray, i);
     tfree(pRaw);
@@ -328,12 +401,21 @@ static void mndTransDropArray(SArray *pArray) {
   taosArrayDestroy(pArray);
 }
 
+static void mndTransDropActions(SArray *pArray) {
+  for (int32_t i = 0; i < pArray->size; ++i) {
+    STransAction *pAction = taosArrayGet(pArray, i);
+    free(pAction->pCont);
+  }
+
+  taosArrayDestroy(pArray);
+}
+
 void mndTransDrop(STrans *pTrans) {
-  mndTransDropArray(pTrans->redoLogs);
-  mndTransDropArray(pTrans->undoLogs);
-  mndTransDropArray(pTrans->commitLogs);
-  mndTransDropArray(pTrans->redoActions);
-  mndTransDropArray(pTrans->undoActions);
+  mndTransDropLogs(pTrans->redoLogs);
+  mndTransDropLogs(pTrans->undoLogs);
+  mndTransDropLogs(pTrans->commitLogs);
+  mndTransDropActions(pTrans->redoActions);
+  mndTransDropActions(pTrans->undoActions);
 
   mDebug("trans:%d, data:%p is dropped", pTrans->id, pTrans);
   tfree(pTrans);
@@ -344,7 +426,7 @@ static void mndTransSetRpcHandle(STrans *pTrans, void *rpcHandle) {
   mTrace("trans:%d, set rpc handle:%p", pTrans->id, rpcHandle);
 }
 
-static int32_t mndTransAppendArray(SArray *pArray, SSdbRaw *pRaw) {
+static int32_t mndTransAppendLog(SArray *pArray, SSdbRaw *pRaw) {
   if (pArray == NULL || pRaw == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
@@ -360,32 +442,44 @@ static int32_t mndTransAppendArray(SArray *pArray, SSdbRaw *pRaw) {
 }
 
 int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
-  int32_t code = mndTransAppendArray(pTrans->redoLogs, pRaw);
+  int32_t code = mndTransAppendLog(pTrans->redoLogs, pRaw);
   mTrace("trans:%d, raw:%p append to redo logs, code:0x%x", pTrans->id, pRaw, code);
   return code;
 }
 
 int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw) {
-  int32_t code = mndTransAppendArray(pTrans->undoLogs, pRaw);
+  int32_t code = mndTransAppendLog(pTrans->undoLogs, pRaw);
   mTrace("trans:%d, raw:%p append to undo logs, code:0x%x", pTrans->id, pRaw, code);
   return code;
 }
 
 int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw) {
-  int32_t code = mndTransAppendArray(pTrans->commitLogs, pRaw);
+  int32_t code = mndTransAppendLog(pTrans->commitLogs, pRaw);
   mTrace("trans:%d, raw:%p append to commit logs, code:0x%x", pTrans->id, pRaw, code);
   return code;
 }
 
-int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *pEpSet, void *pMsg) {
-  int32_t code = mndTransAppendArray(pTrans->redoActions, pMsg);
-  mTrace("trans:%d, msg:%p append to redo actions", pTrans->id, pMsg);
+static int32_t mndTransAppendAction(SArray *pArray, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont) {
+  STransAction action = {.epSet = *pEpSet, .msgType = msgType, .contLen = contLen, .pCont = pCont};
+
+  void *ptr = taosArrayPush(pArray, &action);
+  if (ptr == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+
+  return 0;
+}
+
+int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont) {
+  int32_t code = mndTransAppendAction(pTrans->redoActions, pEpSet, msgType, contLen, pCont);
+  mTrace("trans:%d, msg:%s len:%d append to redo actions", pTrans->id, taosMsg[msgType], contLen);
   return code;
 }
 
-int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *pEpSet, void *pMsg) {
-  int32_t code = mndTransAppendArray(pTrans->undoActions, pMsg);
-  mTrace("trans:%d, msg:%p append to undo actions", pTrans->id, pMsg);
+int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont) {
+  int32_t code = mndTransAppendAction(pTrans->undoActions, pEpSet, msgType, contLen, pCont);
+  mTrace("trans:%d, msg:%s len:%d append to undo actions", pTrans->id, taosMsg[msgType], contLen);
   return code;
 }
 
@@ -502,7 +596,7 @@ void mndTransApply(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg, int32_t code)
   // todo
 }
 
-static int32_t mndTransExecuteArray(SMnode *pMnode, SArray *pArray) {
+static int32_t mndTransExecuteLogs(SMnode *pMnode, SArray *pArray) {
   SSdb   *pSdb = pMnode->pSdb;
   int32_t arraySize = taosArrayGetSize(pArray);
 
@@ -520,7 +614,7 @@ static int32_t mndTransExecuteArray(SMnode *pMnode, SArray *pArray) {
 static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans) {
   int32_t code = 0;
   if (taosArrayGetSize(pTrans->redoLogs) != 0) {
-    code = mndTransExecuteArray(pMnode, pTrans->redoLogs);
+    code = mndTransExecuteLogs(pMnode, pTrans->redoLogs);
     if (code != 0) {
       mError("trans:%d, failed to execute redo logs since %s", pTrans->id, terrstr())
     } else {
@@ -534,7 +628,7 @@ static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans) {
 static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans) {
   int32_t code = 0;
   if (taosArrayGetSize(pTrans->undoLogs) != 0) {
-    code = mndTransExecuteArray(pMnode, pTrans->undoLogs);
+    code = mndTransExecuteLogs(pMnode, pTrans->undoLogs);
     if (code != 0) {
       mError("trans:%d, failed to execute undo logs since %s", pTrans->id, terrstr())
     } else {
@@ -548,7 +642,7 @@ static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans) {
 static int32_t mndTransExecuteCommitLogs(SMnode *pMnode, STrans *pTrans) {
   int32_t code = 0;
   if (taosArrayGetSize(pTrans->commitLogs) != 0) {
-    code = mndTransExecuteArray(pMnode, pTrans->commitLogs);
+    code = mndTransExecuteLogs(pMnode, pTrans->commitLogs);
     if (code != 0) {
       mError("trans:%d, failed to execute commit logs since %s", pTrans->id, terrstr())
     } else {
@@ -559,18 +653,40 @@ static int32_t mndTransExecuteCommitLogs(SMnode *pMnode, STrans *pTrans) {
   return code;
 }
 
-static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans) {
-  if (taosArrayGetSize(pTrans->redoActions) != 0) {
-    mTrace("trans:%d, execute redo actions finished", pTrans->id);
+static int32_t mndTransExecuteActions(SMnode *pMnode, SArray *pArray) {
+#if 0
+  int32_t arraySize = taosArrayGetSize(pArray);
+  for (int32_t i = 0; i < arraySize; ++i) {
+    STransAction *pAction = taosArrayGet(pArray, i);
+
+    SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen};
+    rpcMsg.pCont = rpcMallocCont(pAction->contLen);
+    if (rpcMsg.pCont == NULL) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      return -1;
+    }
+    memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
+    mndSendMsgToDnode(pMnode, &pAction->epSet, &rpcMsg);
   }
+
+  return TSDB_CODE_MND_ACTION_IN_PROGRESS;
+#else
   return 0;
+#endif
+}
+
+static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans) {
+  if (taosArrayGetSize(pTrans->redoActions) <= 0) return 0;
+
+  mTrace("trans:%d, start to execute redo actions", pTrans->id);
+  return mndTransExecuteActions(pMnode, pTrans->redoActions);
 }
 
 static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans) {
-  if (taosArrayGetSize(pTrans->undoActions) != 0) {
-    mTrace("trans:%d, execute undo actions finished", pTrans->id);
-  }
-  return 0;
+  if (taosArrayGetSize(pTrans->undoActions) <= 0) return 0;
+
+  mTrace("trans:%d, start to execute undo actions", pTrans->id);
+  return mndTransExecuteActions(pMnode, pTrans->undoActions);
 }
 
 static int32_t mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans) {

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

@@ -24,9 +24,10 @@
 #define TSDB_VGROUP_VER_NUM 1
 #define TSDB_VGROUP_RESERVE_SIZE 64
 
-static int32_t mndVgroupActionInsert(SSdb *pSdb, SVgObj *pVgroup);
-static int32_t mndVgroupActionDelete(SSdb *pSdb, SVgObj *pVgroup);
-static int32_t mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOldVgroup, SVgObj *pNewVgroup);
+static SSdbRow *mndVgroupActionDecode(SSdbRaw *pRaw);
+static int32_t  mndVgroupActionInsert(SSdb *pSdb, SVgObj *pVgroup);
+static int32_t  mndVgroupActionDelete(SSdb *pSdb, SVgObj *pVgroup);
+static int32_t  mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOldVgroup, SVgObj *pNewVgroup);
 
 static int32_t mndProcessCreateVnodeRsp(SMnodeMsg *pMsg);
 static int32_t mndProcessAlterVnodeRsp(SMnodeMsg *pMsg);
@@ -156,6 +157,80 @@ void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup) {
   sdbRelease(pSdb, pVgroup);
 }
 
+SCreateVnodeMsg *mndBuildCreateVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup) {
+  SCreateVnodeMsg *pCreate = malloc(sizeof(SCreateVnodeMsg));
+  if (pCreate == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return NULL;
+  }
+
+  pCreate->dnodeId = htonl(pDnode->id);
+  pCreate->vgId = htonl(pVgroup->vgId);
+  memcpy(pCreate->db, pDb->name, TSDB_FULL_DB_NAME_LEN);
+  pCreate->dbUid = htobe64(pDb->uid);
+  pCreate->cacheBlockSize = htonl(pDb->cfg.cacheBlockSize);
+  pCreate->totalBlocks = htonl(pDb->cfg.totalBlocks);
+  pCreate->daysPerFile = htonl(pDb->cfg.daysPerFile);
+  pCreate->daysToKeep0 = htonl(pDb->cfg.daysToKeep0);
+  pCreate->daysToKeep1 = htonl(pDb->cfg.daysToKeep1);
+  pCreate->daysToKeep2 = htonl(pDb->cfg.daysToKeep2);
+  pCreate->minRows = htonl(pDb->cfg.minRows);
+  pCreate->maxRows = htonl(pDb->cfg.maxRows);
+  pCreate->commitTime = htonl(pDb->cfg.commitTime);
+  pCreate->fsyncPeriod = htonl(pDb->cfg.fsyncPeriod);
+  pCreate->walLevel = pDb->cfg.walLevel;
+  pCreate->precision = pDb->cfg.precision;
+  pCreate->compression = pDb->cfg.compression;
+  pCreate->quorum = pDb->cfg.quorum;
+  pCreate->update = pDb->cfg.update;
+  pCreate->cacheLastRow = pDb->cfg.cacheLastRow;
+  pCreate->replica = pVgroup->replica;
+  pCreate->selfIndex = -1;
+
+  for (int32_t v = 0; v < pVgroup->replica; ++v) {
+    SReplica  *pReplica = &pCreate->replicas[v];
+    SVnodeGid *pVgid = &pVgroup->vnodeGid[v];
+    SDnodeObj *pVgidDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
+    if (pVgidDnode == NULL) {
+      free(pCreate);
+      terrno = TSDB_CODE_MND_APP_ERROR;
+      return NULL;
+    }
+
+    pReplica->id = htonl(pVgidDnode->id);
+    pReplica->port = htons(pVgidDnode->port);
+    memcpy(pReplica->fqdn, pVgidDnode->fqdn, TSDB_FQDN_LEN);
+    mndReleaseDnode(pMnode, pVgidDnode);
+
+    if (pDnode->id == pVgid->dnodeId) {
+      pCreate->selfIndex = v;
+    }
+  }
+
+  if (pCreate->selfIndex == -1) {
+    free(pCreate);
+    terrno = TSDB_CODE_MND_APP_ERROR;
+    return NULL;
+  }
+
+  return pCreate;
+}
+
+SDropVnodeMsg *mndBuildDropVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup) {
+  SDropVnodeMsg *pDrop = malloc(sizeof(SDropVnodeMsg));
+  if (pDrop == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return NULL;
+  }
+
+  pDrop->dnodeId = htonl(pDnode->id);
+  pDrop->vgId = htonl(pVgroup->vgId);
+  memcpy(pDrop->db, pDb->name, TSDB_FULL_DB_NAME_LEN);
+  pDrop->dbUid = htobe64(pDb->uid);
+
+  return pDrop;
+}
+
 static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup) {
   SSdb   *pSdb = pMnode->pSdb;
   int32_t allocedVnodes = 0;

source/util/src/terror.c

@@ -130,6 +130,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TSC_VALUE_OUT_OF_RANGE,       "Value out of range")
 TAOS_DEFINE_ERROR(TSDB_CODE_TSC_INVALID_INPUT,            "Invalid tsc input")
 
 // mnode-common
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_APP_ERROR,                "Mnode internal error")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_NOT_READY,                "Cluster not ready")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_MSG_NOT_PROCESSED,        "Message not processed")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_ACTION_IN_PROGRESS,       "Message is progressing")