From b21e6034813c3bc6678212921caee5a61c77d345 Mon Sep 17 00:00:00 2001
From: Benguang Zhao <bgzhao@taosdata.com>
Date: Fri, 14 Jul 2023 18:24:53 +0800
Subject: [PATCH 01/13] enh: add the field diskPrimary to vnodeGetPrimaryDir,
 and entries of vnodes.json

---
 source/dnode/mgmt/mgmt_vnode/inc/vmInt.h    |  3 ++
 source/dnode/mgmt/mgmt_vnode/src/vmFile.c   |  3 ++
 source/dnode/mgmt/mgmt_vnode/src/vmHandle.c | 24 ++++++++-----
 source/dnode/mgmt/mgmt_vnode/src/vmInt.c    | 13 ++++++--
 source/dnode/vnode/inc/vnode.h              | 12 ++++---
 source/dnode/vnode/src/inc/vnd.h            |  2 +-
 source/dnode/vnode/src/inc/vnodeInt.h       |  1 +
 source/dnode/vnode/src/meta/metaOpen.c      |  2 +-
 source/dnode/vnode/src/sma/smaUtil.c        |  2 +-
 source/dnode/vnode/src/tsdb/tsdbCache.c     |  2 +-
 source/dnode/vnode/src/tsdb/tsdbFS.c        |  4 +--
 source/dnode/vnode/src/tsdb/tsdbFile.c      |  3 +-
 source/dnode/vnode/src/vnd/vnodeCommit.c    |  8 ++---
 source/dnode/vnode/src/vnd/vnodeOpen.c      | 37 ++++++++++++---------
 source/dnode/vnode/src/vnd/vnodeRetention.c |  4 +--
 source/dnode/vnode/src/vnd/vnodeSnapshot.c  | 10 +++---
 16 files changed, 81 insertions(+), 49 deletions(-)

diff --git a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
index d1dc872f4b..5d08320fab 100644
--- a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
+++ b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
@@ -46,6 +46,7 @@ typedef struct {
   int32_t vgId;
   int32_t vgVersion;
   int8_t  dropped;
+  int32_t diskPrimary;
   int32_t toVgId;
   char    path[PATH_MAX + 20];
 } SWrapperCfg;
@@ -56,6 +57,7 @@ typedef struct {
   int32_t       refCount;
   int8_t        dropped;
   int8_t        disable;
+  int32_t       diskPrimary;
   int32_t       toVgId;
   char         *path;
   SVnode       *pImpl;
@@ -81,6 +83,7 @@ typedef struct {
 } SVnodeThread;
 
 // vmInt.c
+int32_t    vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId);
 SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId);
 void       vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode);
 int32_t    vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl);
diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmFile.c b/source/dnode/mgmt/mgmt_vnode/src/vmFile.c
index 769b274f7f..da7f4d4a56 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmFile.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmFile.c
@@ -71,6 +71,8 @@ static int32_t vmDecodeVnodeList(SJson *pJson, SVnodeMgmt *pMgmt, SWrapperCfg **
     if (code < 0) goto _OVER;
     tjsonGetInt32ValueFromDouble(vnode, "vgVersion", pCfg->vgVersion, code);
     if (code < 0) goto _OVER;
+    tjsonGetInt32ValueFromDouble(vnode, "diskPrimary", pCfg->diskPrimary, code);
+    if (code < 0) goto _OVER;
     tjsonGetInt32ValueFromDouble(vnode, "toVgId", pCfg->toVgId, code);
     if (code < 0) goto _OVER;
 
@@ -167,6 +169,7 @@ static int32_t vmEncodeVnodeList(SJson *pJson, SVnodeObj **ppVnodes, int32_t num
     if (tjsonAddDoubleToObject(vnode, "vgId", pVnode->vgId) < 0) return -1;
     if (tjsonAddDoubleToObject(vnode, "dropped", pVnode->dropped) < 0) return -1;
     if (tjsonAddDoubleToObject(vnode, "vgVersion", pVnode->vgVersion) < 0) return -1;
+    if (tjsonAddDoubleToObject(vnode, "diskPrimary", pVnode->diskPrimary) < 0) return -1;
     if (pVnode->toVgId && tjsonAddDoubleToObject(vnode, "toVgId", pVnode->toVgId) < 0) return -1;
     if (tjsonAddItemToArray(vnodes, vnode) < 0) return -1;
   }
diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
index 7e9f1f795f..bd39ff6a73 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
@@ -263,16 +263,19 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     return 0;
   }
 
+  int32_t diskPrimary = vmAllocPrimaryDisk(pMgmt, vnodeCfg.vgId);
+  wrapperCfg.diskPrimary = diskPrimary;
+
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId);
 
-  if (vnodeCreate(path, &vnodeCfg, pMgmt->pTfs) < 0) {
+  if (vnodeCreate(path, &vnodeCfg, diskPrimary, pMgmt->pTfs) < 0) {
     tFreeSCreateVnodeReq(&req);
     dError("vgId:%d, failed to create vnode since %s", req.vgId, terrstr());
     code = terrno;
     goto _OVER;
   }
 
-  SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
+  SVnode *pImpl = vnodeOpen(path, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to open vnode since %s", req.vgId, terrstr());
     code = terrno;
@@ -399,6 +402,7 @@ int32_t vmProcessAlterVnodeTypeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   }
 
   dInfo("vgId:%d, start to close vnode", vgId);
+  int32_t     diskPrimary = pVnode->diskPrimary;
   SWrapperCfg wrapperCfg = {
       .dropped = pVnode->dropped,
       .vgId = pVnode->vgId,
@@ -411,13 +415,13 @@ int32_t vmProcessAlterVnodeTypeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vgId);
 
   dInfo("vgId:%d, start to alter vnode replica at %s", vgId, path);
-  if (vnodeAlterReplica(path, &req, pMgmt->pTfs) < 0) {
+  if (vnodeAlterReplica(path, &req, diskPrimary, pMgmt->pTfs) < 0) {
     dError("vgId:%d, failed to alter vnode at %s since %s", vgId, path, terrstr());
     return -1;
   }
 
   dInfo("vgId:%d, begin to open vnode", vgId);
-  SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
+  SVnode *pImpl = vnodeOpen(path, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to open vnode at %s since %s", vgId, path, terrstr());
     return -1;
@@ -486,10 +490,12 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     return -1;
   }
 
+  int32_t     diskPrimary = pVnode->diskPrimary;
   SWrapperCfg wrapperCfg = {
       .dropped = pVnode->dropped,
       .vgId = dstVgId,
       .vgVersion = pVnode->vgVersion,
+      .diskPrimary = pVnode->diskPrimary,
   };
   tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
 
@@ -509,13 +515,13 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   snprintf(dstPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, dstVgId);
 
   dInfo("vgId:%d, alter vnode hashrange at %s", srcVgId, srcPath);
-  if (vnodeAlterHashRange(srcPath, dstPath, &req, pMgmt->pTfs) < 0) {
+  if (vnodeAlterHashRange(srcPath, dstPath, &req, diskPrimary, pMgmt->pTfs) < 0) {
     dError("vgId:%d, failed to alter vnode hashrange since %s", srcVgId, terrstr());
     return -1;
   }
 
   dInfo("vgId:%d, open vnode", dstVgId);
-  SVnode *pImpl = vnodeOpen(dstPath, pMgmt->pTfs, pMgmt->msgCb);
+  SVnode *pImpl = vnodeOpen(dstPath, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to open vnode at %s since %s", dstVgId, dstPath, terrstr());
     return -1;
@@ -598,10 +604,12 @@ int32_t vmProcessAlterVnodeReplicaReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   }
 
   dInfo("vgId:%d, start to close vnode", vgId);
+  int32_t     diskPrimary = pVnode->diskPrimary;
   SWrapperCfg wrapperCfg = {
       .dropped = pVnode->dropped,
       .vgId = pVnode->vgId,
       .vgVersion = pVnode->vgVersion,
+      .diskPrimary = pVnode->diskPrimary,
   };
   tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
   vmCloseVnode(pMgmt, pVnode, false);
@@ -610,13 +618,13 @@ int32_t vmProcessAlterVnodeReplicaReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vgId);
 
   dInfo("vgId:%d, start to alter vnode replica at %s", vgId, path);
-  if (vnodeAlterReplica(path, &alterReq, pMgmt->pTfs) < 0) {
+  if (vnodeAlterReplica(path, &alterReq, diskPrimary, pMgmt->pTfs) < 0) {
     dError("vgId:%d, failed to alter vnode at %s since %s", vgId, path, terrstr());
     return -1;
   }
 
   dInfo("vgId:%d, begin to open vnode", vgId);
-  SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
+  SVnode *pImpl = vnodeOpen(path, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to open vnode at %s since %s", vgId, path, terrstr());
     return -1;
diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
index db46ce3ca0..2a144dd23b 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
@@ -17,6 +17,12 @@
 #include "vmInt.h"
 #include "vnd.h"
 
+int32_t vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId) {
+  // search fs
+  // alloc
+  return 0;
+}
+
 SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
   SVnodeObj *pVnode = NULL;
 
@@ -52,6 +58,7 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
 
   pVnode->vgId = pCfg->vgId;
   pVnode->vgVersion = pCfg->vgVersion;
+  pVnode->diskPrimary = pCfg->diskPrimary;
   pVnode->refCount = 0;
   pVnode->dropped = 0;
   pVnode->path = taosStrdup(pCfg->path);
@@ -169,7 +176,8 @@ static int32_t vmRestoreVgroupId(SWrapperCfg *pCfg, STfs *pTfs) {
   snprintf(srcPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, srcVgId);
   snprintf(dstPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, dstVgId);
 
-  int32_t vgId = vnodeRestoreVgroupId(srcPath, dstPath, srcVgId, dstVgId, pTfs);
+  int32_t diskPrimary = pCfg->diskPrimary;
+  int32_t vgId = vnodeRestoreVgroupId(srcPath, dstPath, srcVgId, dstVgId, diskPrimary, pTfs);
   if (vgId <= 0) {
     dError("vgId:%d, failed to restore vgroup id. srcPath: %s", pCfg->vgId, srcPath);
     return -1;
@@ -205,9 +213,10 @@ static void *vmOpenVnodeInThread(void *param) {
       pThread->updateVnodesList = true;
     }
 
+    int32_t diskPrimary = pCfg->diskPrimary;
     snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, pCfg->vgId);
 
-    SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
+    SVnode *pImpl = vnodeOpen(path, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
     if (pImpl == NULL) {
       dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);
       pThread->failed++;
diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h
index 0b7820c030..38216e1414 100644
--- a/source/dnode/vnode/inc/vnode.h
+++ b/source/dnode/vnode/inc/vnode.h
@@ -51,12 +51,14 @@ extern const SVnodeCfg vnodeCfgDefault;
 
 int32_t vnodeInit(int32_t nthreads);
 void    vnodeCleanup();
-int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs);
-int32_t vnodeAlterReplica(const char *path, SAlterVnodeReplicaReq *pReq, STfs *pTfs);
-int32_t vnodeAlterHashRange(const char *srcPath, const char *dstPath, SAlterVnodeHashRangeReq *pReq, STfs *pTfs);
-int32_t vnodeRestoreVgroupId(const char *srcPath, const char *dstPath, int32_t srcVgId, int32_t dstVgId, STfs *pTfs);
+int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, int32_t diskPrimary, STfs *pTfs);
+int32_t vnodeAlterReplica(const char *path, SAlterVnodeReplicaReq *pReq, int32_t diskPrimary, STfs *pTfs);
+int32_t vnodeAlterHashRange(const char *srcPath, const char *dstPath, SAlterVnodeHashRangeReq *pReq,
+                            int32_t diskPrimary, STfs *pTfs);
+int32_t vnodeRestoreVgroupId(const char *srcPath, const char *dstPath, int32_t srcVgId, int32_t dstVgId,
+                             int32_t diskPrimary, STfs *pTfs);
 void    vnodeDestroy(const char *path, STfs *pTfs);
-SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb);
+SVnode *vnodeOpen(const char *path, int32_t diskPrimary, STfs *pTfs, SMsgCb msgCb);
 void    vnodePreClose(SVnode *pVnode);
 void    vnodePostClose(SVnode *pVnode);
 void    vnodeSyncCheckTimeout(SVnode *pVnode);
diff --git a/source/dnode/vnode/src/inc/vnd.h b/source/dnode/vnode/src/inc/vnd.h
index ac352e764c..5e7d522fb9 100644
--- a/source/dnode/vnode/src/inc/vnd.h
+++ b/source/dnode/vnode/src/inc/vnd.h
@@ -87,7 +87,7 @@ void    vnodeBufPoolAddToFreeList(SVBufPool* pPool);
 int32_t vnodeBufPoolRecycle(SVBufPool* pPool);
 
 // vnodeOpen.c
-int32_t vnodeGetPrimaryDir(const char* relPath, STfs* pTfs, char* buf, size_t bufLen);
+int32_t vnodeGetPrimaryDir(const char* relPath, int32_t diskPrimary, STfs* pTfs, char* buf, size_t bufLen);
 
 // vnodeQuery.c
 int32_t vnodeQueryOpen(SVnode* pVnode);
diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h
index 25648d61fa..f79e4650ce 100644
--- a/source/dnode/vnode/src/inc/vnodeInt.h
+++ b/source/dnode/vnode/src/inc/vnodeInt.h
@@ -385,6 +385,7 @@ struct SVnode {
   SVState   state;
   SVStatis  statis;
   STfs*     pTfs;
+  int32_t   diskPrimary;
   SMsgCb    msgCb;
 
   // Buffer Pool
diff --git a/source/dnode/vnode/src/meta/metaOpen.c b/source/dnode/vnode/src/meta/metaOpen.c
index 9d4fdf8b11..517d9692c7 100644
--- a/source/dnode/vnode/src/meta/metaOpen.c
+++ b/source/dnode/vnode/src/meta/metaOpen.c
@@ -41,7 +41,7 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) {
   *ppMeta = NULL;
 
   // create handle
-  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, path, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, path, TSDB_FILENAME_LEN);
   offset = strlen(path);
   snprintf(path + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VNODE_META_DIR);
 
diff --git a/source/dnode/vnode/src/sma/smaUtil.c b/source/dnode/vnode/src/sma/smaUtil.c
index bd06e21053..e45cbac329 100644
--- a/source/dnode/vnode/src/sma/smaUtil.c
+++ b/source/dnode/vnode/src/sma/smaUtil.c
@@ -26,7 +26,7 @@ void tdRSmaGetDirName(SVnode *pVnode, STfs *pTfs, bool endWithSep, char *outputN
   int32_t offset = 0;
 
   // vnode
-  vnodeGetPrimaryDir(pVnode->path, pTfs, outputName, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pTfs, outputName, TSDB_FILENAME_LEN);
   offset = strlen(outputName);
 
   // rsma
diff --git a/source/dnode/vnode/src/tsdb/tsdbCache.c b/source/dnode/vnode/src/tsdb/tsdbCache.c
index 52ad923fca..d0986b25f0 100644
--- a/source/dnode/vnode/src/tsdb/tsdbCache.c
+++ b/source/dnode/vnode/src/tsdb/tsdbCache.c
@@ -59,7 +59,7 @@ typedef struct {
 
 static void tsdbGetRocksPath(STsdb *pTsdb, char *path) {
   SVnode *pVnode = pTsdb->pVnode;
-  vnodeGetPrimaryDir(pTsdb->path, pVnode->pTfs, path, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pTsdb->path, pVnode->diskPrimary, pVnode->pTfs, path, TSDB_FILENAME_LEN);
 
   int32_t offset = strlen(path);
   snprintf(path + offset, TSDB_FILENAME_LEN - offset - 1, "%scache.rdb", TD_DIRSEP);
diff --git a/source/dnode/vnode/src/tsdb/tsdbFS.c b/source/dnode/vnode/src/tsdb/tsdbFS.c
index 1b4461c07b..41fdd05741 100644
--- a/source/dnode/vnode/src/tsdb/tsdbFS.c
+++ b/source/dnode/vnode/src/tsdb/tsdbFS.c
@@ -276,14 +276,14 @@ static void tsdbGetCurrentFName(STsdb *pTsdb, char *current, char *current_t) {
 
   // CURRENT
   if (current) {
-    vnodeGetPrimaryDir(pTsdb->path, pVnode->pTfs, current, TSDB_FILENAME_LEN);
+    vnodeGetPrimaryDir(pTsdb->path, pVnode->diskPrimary, pVnode->pTfs, current, TSDB_FILENAME_LEN);
     offset = strlen(current);
     snprintf(current + offset, TSDB_FILENAME_LEN - offset - 1, "%sCURRENT", TD_DIRSEP);
   }
 
   // CURRENT.t
   if (current_t) {
-    vnodeGetPrimaryDir(pTsdb->path, pVnode->pTfs, current_t, TSDB_FILENAME_LEN);
+    vnodeGetPrimaryDir(pTsdb->path, pVnode->diskPrimary, pVnode->pTfs, current_t, TSDB_FILENAME_LEN);
     offset = strlen(current_t);
     snprintf(current_t + offset, TSDB_FILENAME_LEN - offset - 1, "%sCURRENT.t", TD_DIRSEP);
   }
diff --git a/source/dnode/vnode/src/tsdb/tsdbFile.c b/source/dnode/vnode/src/tsdb/tsdbFile.c
index 8577a42417..9ff4b28779 100644
--- a/source/dnode/vnode/src/tsdb/tsdbFile.c
+++ b/source/dnode/vnode/src/tsdb/tsdbFile.c
@@ -284,8 +284,9 @@ int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet) {
 // SDelFile ===============================================
 void tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile, char fname[]) {
   int32_t offset = 0;
+  SVnode *pVnode = pTsdb->pVnode;
 
-  vnodeGetPrimaryDir(pTsdb->path, pTsdb->pVnode->pTfs, fname, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pTsdb->path, pVnode->diskPrimary, pVnode->pTfs, fname, TSDB_FILENAME_LEN);
   offset = strlen(fname);
   snprintf((char *)fname + offset, TSDB_FILENAME_LEN - offset - 1, "%sv%dver%" PRId64 ".del", TD_DIRSEP,
            TD_VID(pTsdb->pVnode), pFile->commitID);
diff --git a/source/dnode/vnode/src/vnd/vnodeCommit.c b/source/dnode/vnode/src/vnd/vnodeCommit.c
index b2360a57da..1d423beb7b 100644
--- a/source/dnode/vnode/src/vnd/vnodeCommit.c
+++ b/source/dnode/vnode/src/vnd/vnodeCommit.c
@@ -290,7 +290,7 @@ static int32_t vnodePrepareCommit(SVnode *pVnode, SCommitInfo *pInfo) {
   pInfo->txn = metaGetTxn(pVnode->pMeta);
 
   // save info
-  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
 
   vDebug("vgId:%d, save config while prepare commit", TD_VID(pVnode));
   if (vnodeSaveInfo(dir, &pInfo->info) < 0) {
@@ -428,7 +428,7 @@ static int vnodeCommitImpl(SCommitInfo *pInfo) {
     return -1;
   }
 
-  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
 
   syncBeginSnapshot(pVnode->sync, pInfo->info.state.committed);
 
@@ -492,7 +492,7 @@ bool vnodeShouldRollback(SVnode *pVnode) {
   char tFName[TSDB_FILENAME_LEN] = {0};
   int32_t offset = 0;
 
-  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, tFName, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, tFName, TSDB_FILENAME_LEN);
   offset = strlen(tFName);
   snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);
 
@@ -503,7 +503,7 @@ void vnodeRollback(SVnode *pVnode) {
   char tFName[TSDB_FILENAME_LEN] = {0};
   int32_t offset = 0;
 
-  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, tFName, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, tFName, TSDB_FILENAME_LEN);
   offset = strlen(tFName);
   snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);
 
diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c
index 541c695ba0..1134f960a9 100644
--- a/source/dnode/vnode/src/vnd/vnodeOpen.c
+++ b/source/dnode/vnode/src/vnd/vnodeOpen.c
@@ -15,9 +15,11 @@
 
 #include "vnd.h"
 
-int32_t vnodeGetPrimaryDir(const char *relPath, STfs *pTfs, char *buf, size_t bufLen) {
+int32_t vnodeGetPrimaryDir(const char *relPath, int32_t diskPrimary, STfs *pTfs, char *buf, size_t bufLen) {
   if (pTfs) {
-    snprintf(buf, bufLen - 1, "%s%s%s", tfsGetPrimaryPath(pTfs), TD_DIRSEP, relPath);
+    SDiskID diskId = {0};
+    diskId.id = diskPrimary;
+    snprintf(buf, bufLen - 1, "%s%s%s", tfsGetDiskPath(pTfs, diskId), TD_DIRSEP, relPath);
   } else {
     snprintf(buf, bufLen - 1, "%s", relPath);
   }
@@ -25,7 +27,7 @@ int32_t vnodeGetPrimaryDir(const char *relPath, STfs *pTfs, char *buf, size_t bu
   return 0;
 }
 
-int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
+int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, int32_t diskPrimary, STfs *pTfs) {
   SVnodeInfo info = {0};
   char       dir[TSDB_FILENAME_LEN] = {0};
 
@@ -36,7 +38,7 @@ int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
   }
 
   // create vnode env
-  vnodeGetPrimaryDir(path, pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(path, diskPrimary, pTfs, dir, TSDB_FILENAME_LEN);
   if (taosMkDir(dir)) {
     return TAOS_SYSTEM_ERROR(errno);
   }
@@ -60,12 +62,12 @@ int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
   return 0;
 }
 
-int32_t vnodeAlterReplica(const char *path, SAlterVnodeReplicaReq *pReq, STfs *pTfs) {
+int32_t vnodeAlterReplica(const char *path, SAlterVnodeReplicaReq *pReq, int32_t diskPrimary, STfs *pTfs) {
   SVnodeInfo info = {0};
   char       dir[TSDB_FILENAME_LEN] = {0};
   int32_t    ret = 0;
 
-  vnodeGetPrimaryDir(path, pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(path, diskPrimary, pTfs, dir, TSDB_FILENAME_LEN);
 
   ret = vnodeLoadInfo(dir, &info);
   if (ret < 0) {
@@ -133,7 +135,8 @@ static int32_t vnodeVgroupIdLen(int32_t vgId) {
   return strlen(tmp);
 }
 
-int32_t vnodeRenameVgroupId(const char *srcPath, const char *dstPath, int32_t srcVgId, int32_t dstVgId, STfs *pTfs) {
+int32_t vnodeRenameVgroupId(const char *srcPath, const char *dstPath, int32_t srcVgId, int32_t dstVgId,
+                            int32_t diskPrimary, STfs *pTfs) {
   int32_t ret = 0;
 
   char oldRname[TSDB_FILENAME_LEN] = {0};
@@ -183,12 +186,13 @@ int32_t vnodeRenameVgroupId(const char *srcPath, const char *dstPath, int32_t sr
   return ret;
 }
 
-int32_t vnodeAlterHashRange(const char *srcPath, const char *dstPath, SAlterVnodeHashRangeReq *pReq, STfs *pTfs) {
+int32_t vnodeAlterHashRange(const char *srcPath, const char *dstPath, SAlterVnodeHashRangeReq *pReq,
+                            int32_t diskPrimary, STfs *pTfs) {
   SVnodeInfo info = {0};
   char       dir[TSDB_FILENAME_LEN] = {0};
   int32_t    ret = 0;
 
-  vnodeGetPrimaryDir(srcPath, pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(srcPath, diskPrimary, pTfs, dir, TSDB_FILENAME_LEN);
 
   ret = vnodeLoadInfo(dir, &info);
   if (ret < 0) {
@@ -232,7 +236,7 @@ int32_t vnodeAlterHashRange(const char *srcPath, const char *dstPath, SAlterVnod
   }
 
   vInfo("vgId:%d, rename %s to %s", pReq->dstVgId, srcPath, dstPath);
-  ret = vnodeRenameVgroupId(srcPath, dstPath, pReq->srcVgId, pReq->dstVgId, pTfs);
+  ret = vnodeRenameVgroupId(srcPath, dstPath, pReq->srcVgId, pReq->dstVgId, diskPrimary, pTfs);
   if (ret < 0) {
     vError("vgId:%d, failed to rename vnode from %s to %s since %s", pReq->dstVgId, srcPath, dstPath,
            tstrerror(terrno));
@@ -243,11 +247,12 @@ int32_t vnodeAlterHashRange(const char *srcPath, const char *dstPath, SAlterVnod
   return 0;
 }
 
-int32_t vnodeRestoreVgroupId(const char *srcPath, const char *dstPath, int32_t srcVgId, int32_t dstVgId, STfs *pTfs) {
+int32_t vnodeRestoreVgroupId(const char *srcPath, const char *dstPath, int32_t srcVgId, int32_t dstVgId,
+                             int32_t diskPrimary, STfs *pTfs) {
   SVnodeInfo info = {0};
   char       dir[TSDB_FILENAME_LEN] = {0};
 
-  vnodeGetPrimaryDir(dstPath, pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(dstPath, diskPrimary, pTfs, dir, TSDB_FILENAME_LEN);
   if (vnodeLoadInfo(dir, &info) == 0) {
     if (info.config.vgId != dstVgId) {
       vError("vgId:%d, unexpected vnode config.vgId:%d", dstVgId, info.config.vgId);
@@ -256,7 +261,7 @@ int32_t vnodeRestoreVgroupId(const char *srcPath, const char *dstPath, int32_t s
     return dstVgId;
   }
 
-  vnodeGetPrimaryDir(srcPath, pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(srcPath, diskPrimary, pTfs, dir, TSDB_FILENAME_LEN);
   if (vnodeLoadInfo(dir, &info) < 0) {
     vError("vgId:%d, failed to read vnode config from %s since %s", srcVgId, srcPath, tstrerror(terrno));
     return -1;
@@ -271,7 +276,7 @@ int32_t vnodeRestoreVgroupId(const char *srcPath, const char *dstPath, int32_t s
   }
 
   vInfo("vgId:%d, rename %s to %s", dstVgId, srcPath, dstPath);
-  if (vnodeRenameVgroupId(srcPath, dstPath, srcVgId, dstVgId, pTfs) < 0) {
+  if (vnodeRenameVgroupId(srcPath, dstPath, srcVgId, dstVgId, diskPrimary, pTfs) < 0) {
     vError("vgId:%d, failed to rename vnode from %s to %s since %s", dstVgId, srcPath, dstPath, tstrerror(terrno));
     return -1;
   }
@@ -284,14 +289,14 @@ void vnodeDestroy(const char *path, STfs *pTfs) {
   tfsRmdir(pTfs, path);
 }
 
-SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) {
+SVnode *vnodeOpen(const char *path, int32_t diskPrimary, STfs *pTfs, SMsgCb msgCb) {
   SVnode    *pVnode = NULL;
   SVnodeInfo info = {0};
   char       dir[TSDB_FILENAME_LEN] = {0};
   char       tdir[TSDB_FILENAME_LEN * 2] = {0};
   int32_t    ret = 0;
 
-  vnodeGetPrimaryDir(path, pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(path, diskPrimary, pTfs, dir, TSDB_FILENAME_LEN);
 
   info.config = vnodeCfgDefault;
 
diff --git a/source/dnode/vnode/src/vnd/vnodeRetention.c b/source/dnode/vnode/src/vnd/vnodeRetention.c
index 5a2f612ef5..f582d5e4a4 100644
--- a/source/dnode/vnode/src/vnd/vnodeRetention.c
+++ b/source/dnode/vnode/src/vnd/vnodeRetention.c
@@ -35,7 +35,7 @@ static int32_t vnodePrepareRentention(SVnode *pVnode, SRetentionInfo *pInfo) {
   pInfo->commitID = ++pVnode->state.commitID;
 
   char dir[TSDB_FILENAME_LEN] = {0};
-  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
 
   if (vnodeLoadInfo(dir, &pInfo->info) < 0) {
     code = terrno;
@@ -60,7 +60,7 @@ static int32_t vnodeRetentionTask(void *param) {
   SVnode         *pVnode = pInfo->pVnode;
   char            dir[TSDB_FILENAME_LEN] = {0};
 
-  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
 
   // save info
   pInfo->info.state.commitID = pInfo->commitID;
diff --git a/source/dnode/vnode/src/vnd/vnodeSnapshot.c b/source/dnode/vnode/src/vnd/vnodeSnapshot.c
index c80792490a..d559783c2f 100644
--- a/source/dnode/vnode/src/vnd/vnodeSnapshot.c
+++ b/source/dnode/vnode/src/vnd/vnodeSnapshot.c
@@ -86,6 +86,7 @@ void vnodeSnapReaderClose(SVSnapReader *pReader) {
 
 int32_t vnodeSnapRead(SVSnapReader *pReader, uint8_t **ppData, uint32_t *nData) {
   int32_t code = 0;
+  SVnode *pVnode = pReader->pVnode;
 
   // CONFIG ==============
   // FIXME: if commit multiple times and the config changed?
@@ -93,7 +94,7 @@ int32_t vnodeSnapRead(SVSnapReader *pReader, uint8_t **ppData, uint32_t *nData)
     char fName[TSDB_FILENAME_LEN];
     int32_t offset = 0;
 
-    vnodeGetPrimaryDir(pReader->pVnode->path, pReader->pVnode->pTfs, fName, TSDB_FILENAME_LEN);
+    vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, fName, TSDB_FILENAME_LEN);
     offset = strlen(fName);
     snprintf(fName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME);
 
@@ -343,7 +344,7 @@ int32_t vnodeSnapWriterClose(SVSnapWriter *pWriter, int8_t rollback, SSnapshot *
                               .applyTerm = pWriter->info.state.commitTerm};
     pVnode->statis = pWriter->info.statis;
     char dir[TSDB_FILENAME_LEN] = {0};
-    vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
+    vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
 
     vnodeCommitInfo(dir);
   } else {
@@ -381,7 +382,7 @@ _exit:
 
 static int32_t vnodeSnapWriteInfo(SVSnapWriter *pWriter, uint8_t *pData, uint32_t nData) {
   int32_t code = 0;
-
+  SVnode       *pVnode = pWriter->pVnode;
   SSnapDataHdr *pHdr = (SSnapDataHdr *)pData;
 
   // decode info
@@ -395,10 +396,9 @@ static int32_t vnodeSnapWriteInfo(SVSnapWriter *pWriter, uint8_t *pData, uint32_
 
   // modify info as needed
   char dir[TSDB_FILENAME_LEN] = {0};
-  vnodeGetPrimaryDir(pWriter->pVnode->path, pWriter->pVnode->pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
 
   SVnodeStats vndStats = pWriter->info.config.vndStats;
-  SVnode     *pVnode = pWriter->pVnode;
   pWriter->info.config = pVnode->config;
   pWriter->info.config.vndStats = vndStats;
   vDebug("vgId:%d, save config while write snapshot", pWriter->pVnode->config.vgId);

From 1d0c73feb679b0e0d25b2b83b131d101bf9f76d0 Mon Sep 17 00:00:00 2001
From: Shungang Li <sgli@taosdata.com>
Date: Mon, 17 Jul 2023 18:37:00 +0800
Subject: [PATCH 02/13] fix: alter dnode keeptimeoffset err

---
 source/common/src/tglobal.c            |  6 ------
 source/dnode/mnode/impl/src/mndDnode.c | 21 +++++++++++++++++++--
 2 files changed, 19 insertions(+), 8 deletions(-)

diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c
index 3545ece6d8..b1d2502e2f 100644
--- a/source/common/src/tglobal.c
+++ b/source/common/src/tglobal.c
@@ -1490,14 +1490,8 @@ void taosCfgDynamicOptions(const char *option, const char *value) {
 
   if (strcasecmp(option, "keepTimeOffset") == 0) {
     int32_t newKeepTimeOffset = atoi(value);
-    if (newKeepTimeOffset < 0 || newKeepTimeOffset > 23) {
-      uError("failed to set keepTimeOffset from %d to %d. Valid range: [0, 23]", tsKeepTimeOffset, newKeepTimeOffset);
-      return;
-    }
-
     uInfo("keepTimeOffset set from %d to %d", tsKeepTimeOffset, newKeepTimeOffset);
     tsKeepTimeOffset = newKeepTimeOffset;
-
     return;
   }
 
diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c
index bb92bfb4c7..11825a4e50 100644
--- a/source/dnode/mnode/impl/src/mndDnode.c
+++ b/source/dnode/mnode/impl/src/mndDnode.c
@@ -947,7 +947,7 @@ static int32_t mndProcessDropDnodeReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
-  mInfo("dnode:%d, start to drop, ep:%s:%d, force:%s, unsafe:%s", 
+  mInfo("dnode:%d, start to drop, ep:%s:%d, force:%s, unsafe:%s",
           dropReq.dnodeId, dropReq.fqdn, dropReq.port, dropReq.force?"true":"false", dropReq.unsafe?"true":"false");
   if (mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_DROP_MNODE) != 0) {
     goto _OVER;
@@ -987,7 +987,7 @@ static int32_t mndProcessDropDnodeReq(SRpcMsg *pReq) {
 
   int32_t numOfVnodes = mndGetVnodesNum(pMnode, pDnode->id);
   bool isonline = mndIsDnodeOnline(pDnode, taosGetTimestampMs());
-  
+
   if (isonline && force) {
     terrno = TSDB_CODE_DNODE_ONLY_USE_WHEN_OFFLINE;
     mError("dnode:%d, failed to drop since %s, vnodes:%d mnode:%d qnode:%d snode:%d", pDnode->id, terrstr(),
@@ -1060,6 +1060,23 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
 
     strcpy(dcfgReq.config, "monitor");
     snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
+  } else if (strncasecmp(cfgReq.config, "keeptimeoffset", 14) == 0) {
+    if (' ' != cfgReq.config[14] && 0 != cfgReq.config[14]) {
+      mError("dnode:%d, failed to config keeptimeoffset since invalid conf:%s", cfgReq.dnodeId, cfgReq.config);
+      terrno = TSDB_CODE_INVALID_CFG;
+      return -1;
+    }
+
+    const char *value = cfgReq.value;
+    int32_t     offset = atoi(value);
+    if (offset < 0 || offset > 23) {
+      mError("dnode:%d, failed to config keepTimeOffset since value:%d. Valid range: [0, 23]", cfgReq.dnodeId, offset);
+      terrno = TSDB_CODE_INVALID_CFG;
+      return -1;
+    }
+
+    strcpy(dcfgReq.config, "keeptimeoffset");
+    snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", offset);
 #ifdef TD_ENTERPRISE
   } else if (strncasecmp(cfgReq.config, "activeCode", 10) == 0 || strncasecmp(cfgReq.config, "cActiveCode", 11) == 0) {
     int8_t opt = strncasecmp(cfgReq.config, "a", 1) == 0 ? DND_ACTIVE_CODE : DND_CONN_ACTIVE_CODE;

From cf39fcdd12da3c1f847f04720335d180014156ef Mon Sep 17 00:00:00 2001
From: kailixu <klxu@taosdata.com>
Date: Mon, 17 Jul 2023 22:24:27 +0800
Subject: [PATCH 03/13] fix: add sma option when show create table for stb/ntb

---
 source/dnode/mnode/impl/src/mndStb.c |  3 +++
 source/libs/command/src/command.c    | 24 ++++++++++++++++++++++++
 2 files changed, 27 insertions(+)

diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c
index 162e75d783..d6537ef992 100644
--- a/source/dnode/mnode/impl/src/mndStb.c
+++ b/source/dnode/mnode/impl/src/mndStb.c
@@ -1738,6 +1738,7 @@ static int32_t mndBuildStbSchemaImp(SDbObj *pDb, SStbObj *pStb, const char *tbNa
     SSchema *pSrcSchema = &pStb->pColumns[i];
     memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
     pSchema->type = pSrcSchema->type;
+    pSchema->flags = pSrcSchema->flags;
     pSchema->colId = pSrcSchema->colId;
     pSchema->bytes = pSrcSchema->bytes;
   }
@@ -1788,6 +1789,7 @@ static int32_t mndBuildStbCfgImp(SDbObj *pDb, SStbObj *pStb, const char *tbName,
     SSchema *pSrcSchema = &pStb->pColumns[i];
     memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
     pSchema->type = pSrcSchema->type;
+    pSchema->flags = pSrcSchema->flags;
     pSchema->colId = pSrcSchema->colId;
     pSchema->bytes = pSrcSchema->bytes;
   }
@@ -1797,6 +1799,7 @@ static int32_t mndBuildStbCfgImp(SDbObj *pDb, SStbObj *pStb, const char *tbName,
     SSchema *pSrcSchema = &pStb->pTags[i];
     memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
     pSchema->type = pSrcSchema->type;
+    pSchema->flags = pSrcSchema->flags;
     pSchema->colId = pSrcSchema->colId;
     pSchema->bytes = pSrcSchema->bytes;
   }
diff --git a/source/libs/command/src/command.c b/source/libs/command/src/command.c
index ecda1d596a..172960653d 100644
--- a/source/libs/command/src/command.c
+++ b/source/libs/command/src/command.c
@@ -615,6 +615,30 @@ void appendTableOptions(char* buf, int32_t* len, SDbCfgInfo* pDbCfg, STableCfg*
   if (pCfg->ttl > 0) {
     *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, " TTL %d", pCfg->ttl);
   }
+
+  if (TSDB_SUPER_TABLE == pCfg->tableType || TSDB_NORMAL_TABLE == pCfg->tableType) {
+    int32_t nSma = 0;
+    for (int32_t i = 0; i < pCfg->numOfColumns; ++i) {
+      if (IS_BSMA_ON(pCfg->pSchemas + i)) {
+        ++nSma;
+      }
+    }
+
+    if (nSma < pCfg->numOfColumns && nSma > 0) {
+      bool smaOn = false;
+      for (int32_t i = 0; i < pCfg->numOfColumns; ++i) {
+        if (IS_BSMA_ON(pCfg->pSchemas + i)) {
+          if (smaOn) {
+            *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, ",`%s`", (pCfg->pSchemas + i)->name);
+          } else {
+            smaOn = true;
+            *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, " SMA(`%s`", (pCfg->pSchemas + i)->name);
+          }
+        }
+      }
+      *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, ")");
+    }
+  }
 }
 
 static int32_t setCreateTBResultIntoDataBlock(SSDataBlock* pBlock, SDbCfgInfo* pDbCfg, char* tbName, STableCfg* pCfg) {

From 52c56f4bd7a3193125d9cc8bc8b331fd9d73f29b Mon Sep 17 00:00:00 2001
From: kailixu <klxu@taosdata.com>
Date: Mon, 17 Jul 2023 22:40:22 +0800
Subject: [PATCH 04/13] chore: support non sma option () if supported

---
 source/libs/command/src/command.c | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/source/libs/command/src/command.c b/source/libs/command/src/command.c
index 172960653d..54352d0a53 100644
--- a/source/libs/command/src/command.c
+++ b/source/libs/command/src/command.c
@@ -624,15 +624,16 @@ void appendTableOptions(char* buf, int32_t* len, SDbCfgInfo* pDbCfg, STableCfg*
       }
     }
 
-    if (nSma < pCfg->numOfColumns && nSma > 0) {
+    if (nSma < pCfg->numOfColumns) {
       bool smaOn = false;
+      *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, " SMA(");
       for (int32_t i = 0; i < pCfg->numOfColumns; ++i) {
         if (IS_BSMA_ON(pCfg->pSchemas + i)) {
           if (smaOn) {
             *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, ",`%s`", (pCfg->pSchemas + i)->name);
           } else {
             smaOn = true;
-            *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, " SMA(`%s`", (pCfg->pSchemas + i)->name);
+            *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "`%s`", (pCfg->pSchemas + i)->name);
           }
         }
       }

From 6b6b7d8d8f77b0996832dedf333f443950601b64 Mon Sep 17 00:00:00 2001
From: Benguang Zhao <bgzhao@taosdata.com>
Date: Mon, 17 Jul 2023 20:43:31 +0800
Subject: [PATCH 05/13] enh: alloc disk of vnode primary dir in
 vmAllocPrimaryDisk

---
 include/libs/tfs/tfs.h                   | 17 ++++++++
 source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 52 +++++++++++++++++++++++-
 source/dnode/vnode/src/inc/vnodeInt.h    |  1 +
 source/dnode/vnode/src/vnd/vnodeCommit.c |  2 -
 source/dnode/vnode/src/vnd/vnodeOpen.c   |  6 ++-
 source/libs/tfs/src/tfs.c                | 28 ++++++++++++-
 6 files changed, 100 insertions(+), 6 deletions(-)

diff --git a/include/libs/tfs/tfs.h b/include/libs/tfs/tfs.h
index 622cd615b8..988294be4b 100644
--- a/include/libs/tfs/tfs.h
+++ b/include/libs/tfs/tfs.h
@@ -69,6 +69,13 @@ void tfsUpdateSize(STfs *pTfs);
  */
 SDiskSize tfsGetSize(STfs *pTfs);
 
+/**
+ * @brief Get the number of disks at level of multi-tier storage.
+ *
+ * @param pTfs
+ * @return int32_t
+ */
+int32_t tfsGetDisksAtLevel(STfs *pTfs, int32_t level);
 /**
  * @brief Get level of multi-tier storage.
  *
@@ -162,6 +169,16 @@ int32_t tfsRmdir(STfs *pTfs, const char *rname);
  */
 int32_t tfsRename(STfs *pTfs, const char *orname, const char *nrname);
 
+/**
+ * @brief Search fname in level of tfs
+ *
+ * @param pTfs The fs object.
+ * @param level The level to search on
+ * @param fname The relative file name to be searched
+ * @param int32_t diskId for successs, -1 for failure
+ */
+int32_t tfsSearch(STfs *pTfs, int32_t level, const char *fname);
+
 /**
  * @brief Init file object in tfs.
  *
diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
index 2a144dd23b..a89a3d0781 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
@@ -15,12 +15,62 @@
 
 #define _DEFAULT_SOURCE
 #include "vmInt.h"
+#include "tfs.h"
 #include "vnd.h"
 
 int32_t vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId) {
+  STfs   *pTfs = pMgmt->pTfs;
+  int32_t diskId = 0;
+  if (!pTfs) {
+    return diskId;
+  }
+
   // search fs
+  char vnodePath[TSDB_FILENAME_LEN] = {0};
+  snprintf(vnodePath, TSDB_FILENAME_LEN - 1, "vnode%svnode%d", TD_DIRSEP, vgId);
+  char fname[TSDB_FILENAME_LEN] = {0};
+  char fnameTmp[TSDB_FILENAME_LEN] = {0};
+  snprintf(fname, TSDB_FILENAME_LEN - 1, "%s%s%s", vnodePath, TD_DIRSEP, VND_INFO_FNAME);
+  snprintf(fnameTmp, TSDB_FILENAME_LEN - 1, "%s%s%s", vnodePath, TD_DIRSEP, VND_INFO_FNAME_TMP);
+
+  diskId = tfsSearch(pTfs, 0, fname);
+  if (diskId >= 0) {
+    return diskId;
+  }
+  diskId = tfsSearch(pTfs, 0, fnameTmp);
+  if (diskId >= 0) {
+    return diskId;
+  }
+
   // alloc
-  return 0;
+  int32_t     disks[TFS_MAX_DISKS_PER_TIER] = {0};
+  int32_t     numOfVnodes = 0;
+  SVnodeObj **ppVnodes = vmGetVnodeListFromHash(pMgmt, &numOfVnodes);
+  for (int32_t v = 0; v < numOfVnodes; v++) {
+    SVnodeObj *pVnode = ppVnodes[v];
+    disks[pVnode->diskPrimary] += 1;
+  }
+
+  int32_t minVal = INT_MAX;
+  int32_t ndisk = tfsGetDisksAtLevel(pTfs, 0);
+  diskId = 0;
+  for (int32_t id = 0; id < ndisk; id++) {
+    if (minVal > disks[id]) {
+      minVal = disks[id];
+      diskId = id;
+    }
+  }
+
+  for (int32_t i = 0; i < numOfVnodes; ++i) {
+    if (ppVnodes == NULL || ppVnodes[i] == NULL) continue;
+    vmReleaseVnode(pMgmt, ppVnodes[i]);
+  }
+  if (ppVnodes != NULL) {
+    taosMemoryFree(ppVnodes);
+  }
+
+  dInfo("vgId:%d, alloc disk:%d of level 0. ndisk:%d, vnodes: %d", vgId, diskId, ndisk, numOfVnodes);
+  return diskId;
 }
 
 SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h
index f79e4650ce..f5a610efc7 100644
--- a/source/dnode/vnode/src/inc/vnodeInt.h
+++ b/source/dnode/vnode/src/inc/vnodeInt.h
@@ -93,6 +93,7 @@ typedef struct SQueryNode         SQueryNode;
 #define VNODE_BUFPOOL_SEGMENTS 3
 
 #define VND_INFO_FNAME "vnode.json"
+#define VND_INFO_FNAME_TMP "vnode_tmp.json"
 
 // vnd.h
 typedef int32_t (*_query_reseek_func_t)(void* pQHandle);
diff --git a/source/dnode/vnode/src/vnd/vnodeCommit.c b/source/dnode/vnode/src/vnd/vnodeCommit.c
index 1d423beb7b..7109f8466e 100644
--- a/source/dnode/vnode/src/vnd/vnodeCommit.c
+++ b/source/dnode/vnode/src/vnd/vnodeCommit.c
@@ -16,8 +16,6 @@
 #include "vnd.h"
 #include "vnodeInt.h"
 
-#define VND_INFO_FNAME_TMP "vnode_tmp.json"
-
 static int vnodeEncodeInfo(const SVnodeInfo *pInfo, char **ppData);
 static int vnodeCommitImpl(SCommitInfo *pInfo);
 
diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c
index 1134f960a9..6aa9be7b8a 100644
--- a/source/dnode/vnode/src/vnd/vnodeOpen.c
+++ b/source/dnode/vnode/src/vnd/vnodeOpen.c
@@ -38,10 +38,11 @@ int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, int32_t diskPrimary, STfs
   }
 
   // create vnode env
-  vnodeGetPrimaryDir(path, diskPrimary, pTfs, dir, TSDB_FILENAME_LEN);
-  if (taosMkDir(dir)) {
+  if ((pTfs) ? tfsMkdir(pTfs, path) : taosMkDir(path)) {
+    vError("vgId:%d, failed to mkdir since %s, dir: %s", pCfg->vgId, strerror(errno), path);
     return TAOS_SYSTEM_ERROR(errno);
   }
+  vnodeGetPrimaryDir(path, diskPrimary, pTfs, dir, TSDB_FILENAME_LEN);
 
   if (pCfg) {
     info.config = *pCfg;
@@ -339,6 +340,7 @@ SVnode *vnodeOpen(const char *path, int32_t diskPrimary, STfs *pTfs, SMsgCb msgC
   pVnode->state.applied = info.state.committed;
   pVnode->state.applyTerm = info.state.commitTerm;
   pVnode->pTfs = pTfs;
+  pVnode->diskPrimary = diskPrimary;
   pVnode->msgCb = msgCb;
   taosThreadMutexInit(&pVnode->lock, NULL);
   pVnode->blocked = false;
diff --git a/source/libs/tfs/src/tfs.c b/source/libs/tfs/src/tfs.c
index 4383c46c17..7747f158b1 100644
--- a/source/libs/tfs/src/tfs.c
+++ b/source/libs/tfs/src/tfs.c
@@ -113,6 +113,15 @@ SDiskSize tfsGetSize(STfs *pTfs) {
   return size;
 }
 
+int32_t tfsGetDisksAtLevel(STfs *pTfs, int32_t level) {
+  if (level < 0 || level >= pTfs->nlevel) {
+    return 0;
+  }
+
+  STfsTier *pTier = TFS_TIER_AT(pTfs, level);
+  return pTier->ndisk;
+}
+
 int32_t tfsGetLevel(STfs *pTfs) { return pTfs->nlevel; }
 
 int32_t tfsAllocDisk(STfs *pTfs, int32_t expLevel, SDiskID *pDiskId) {
@@ -277,7 +286,7 @@ int32_t tfsMkdir(STfs *pTfs, const char *rname) {
     STfsTier *pTier = TFS_TIER_AT(pTfs, level);
     for (int32_t id = 0; id < pTier->ndisk; id++) {
       SDiskID did = {.id = id, .level = level};
-      if (tfsMkdirAt(pTfs, rname, did) < 0) {
+      if (tfsMkdirRecurAt(pTfs, rname, did) < 0) {
         return -1;
       }
     }
@@ -335,6 +344,23 @@ int32_t tfsRename(STfs *pTfs, const char *orname, const char *nrname) {
   return 0;
 }
 
+int32_t tfsSearch(STfs *pTfs, int32_t level, const char *fname) {
+  if (level < 0 || level >= pTfs->nlevel) {
+    return -1;
+  }
+  char      path[TMPNAME_LEN] = {0};
+  STfsTier *pTier = TFS_TIER_AT(pTfs, level);
+
+  for (int32_t id = 0; id < pTier->ndisk; id++) {
+    STfsDisk *pDisk = pTier->disks[id];
+    snprintf(path, TMPNAME_LEN - 1, "%s%s%s", pDisk->path, TD_DIRSEP, fname);
+    if (taosCheckExistFile(path)) {
+      return id;
+    }
+  }
+  return -1;
+}
+
 STfsDir *tfsOpendir(STfs *pTfs, const char *rname) {
   STfsDir *pDir = taosMemoryCalloc(1, sizeof(STfsDir));
   if (pDir == NULL) {

From 5624ecc8dfc5ce811fd8abffc47269cf72bfb2ee Mon Sep 17 00:00:00 2001
From: Benguang Zhao <bgzhao@taosdata.com>
Date: Tue, 18 Jul 2023 10:04:00 +0800
Subject: [PATCH 06/13] fix: prepare vnode dir recursively in vnodeCreate

---
 include/libs/tfs/tfs.h                 |  9 +++++++++
 source/dnode/vnode/src/vnd/vnodeOpen.c | 12 ++++++++++--
 source/libs/tfs/src/tfs.c              | 16 +++++++++++++++-
 3 files changed, 34 insertions(+), 3 deletions(-)

diff --git a/include/libs/tfs/tfs.h b/include/libs/tfs/tfs.h
index 988294be4b..ef8a4cfe1f 100644
--- a/include/libs/tfs/tfs.h
+++ b/include/libs/tfs/tfs.h
@@ -130,6 +130,15 @@ int32_t tfsMkdir(STfs *pTfs, const char *rname);
  */
 int32_t tfsMkdirAt(STfs *pTfs, const char *rname, SDiskID diskId);
 
+/**
+ * @brief Recursive make directory at all levels in tfs.
+ *
+ * @param pTfs The fs object.
+ * @param rname The rel name of directory.
+ * @return int32_t 0 for success, -1 for failure.
+ */
+int32_t tfsMkdirRecur(STfs *pTfs, const char *rname);
+
 /**
  * @brief Recursive create directories in tfs.
  *
diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c
index 6aa9be7b8a..4c30011d89 100644
--- a/source/dnode/vnode/src/vnd/vnodeOpen.c
+++ b/source/dnode/vnode/src/vnd/vnodeOpen.c
@@ -27,6 +27,14 @@ int32_t vnodeGetPrimaryDir(const char *relPath, int32_t diskPrimary, STfs *pTfs,
   return 0;
 }
 
+static int32_t vnodeMkDir(STfs *pTfs, const char *path) {
+  if (pTfs) {
+    return tfsMkdirRecur(pTfs, path);
+  } else {
+    return taosMkDir(path);
+  }
+}
+
 int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, int32_t diskPrimary, STfs *pTfs) {
   SVnodeInfo info = {0};
   char       dir[TSDB_FILENAME_LEN] = {0};
@@ -38,8 +46,8 @@ int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, int32_t diskPrimary, STfs
   }
 
   // create vnode env
-  if ((pTfs) ? tfsMkdir(pTfs, path) : taosMkDir(path)) {
-    vError("vgId:%d, failed to mkdir since %s, dir: %s", pCfg->vgId, strerror(errno), path);
+  if (vnodeMkDir(pTfs, path)) {
+    vError("vgId:%d, failed to prepare vnode dir since %s, path: %s", pCfg->vgId, strerror(errno), path);
     return TAOS_SYSTEM_ERROR(errno);
   }
   vnodeGetPrimaryDir(path, diskPrimary, pTfs, dir, TSDB_FILENAME_LEN);
diff --git a/source/libs/tfs/src/tfs.c b/source/libs/tfs/src/tfs.c
index 7747f158b1..a0a95114ae 100644
--- a/source/libs/tfs/src/tfs.c
+++ b/source/libs/tfs/src/tfs.c
@@ -281,7 +281,7 @@ int32_t tfsMkdirRecurAt(STfs *pTfs, const char *rname, SDiskID diskId) {
   return 0;
 }
 
-int32_t tfsMkdir(STfs *pTfs, const char *rname) {
+int32_t tfsMkdirRecur(STfs *pTfs, const char *rname) {
   for (int32_t level = 0; level < pTfs->nlevel; level++) {
     STfsTier *pTier = TFS_TIER_AT(pTfs, level);
     for (int32_t id = 0; id < pTier->ndisk; id++) {
@@ -295,6 +295,20 @@ int32_t tfsMkdir(STfs *pTfs, const char *rname) {
   return 0;
 }
 
+int32_t tfsMkdir(STfs *pTfs, const char *rname) {
+  for (int32_t level = 0; level < pTfs->nlevel; level++) {
+    STfsTier *pTier = TFS_TIER_AT(pTfs, level);
+    for (int32_t id = 0; id < pTier->ndisk; id++) {
+      SDiskID did = {.id = id, .level = level};
+      if (tfsMkdirAt(pTfs, rname, did) < 0) {
+        return -1;
+      }
+    }
+  }
+
+  return 0;
+}
+
 bool tfsDirExistAt(STfs *pTfs, const char *rname, SDiskID diskId) {
   STfsDisk *pDisk = TFS_DISK_AT(pTfs, diskId);
   char      aname[TMPNAME_LEN];

From 24fed5113704781e65a024501abbf41d9b1b56eb Mon Sep 17 00:00:00 2001
From: Benguang Zhao <bgzhao@taosdata.com>
Date: Tue, 18 Jul 2023 13:58:43 +0800
Subject: [PATCH 07/13] enh: tfsRename dir on primary disk at last

---
 include/libs/tfs/tfs.h                 |  2 +-
 source/dnode/vnode/src/vnd/vnodeOpen.c |  4 +--
 source/libs/tfs/src/tfs.c              | 34 ++++++++++++++++++++------
 source/libs/tfs/test/tfsTest.cpp       |  6 ++---
 4 files changed, 32 insertions(+), 14 deletions(-)

diff --git a/include/libs/tfs/tfs.h b/include/libs/tfs/tfs.h
index ef8a4cfe1f..509f8dc9e8 100644
--- a/include/libs/tfs/tfs.h
+++ b/include/libs/tfs/tfs.h
@@ -176,7 +176,7 @@ int32_t tfsRmdir(STfs *pTfs, const char *rname);
  * @param nrname The rel name of new file.
  * @return int32_t 0 for success, -1 for failure.
  */
-int32_t tfsRename(STfs *pTfs, const char *orname, const char *nrname);
+int32_t tfsRename(STfs *pTfs, int32_t diskPrimary, const char *orname, const char *nrname);
 
 /**
  * @brief Search fname in level of tfs
diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c
index 4c30011d89..8f613fd0d8 100644
--- a/source/dnode/vnode/src/vnd/vnodeOpen.c
+++ b/source/dnode/vnode/src/vnd/vnodeOpen.c
@@ -176,7 +176,7 @@ int32_t vnodeRenameVgroupId(const char *srcPath, const char *dstPath, int32_t sr
       snprintf(newRname, TSDB_FILENAME_LEN, "%s%d%s", oldRname, dstVgId, tsdbFileSurfixPos);
       vInfo("vgId:%d, rename file from %s to %s", dstVgId, tsdbFile->rname, newRname);
 
-      ret = tfsRename(pTfs, tsdbFile->rname, newRname);
+      ret = tfsRename(pTfs, diskPrimary, tsdbFile->rname, newRname);
       if (ret != 0) {
         vError("vgId:%d, failed to rename file from %s to %s since %s", dstVgId, tsdbFile->rname, newRname, terrstr());
         tfsClosedir(tsdbDir);
@@ -188,7 +188,7 @@ int32_t vnodeRenameVgroupId(const char *srcPath, const char *dstPath, int32_t sr
   tfsClosedir(tsdbDir);
 
   vInfo("vgId:%d, rename dir from %s to %s", dstVgId, srcPath, dstPath);
-  ret = tfsRename(pTfs, srcPath, dstPath);
+  ret = tfsRename(pTfs, diskPrimary, srcPath, dstPath);
   if (ret != 0) {
     vError("vgId:%d, failed to rename dir from %s to %s since %s", dstVgId, srcPath, dstPath, terrstr());
   }
diff --git a/source/libs/tfs/src/tfs.c b/source/libs/tfs/src/tfs.c
index a0a95114ae..8adaab91a1 100644
--- a/source/libs/tfs/src/tfs.c
+++ b/source/libs/tfs/src/tfs.c
@@ -337,25 +337,43 @@ int32_t tfsRmdir(STfs *pTfs, const char *rname) {
   return 0;
 }
 
-int32_t tfsRename(STfs *pTfs, const char *orname, const char *nrname) {
+static int32_t tfsRenameAt(STfs *pTfs, SDiskID diskId, const char *orname, const char *nrname) {
   char oaname[TMPNAME_LEN] = "\0";
   char naname[TMPNAME_LEN] = "\0";
 
+  int32_t   level = diskId.level;
+  int32_t   id = diskId.id;
+  STfsTier *pTier = TFS_TIER_AT(pTfs, level);
+  STfsDisk *pDisk = pTier->disks[id];
+  snprintf(oaname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, orname);
+  snprintf(naname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, nrname);
+
+  if (taosRenameFile(oaname, naname) != 0 && errno != ENOENT) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    fError("failed to rename %s to %s since %s", oaname, naname, terrstr());
+    return -1;
+  }
+
+  return 0;
+}
+
+int32_t tfsRename(STfs *pTfs, int32_t diskPrimary, const char *orname, const char *nrname) {
   for (int32_t level = pTfs->nlevel - 1; level >= 0; level--) {
     STfsTier *pTier = TFS_TIER_AT(pTfs, level);
     for (int32_t id = pTier->ndisk - 1; id >= 0; id--) {
-      STfsDisk *pDisk = pTier->disks[id];
-      snprintf(oaname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, orname);
-      snprintf(naname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, nrname);
-      if (taosRenameFile(oaname, naname) != 0 && errno != ENOENT) {
-        terrno = TAOS_SYSTEM_ERROR(errno);
-        fError("failed to rename %s to %s since %s", oaname, naname, terrstr());
+      if (level == 0 && id == diskPrimary) {
+        continue;
+      }
+
+      SDiskID diskId = {.level = level, .id = id};
+      if (tfsRenameAt(pTfs, diskId, orname, nrname)) {
         return -1;
       }
     }
   }
 
-  return 0;
+  SDiskID diskId = {.level = 0, .id = diskPrimary};
+  return tfsRenameAt(pTfs, diskId, orname, nrname);
 }
 
 int32_t tfsSearch(STfs *pTfs, int32_t level, const char *fname) {
diff --git a/source/libs/tfs/test/tfsTest.cpp b/source/libs/tfs/test/tfsTest.cpp
index df37630fd7..9bbf6bc729 100644
--- a/source/libs/tfs/test/tfsTest.cpp
+++ b/source/libs/tfs/test/tfsTest.cpp
@@ -156,7 +156,7 @@ TEST_F(TfsTest, 03_Dir) {
   EXPECT_NE(taosDirExist(ap4), 1);
   EXPECT_EQ(tfsMkdirRecurAt(pTfs, p4, did), 0);
   EXPECT_EQ(taosDirExist(ap4), 1);
-  EXPECT_EQ(tfsRename(pTfs, p44, p45), 0);
+  EXPECT_EQ(tfsRename(pTfs, 0, p44, p45), 0);
   EXPECT_EQ(tfsRmdir(pTfs, p4), 0);
   EXPECT_NE(taosDirExist(ap4), 1);
 
@@ -609,7 +609,7 @@ TEST_F(TfsTest, 05_MultiDisk) {
     EXPECT_NE(taosDirExist(_ap22), 1);
     EXPECT_EQ(tfsMkdirRecurAt(pTfs, p4, did), 0);
     EXPECT_EQ(taosDirExist(_ap22), 1);
-    EXPECT_EQ(tfsRename(pTfs, p44, p45), 0);
+    EXPECT_EQ(tfsRename(pTfs, 0, p44, p45), 0);
     EXPECT_EQ(tfsRmdir(pTfs, p4), 0);
     EXPECT_NE(taosDirExist(_ap22), 1);
   }
@@ -721,4 +721,4 @@ TEST_F(TfsTest, 05_MultiDisk) {
   }
 
   tfsClose(pTfs);
-}
\ No newline at end of file
+}

From 2faa3d2e7b5a9b6bf7ab4ac0d6b4ee1c024c1017 Mon Sep 17 00:00:00 2001
From: Benguang Zhao <bgzhao@taosdata.com>
Date: Tue, 18 Jul 2023 15:39:29 +0800
Subject: [PATCH 08/13] enh: check validity of diskPrimary during vnodeOpen

---
 source/dnode/vnode/src/vnd/vnodeOpen.c | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c
index 8f613fd0d8..3b07ef9ea9 100644
--- a/source/dnode/vnode/src/vnd/vnodeOpen.c
+++ b/source/dnode/vnode/src/vnd/vnodeOpen.c
@@ -298,6 +298,19 @@ void vnodeDestroy(const char *path, STfs *pTfs) {
   tfsRmdir(pTfs, path);
 }
 
+static int32_t vnodeCheckDisk(int32_t diskPrimary, STfs *pTfs) {
+  int32_t ndisk = 1;
+  if (pTfs) {
+    ndisk = tfsGetDisksAtLevel(pTfs, 0);
+  }
+  if (diskPrimary < 0 || diskPrimary >= ndisk) {
+    vError("disk:%d is unavailable from the %d disks mounted at level 0", diskPrimary, ndisk);
+    terrno = TSDB_CODE_FS_INVLD_CFG;
+    return -1;
+  }
+  return 0;
+}
+
 SVnode *vnodeOpen(const char *path, int32_t diskPrimary, STfs *pTfs, SMsgCb msgCb) {
   SVnode    *pVnode = NULL;
   SVnodeInfo info = {0};
@@ -305,6 +318,10 @@ SVnode *vnodeOpen(const char *path, int32_t diskPrimary, STfs *pTfs, SMsgCb msgC
   char       tdir[TSDB_FILENAME_LEN * 2] = {0};
   int32_t    ret = 0;
 
+  if (vnodeCheckDisk(diskPrimary, pTfs)) {
+    vError("failed to open vnode from %s since %s. diskPrimary:%d", path, terrstr(), diskPrimary);
+    return NULL;
+  }
   vnodeGetPrimaryDir(path, diskPrimary, pTfs, dir, TSDB_FILENAME_LEN);
 
   info.config = vnodeCfgDefault;

From 02f4613154eb3237058d69811e753009a3afd11a Mon Sep 17 00:00:00 2001
From: Benguang Zhao <bgzhao@taosdata.com>
Date: Tue, 18 Jul 2023 15:44:51 +0800
Subject: [PATCH 09/13] enh: report VND_INIT_FAILED on vnodes open failure in
 vmOpenVnodes

---
 include/util/taoserror.h                 | 2 +-
 source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 5 +++--
 source/util/src/terror.c                 | 1 +
 3 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/include/util/taoserror.h b/include/util/taoserror.h
index 06fdfa67d5..ae105aa1dc 100644
--- a/include/util/taoserror.h
+++ b/include/util/taoserror.h
@@ -416,7 +416,7 @@ int32_t* taosGetErrno();
 // #define TSDB_CODE_VND_MSG_NOT_PROCESSED      TAOS_DEF_ERROR_CODE(0, 0x0501) // 2.x
 // #define TSDB_CODE_VND_ACTION_NEED_REPROCESS. TAOS_DEF_ERROR_CODE(0, 0x0502) // 2.x
 #define TSDB_CODE_VND_INVALID_VGROUP_ID         TAOS_DEF_ERROR_CODE(0, 0x0503)
-// #define TSDB_CODE_VND_INIT_FAILED            TAOS_DEF_ERROR_CODE(0, 0x0504) // 2.x
+#define TSDB_CODE_VND_INIT_FAILED               TAOS_DEF_ERROR_CODE(0, 0x0504)
 // #define TSDB_CODE_VND_NO_DISKSPACE           TAOS_DEF_ERROR_CODE(0, 0x0505) // 2.x
 // #define TSDB_CODE_VND_NO_DISK_PERMISSIONS    TAOS_DEF_ERROR_CODE(0, 0x0506) // 2.x
 // #define TSDB_CODE_VND_NO_SUCH_FILE_OR_DIR    TAOS_DEF_ERROR_CODE(0, 0x0507) // 2.x
diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
index a89a3d0781..94a753062c 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
@@ -268,7 +268,7 @@ static void *vmOpenVnodeInThread(void *param) {
 
     SVnode *pImpl = vnodeOpen(path, diskPrimary, pMgmt->pTfs, pMgmt->msgCb);
     if (pImpl == NULL) {
-      dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);
+      dError("vgId:%d, failed to open vnode by thread:%d since %s", pCfg->vgId, pThread->threadIndex, terrstr());
       pThread->failed++;
       continue;
     }
@@ -355,6 +355,7 @@ static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) {
 
   if (pMgmt->state.openVnodes != pMgmt->state.totalVnodes) {
     dError("there are total vnodes:%d, opened:%d", pMgmt->state.totalVnodes, pMgmt->state.openVnodes);
+    terrno = TSDB_CODE_VND_INIT_FAILED;
     return -1;
   }
 
@@ -577,7 +578,7 @@ static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
   tmsgReportStartup("vnode-worker", "initialized");
 
   if (vmOpenVnodes(pMgmt) != 0) {
-    dError("failed to open vnode since %s", terrstr());
+    dError("failed to open all vnodes since %s", terrstr());
     goto _OVER;
   }
   tmsgReportStartup("vnode-vnodes", "initialized");
diff --git a/source/util/src/terror.c b/source/util/src/terror.c
index d043d22445..8f5fbc0844 100644
--- a/source/util/src/terror.c
+++ b/source/util/src/terror.c
@@ -329,6 +329,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_DNODE_ONLY_USE_WHEN_OFFLINE,  "Please use this comma
 
 // vnode
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_INVALID_VGROUP_ID,        "Vnode is closed or removed")
+TAOS_DEFINE_ERROR(TSDB_CODE_VND_INIT_FAILED,              "Vnode init failure")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_WRITE_AUTH,            "Database write operation denied")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_NOT_EXIST,                "Vnode not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_ALREADY_EXIST,            "Vnode already exist")

From dfaf9134c65da5303537a3c1467f55931ead3d83 Mon Sep 17 00:00:00 2001
From: Shungang Li <sgli@taosdata.com>
Date: Tue, 18 Jul 2023 16:04:38 +0800
Subject: [PATCH 10/13] fix: alter keeptimeoffset by 'k v' or 'k' 'v'

---
 source/dnode/mnode/impl/src/mndDnode.c | 44 ++++++++++++++++++++------
 1 file changed, 34 insertions(+), 10 deletions(-)

diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c
index 11825a4e50..100513b932 100644
--- a/source/dnode/mnode/impl/src/mndDnode.c
+++ b/source/dnode/mnode/impl/src/mndDnode.c
@@ -70,6 +70,8 @@ static void    mndCancelGetNextConfig(SMnode *pMnode, void *pIter);
 static int32_t mndRetrieveDnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
 static void    mndCancelGetNextDnode(SMnode *pMnode, void *pIter);
 
+static int32_t mndMCfgGetValInt32(SMCfgDnodeReq *pInMCfgReq, int32_t opLen, int32_t *pOutValue);
+
 int32_t mndInitDnode(SMnode *pMnode) {
   SSdbTable table = {
       .sdbType = SDB_DNODE,
@@ -1061,22 +1063,19 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
     strcpy(dcfgReq.config, "monitor");
     snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
   } else if (strncasecmp(cfgReq.config, "keeptimeoffset", 14) == 0) {
-    if (' ' != cfgReq.config[14] && 0 != cfgReq.config[14]) {
-      mError("dnode:%d, failed to config keeptimeoffset since invalid conf:%s", cfgReq.dnodeId, cfgReq.config);
-      terrno = TSDB_CODE_INVALID_CFG;
-      return -1;
-    }
+    int32_t optLen = strlen("keeptimeoffset");
+    int32_t flag = -1;
+    int32_t code = mndMCfgGetValInt32(&cfgReq, optLen, &flag);
+    if (code < 0) return code;
 
-    const char *value = cfgReq.value;
-    int32_t     offset = atoi(value);
-    if (offset < 0 || offset > 23) {
-      mError("dnode:%d, failed to config keepTimeOffset since value:%d. Valid range: [0, 23]", cfgReq.dnodeId, offset);
+    if (flag < 0 || flag > 23) {
+      mError("dnode:%d, failed to config keepTimeOffset since value:%d. Valid range: [0, 23]", cfgReq.dnodeId, flag);
       terrno = TSDB_CODE_INVALID_CFG;
       return -1;
     }
 
     strcpy(dcfgReq.config, "keeptimeoffset");
-    snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", offset);
+    snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
 #ifdef TD_ENTERPRISE
   } else if (strncasecmp(cfgReq.config, "activeCode", 10) == 0 || strncasecmp(cfgReq.config, "cActiveCode", 11) == 0) {
     int8_t opt = strncasecmp(cfgReq.config, "a", 1) == 0 ? DND_ACTIVE_CODE : DND_CONN_ACTIVE_CODE;
@@ -1309,3 +1308,28 @@ static void mndCancelGetNextDnode(SMnode *pMnode, void *pIter) {
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetch(pSdb, pIter);
 }
+
+// get int32_t value from 'SMCfgDnodeReq'
+static int32_t mndMCfgGetValInt32(SMCfgDnodeReq *pMCfgReq, int32_t opLen, int32_t *pOutValue) {
+  terrno = 0;
+  if (' ' != pMCfgReq->config[opLen] && 0 != pMCfgReq->config[opLen]) {
+    goto _err;
+  }
+
+  if (' ' == pMCfgReq->config[opLen]) {
+    // 'key value'
+    if (strlen(pMCfgReq->value) != 0) goto _err;
+    *pOutValue = atoi(pMCfgReq->config + opLen + 1);
+  } else {
+    // 'key' 'value'
+    if (strlen(pMCfgReq->value) == 0) goto _err;
+    *pOutValue = atoi(pMCfgReq->value);
+  }
+
+  return 0;
+
+_err:
+  mError("dnode:%d, failed to config keeptimeoffset since invalid conf:%s", pMCfgReq->dnodeId, pMCfgReq->config);
+  terrno = TSDB_CODE_INVALID_CFG;
+  return -1;
+}

From 224f90761e402d218f12528e633a79a08f818584 Mon Sep 17 00:00:00 2001
From: Benguang Zhao <bgzhao@taosdata.com>
Date: Tue, 18 Jul 2023 16:57:52 +0800
Subject: [PATCH 11/13] fix: set wrapperCfg.diskPrimary in
 vmProcessAlterVnodeTypeReq

---
 source/dnode/mgmt/mgmt_vnode/src/vmHandle.c | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
index bd39ff6a73..d48bd3f847 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
@@ -263,8 +263,8 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     return 0;
   }
 
-  int32_t diskPrimary = vmAllocPrimaryDisk(pMgmt, vnodeCfg.vgId);
-  wrapperCfg.diskPrimary = diskPrimary;
+  wrapperCfg.diskPrimary = vmAllocPrimaryDisk(pMgmt, vnodeCfg.vgId);
+  int32_t diskPrimary = wrapperCfg.diskPrimary;
 
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId);
 
@@ -402,15 +402,16 @@ int32_t vmProcessAlterVnodeTypeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   }
 
   dInfo("vgId:%d, start to close vnode", vgId);
-  int32_t     diskPrimary = pVnode->diskPrimary;
   SWrapperCfg wrapperCfg = {
       .dropped = pVnode->dropped,
       .vgId = pVnode->vgId,
       .vgVersion = pVnode->vgVersion,
+      .diskPrimary = pVnode->diskPrimary,
   };
   tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
   vmCloseVnode(pMgmt, pVnode, false);
 
+  int32_t diskPrimary = wrapperCfg.diskPrimary;
   char path[TSDB_FILENAME_LEN] = {0};
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vgId);
 
@@ -490,7 +491,6 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     return -1;
   }
 
-  int32_t     diskPrimary = pVnode->diskPrimary;
   SWrapperCfg wrapperCfg = {
       .dropped = pVnode->dropped,
       .vgId = dstVgId,
@@ -509,6 +509,7 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   dInfo("vgId:%d, close vnode", srcVgId);
   vmCloseVnode(pMgmt, pVnode, true);
 
+  int32_t diskPrimary = wrapperCfg.diskPrimary;
   char srcPath[TSDB_FILENAME_LEN] = {0};
   char dstPath[TSDB_FILENAME_LEN] = {0};
   snprintf(srcPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, srcVgId);
@@ -604,7 +605,6 @@ int32_t vmProcessAlterVnodeReplicaReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   }
 
   dInfo("vgId:%d, start to close vnode", vgId);
-  int32_t     diskPrimary = pVnode->diskPrimary;
   SWrapperCfg wrapperCfg = {
       .dropped = pVnode->dropped,
       .vgId = pVnode->vgId,
@@ -614,6 +614,7 @@ int32_t vmProcessAlterVnodeReplicaReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
   vmCloseVnode(pMgmt, pVnode, false);
 
+  int32_t diskPrimary = wrapperCfg.diskPrimary;
   char path[TSDB_FILENAME_LEN] = {0};
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vgId);
 

From caa45033526d8a29293d3be5b55d9664d0f0dc30 Mon Sep 17 00:00:00 2001
From: liuyao <54liuyao@163.com>
Date: Tue, 18 Jul 2023 18:02:39 +0800
Subject: [PATCH 12/13] reset state key memory

---
 source/libs/executor/src/timewindowoperator.c | 37 ++++++++++++++++---
 1 file changed, 32 insertions(+), 5 deletions(-)

diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c
index c4111ded92..51da73d4d7 100644
--- a/source/libs/executor/src/timewindowoperator.c
+++ b/source/libs/executor/src/timewindowoperator.c
@@ -4036,7 +4036,7 @@ bool isEqualStateKey(SStateWindowInfo* pWin, char* pKeyData) {
 
 bool compareStateKey(void* data, void* key) {
   if (!data || !key) {
-    return true;
+    return false;
   }
   SStateKeys* stateKey = (SStateKeys*)key;
   stateKey->pData = (char*)key + sizeof(SStateKeys);
@@ -4062,7 +4062,13 @@ void setStateOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId,
   if (code == TSDB_CODE_SUCCESS && !inWinRange(&pAggSup->winRange, &pCurWin->winInfo.sessionWin.win)) {
     code = TSDB_CODE_FAILED;
     releaseOutputBuf(pAggSup->pState, NULL, (SResultRow*)pCurWin->winInfo.pOutputBuf, &pAggSup->pSessionAPI->stateStore);
-    pCurWin->winInfo.pOutputBuf = taosMemoryMalloc(size);
+    pCurWin->winInfo.pOutputBuf = taosMemoryCalloc(1, size);
+    pCurWin->pStateKey =
+      (SStateKeys*)((char*)pCurWin->winInfo.pOutputBuf + (pAggSup->resultRowSize - pAggSup->stateKeySize));
+      pCurWin->pStateKey->bytes = pAggSup->stateKeySize - sizeof(SStateKeys);
+      pCurWin->pStateKey->type = pAggSup->stateKeyType;
+      pCurWin->pStateKey->pData = (char*)pCurWin->pStateKey + sizeof(SStateKeys);
+      pCurWin->pStateKey->isNull = false;
   }
 
   if (code == TSDB_CODE_SUCCESS) {
@@ -4076,11 +4082,19 @@ void setStateOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId,
   }
 
   pNextWin->winInfo.sessionWin = pCurWin->winInfo.sessionWin;
-  pNextWin->winInfo.pOutputBuf = NULL;
-  SStreamStateCur* pCur = pAggSup->stateStore.streamStateSessionSeekKeyNext(pAggSup->pState, &pCurWin->winInfo.sessionWin);
-  code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &pNextWin->winInfo.sessionWin, NULL, 0);
+  SStreamStateCur* pCur = pAggSup->stateStore.streamStateSessionSeekKeyNext(pAggSup->pState, &pNextWin->winInfo.sessionWin);
+  int32_t nextSize = pAggSup->resultRowSize;
+  code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &pNextWin->winInfo.sessionWin, &pNextWin->winInfo.pOutputBuf, &nextSize);
   if (code != TSDB_CODE_SUCCESS) {
     SET_SESSION_WIN_INVALID(pNextWin->winInfo);
+  } else {
+    pNextWin->pStateKey =
+      (SStateKeys*)((char*)pNextWin->winInfo.pOutputBuf + (pAggSup->resultRowSize - pAggSup->stateKeySize));
+    pNextWin->pStateKey->bytes = pAggSup->stateKeySize - sizeof(SStateKeys);
+    pNextWin->pStateKey->type = pAggSup->stateKeyType;
+    pNextWin->pStateKey->pData = (char*)pNextWin->pStateKey + sizeof(SStateKeys);
+    pNextWin->pStateKey->isNull = false;
+    pNextWin->winInfo.isOutput = true;
   }
   pAggSup->stateStore.streamStateFreeCur(pCur);
 }
@@ -4156,6 +4170,9 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
     SStateWindowInfo curWin = {0};
     SStateWindowInfo nextWin = {0};
     setStateOutputBuf(pAggSup, tsCols[i], groupId, pKeyData, &curWin, &nextWin);
+    if (IS_VALID_SESSION_WIN(nextWin.winInfo)) {
+      releaseOutputBuf(pAggSup->pState, NULL, (SResultRow*)nextWin.winInfo.pOutputBuf, &pAPI->stateStore);
+    }
     setSessionWinOutputInfo(pSeUpdated, &curWin.winInfo);
     winRows = updateStateWindowInfo(&curWin, &nextWin, tsCols, groupId, pKeyColInfo, rows, i, &allEqual,
                                     pAggSup->pResultRows, pSeUpdated, pStDeleted);
@@ -4346,9 +4363,19 @@ void streamStateReloadState(SOperatorInfo* pOperator) {
   for (int32_t i = 0; i < num; i++) {
     SStateWindowInfo curInfo = {0};
     SStateWindowInfo nextInfo = {0};
+    SStateWindowInfo dummy = {0};
     setStateOutputBuf(pAggSup, pSeKeyBuf[i].win.skey, pSeKeyBuf[i].groupId, NULL, &curInfo, &nextInfo);
     if (compareStateKey(curInfo.pStateKey,nextInfo.pStateKey)) {
       compactStateWindow(pOperator, &curInfo.winInfo, &nextInfo.winInfo, pInfo->pStUpdated, pInfo->pStDeleted);
+      saveResult(curInfo.winInfo, pInfo->pStUpdated);
+    }
+
+    if (IS_VALID_SESSION_WIN(curInfo.winInfo)) {
+      releaseOutputBuf(pAggSup->pState, NULL, (SResultRow*)curInfo.winInfo.pOutputBuf, &pAggSup->pSessionAPI->stateStore);
+    }
+
+    if (IS_VALID_SESSION_WIN(nextInfo.winInfo)) {
+      releaseOutputBuf(pAggSup->pState, NULL, (SResultRow*)nextInfo.winInfo.pOutputBuf, &pAggSup->pSessionAPI->stateStore);
     }
   }
   taosMemoryFree(pBuf);

From 9513f761721caab080519a3b5eb4ed176e18181e Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Tue, 18 Jul 2023 19:02:24 +0800
Subject: [PATCH 13/13] fix(stream): set downstream task ready state.

---
 source/dnode/vnode/src/tq/tq.c         | 7 ++++---
 source/libs/stream/src/streamRecover.c | 1 +
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index ec9a796cfd..bbdd98e356 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -1137,14 +1137,15 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
       // wait for the stream task get ready for scan history data
       while (((pStreamTask->status.downstreamReady == 0) && (pStreamTask->status.taskStatus != TASK_STATUS__STOP)) ||
              pStreamTask->status.taskStatus == TASK_STATUS__SCAN_HISTORY) {
-        tqDebug("s-task:%s level:%d related stream task:%s not ready for halt, wait for it and recheck in 100ms", pId,
-                pTask->info.taskLevel, pId);
+        tqDebug(
+            "s-task:%s level:%d related stream task:%s(status:%s) not ready for halt, wait for it and recheck in 100ms",
+            pId, pTask->info.taskLevel, pStreamTask->id.idStr, streamGetTaskStatusStr(pStreamTask->status.taskStatus));
         taosMsleep(100);
       }
 
       // now we can stop the stream task execution
       pStreamTask->status.taskStatus = TASK_STATUS__HALT;
-      tqDebug("s-task:%s level:%d status is set to halt by history scan task:%s", pId,
+      tqDebug("s-task:%s level:%d status is set to halt by history scan task:%s", pStreamTask->id.idStr,
               pStreamTask->info.taskLevel, pId);
 
       // if it's an source task, extract the last version in wal.
diff --git a/source/libs/stream/src/streamRecover.c b/source/libs/stream/src/streamRecover.c
index 0f2281ea73..f51efb23d1 100644
--- a/source/libs/stream/src/streamRecover.c
+++ b/source/libs/stream/src/streamRecover.c
@@ -201,6 +201,7 @@ int32_t streamProcessCheckRsp(SStreamTask* pTask, const SStreamTaskCheckRsp* pRs
       if (left == 0) {
         taosArrayDestroy(pTask->checkReqIds);
         pTask->checkReqIds = NULL;
+        pTask->status.downstreamReady = 1;
 
         if (pTask->status.taskStatus == TASK_STATUS__SCAN_HISTORY) {
           qDebug("s-task:%s all %d downstream tasks are ready, now enter into scan-history-data stage, status:%s", id,