Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/row_refact

docs-cn/12-taos-sql/13-operators.md

@@ -35,8 +35,8 @@ TDengine 支持 `UNION ALL` 和 `UNION` 操作符。UNION ALL 将查询返回的
 | --- | :---------------: | -------------------------------------------------------------------- | -------------------- |
 | 1   |         =         | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 相等                 |
 | 2   |      <\>, !=      | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型，且不可以为表的时间戳主键列 | 不相等               |
-| 3   |      \>, \<       | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 大于，小于           |
-| 4   |     \>=, \<=      | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 大于等于，小于等于   |
+| 3   |      \>, <       | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 大于，小于           |
+| 4   |     \>=, <=      | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 大于等于，小于等于   |
 | 5   |   IS [NOT] NULL   | 所有类型                                                             | 是否为空值           |
 | 6   | [NOT] BETWEEN AND | 除 BOOL、BLOB、MEDIUMBLOB 和 JSON 外的所有类型                       | 闭区间比较           |
 | 7   |        IN         | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型，且不可以为表的时间戳主键列 | 与列表内的任意值相等 |

docs-en/12-taos-sql/13-operators.md

@@ -35,8 +35,8 @@ TDengine provides 2 set operators: `UNION ALL` and `UNION`. `UNION ALL` combines
 | --- | :---------------: | ------------------------------------------------------------------- | ----------------------------------------------- |
 | 1   |         =         | Except for BLOB, MEDIUMBLOB and JSON                                | Equal                                           |
 | 2   |      <\>, !=      | Except for BLOB, MEDIUMBLOB, JSON and primary key of timestamp type | Not equal                                       |
-| 3   |      \>, \<       | Except for BLOB, MEDIUMBLOB and JSON                                | Greater than, less than                         |
-| 4   |     \>=, \<=      | Except for BLOB, MEDIUMBLOB and JSON                                | Greater than or equal to, less than or equal to |
+| 3   |      \>, <       | Except for BLOB, MEDIUMBLOB and JSON                                | Greater than, less than                         |
+| 4   |     \>=, <=      | Except for BLOB, MEDIUMBLOB and JSON                                | Greater than or equal to, less than or equal to |
 | 5   |   IS [NOT] NULL   | Any types                                                           | Is NULL or NOT                                  |
 | 6   | [NOT] BETWEEN AND | Except for BLOB, MEDIUMBLOB and JSON                                | In a value range or not                         |
 | 7   |        IN         | Except for BLOB, MEDIUMBLOB, JSON and primary key of timestamp type | In a list of values or not                      |

example/CMakeLists.txt

-add_executable(tmq "")
-add_executable(tstream "")
-add_executable(demoapi "")
-
-target_sources(tmq
-    PRIVATE
-    "src/tmq.c"
-)
-
-target_sources(tstream
-    PRIVATE
-    "src/tstream.c"
-)
-
-target_sources(demoapi
-    PRIVATE
-    "src/demoapi.c"
-)
-
-target_link_libraries(tmq
-    taos_static
-)
-
-target_link_libraries(tstream
-    taos_static
-)
-
-target_link_libraries(demoapi
-    taos_static
-)
-
-target_include_directories(tmq
-    PUBLIC "${TD_SOURCE_DIR}/include/os"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
-
-target_include_directories(tstream
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
-
-target_include_directories(demoapi
-    PUBLIC "${TD_SOURCE_DIR}/include/client"
-    PUBLIC "${TD_SOURCE_DIR}/include/os"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
-
-SET_TARGET_PROPERTIES(tmq PROPERTIES OUTPUT_NAME tmq)
-SET_TARGET_PROPERTIES(tstream PROPERTIES OUTPUT_NAME tstream)
-SET_TARGET_PROPERTIES(demoapi PROPERTIES OUTPUT_NAME demoapi)

examples/c/CMakeLists.txt

@@ -11,6 +11,56 @@ IF (TD_LINUX)
   #TARGET_LINK_LIBRARIES(subscribe taos_static trpc tutil pthread )
   #ADD_EXECUTABLE(epoll epoll.c)
   #TARGET_LINK_LIBRARIES(epoll taos_static trpc tutil pthread lua)
+
+  add_executable(tmq "")
+  add_executable(tstream "")
+  add_executable(demoapi "")
+
+  target_sources(tmq
+      PRIVATE
+      "tmq.c"
+      )
+
+  target_sources(tstream
+      PRIVATE
+      "tstream.c"
+      )
+
+  target_sources(demoapi
+      PRIVATE
+      "demoapi.c"
+      )
+
+  target_link_libraries(tmq
+      taos_static
+      )
+
+  target_link_libraries(tstream
+      taos_static
+      )
+
+  target_link_libraries(demoapi
+      taos_static
+      )
+
+  target_include_directories(tmq
+      PUBLIC "${TD_SOURCE_DIR}/include/os"
+      PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+      )
+
+  target_include_directories(tstream
+      PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+      )
+
+  target_include_directories(demoapi
+      PUBLIC "${TD_SOURCE_DIR}/include/client"
+      PUBLIC "${TD_SOURCE_DIR}/include/os"
+      PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+      )
+
+  SET_TARGET_PROPERTIES(tmq PROPERTIES OUTPUT_NAME tmq)
+  SET_TARGET_PROPERTIES(tstream PROPERTIES OUTPUT_NAME tstream)
+  SET_TARGET_PROPERTIES(demoapi PROPERTIES OUTPUT_NAME demoapi)
 ENDIF ()
 IF (TD_DARWIN)
   INCLUDE_DIRECTORIES(. ${TD_SOURCE_DIR}/src/inc ${TD_SOURCE_DIR}/src/client/inc  ${TD_SOURCE_DIR}/inc)

include/common/tmsg.h

@@ -244,7 +244,7 @@ typedef struct {
   const void* pMsg;
 } SSubmitMsgIter;
 
-int32_t tInitSubmitMsgIter(SSubmitReq* pMsg, SSubmitMsgIter* pIter);
+int32_t tInitSubmitMsgIter(const SSubmitReq* pMsg, SSubmitMsgIter* pIter);
 int32_t tGetSubmitMsgNext(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
 int32_t tInitSubmitBlkIter(SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
 STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
@@ -1001,7 +1001,6 @@ typedef struct {
 
 typedef struct {
   int32_t  vgId;
-  int32_t  dnodeId;
   char     db[TSDB_DB_FNAME_LEN];
   int64_t  dbUid;
   int32_t  vgVersion;
@@ -1024,16 +1023,14 @@ typedef struct {
   int8_t   compression;
   int8_t   strict;
   int8_t   cacheLastRow;
+  int8_t   isTsma;
+  int8_t   standby;
   int8_t   replica;
   int8_t   selfIndex;
   SReplica replicas[TSDB_MAX_REPLICA];
   int32_t  numOfRetensions;
   SArray*  pRetensions;  // SRetention
-
-  // for tsma
-  int8_t isTsma;
   void*    pTsma;
-
 } SCreateVnodeReq;
 
 int32_t tSerializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
@@ -1071,8 +1068,8 @@ typedef struct {
   int8_t   walLevel;
   int8_t   strict;
   int8_t   cacheLastRow;
-  int8_t   replica;
   int8_t   selfIndex;
+  int8_t   replica;
   SReplica replicas[TSDB_MAX_REPLICA];
 } SAlterVnodeReq;
 

include/common/tmsgdef.h

@@ -221,9 +221,9 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_APPLY_MSG, "vnode-sync-apply-msg", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_CONFIG_CHANGE, "vnode-sync-config-change", NULL, NULL)
   
-  TD_DEF_MSG_TYPE(TDMT_VND_SYNC_VNODE, "vnode-sync-vnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_ALTER_VNODE, "vnode-alter-vnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_COMPACT_VNODE, "vnode-compact-vnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_ALTER_CONFIG, "vnode-alter-config", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_ALTER_REPLICA, "vnode-alter-replica", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_COMPACT, "vnode-compact", NULL, NULL)
 
   TD_DEF_MSG_TYPE(TDMT_VND_DELETE, "vnode-delete-data", SVDeleteReq, SVDeleteRsp)
 

include/libs/stream/tstream.h

@@ -62,10 +62,7 @@ enum {
 
 typedef struct {
   int8_t      type;
-
-  int32_t sourceVg;
-  int64_t sourceVer;
-
+  int64_t     ver;
   int32_t*    dataRef;
   SSubmitReq* data;
 } SStreamDataSubmit;
@@ -111,6 +108,8 @@ static FORCE_INLINE void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit)
   }
 }
 
+SStreamDataSubmit* streamSubmitRefClone(SStreamDataSubmit* pSubmit);
+
 int32_t streamDataBlockEncode(void** buf, const SStreamDataBlock* pOutput);
 void*   streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput);
 
@@ -209,8 +208,6 @@ struct SStreamTask {
   int32_t nodeId;
   SEpSet  epSet;
 
-  // source preprocess
-
   // exec
   STaskExec exec;
 
@@ -318,8 +315,6 @@ int32_t streamDequeueOutput(SStreamTask* pTask, void** output);
 
 int32_t streamTaskRun(SStreamTask* pTask);
 
-int32_t streamTaskHandleInput(SStreamTask* pTask, void* data);
-
 int32_t streamTaskProcessRunReq(SStreamTask* pTask, SMsgCb* pMsgCb);
 int32_t streamProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchReq* pReq, SRpcMsg* pMsg);
 int32_t streamProcessDispatchRsp(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchRsp* pRsp);

source/common/src/tmsg.c

@@ -28,7 +28,7 @@
 #undef TD_MSG_SEG_CODE_
 #include "tmsgdef.h"
 
-int32_t tInitSubmitMsgIter(SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
+int32_t tInitSubmitMsgIter(const SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
   if (pMsg == NULL) {
     terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
     return -1;
@@ -165,7 +165,6 @@ int32_t tDecodeSQueryNodeLoad(SDecoder *pDecoder, SQueryNodeLoad *pLoad) {
   return 0;
 }
 
-
 int32_t taosEncodeSEpSet(void **buf, const SEpSet *pEp) {
   int32_t tlen = 0;
   tlen += taosEncodeFixedI8(buf, pEp->inUse);
@@ -2934,7 +2933,6 @@ int32_t tSerializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *pR
 
   if (tStartEncode(&encoder) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->vgId) < 0) return -1;
-  if (tEncodeI32(&encoder, pReq->dnodeId) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->db) < 0) return -1;
   if (tEncodeI64(&encoder, pReq->dbUid) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->vgVersion) < 0) return -1;
@@ -2957,6 +2955,7 @@ int32_t tSerializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *pR
   if (tEncodeI8(&encoder, pReq->compression) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->strict) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->cacheLastRow) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->standby) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->replica) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->selfIndex) < 0) return -1;
   for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
@@ -2991,7 +2990,6 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
 
   if (tStartDecode(&decoder) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->vgId) < 0) return -1;
-  if (tDecodeI32(&decoder, &pReq->dnodeId) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->db) < 0) return -1;
   if (tDecodeI64(&decoder, &pReq->dbUid) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->vgVersion) < 0) return -1;
@@ -3014,6 +3012,7 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
   if (tDecodeI8(&decoder, &pReq->compression) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->strict) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->cacheLastRow) < 0) return -1;
+  if (tDecodeI8(&decoder, &pReq->standby) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->replica) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->selfIndex) < 0) return -1;
   for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
@@ -3053,7 +3052,7 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
 int32_t tFreeSCreateVnodeReq(SCreateVnodeReq *pReq) {
   taosArrayDestroy(pReq->pRetensions);
   pReq->pRetensions = NULL;
-  if(pReq->isTsma) {
+  if (pReq->isTsma) {
     taosMemoryFreeClear(pReq->pTsma);
   }
   return 0;
@@ -3134,8 +3133,8 @@ int32_t tSerializeSAlterVnodeReq(void *buf, int32_t bufLen, SAlterVnodeReq *pReq
   if (tEncodeI8(&encoder, pReq->walLevel) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->strict) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->cacheLastRow) < 0) return -1;
-  if (tEncodeI8(&encoder, pReq->replica) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->selfIndex) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->replica) < 0) return -1;
   for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
     SReplica *pReplica = &pReq->replicas[i];
     if (tEncodeSReplica(&encoder, pReplica) < 0) return -1;
@@ -3165,8 +3164,8 @@ int32_t tDeserializeSAlterVnodeReq(void *buf, int32_t bufLen, SAlterVnodeReq *pR
   if (tDecodeI8(&decoder, &pReq->walLevel) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->strict) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->cacheLastRow) < 0) return -1;
-  if (tDecodeI8(&decoder, &pReq->replica) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->selfIndex) < 0) return -1;
+  if (tDecodeI8(&decoder, &pReq->replica) < 0) return -1;
   for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
     SReplica *pReplica = &pReq->replicas[i];
     if (tDecodeSReplica(&decoder, pReplica) < 0) return -1;

source/dnode/mgmt/mgmt_mnode/src/mmHandle.c

@@ -219,9 +219,9 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TASK, mmPutNodeMsgToQueryQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_HEARTBEAT, mmPutNodeMsgToQueryQueue, 1) == NULL) goto _OVER;
 
-  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_VNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_VNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT_VNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
 
   if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_TIMEOUT, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;

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

@@ -162,13 +162,14 @@ static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
   pCfg->hashEnd = pCreate->hashEnd;
   pCfg->hashMethod = pCreate->hashMethod;
 
+  pCfg->standby = pCfg->standby;
   pCfg->syncCfg.myIndex = pCreate->selfIndex;
   pCfg->syncCfg.replicaNum = pCreate->replica;
   memset(&pCfg->syncCfg.nodeInfo, 0, sizeof(pCfg->syncCfg.nodeInfo));
   for (int i = 0; i < pCreate->replica; ++i) {
-    pCfg->syncCfg.nodeInfo[i].nodePort = pCreate->replicas[i].port;
-    snprintf(pCfg->syncCfg.nodeInfo[i].nodeFqdn, sizeof(pCfg->syncCfg.nodeInfo[i].nodeFqdn), "%s",
-             pCreate->replicas[i].fqdn);
+    SNodeInfo *pNode = &pCfg->syncCfg.nodeInfo[i];
+    pNode->nodePort = pCreate->replicas[i].port;
+    tstrncpy(pNode->nodeFqdn, pCreate->replicas[i].fqdn, sizeof(pNode->nodeFqdn));
   }
 }
 
@@ -181,6 +182,8 @@ static void vmGenerateWrapperCfg(SVnodeMgmt *pMgmt, SCreateVnodeReq *pCreate, SW
 
 int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SCreateVnodeReq createReq = {0};
+  SVnodeCfg       vnodeCfg = {0};
+  SWrapperCfg     wrapperCfg = {0};
   int32_t         code = -1;
   char            path[TSDB_FILENAME_LEN] = {0};
 
@@ -189,12 +192,9 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     return -1;
   }
 
-  dDebug("vgId:%d, create vnode req is received, tsma:%d", createReq.vgId, createReq.isTsma);
-
-  SVnodeCfg vnodeCfg = {0};
+  dDebug("vgId:%d, create vnode req is received, tsma:%d standby:%d", createReq.vgId, createReq.isTsma,
+         createReq.standby);
   vmGenerateVnodeCfg(&createReq, &vnodeCfg);
-
-  SWrapperCfg wrapperCfg = {0};
   vmGenerateWrapperCfg(pMgmt, &createReq, &wrapperCfg);
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, createReq.vgId);
@@ -333,8 +333,9 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DISPATCH, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_RECOVER, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
 
-  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_VNODE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT_VNODE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
 

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

@@ -61,7 +61,7 @@ int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction);
 int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction);
 void    mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen);
 void    mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen);
-void    mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb);
+void    mndTransSetDbName(STrans *pTrans, const char *dbname);
 void    mndTransSetSerial(STrans *pTrans);
 
 int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);

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

@@ -36,7 +36,7 @@ SArray *mndBuildDnodesArray(SMnode *pMnode);
 int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray);
 int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray, SVnodeGid *del1, SVnodeGid *del2);
 
-void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
+void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen, bool standby);
 void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
 void *mndBuildAlterVnodeReq(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
 

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

@@ -261,7 +261,7 @@ void mndReleaseDb(SMnode *pMnode, SDbObj *pDb) {
   sdbRelease(pSdb, pDb);
 }
 
-static int32_t mndAddCreateVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid) {
+static int32_t mndAddCreateVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid, bool standby) {
   STransAction action = {0};
 
   SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
@@ -270,7 +270,7 @@ static int32_t mndAddCreateVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *p
   mndReleaseDnode(pMnode, pDnode);
 
   int32_t contLen = 0;
-  void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
+  void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen, standby);
   if (pReq == NULL) return -1;
 
   action.pCont = pReq;
@@ -286,7 +286,7 @@ static int32_t mndAddCreateVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *p
   return 0;
 }
 
-static int32_t mndAddAlterVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup) {
+static int32_t mndAddAlterVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, tmsg_t msgType) {
   STransAction action = {0};
   action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
 
@@ -296,7 +296,7 @@ static int32_t mndAddAlterVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pD
 
   action.pCont = pReq;
   action.contLen = contLen;
-  action.msgType = TDMT_VND_ALTER_VNODE;
+  action.msgType = msgType;
 
   if (mndTransAppendRedoAction(pTrans, &action) != 0) {
     taosMemoryFree(pReq);
@@ -388,7 +388,7 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
   }
 
   terrno = 0;
-  return TSDB_CODE_SUCCESS;
+  return terrno;
 }
 
 static void mndSetDefaultDbCfg(SDbCfg *pCfg) {
@@ -467,7 +467,7 @@ static int32_t mndSetCreateDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
 
     for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
       SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
-      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid) != 0) {
+      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, false) != 0) {
         return -1;
       }
     }
@@ -550,7 +550,7 @@ static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate,
 
   mDebug("trans:%d, used to create db:%s", pTrans->id, pCreate->db);
 
-  mndTransSetDbInfo(pTrans, &dbObj);
+  mndTransSetDbName(pTrans, dbObj.name);
   if (mndSetCreateDbRedoLogs(pMnode, pTrans, &dbObj, pVgroups) != 0) goto _OVER;
   if (mndSetCreateDbUndoLogs(pMnode, pTrans, &dbObj, pVgroups) != 0) goto _OVER;
   if (mndSetCreateDbCommitLogs(pMnode, pTrans, &dbObj, pVgroups) != 0) goto _OVER;
@@ -688,29 +688,37 @@ static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
 static int32_t mndSetAlterDbRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pOld, SDbObj *pNew) {
   SSdbRaw *pRedoRaw = mndDbActionEncode(pOld);
   if (pRedoRaw == NULL) return -1;
-  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
-  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY) != 0) return -1;
+  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
+    sdbFreeRaw(pRedoRaw);
+    return -1;
+  }
 
+  sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
   return 0;
 }
 
 static int32_t mndSetAlterDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pOld, SDbObj *pNew) {
   SSdbRaw *pCommitRaw = mndDbActionEncode(pNew);
   if (pCommitRaw == NULL) return -1;
-  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
-  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;
+  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    sdbFreeRaw(pCommitRaw);
+    return -1;
+  }
 
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
   return 0;
 }
 
 static int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SArray *pArray) {
   if (pVgroup->replica <= 0 || pVgroup->replica == pDb->cfg.replications) {
-    if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, pVgroup) != 0) {
+    if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, pVgroup, TDMT_VND_ALTER_CONFIG) != 0) {
       return -1;
     }
   } else {
     SVgObj newVgroup = {0};
     memcpy(&newVgroup, pVgroup, sizeof(SVgObj));
+    mndTransSetSerial(pTrans);
+
     if (newVgroup.replica < pDb->cfg.replications) {
       mInfo("db:%s, vgId:%d, will add 2 vnodes, vn:0 dnode:%d", pVgroup->dbName, pVgroup->vgId,
             pVgroup->vnodeGid[0].dnodeId);
@@ -720,9 +728,9 @@ static int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj
         return -1;
       }
       newVgroup.replica = pDb->cfg.replications;
-      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup) != 0) return -1;
-      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[1]) != 0) return -1;
-      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[2]) != 0) return -1;
+      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[1], true) != 0) return -1;
+      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[2], true) != 0) return -1;
+      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
     } else {
       mInfo("db:%s, vgId:%d, will remove 2 vnodes", pVgroup->dbName, pVgroup->vgId);
 
@@ -733,15 +741,18 @@ static int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj
         return -1;
       }
       newVgroup.replica = pDb->cfg.replications;
-      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup) != 0) return -1;
+      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
       if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del1, true) != 0) return -1;
       if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del2, true) != 0) return -1;
     }
 
     SSdbRaw *pVgRaw = mndVgroupActionEncode(&newVgroup);
     if (pVgRaw == NULL) return -1;
-    if (mndTransAppendCommitlog(pTrans, pVgRaw) != 0) return -1;
-    if (sdbSetRawStatus(pVgRaw, SDB_STATUS_READY) != 0) return -1;
+    if (mndTransAppendCommitlog(pTrans, pVgRaw) != 0) {
+      sdbFreeRaw(pVgRaw);
+      return -1;
+    }
+    sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
   }
 
   return 0;
@@ -774,18 +785,16 @@ static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
 }
 
 static int32_t mndAlterDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pOld, SDbObj *pNew) {
-  int32_t code = -1;
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq);
-  if (pTrans == NULL) goto _OVER;
-
+  if (pTrans == NULL) return -1;
   mDebug("trans:%d, used to alter db:%s", pTrans->id, pOld->name);
 
-  mndTransSetDbInfo(pTrans, pOld);
+  int32_t code = -1;
+  mndTransSetDbName(pTrans, pOld->name);
   if (mndSetAlterDbRedoLogs(pMnode, pTrans, pOld, pNew) != 0) goto _OVER;
   if (mndSetAlterDbCommitLogs(pMnode, pTrans, pOld, pNew) != 0) goto _OVER;
   if (mndSetAlterDbRedoActions(pMnode, pTrans, pOld, pNew) != 0) goto _OVER;
   if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
-
   code = 0;
 
 _OVER:
@@ -1040,7 +1049,7 @@ static int32_t mndDropDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
   if (pTrans == NULL) goto _OVER;
 
   mDebug("trans:%d, used to drop db:%s", pTrans->id, pDb->name);
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
 
   if (mndSetDropDbRedoLogs(pMnode, pTrans, pDb) != 0) goto _OVER;
   if (mndSetDropDbCommitLogs(pMnode, pTrans, pDb) != 0) goto _OVER;

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

@@ -426,7 +426,7 @@ static int32_t mndSetCreateSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans,
   pVgroup->pTsma = pSmaReq;
 
   int32_t contLen = 0;
-  void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
+  void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen, false);
   taosMemoryFreeClear(pSmaReq);
   if (pReq == NULL) return -1;
 
@@ -512,7 +512,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   if (pTrans == NULL) goto _OVER;
 
   mDebug("trans:%d, used to create sma:%s", pTrans->id, pCreate->name);
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
   mndTransSetSerial(pTrans);
 
   if (mndSetCreateSmaRedoLogs(pMnode, pTrans, &smaObj) != 0) goto _OVER;
@@ -757,7 +757,7 @@ static int32_t mndDropSma(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SSmaObj *p
   if (pTrans == NULL) goto _OVER;
 
   mDebug("trans:%d, used to drop sma:%s", pTrans->id, pSma->name);
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
 
   if (mndSetDropSmaRedoLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
   if (mndSetDropSmaVgroupRedoLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;

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

@@ -754,7 +754,7 @@ _OVER:
 }
 
 int32_t mndAddStbToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
   if (mndSetCreateStbRedoLogs(pMnode, pTrans, pDb, pStb) != 0) return -1;
   if (mndSetCreateStbUndoLogs(pMnode, pTrans, pDb, pStb) != 0) return -1;
   if (mndSetCreateStbCommitLogs(pMnode, pTrans, pDb, pStb) != 0) return -1;
@@ -1261,7 +1261,7 @@ static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *p
   if (pTrans == NULL) goto _OVER;
 
   mDebug("trans:%d, used to alter stb:%s", pTrans->id, pAlter->name);
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
 
   if (mndSetAlterStbRedoLogs(pMnode, pTrans, pDb, &stbObj) != 0) goto _OVER;
   if (mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, &stbObj) != 0) goto _OVER;
@@ -1407,7 +1407,7 @@ static int32_t mndDropStb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *p
   if (pTrans == NULL) goto _OVER;
 
   mDebug("trans:%d, used to drop stb:%s", pTrans->id, pStb->name);
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
 
   if (mndSetDropStbRedoLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
   if (mndSetDropStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER;

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

@@ -619,8 +619,8 @@ void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *
   pTrans->paramLen = paramLen;
 }
 
-void mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb) {
-  memcpy(pTrans->dbname, pDb->name, TSDB_DB_FNAME_LEN);
+void mndTransSetDbName(STrans *pTrans, const char *dbname) {
+  memcpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
 }
 
 void mndTransSetSerial(STrans *pTrans) { pTrans->exec = TRN_EXEC_SERIAL; }

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

@@ -51,9 +51,10 @@ int32_t mndInitVgroup(SMnode *pMnode) {
   };
 
   mndSetMsgHandle(pMnode, TDMT_DND_CREATE_VNODE_RSP, mndProcessCreateVnodeRsp);
-  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_VNODE_RSP, mndProcessAlterVnodeRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_REPLICA_RSP, mndProcessAlterVnodeRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_CONFIG_RSP, mndProcessAlterVnodeRsp);
   mndSetMsgHandle(pMnode, TDMT_DND_DROP_VNODE_RSP, mndProcessDropVnodeRsp);
-  mndSetMsgHandle(pMnode, TDMT_VND_COMPACT_VNODE_RSP, mndProcessCompactVnodeRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_COMPACT_RSP, mndProcessCompactVnodeRsp);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_VGROUP, mndRetrieveVgroups);
   mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_VGROUP, mndCancelGetNextVgroup);
@@ -188,10 +189,10 @@ void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup) {
   sdbRelease(pSdb, pVgroup);
 }
 
-void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen) {
+void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen,
+                             bool standby) {
   SCreateVnodeReq createReq = {0};
   createReq.vgId = pVgroup->vgId;
-  createReq.dnodeId = pDnode->id;
   memcpy(createReq.db, pDb->name, TSDB_DB_FNAME_LEN);
   createReq.dbUid = pDb->uid;
   createReq.vgVersion = pVgroup->version;
@@ -218,6 +219,7 @@ void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVg
   createReq.hashMethod = pDb->cfg.hashMethod;
   createReq.numOfRetensions = pDb->cfg.numOfRetensions;
   createReq.pRetensions = pDb->cfg.pRetensions;
+  createReq.standby = standby;
   createReq.isTsma = pVgroup->isTsma;
   createReq.pTsma = pVgroup->pTsma;
 
@@ -276,7 +278,6 @@ void *mndBuildAlterVnodeReq(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup, int32_
   alterReq.strict = pDb->cfg.strict;
   alterReq.cacheLastRow = pDb->cfg.cacheLastRow;
   alterReq.replica = pVgroup->replica;
-  alterReq.selfIndex = -1;
 
   for (int32_t v = 0; v < pVgroup->replica; ++v) {
     SReplica  *pReplica = &alterReq.replicas[v];
@@ -292,13 +293,6 @@ void *mndBuildAlterVnodeReq(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup, int32_
     mndReleaseDnode(pMnode, pVgidDnode);
   }
 
-#if 0
-  if (alterReq.selfIndex == -1) {
-    terrno = TSDB_CODE_MND_APP_ERROR;
-    return NULL;
-  }
-#endif
-
   int32_t contLen = tSerializeSAlterVnodeReq(NULL, 0, &alterReq);
   if (contLen < 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -510,7 +504,7 @@ _OVER:
   taosArrayDestroy(pArray);
   return code;
 }
-
+//--->
 int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
   taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
   for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
@@ -538,7 +532,7 @@ int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
 
     SVnodeGid *pVgid = &pVgroup->vnodeGid[maxPos];
     pVgid->dnodeId = pDnode->id;
-    pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
+    pVgid->role = TAOS_SYNC_STATE_ERROR;
     pDnode->numOfVnodes++;
 
     mInfo("db:%s, vgId:%d, vnode_index:%d dnode:%d is added", pVgroup->dbName, pVgroup->vgId, maxPos, pVgid->dnodeId);
@@ -549,16 +543,15 @@ int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
   terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
   return -1;
 }
-
+//--->
 int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray, SVnodeGid *del1, SVnodeGid *del2) {
-  int32_t removedNum = 0;
-
   taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
   for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
     SDnodeObj *pDnode = taosArrayGet(pArray, i);
     mDebug("dnode:%d, equivalent vnodes:%d", pDnode->id, pDnode->numOfVnodes);
   }
 
+  int32_t removedNum = 0;
   for (int32_t d = taosArrayGetSize(pArray) - 1; d >= 0; --d) {
     SDnodeObj *pDnode = taosArrayGet(pArray, d);
 
@@ -664,6 +657,7 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
   int32_t numOfRows = 0;
   SVgObj *pVgroup = NULL;
   int32_t cols = 0;
+  int64_t curMs = taosGetTimestampMs();
 
   SDbObj *pDb = NULL;
   if (strlen(pShow->db) > 0) {
@@ -703,12 +697,15 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
       if (i < pVgroup->replica) {
         colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->vnodeGid[i].dnodeId, false);
 
-        char        buf1[20] = {0};
-        SDnodeObj *pDnodeObj = mndAcquireDnode(pMnode, pVgroup->vnodeGid[i].dnodeId);
-        ASSERT(pDnodeObj != NULL);
-        bool isOffLine = !mndIsDnodeOnline(pMnode, pDnodeObj, taosGetTimestampMs());
-        const char *role = isOffLine ? "OFFLINE" : syncStr(pVgroup->vnodeGid[i].role);
+        bool       online = false;
+        SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgroup->vnodeGid[i].dnodeId);
+        if (pDnode != NULL) {
+          online = mndIsDnodeOnline(pMnode, pDnode, curMs);
+          mndReleaseDnode(pMnode, pDnode);
+        }
 
+        char        buf1[20] = {0};
+        const char *role = online ? syncStr(pVgroup->vnodeGid[i].role) : "OFFLINE";
         STR_WITH_MAXSIZE_TO_VARSTR(buf1, role, pShow->pMeta->pSchemas[cols].bytes);
 
         pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

source/dnode/mnode/impl/test/trans/trans2.cpp

@@ -128,7 +128,7 @@ class MndTestTrans2 : public ::testing::Test {
     mndTransSetCb(pTrans, TRANS_START_FUNC_TEST, TRANS_STOP_FUNC_TEST, param, strlen(param) + 1);
 
     if (pDb != NULL) {
-      mndTransSetDbInfo(pTrans, pDb);
+      mndTransSetDbName(pTrans, pDb->name);
     }
 
     int32_t code = mndTransPrepare(pMnode, pTrans);
@@ -201,7 +201,7 @@ class MndTestTrans2 : public ::testing::Test {
     }
 
     if (pDb != NULL) {
-      mndTransSetDbInfo(pTrans, pDb);
+      mndTransSetDbName(pTrans, pDb->name);
     }
 
     int32_t code = mndTransPrepare(pMnode, pTrans);

source/dnode/vnode/CMakeLists.txt

@@ -52,8 +52,9 @@ target_sources(
     "src/tq/tqMeta.c"
     "src/tq/tqRead.c"
     "src/tq/tqOffset.c"
-    #"src/tq/tqPush.c"
-    #"src/tq/tqCommit.c"
+    "src/tq/tqPush.c"
+    "src/tq/tqSink.c"
+    "src/tq/tqCommit.c"
 )
 target_include_directories(
     vnode

source/dnode/vnode/inc/vnode.h

@@ -175,6 +175,7 @@ struct SVnodeCfg {
   int8_t   isTsma;
   int8_t   isRsma;
   int8_t   hashMethod;
+  int8_t   standby;
   STsdbCfg tsdbCfg;
   SWalCfg  walCfg;
   SSyncCfg syncCfg;

source/dnode/vnode/src/inc/sma.h

@@ -219,7 +219,7 @@ static int32_t  tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SDisk
 void *tdFreeRSmaInfo(SRSmaInfo *pInfo);
 
 int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg);
-int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version);
+int32_t tdUpdateExpiredWindowImpl(SSma *pSma, const SSubmitReq *pMsg, int64_t version);
 // TODO: This is the basic params, and should wrap the params to a queryHandle.
 int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult);
 

source/dnode/vnode/src/inc/tq.h

@@ -66,34 +66,28 @@ struct STqReadHandle {
 // tqPush
 
 typedef struct {
+  STaosQueue* queue;
+  STaosQall*  qall;
+  void*       qItem;
+} STqInputQ;
+
+typedef struct {
+  // msg info
   int64_t consumerId;
+  int64_t reqOffset;
+  int64_t processedVer;
   int32_t epoch;
   int32_t skipLogNum;
-  int64_t        reqOffset;
-  SRpcHandleInfo info;
+  // rpc info
+  int64_t        reqId;
+  SRpcHandleInfo rpcInfo;
+  // exec
+  int8_t    inputStatus;
+  int8_t    execStatus;
+  STqInputQ inputQ;
   SRWLatch  lock;
 } STqPushHandle;
 
-#if 0
-typedef struct {
-  char    subKey[TSDB_SUBSCRIBE_KEY_LEN];
-  int64_t consumerId;
-  int32_t epoch;
-  int8_t  subType;
-  // int8_t        withTbName;
-  // int8_t        withSchema;
-  // int8_t        withTag;
-  char*         qmsg;
-  SHashObj*     pDropTbUid;
-  STqPushHandle pushHandle;
-  // SRWLatch        lock;
-  SWalReadHandle* pWalReader;
-  // task number should be the same with fetch thread
-  STqReadHandle* pExecReader[5];
-  qTaskInfo_t    task[5];
-} STqExec;
-#endif
-
 // tqExec
 
 typedef struct {
@@ -154,27 +148,21 @@ typedef struct {
 
 static STqMgmt tqMgmt = {0};
 
-// init once
-int  tqInit();
-void tqCleanUp();
-
-// int32_t tEncodeSTqExec(SEncoder* pEncoder, const STqExec* pExec);
-// int32_t tDecodeSTqExec(SDecoder* pDecoder, STqExec* pExec);
-
-int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle);
-int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle);
-
+// tqRead
 int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalHead** pHeadWithCkSum);
 
+// tqExec
 int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkRsp* pRsp, int32_t workerId);
 
 // tqMeta
-
 int32_t tqMetaOpen(STQ* pTq);
 int32_t tqMetaClose(STQ* pTq);
 int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle);
 int32_t tqMetaDeleteHandle(STQ* pTq, const char* key);
 
+// tqSink
+void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
+
 // tqOffset
 STqOffsetStore* STqOffsetOpen(STqOffsetCfg*);
 void            STqOffsetClose(STqOffsetStore*);

source/dnode/vnode/src/inc/vnd.h

@@ -84,6 +84,7 @@ int32_t vnodeAsyncCommit(SVnode* pVnode);
 int32_t vnodeSyncOpen(SVnode* pVnode, char* path);
 void    vnodeSyncStart(SVnode* pVnode);
 void    vnodeSyncClose(SVnode* pVnode);
+void    vnodeSyncAlter(SVnode* pVnode, SRpcMsg* pMsg);
 
 #ifdef __cplusplus
 }

source/dnode/vnode/src/inc/vnodeInt.h

@@ -125,6 +125,8 @@ int32_t      tsdbSnapshotReaderClose(STsdbSnapshotReader* pReader);
 int32_t      tsdbSnapshotRead(STsdbSnapshotReader* pReader, void** ppData, uint32_t* nData);
 
 // tq
+int     tqInit();
+void    tqCleanUp();
 STQ*    tqOpen(const char* path, SVnode* pVnode, SWal* pWal);
 void    tqClose(STQ*);
 int     tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver);
@@ -145,11 +147,11 @@ int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t smaOpen(SVnode* pVnode);
 int32_t smaClose(SSma* pSma);
 
-int32_t tdUpdateExpireWindow(SSma* pSma, SSubmitReq* pMsg, int64_t version);
+int32_t tdUpdateExpireWindow(SSma* pSma, const SSubmitReq* pMsg, int64_t version);
 int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg);
 int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg);
 
-int32_t tdProcessRSmaCreate(SVnode *pVnode, SVCreateStbReq* pReq);
+int32_t tdProcessRSmaCreate(SVnode* pVnode, SVCreateStbReq* pReq);
 int32_t tdProcessRSmaSubmit(SSma* pSma, void* pMsg, int32_t inputType);
 int32_t tdFetchTbUidList(SSma* pSma, STbUidStore** ppStore, tb_uid_t suid, tb_uid_t uid);
 int32_t tdUpdateTbUidList(SSma* pSma, STbUidStore* pUidStore);

source/dnode/vnode/src/sma/sma.c

@@ -15,7 +15,6 @@
 
 #include "sma.h"
 
-
 // TODO: Who is responsible for resource allocate and release?
 int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg) {
   int32_t code = TSDB_CODE_SUCCESS;
@@ -37,7 +36,7 @@ int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg) {
   return code;
 }
 
-int32_t tdUpdateExpireWindow(SSma* pSma, SSubmitReq* pMsg, int64_t version) {
+int32_t tdUpdateExpireWindow(SSma* pSma, const SSubmitReq* pMsg, int64_t version) {
   int32_t code = TSDB_CODE_SUCCESS;
   if ((code = tdUpdateExpiredWindowImpl(pSma, pMsg, version)) < 0) {
     smaWarn("vgId:%d update expired sma window failed since %s", SMA_VID(pSma), tstrerror(terrno));

source/dnode/vnode/src/sma/smaTimeRange.c

@@ -932,7 +932,7 @@ static int32_t tdSetExpiredWindow(SSma *pSma, SHashObj *pItemsHash, int64_t inde
  * @param msg SSubmitReq
  * @return int32_t
  */
-int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version) {
+int32_t tdUpdateExpiredWindowImpl(SSma *pSma, const SSubmitReq *pMsg, int64_t version) {
   // no time-range-sma, just return success
   if (atomic_load_16(&SMA_TSMA_NUM(pSma)) <= 0) {
     smaTrace("vgId:%d not update expire window since no tSma", SMA_VID(pSma));

source/dnode/vnode/src/tq/tq.c

@@ -81,169 +81,10 @@ void tqClose(STQ* pTq) {
   // TODO
 }
 
-int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle) {
-  if (tStartEncode(pEncoder) < 0) return -1;
-  if (tEncodeCStr(pEncoder, pHandle->subKey) < 0) return -1;
-  if (tEncodeI64(pEncoder, pHandle->consumerId) < 0) return -1;
-  if (tEncodeI32(pEncoder, pHandle->epoch) < 0) return -1;
-  if (tEncodeI8(pEncoder, pHandle->execHandle.subType) < 0) return -1;
-  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
-    if (tEncodeCStr(pEncoder, pHandle->execHandle.exec.execCol.qmsg) < 0) return -1;
-  }
-  tEndEncode(pEncoder);
-  return pEncoder->pos;
-}
-
-int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle) {
-  if (tStartDecode(pDecoder) < 0) return -1;
-  if (tDecodeCStrTo(pDecoder, pHandle->subKey) < 0) return -1;
-  if (tDecodeI64(pDecoder, &pHandle->consumerId) < 0) return -1;
-  if (tDecodeI32(pDecoder, &pHandle->epoch) < 0) return -1;
-  if (tDecodeI8(pDecoder, &pHandle->execHandle.subType) < 0) return -1;
-  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
-    if (tDecodeCStrAlloc(pDecoder, &pHandle->execHandle.exec.execCol.qmsg) < 0) return -1;
-  }
-  tEndDecode(pDecoder);
-  return 0;
-}
-
-int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
-  void* pIter = NULL;
-  while (1) {
-    pIter = taosHashIterate(pTq->handles, pIter);
-    if (pIter == NULL) break;
-    STqHandle* pExec = (STqHandle*)pIter;
-    if (pExec->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
-      for (int32_t i = 0; i < 5; i++) {
-        int32_t code = qUpdateQualifiedTableId(pExec->execHandle.exec.execCol.task[i], tbUidList, isAdd);
-        ASSERT(code == 0);
-      }
-    } else if (pExec->execHandle.subType == TOPIC_SUB_TYPE__DB) {
-      if (!isAdd) {
-        int32_t sz = taosArrayGetSize(tbUidList);
-        for (int32_t i = 0; i < sz; i++) {
-          int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i);
-          taosHashPut(pExec->execHandle.exec.execDb.pFilterOutTbUid, &tbUid, sizeof(int64_t), NULL, 0);
-        }
-      }
-    } else {
-      // tq update id
-    }
-  }
-  while (1) {
-    pIter = taosHashIterate(pTq->pStreamTasks, pIter);
-    if (pIter == NULL) break;
-    SStreamTask* pTask = (SStreamTask*)pIter;
-    if (pTask->inputType == STREAM_INPUT__DATA_SUBMIT) {
-      int32_t code = qUpdateQualifiedTableId(pTask->exec.executor, tbUidList, isAdd);
-      ASSERT(code == 0);
-    }
-  }
-  return 0;
-}
-
-int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver, SRpcHandleInfo handleInfo) {
-  if (msgType != TDMT_VND_SUBMIT) return 0;
-  void*       pIter = NULL;
-  STqHandle*  pHandle = NULL;
-  SSubmitReq* pReq = (SSubmitReq*)msg;
-  int32_t     workerId = 4;
-  int64_t     fetchOffset = ver;
-
-  while (1) {
-    pIter = taosHashIterate(pTq->pushMgr, pIter);
-    if (pIter == NULL) break;
-    pHandle = *(STqHandle**)pIter;
-
-    taosWLockLatch(&pHandle->pushHandle.lock);
-
-    SMqDataBlkRsp rsp = {0};
-    rsp.reqOffset = pHandle->pushHandle.reqOffset;
-    rsp.blockData = taosArrayInit(0, sizeof(void*));
-    rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
-
-    if (msgType == TDMT_VND_SUBMIT) {
-      tqDataExec(pTq, &pHandle->execHandle, pReq, &rsp, workerId);
-    } else {
-      // TODO
-      ASSERT(0);
-    }
-
-    if (rsp.blockNum == 0) {
-      taosWUnLockLatch(&pHandle->pushHandle.lock);
-      continue;
-    }
-
-    ASSERT(taosArrayGetSize(rsp.blockData) == rsp.blockNum);
-    ASSERT(taosArrayGetSize(rsp.blockDataLen) == rsp.blockNum);
-
-    rsp.rspOffset = fetchOffset;
-
-    int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqDataBlkRsp(NULL, &rsp);
-    void*   buf = rpcMallocCont(tlen);
-    if (buf == NULL) {
-      // todo free
-      return -1;
-    }
-
-    ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
-    ((SMqRspHead*)buf)->epoch = pHandle->pushHandle.epoch;
-    ((SMqRspHead*)buf)->consumerId = pHandle->pushHandle.consumerId;
-
-    void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
-    tEncodeSMqDataBlkRsp(&abuf, &rsp);
-
-    SRpcMsg resp = {
-        .info = pHandle->pushHandle.info,
-        .pCont = buf,
-        .contLen = tlen,
-        .code = 0,
-    };
-    tmsgSendRsp(&resp);
-
-    memset(&pHandle->pushHandle.info, 0, sizeof(SRpcHandleInfo));
-    taosWUnLockLatch(&pHandle->pushHandle.lock);
-
-    tqDebug("vg %d offset %ld from consumer %ld (epoch %d) send rsp, block num: %d, reqOffset: %ld, rspOffset: %ld",
-            TD_VID(pTq->pVnode), fetchOffset, pHandle->pushHandle.consumerId, pHandle->pushHandle.epoch, rsp.blockNum,
-            rsp.reqOffset, rsp.rspOffset);
-
-    // TODO destroy
-    taosArrayDestroy(rsp.blockData);
-    taosArrayDestroy(rsp.blockDataLen);
-  }
-
-  return 0;
-}
-
-int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
-  if (msgType == TDMT_VND_SUBMIT) {
-    if (taosHashGetSize(pTq->pStreamTasks) == 0) return 0;
-
-    if (tdUpdateExpireWindow(pTq->pVnode->pSma, msg, ver) != 0) {
-      // TODO handle sma error
-    }
-    void* data = taosMemoryMalloc(msgLen);
-    if (data == NULL) {
-      return -1;
-    }
-    memcpy(data, msg, msgLen);
-
-    tqProcessStreamTrigger(pTq, data);
-  }
-
-  return 0;
-}
-
-int tqCommit(STQ* pTq) {
-  // do nothing
-  return 0;
-}
-
 int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
   SMqPollReq* pReq = pMsg->pCont;
   int64_t     consumerId = pReq->consumerId;
-  int64_t     waitTime = pReq->timeout;
+  int64_t     timeout = pReq->timeout;
   int32_t     reqEpoch = pReq->epoch;
   int64_t     fetchOffset;
 
@@ -286,19 +127,17 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
   }
   if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
     rsp.withSchema = false;
-
     rsp.withTag = false;
   } else {
     rsp.withSchema = true;
-    rsp.blockSchema = taosArrayInit(0, sizeof(void*));
-
     rsp.withTag = false;
+    rsp.blockSchema = taosArrayInit(0, sizeof(void*));
   }
 
   while (1) {
     consumerEpoch = atomic_load_32(&pHandle->epoch);
     if (consumerEpoch > reqEpoch) {
-      tqDebug("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, found new consumer epoch %d discard req epoch %d",
+      tqWarn("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, found new consumer epoch %d, discard req epoch %d",
              consumerId, pReq->epoch, TD_VID(pTq->pVnode), fetchOffset, consumerEpoch, reqEpoch);
       break;
     }
@@ -310,27 +149,6 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
 
     SWalReadHead* pHead = &pHeadWithCkSum->head;
 
-#if 0
-      // add to pushMgr
-      taosWLockLatch(&pExec->pushHandle.lock);
-
-      pExec->pushHandle.consumerId = consumerId;
-      pExec->pushHandle.epoch = reqEpoch;
-      pExec->pushHandle.reqOffset = rsp.reqOffset;
-      pExec->pushHandle.skipLogNum = rsp.skipLogNum;
-      pExec->pushHandle.handle = pMsg;
-
-      taosWUnLockLatch(&pExec->pushHandle.lock);
-
-      // TODO add timer
-
-      // TODO: the pointer will always be valid?
-      taosHashPut(pTq->pushMgr, &consumerId, sizeof(int64_t), &pExec, sizeof(void*));
-      taosArrayDestroy(rsp.blockData);
-      taosArrayDestroy(rsp.blockDataLen);
-      return 0;
-#endif
-
     tqDebug("tmq poll: consumer %ld (epoch %d) iter log, vg %d offset %ld msgType %d", consumerId, pReq->epoch,
             TD_VID(pTq->pVnode), fetchOffset, pHead->msgType);
 
@@ -471,24 +289,6 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
   return 0;
 }
 
-void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
-  const SArray* pRes = (const SArray*)data;
-  SVnode*       pVnode = (SVnode*)vnode;
-
-  ASSERT(pTask->tbSink.pTSchema);
-  SSubmitReq* pReq = tdBlockToSubmit(pRes, pTask->tbSink.pTSchema, true, pTask->tbSink.stbUid,
-                                     pTask->tbSink.stbFullName, pVnode->config.vgId);
-  /*tPrintFixedSchemaSubmitReq(pReq, pTask->tbSink.pTSchema);*/
-  // build write msg
-  SRpcMsg msg = {
-      .msgType = TDMT_VND_SUBMIT,
-      .pCont = pReq,
-      .contLen = ntohl(pReq->length),
-  };
-
-  ASSERT(tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) == 0);
-}
-
 int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
   SStreamTask* pTask = taosMemoryCalloc(1, sizeof(SStreamTask));
   if (pTask == NULL) {
@@ -579,9 +379,11 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
         continue;
       }
 
-      streamDataSubmitRefInc(pSubmit);
-      SStreamDataSubmit* pSubmitClone = taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM);
-      memcpy(pSubmitClone, pSubmit, sizeof(SStreamDataSubmit));
+      SStreamDataSubmit* pSubmitClone = streamSubmitRefClone(pSubmit);
+      if (pSubmitClone == NULL) {
+        atomic_store_8(&pTask->inputStatus, TASK_INPUT_STATUS__FAILED);
+        continue;
+      }
       taosWriteQitem(pTask->inputQ, pSubmitClone);
 
       int8_t execStatus = atomic_load_8(&pTask->status);

source/dnode/vnode/src/tq/tqCommit.c

@@ -14,3 +14,8 @@
  */
 
 #include "tq.h"
+
+int tqCommit(STQ* pTq) {
+  // do nothing
+  return 0;
+}

source/dnode/vnode/src/tq/tqMeta.c

@@ -15,6 +15,32 @@
 #include "tdbInt.h"
 #include "tq.h"
 
+static int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle) {
+  if (tStartEncode(pEncoder) < 0) return -1;
+  if (tEncodeCStr(pEncoder, pHandle->subKey) < 0) return -1;
+  if (tEncodeI64(pEncoder, pHandle->consumerId) < 0) return -1;
+  if (tEncodeI32(pEncoder, pHandle->epoch) < 0) return -1;
+  if (tEncodeI8(pEncoder, pHandle->execHandle.subType) < 0) return -1;
+  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
+    if (tEncodeCStr(pEncoder, pHandle->execHandle.exec.execCol.qmsg) < 0) return -1;
+  }
+  tEndEncode(pEncoder);
+  return pEncoder->pos;
+}
+
+static int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle) {
+  if (tStartDecode(pDecoder) < 0) return -1;
+  if (tDecodeCStrTo(pDecoder, pHandle->subKey) < 0) return -1;
+  if (tDecodeI64(pDecoder, &pHandle->consumerId) < 0) return -1;
+  if (tDecodeI32(pDecoder, &pHandle->epoch) < 0) return -1;
+  if (tDecodeI8(pDecoder, &pHandle->execHandle.subType) < 0) return -1;
+  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
+    if (tDecodeCStrAlloc(pDecoder, &pHandle->execHandle.exec.execCol.qmsg) < 0) return -1;
+  }
+  tEndDecode(pDecoder);
+  return 0;
+}
+
 int tqExecKeyCompare(const void* pKey1, int32_t kLen1, const void* pKey2, int32_t kLen2) {
   return strcmp(pKey1, pKey2);
 }

source/dnode/vnode/src/tq/tqPush.c

@@ -14,3 +14,183 @@
  */
 
 #include "tq.h"
+
+int32_t tqExecFromInputQ(STQ* pTq, STqHandle* pHandle) {
+  // 1. guard and set status executing
+  // 2. check processedVer
+  // 2.1. if not missed, get msg from queue
+  // 2.2. if missed, scan wal
+  //
+  // 3. exec, after each success, update processed ver
+  // first run
+  // set exec status closing
+  // second run
+  // set exec status idle
+  //
+  // 4. if get result
+  // 4.1 set exec input status blocked and exec status idle
+  // 4.2 rpc send
+  // 4.3 clear rpc info
+  return 0;
+}
+
+int32_t tqOpenPushHandle(STQ* pTq, STqHandle* pHandle) {
+  memset(&pHandle->pushHandle, 0, sizeof(STqPushHandle));
+  pHandle->pushHandle.inputQ.queue = taosOpenQueue();
+  pHandle->pushHandle.inputQ.qall = taosAllocateQall();
+  if (pHandle->pushHandle.inputQ.queue == NULL || pHandle->pushHandle.inputQ.qall == NULL) {
+    if (pHandle->pushHandle.inputQ.queue) {
+      taosCloseQueue(pHandle->pushHandle.inputQ.queue);
+    }
+    if (pHandle->pushHandle.inputQ.qall) {
+      taosFreeQall(pHandle->pushHandle.inputQ.qall);
+    }
+    return -1;
+  }
+  return 0;
+}
+
+void tqPreparePush(STQ* pTq, STqHandle* pHandle, int64_t reqId, const SRpcHandleInfo* pInfo, int64_t processedVer) {
+  memcpy(&pHandle->pushHandle.rpcInfo, pInfo, sizeof(SRpcHandleInfo));
+  atomic_store_64(&pHandle->pushHandle.reqId, reqId);
+  atomic_store_64(&pHandle->pushHandle.processedVer, processedVer);
+  atomic_store_8(&pHandle->pushHandle.inputStatus, TASK_INPUT_STATUS__NORMAL);
+  // set timeout timer
+}
+
+int32_t tqEnqueue(STqHandle* pHandle, SStreamDataSubmit* pSubmit) {
+  int8_t inputStatus = atomic_load_8(&pHandle->pushHandle.inputStatus);
+  if (inputStatus == TASK_INPUT_STATUS__NORMAL) {
+    SStreamDataSubmit* pSubmitClone = streamSubmitRefClone(pSubmit);
+    if (pSubmitClone == NULL) {
+      return -1;
+    }
+    taosWriteQitem(pHandle->pushHandle.inputQ.queue, pSubmitClone);
+    return 0;
+  }
+  return -1;
+}
+
+int32_t tqSendExecReq(STQ* pTq, STqHandle* pHandle) {
+  //
+  return 0;
+}
+
+int32_t tqEnqueueAll(STQ* pTq, SSubmitReq* pReq) {
+  void*              pIter = NULL;
+  SStreamDataSubmit* pSubmit = streamDataSubmitNew(pReq);
+  if (pSubmit == NULL) {
+    return -1;
+  }
+
+  while (1) {
+    pIter = taosHashIterate(pTq->handles, pIter);
+    if (pIter == NULL) break;
+    STqHandle* pHandle = (STqHandle*)pIter;
+    if (tqEnqueue(pHandle, pSubmit) < 0) {
+      continue;
+    }
+    int8_t execStatus = atomic_load_8(&pHandle->pushHandle.execStatus);
+    if (execStatus == TASK_STATUS__IDLE || execStatus == TASK_STATUS__CLOSING) {
+      tqSendExecReq(pTq, pHandle);
+    }
+  }
+
+  streamDataSubmitRefDec(pSubmit);
+
+  return 0;
+}
+
+int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver, SRpcHandleInfo handleInfo) {
+  if (msgType != TDMT_VND_SUBMIT) return 0;
+  void*       pIter = NULL;
+  STqHandle*  pHandle = NULL;
+  SSubmitReq* pReq = (SSubmitReq*)msg;
+  int32_t     workerId = 4;
+  int64_t     fetchOffset = ver;
+
+  while (1) {
+    pIter = taosHashIterate(pTq->pushMgr, pIter);
+    if (pIter == NULL) break;
+    pHandle = *(STqHandle**)pIter;
+
+    taosWLockLatch(&pHandle->pushHandle.lock);
+
+    SMqDataBlkRsp rsp = {0};
+    rsp.reqOffset = pHandle->pushHandle.reqOffset;
+    rsp.blockData = taosArrayInit(0, sizeof(void*));
+    rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
+
+    if (msgType == TDMT_VND_SUBMIT) {
+      tqDataExec(pTq, &pHandle->execHandle, pReq, &rsp, workerId);
+    } else {
+      // TODO
+      ASSERT(0);
+    }
+
+    if (rsp.blockNum == 0) {
+      taosWUnLockLatch(&pHandle->pushHandle.lock);
+      continue;
+    }
+
+    ASSERT(taosArrayGetSize(rsp.blockData) == rsp.blockNum);
+    ASSERT(taosArrayGetSize(rsp.blockDataLen) == rsp.blockNum);
+
+    rsp.rspOffset = fetchOffset;
+
+    int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqDataBlkRsp(NULL, &rsp);
+    void*   buf = rpcMallocCont(tlen);
+    if (buf == NULL) {
+      // todo free
+      return -1;
+    }
+
+    ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
+    ((SMqRspHead*)buf)->epoch = pHandle->pushHandle.epoch;
+    ((SMqRspHead*)buf)->consumerId = pHandle->pushHandle.consumerId;
+
+    void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
+    tEncodeSMqDataBlkRsp(&abuf, &rsp);
+
+    SRpcMsg resp = {
+        .info = pHandle->pushHandle.rpcInfo,
+        .pCont = buf,
+        .contLen = tlen,
+        .code = 0,
+    };
+    tmsgSendRsp(&resp);
+
+    memset(&pHandle->pushHandle.rpcInfo, 0, sizeof(SRpcHandleInfo));
+    taosWUnLockLatch(&pHandle->pushHandle.lock);
+
+    tqDebug("vg %d offset %ld from consumer %ld (epoch %d) send rsp, block num: %d, reqOffset: %ld, rspOffset: %ld",
+            TD_VID(pTq->pVnode), fetchOffset, pHandle->pushHandle.consumerId, pHandle->pushHandle.epoch, rsp.blockNum,
+            rsp.reqOffset, rsp.rspOffset);
+
+    // TODO destroy
+    taosArrayDestroy(rsp.blockData);
+    taosArrayDestroy(rsp.blockDataLen);
+  }
+
+  return 0;
+}
+
+int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
+  if (msgType == TDMT_VND_SUBMIT) {
+    if (taosHashGetSize(pTq->pStreamTasks) == 0) return 0;
+
+    if (tdUpdateExpireWindow(pTq->pVnode->pSma, msg, ver) != 0) {
+      // TODO handle sma error
+    }
+    void* data = taosMemoryMalloc(msgLen);
+    if (data == NULL) {
+      return -1;
+    }
+    memcpy(data, msg, msgLen);
+
+    tqProcessStreamTrigger(pTq, data);
+  }
+
+  return 0;
+}
+

source/dnode/vnode/src/tq/tqRead.c

@@ -298,3 +298,38 @@ int tqReadHandleRemoveTbUidList(STqReadHandle* pHandle, const SArray* tbUidList)
 
   return 0;
 }
+
+int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
+  void* pIter = NULL;
+  while (1) {
+    pIter = taosHashIterate(pTq->handles, pIter);
+    if (pIter == NULL) break;
+    STqHandle* pExec = (STqHandle*)pIter;
+    if (pExec->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
+      for (int32_t i = 0; i < 5; i++) {
+        int32_t code = qUpdateQualifiedTableId(pExec->execHandle.exec.execCol.task[i], tbUidList, isAdd);
+        ASSERT(code == 0);
+      }
+    } else if (pExec->execHandle.subType == TOPIC_SUB_TYPE__DB) {
+      if (!isAdd) {
+        int32_t sz = taosArrayGetSize(tbUidList);
+        for (int32_t i = 0; i < sz; i++) {
+          int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i);
+          taosHashPut(pExec->execHandle.exec.execDb.pFilterOutTbUid, &tbUid, sizeof(int64_t), NULL, 0);
+        }
+      }
+    } else {
+      // tq update id
+    }
+  }
+  while (1) {
+    pIter = taosHashIterate(pTq->pStreamTasks, pIter);
+    if (pIter == NULL) break;
+    SStreamTask* pTask = (SStreamTask*)pIter;
+    if (pTask->inputType == STREAM_INPUT__DATA_SUBMIT) {
+      int32_t code = qUpdateQualifiedTableId(pTask->exec.executor, tbUidList, isAdd);
+      ASSERT(code == 0);
+    }
+  }
+  return 0;
+}

source/dnode/vnode/src/tq/tqSink.c

+/*
+ * 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 "tq.h"
+
+void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
+  const SArray* pRes = (const SArray*)data;
+  SVnode*       pVnode = (SVnode*)vnode;
+
+  ASSERT(pTask->tbSink.pTSchema);
+  SSubmitReq* pReq = tdBlockToSubmit(pRes, pTask->tbSink.pTSchema, true, pTask->tbSink.stbUid,
+                                     pTask->tbSink.stbFullName, pVnode->config.vgId);
+  /*tPrintFixedSchemaSubmitReq(pReq, pTask->tbSink.pTSchema);*/
+  // build write msg
+  SRpcMsg msg = {
+      .msgType = TDMT_VND_SUBMIT,
+      .pCont = pReq,
+      .contLen = ntohl(pReq->length),
+  };
+
+  ASSERT(tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) == 0);
+}

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -22,7 +22,7 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
 static int vnodeProcessAlterTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
 static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
 static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
-static int vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp);
+static int vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
 
 int32_t vnodePreprocessReq(SVnode *pVnode, SRpcMsg *pMsg) {
   SDecoder dc = {0};
@@ -88,6 +88,9 @@ int32_t vnodePreprocessReq(SVnode *pVnode, SRpcMsg *pMsg) {
       }
 
     } break;
+    case TDMT_VND_ALTER_REPLICA: {
+      vnodeSyncAlter(pVnode, pMsg);
+    } break;
     default:
       break;
   }
@@ -154,7 +157,7 @@ int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg
                               pMsg->contLen - sizeof(SMsgHead)) < 0) {
       }
     } break;
-    case TDMT_VND_ALTER_VNODE:
+    case TDMT_VND_ALTER_CONFIG:
       break;
     default:
       ASSERT(0);
@@ -783,7 +786,7 @@ _exit:
   return 0;
 }
 
-static int vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp) {
+static int vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
   SVCreateTSmaReq req = {0};
   SDecoder        coder;
 

source/dnode/vnode/src/vnd/vnodeSync.c

@@ -27,6 +27,7 @@ static int32_t   vnodeSyncGetSnapshot(SSyncFSM *pFsm, SSnapshot *pSnapshot);
 int32_t vnodeSyncOpen(SVnode *pVnode, char *path) {
   SSyncInfo syncInfo = {
       .vgId = pVnode->config.vgId,
+      .isStandBy = pVnode->config.standby,
       .syncCfg = pVnode->config.syncCfg,
       .pWal = pVnode->pWal,
       .msgcb = NULL,
@@ -49,9 +50,38 @@ int32_t vnodeSyncOpen(SVnode *pVnode, char *path) {
   return 0;
 }
 
+void vnodeSyncAlter(SVnode *pVnode, SRpcMsg *pMsg) {
+  SAlterVnodeReq req = {0};
+  if (tDeserializeSAlterVnodeReq((char *)pMsg->pCont + sizeof(SMsgHead), pMsg->contLen - sizeof(SMsgHead), &req) != 0) {
+    terrno = TSDB_CODE_INVALID_MSG;
+    vError("vgId:%d, failed to alter replica since %s", TD_VID(pVnode), terrstr());
+    SRpcMsg rsp = {.info = pMsg->info, .code = terrno};
+    tmsgSendRsp(&rsp);
+  }
+
+  vInfo("vgId:%d, start to alter vnode replica to %d", TD_VID(pVnode), req.replica);
+  SSyncCfg cfg = {.replicaNum = req.replica, .myIndex = req.selfIndex};
+  for (int32_t r = 0; r < req.replica; ++r) {
+    SNodeInfo *pNode = &cfg.nodeInfo[r];
+    tstrncpy(pNode->nodeFqdn, req.replicas[r].fqdn, sizeof(pNode->nodeFqdn));
+    pNode->nodePort = req.replicas[r].port;
+    vInfo("vgId:%d, replica:%d %s:%u", TD_VID(pVnode), r, pNode->nodeFqdn, pNode->nodePort);
+  }
+
+  if (syncReconfig(pVnode->sync, &cfg) != 0) {
+    vError("vgId:%d, failed to propose sync reconfig since %s", TD_VID(pVnode), terrstr());
+    SRpcMsg rsp = {.info = pMsg->info, .code = terrno};
+    tmsgSendRsp(&rsp);
+  }
+}
+
 void vnodeSyncStart(SVnode *pVnode) {
   syncSetMsgCb(pVnode->sync, &pVnode->msgCb);
+  if (pVnode->config.standby) {
+    syncStartStandBy(pVnode->sync);
+  } else {
     syncStart(pVnode->sync);
+  }
 }
 
 void vnodeSyncClose(SVnode *pVnode) { syncStop(pVnode->sync); }
@@ -60,17 +90,32 @@ int32_t vnodeSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
   int32_t code = tmsgPutToQueue(msgcb, SYNC_QUEUE, pMsg);
   if (code != 0) {
     rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
   }
   return code;
 }
 
-int32_t vnodeSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) { return tmsgSendReq(pEpSet, pMsg); }
+int32_t vnodeSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) {
+  int32_t code = tmsgSendReq(pEpSet, pMsg);
+  if (code != 0) {
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
+  }
+  return code;
+}
 
 int32_t vnodeSyncGetSnapshot(SSyncFSM *pFsm, SSnapshot *pSnapshot) {
   vnodeGetSnapshot(pFsm->data, pSnapshot);
   return 0;
 }
 
+void vnodeSyncReconfig(struct SSyncFSM *pFsm, SSyncCfg newCfg, SReConfigCbMeta cbMeta) {
+  SVnode *pVnode = pFsm->data;
+  vInfo("vgId:%d, sync reconfig is confirmed", TD_VID(pVnode));
+
+  // todo rpc response here
+}
+
 void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
   SyncIndex beginIndex = SYNC_INDEX_INVALID;
   if (pFsm->FpGetSnapshot != NULL) {
@@ -87,20 +132,12 @@ void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta)
         pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), beginIndex);
     syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg);
 
-    SVnode       *pVnode = (SVnode *)(pFsm->data);
+    SVnode       *pVnode = pFsm->data;
     SyncApplyMsg *pSyncApplyMsg = syncApplyMsgBuild2(pMsg, pVnode->config.vgId, &cbMeta);
     SRpcMsg       applyMsg;
     syncApplyMsg2RpcMsg(pSyncApplyMsg, &applyMsg);
     syncApplyMsgDestroy(pSyncApplyMsg);
 
-    /*
-        SRpcMsg applyMsg;
-        applyMsg = *pMsg;
-        applyMsg.pCont = rpcMallocCont(applyMsg.contLen);
-        assert(applyMsg.contLen == pMsg->contLen);
-        memcpy(applyMsg.pCont, pMsg->pCont, applyMsg.contLen);
-    */
-
     // recover handle for response
     SRpcMsg saveRpcMsg;
     int32_t ret = syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &saveRpcMsg);
@@ -142,14 +179,13 @@ void vnodeSyncRollBackMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta
 
 SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode) {
   SSyncFSM *pFsm = taosMemoryCalloc(1, sizeof(SSyncFSM));
-  memset(pFsm, 0, sizeof(*pFsm));
   pFsm->data = pVnode;
   pFsm->FpCommitCb = vnodeSyncCommitMsg;
   pFsm->FpPreCommitCb = vnodeSyncPreCommitMsg;
   pFsm->FpRollBackCb = vnodeSyncRollBackMsg;
   pFsm->FpGetSnapshot = vnodeSyncGetSnapshot;
   pFsm->FpRestoreFinishCb = NULL;
-  pFsm->FpReConfigCb = NULL;
+  pFsm->FpReConfigCb = vnodeSyncReconfig;
 
   return pFsm;
 }
\ No newline at end of file

source/libs/scalar/test/scalar/CMakeLists.txt

@@ -17,9 +17,7 @@ TARGET_INCLUDE_DIRECTORIES(
         PUBLIC "${TD_SOURCE_DIR}/source/libs/parser/inc"
         PRIVATE "${TD_SOURCE_DIR}/source/libs/scalar/inc"
 )
-if(NOT TD_WINDOWS)
-        add_test(
+add_test(
         NAME scalarTest
         COMMAND scalarTest
-        )
-endif(NOT TD_WINDOWS)
+)

source/libs/stream/src/tstream.c

@@ -35,6 +35,16 @@ void* streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput) {
   return (void*)buf;
 }
 
+SStreamDataSubmit* streamSubmitRefClone(SStreamDataSubmit* pSubmit) {
+  SStreamDataSubmit* pSubmitClone = taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM);
+  if (pSubmitClone == NULL) {
+    return NULL;
+  }
+  streamDataSubmitRefInc(pSubmit);
+  memcpy(pSubmitClone, pSubmit, sizeof(SStreamDataSubmit));
+  return pSubmitClone;
+}
+
 static int32_t streamBuildDispatchMsg(SStreamTask* pTask, SArray* data, SRpcMsg* pMsg, SEpSet** ppEpSet) {
   SStreamDispatchReq req = {
       .streamId = pTask->streamId,
@@ -207,7 +217,6 @@ int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb) {
   if (pRes == NULL) return -1;
   while (1) {
     int8_t execStatus = atomic_val_compare_exchange_8(&pTask->status, TASK_STATUS__IDLE, TASK_STATUS__EXECUTING);
-    void*  exec = pTask->exec.executor;
     if (execStatus == TASK_STATUS__IDLE) {
       // first run, from qall, handle failure from last exec
       pRes = streamExecForQall(pTask, pRes);

tests/pytest/util/dnodes.py

@@ -397,6 +397,7 @@ class TDDnode:
     def stop(self):
         if (not self.remoteIP == ""):
             self.remoteExec(self.cfgDict, "tdDnodes.stop(%d)"%self.index)
+            tdLog.info("stop dnode%d"%self.index)
             return
         if self.valgrind == 0:
             toBeKilled = "taosd"

tests/script/tsim/db/alter_replica_13.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/deploy.sh -n dnode2 -i 2
+system sh/deploy.sh -n dnode3 -i 3
+system sh/deploy.sh -n dnode4 -i 4
+system sh/exec.sh -n dnode1 -s start
+system sh/exec.sh -n dnode2 -s start
+system sh/exec.sh -n dnode3 -s start
+system sh/exec.sh -n dnode4 -s start
+sql connect
+
+print =============== step1: create dnodes
+sql create dnode $hostname port 7200
+
+$loop_cnt = 0
+step1:
+	$loop_cnt = $loop_cnt + 1
+	sleep 1000
+	if $loop_cnt == 10 then
+	  print ====> dnode not ready!
+		return -1
+	endi
+sql show dnodes
+print ===> $data00 $data01 $data02 $data03 $data04 $data05
+print ===> $data10 $data11 $data12 $data13 $data14 $data15
+if $rows != 2 then
+  return -1
+endi
+if $data(1)[4] != ready then
+  goto step1
+endi
+if $data(2)[4] != ready then
+  goto step1
+endi
+
+print =============== step2: create database
+sql create database db vgroups 1
+sql show databases
+if $rows != 3 then 
+  return -1
+endi
+if $data(db)[4] != 1 then 
+  return -1
+endi
+
+sql show dnodes
+if $data(2)[2] != 1 then
+  return -1
+endi
+
+# vnodes
+sql show dnodes
+if $data(2)[2] != 1 then
+  return -1
+endi
+
+# v1_dnode
+sql show db.vgroups
+if $data(2)[3] != 2 then
+  return -1
+endi
+
+sql_error alter database db replica 3
+sql create table db.stb (ts timestamp, c1 int, c2 binary(4)) tags(t1 int, t2 binary(16)) comment "abd"
+sql create table db.ctb using db.stb tags(101, "102")
+sql insert into db.ctb values(now, 1, "2")
+sql select * from db.stb
+if $rows != 1 then
+  return -1
+endi
+
+print =============== step3: create dnodes
+sql create dnode $hostname port 7300
+sql create dnode $hostname port 7400
+
+$loop_cnt = 0
+step3:
+	$loop_cnt = $loop_cnt + 1
+	sleep 1000
+	if $loop_cnt == 10 then
+	  print ====> dnode not ready!
+		return -1
+	endi
+sql show dnodes
+print ===> rows: $rows 
+print ===> $data00 $data01 $data02 $data03 $data04 $data05
+print ===> $data10 $data11 $data12 $data13 $data14 $data15
+print ===> $data20 $data21 $data22 $data23 $data24 $data25
+print ===> $data30 $data31 $data32 $data33 $data24 $data35
+if $rows != 4 then
+  return -1
+endi
+if $data(1)[4] != ready then
+  goto step3
+endi
+if $data(2)[4] != ready then
+  goto step3
+endi
+if $data(3)[4] != ready then
+  goto step3
+endi
+if $data(4)[4] != ready then
+  goto step3
+endi
+
+return
+print ============= step4: alter database
+sql alter database db replica 3
+if $rows != 3 then 
+  return -1
+endi
+if $data(db)[4] != 3 then 
+  return -1
+endi
+
+sql select * from db.stb
+if $rows != 1 then
+  return -1
+endi
+
+
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
+system sh/exec.sh -n dnode2 -s stop -x SIGINT
+system sh/exec.sh -n dnode3 -s stop -x SIGINT

tests/system-test/0-others/taosShell.py

@@ -84,6 +84,12 @@ class TDTestCase:
     #updatecfgDict = {'clientCfg': {'serverPort': 7080, 'firstEp': 'trd02:7080', 'secondEp':'trd02:7080'},\
     #                 'serverPort': 7080, 'firstEp': 'trd02:7080'}
     hostname = socket.gethostname()
+    if (platform.system().lower() == 'windows' and not tdDnodes.dnodes[0].remoteIP == ""):
+        try:
+            config = eval(tdDnodes.dnodes[0].remoteIP)
+            hostname = config["host"]
+        except Exception:
+            hostname = tdDnodes.dnodes[0].remoteIP
     serverPort = '7080'
     rpcDebugFlagVal = '143'
     clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}

tests/system-test/0-others/taosShellError.py

@@ -86,6 +86,12 @@ class TDTestCase:
     #updatecfgDict = {'clientCfg': {'serverPort': 7080, 'firstEp': 'trd02:7080', 'secondEp':'trd02:7080'},\
     #                 'serverPort': 7080, 'firstEp': 'trd02:7080'}
     hostname = socket.gethostname()
+    if (platform.system().lower() == 'windows' and not tdDnodes.dnodes[0].remoteIP == ""):
+        try:
+            config = eval(tdDnodes.dnodes[0].remoteIP)
+            hostname = config["host"]
+        except Exception:
+            hostname = tdDnodes.dnodes[0].remoteIP
     serverPort = '7080'
     rpcDebugFlagVal = '143'
     clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}

tests/system-test/0-others/taosShellNetChk.py

@@ -86,6 +86,12 @@ class TDTestCase:
     #updatecfgDict = {'clientCfg': {'serverPort': 7080, 'firstEp': 'trd02:7080', 'secondEp':'trd02:7080'},\
     #                 'serverPort': 7080, 'firstEp': 'trd02:7080'}
     hostname = socket.gethostname()
+    if (platform.system().lower() == 'windows' and not tdDnodes.dnodes[0].remoteIP == ""):
+        try:
+            config = eval(tdDnodes.dnodes[0].remoteIP )
+            hostname = config["host"]
+        except Exception:
+            hostname = tdDnodes.dnodes[0].remoteIP
     serverPort = '7080'
     rpcDebugFlagVal = '143'
     clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
@@ -196,7 +202,7 @@ class TDTestCase:
         pktNum = '10'
         role   = 'client'
         if platform.system().lower() == 'windows':
-            taosCmd = buildPath + '\\build\\bin\\taos.exe -c ' + keyDict['c']
+            taosCmd = buildPath + '\\build\\bin\\taos.exe -h 127.0.0.1 -c ' + keyDict['c']
             taosCmd = taosCmd.replace('\\','\\\\')
         else:
             taosCmd = buildPath + '/build/bin/taos -c ' + keyDict['c']