From 93dc0aff07e9b72d861793726cda298e459e6e20 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Thu, 15 Dec 2022 17:07:11 +0800
Subject: [PATCH 01/13] change err msg while vnode is stoped

---
 include/libs/qcom/query.h                | 22 ++++++++++++----------
 source/dnode/mgmt/node_mgmt/src/dmMgmt.c |  3 +++
 source/libs/transport/src/transCli.c     |  2 +-
 3 files changed, 16 insertions(+), 11 deletions(-)

diff --git a/include/libs/qcom/query.h b/include/libs/qcom/query.h
index 91ec5f52e5..3f11d2a218 100644
--- a/include/libs/qcom/query.h
+++ b/include/libs/qcom/query.h
@@ -58,8 +58,7 @@ typedef enum {
 #define QUERY_RSP_POLICY_QUICK 1
 
 #define QUERY_MSG_MASK_SHOW_REWRITE() (1 << 0)
-#define TEST_SHOW_REWRITE_MASK(m) (((m) & QUERY_MSG_MASK_SHOW_REWRITE()) != 0)
-
+#define TEST_SHOW_REWRITE_MASK(m)     (((m)&QUERY_MSG_MASK_SHOW_REWRITE()) != 0)
 
 typedef struct STableComInfo {
   uint8_t  numOfTags;     // the number of tags in schema
@@ -128,7 +127,7 @@ typedef struct SDBVgInfo {
   int8_t    hashMethod;
   int32_t   numOfTable;  // DB's table num, unit is TSDB_TABLE_NUM_UNIT
   int64_t   stateTs;
-  SHashObj* vgHash;      // key:vgId, value:SVgroupInfo
+  SHashObj* vgHash;  // key:vgId, value:SVgroupInfo
   SArray*   vgArray;
 } SDBVgInfo;
 
@@ -262,23 +261,26 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
   (NEED_CLIENT_RM_TBLMETA_ERROR(_code) || NEED_CLIENT_REFRESH_VG_ERROR(_code) || \
    NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code))
 
-#define SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR || (_code) == TSDB_CODE_VND_STOPPED || (_code) == TSDB_CODE_APP_IS_STARTING || (_code) == TSDB_CODE_APP_IS_STOPPING)
-#define SYNC_SELF_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_RESTORING || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR)
-#define SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_MNODE_NOT_FOUND) 
+#define SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code)                                    \
+  ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR || \
+   (_code) == TSDB_CODE_VND_STOPPED || (_code) == TSDB_CODE_APP_IS_STARTING || (_code) == TSDB_CODE_APP_IS_STOPPING)
+#define SYNC_SELF_LEADER_REDIRECT_ERROR(_code) \
+  ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_RESTORING || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR)
+#define SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_MNODE_NOT_FOUND)
 
 #define NO_RET_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_RPC_BROKEN_LINK || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL)
 
-#define NEED_REDIRECT_ERROR(_code)                                                      \
+#define NEED_REDIRECT_ERROR(_code)                                              \
   (NO_RET_REDIRECT_ERROR(_code) || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \
    SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || SYNC_OTHER_LEADER_REDIRECT_ERROR(_code))
-   
 
 #define NEED_CLIENT_RM_TBLMETA_REQ(_type)                                                                  \
   ((_type) == TDMT_VND_CREATE_TABLE || (_type) == TDMT_MND_CREATE_STB || (_type) == TDMT_VND_DROP_TABLE || \
    (_type) == TDMT_MND_DROP_STB)
 
-#define NEED_SCHEDULER_REDIRECT_ERROR(_code)                                            \
-  (SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || SYNC_OTHER_LEADER_REDIRECT_ERROR(_code))
+#define NEED_SCHEDULER_REDIRECT_ERROR(_code)                                              \
+  (SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || \
+   SYNC_OTHER_LEADER_REDIRECT_ERROR(_code))
 
 #define REQUEST_TOTAL_EXEC_TIMES 2
 
diff --git a/source/dnode/mgmt/node_mgmt/src/dmMgmt.c b/source/dnode/mgmt/node_mgmt/src/dmMgmt.c
index 02a268afda..b6cce249ea 100644
--- a/source/dnode/mgmt/node_mgmt/src/dmMgmt.c
+++ b/source/dnode/mgmt/node_mgmt/src/dmMgmt.c
@@ -214,6 +214,9 @@ int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) {
       case SNODE:
         terrno = TSDB_CODE_SNODE_NOT_FOUND;
         break;
+      case VNODE:
+        terrno = TSDB_CODE_VND_STOPPED;
+        break;
       default:
         terrno = TSDB_CODE_APP_IS_STOPPING;
         break;
diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c
index 0736b763a5..6ad126162e 100644
--- a/source/libs/transport/src/transCli.c
+++ b/source/libs/transport/src/transCli.c
@@ -1598,7 +1598,7 @@ bool cliGenRetryRule(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
   } else if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_INTERNAL_ERROR ||
              code == TSDB_CODE_SYN_PROPOSE_NOT_READY || code == TSDB_CODE_VND_STOPPED ||
              code == TSDB_CODE_MNODE_NOT_FOUND || code == TSDB_CODE_APP_IS_STARTING ||
-             code == TSDB_CODE_APP_IS_STOPPING) {
+             code == TSDB_CODE_APP_IS_STOPPING || code == TSDB_CODE_VND_STOPPED) {
     tTrace("code str %s, contlen:%d 1", tstrerror(code), pResp->contLen);
     noDelay = cliResetEpset(pCtx, pResp, true);
     transFreeMsg(pResp->pCont);

From 4402bcef947af1cf513468c4503b63b81d4137a7 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Fri, 16 Dec 2022 08:53:33 +0800
Subject: [PATCH 02/13] enh(stream): forbid source db replica > 1

---
 include/util/taoserror.h                |  1 +
 source/client/src/clientTmq.c           |  6 ++----
 source/dnode/mnode/impl/src/mndStream.c | 13 +++++++++++--
 source/dnode/vnode/src/tq/tq.c          |  2 +-
 source/util/src/terror.c                |  1 +
 5 files changed, 16 insertions(+), 7 deletions(-)

diff --git a/include/util/taoserror.h b/include/util/taoserror.h
index e352cfb569..52221bdd44 100644
--- a/include/util/taoserror.h
+++ b/include/util/taoserror.h
@@ -342,6 +342,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_INVALID_STREAM_OPTION     TAOS_DEF_ERROR_CODE(0, 0x03F2)
 #define TSDB_CODE_MND_STREAM_MUST_BE_DELETED    TAOS_DEF_ERROR_CODE(0, 0x03F3)
 #define TSDB_CODE_MND_STREAM_TASK_DROPPED       TAOS_DEF_ERROR_CODE(0, 0x03F4)
+#define TSDB_CODE_MND_MULTI_REPLICA_SOURCE_DB   TAOS_DEF_ERROR_CODE(0, 0x03F5)
 
 // mnode-sma
 #define TSDB_CODE_MND_SMA_ALREADY_EXIST         TAOS_DEF_ERROR_CODE(0, 0x0480)
diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c
index 4352ec69d3..d94a049722 100644
--- a/source/client/src/clientTmq.c
+++ b/source/client/src/clientTmq.c
@@ -1035,7 +1035,7 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
   SCMSubscribeReq req = {0};
   int32_t         code = -1;
 
-  tscDebug("call tmq subscribe, consumer: %" PRId64 ", topic num %d", tmq->consumerId, sz);
+  tscDebug("tmq subscribe, consumer: %" PRId64 ", topic num %d", tmq->consumerId, sz);
 
   req.consumerId = tmq->consumerId;
   tstrncpy(req.clientId, tmq->clientId, 256);
@@ -1043,7 +1043,7 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
   req.topicNames = taosArrayInit(sz, sizeof(void*));
   if (req.topicNames == NULL) goto FAIL;
 
-  tscDebug("call tmq subscribe, consumer: %" PRId64 ", topic num %d", tmq->consumerId, sz);
+  tscDebug("tmq subscribe, consumer: %" PRId64 ", topic num %d", tmq->consumerId, sz);
 
   for (int32_t i = 0; i < sz; i++) {
     char* topic = taosArrayGetP(container, i);
@@ -1570,7 +1570,6 @@ SMqTaosxRspObj* tmqBuildTaosxRspFromWrapper(SMqPollRspWrapper* pWrapper) {
 }
 
 int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
-  /*tscDebug("call poll");*/
   for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
     SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
     for (int j = 0; j < taosArrayGetSize(pTopic->vgs); j++) {
@@ -1794,7 +1793,6 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
 }
 
 TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
-  /*tscDebug("call poll1");*/
   void*   rspObj;
   int64_t startTime = taosGetTimestampMs();
 
diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c
index 7ee688d220..6ada70e1b2 100644
--- a/source/dnode/mnode/impl/src/mndStream.c
+++ b/source/dnode/mnode/impl/src/mndStream.c
@@ -164,7 +164,8 @@ SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw) {
 STREAM_DECODE_OVER:
   taosMemoryFreeClear(buf);
   if (terrno != TSDB_CODE_SUCCESS) {
-    mError("stream:%s, failed to decode from raw:%p since %s", pStream == NULL ? "null" : pStream->name, pRaw, terrstr());
+    mError("stream:%s, failed to decode from raw:%p since %s", pStream == NULL ? "null" : pStream->name, pRaw,
+           terrstr());
     taosMemoryFreeClear(pRow);
     return NULL;
   }
@@ -624,6 +625,15 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
+  pDb = mndAcquireDb(pMnode, streamObj.sourceDb);
+  if (pDb->cfg.replications != 1) {
+    mError("stream source db must have only 1 replica, but %s has %d", pDb->name, pDb->cfg.replications);
+    terrno = TSDB_CODE_MND_MULTI_REPLICA_SOURCE_DB;
+    mndReleaseDb(pMnode, pDb);
+    pDb = NULL;
+    goto _OVER;
+  }
+
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq, "create-stream");
   if (pTrans == NULL) {
     mError("stream:%s, failed to create since %s", createStreamReq.name, terrstr());
@@ -680,7 +690,6 @@ _OVER:
   }
 
   mndReleaseStream(pMnode, pStream);
-  mndReleaseDb(pMnode, pDb);
 
   tFreeSCMCreateStreamReq(&createStreamReq);
   tFreeStreamObj(&streamObj);
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 389c8013f9..356cbd6332 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -738,7 +738,7 @@ int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t version, char* msg, int32_t msgL
 
   code = tqOffsetDelete(pTq->pOffsetStore, pReq->subKey);
   if (code != 0) {
-    tqError("cannot process tq delete req %s, since no such offset", pReq->subKey);
+    tqError("cannot process tq delete req %s, since no such offset in cache", pReq->subKey);
   }
 
   if (tqMetaDeleteHandle(pTq, pReq->subKey) < 0) {
diff --git a/source/util/src/terror.c b/source/util/src/terror.c
index 656d775ea4..4b9dde5059 100644
--- a/source/util/src/terror.c
+++ b/source/util/src/terror.c
@@ -281,6 +281,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_STREAM_ALREADY_EXIST,     "Stream already exists
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_STREAM_NOT_EXIST,         "Stream not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_STREAM_OPTION,    "Invalid stream option")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_STREAM_MUST_BE_DELETED,   "Stream must be dropped first")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_MULTI_REPLICA_SOURCE_DB,  "Stream temporarily does not support source db having replica > 1")
 
 // mnode-sma
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_SMA_ALREADY_EXIST,        "SMA already exists")

From 9adf7d89a7845f38b51045ac071e348ba7aa961d Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Fri, 16 Dec 2022 12:28:22 +0800
Subject: [PATCH 03/13] fix: incorrect error code for repeated dnode creation

---
 source/dnode/mnode/impl/src/mndDnode.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c
index 0538d70101..5772864ba8 100644
--- a/source/dnode/mnode/impl/src/mndDnode.c
+++ b/source/dnode/mnode/impl/src/mndDnode.c
@@ -672,6 +672,7 @@ static int32_t mndProcessCreateDnodeReq(SRpcMsg *pReq) {
   snprintf(ep, TSDB_EP_LEN, "%s:%d", createReq.fqdn, createReq.port);
   pDnode = mndAcquireDnodeByEp(pMnode, ep);
   if (pDnode != NULL) {
+    terrno = TSDB_CODE_MND_DNODE_ALREADY_EXIST;
     goto _OVER;
   }
 

From 4ab9638f9a557b73dc07bc3b30b55c2273a4d44b Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Fri, 16 Dec 2022 14:45:52 +0800
Subject: [PATCH 04/13] enh(stream): forbid source db replica > 1

---
 source/dnode/mnode/impl/src/mndStream.c | 1 +
 tests/system-test/1-insert/drop.py      | 4 ++--
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c
index 6ada70e1b2..05c0594339 100644
--- a/source/dnode/mnode/impl/src/mndStream.c
+++ b/source/dnode/mnode/impl/src/mndStream.c
@@ -633,6 +633,7 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
     pDb = NULL;
     goto _OVER;
   }
+  mndReleaseDb(pMnode, pDb);
 
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq, "create-stream");
   if (pTrans == NULL) {
diff --git a/tests/system-test/1-insert/drop.py b/tests/system-test/1-insert/drop.py
index f8796bcf6a..c15a9bbc35 100644
--- a/tests/system-test/1-insert/drop.py
+++ b/tests/system-test/1-insert/drop.py
@@ -132,7 +132,7 @@ class TDTestCase:
         tdSql.execute(f'drop database {self.dbname}')
 
     def drop_stream_check(self):
-        tdSql.execute(f'create database {self.dbname} replica {self.replicaVar}')
+        tdSql.execute(f'create database {self.dbname} replica 1')
         tdSql.execute(f'use {self.dbname}')
         stbname = tdCom.getLongName(5,"letters")
         stream_name = tdCom.getLongName(5,"letters")
@@ -158,4 +158,4 @@ class TDTestCase:
         tdLog.success("%s successfully executed" % __file__)
 
 tdCases.addWindows(__file__, TDTestCase())
-tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file
+tdCases.addLinux(__file__, TDTestCase())

From 12088812dd0d4430aea237512d8f8964d4d0dc8f Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Fri, 16 Dec 2022 15:54:19 +0800
Subject: [PATCH 05/13] fix: should not fsync in log level is error

---
 source/os/src/osFile.c |  4 +++-
 source/util/src/tlog.c | 10 ++++++++++
 2 files changed, 13 insertions(+), 1 deletion(-)

diff --git a/source/os/src/osFile.c b/source/os/src/osFile.c
index 8c2170239f..d8cccc83ed 100644
--- a/source/os/src/osFile.c
+++ b/source/os/src/osFile.c
@@ -464,7 +464,9 @@ int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count) {
 #if FILE_WITH_LOCK
   taosThreadRwlockWrlock(&(pFile->rwlock));
 #endif
-  assert(pFile->fd >= 0);  // Please check if you have closed the file.
+  if (pFile->fd < 0) {
+    return 0;
+  }
 
   int64_t nleft = count;
   int64_t nwritten = 0;
diff --git a/source/util/src/tlog.c b/source/util/src/tlog.c
index 747187254f..f6f814d82b 100644
--- a/source/util/src/tlog.c
+++ b/source/util/src/tlog.c
@@ -445,6 +445,9 @@ static inline int32_t taosBuildLogHead(char *buffer, const char *flags) {
 static inline void taosPrintLogImp(ELogLevel level, int32_t dflag, const char *buffer, int32_t len) {
   if ((dflag & DEBUG_FILE) && tsLogObj.logHandle && tsLogObj.logHandle->pFile != NULL && osLogSpaceAvailable()) {
     taosUpdateLogNums(level);
+#if 0 
+    // DEBUG_FATAL and DEBUG_ERROR are duplicated
+    // fsync will cause thread blocking and may also generate log misalignment in case of asyncLog
     if (tsAsyncLog && level != DEBUG_FATAL) {
       taosPushLogBuffer(tsLogObj.logHandle, buffer, len);
     } else {
@@ -453,6 +456,13 @@ static inline void taosPrintLogImp(ELogLevel level, int32_t dflag, const char *b
         taosFsyncFile(tsLogObj.logHandle->pFile);
       }
     }
+#else
+    if (tsAsyncLog) {
+      taosPushLogBuffer(tsLogObj.logHandle, buffer, len);
+    } else {
+      taosWriteFile(tsLogObj.logHandle->pFile, buffer, len);
+    }
+#endif
 
     if (tsLogObj.maxLines > 0) {
       atomic_add_fetch_32(&tsLogObj.lines, 1);

From 728cbbac5eb7b30e1646ce2725a0e9cb1a1a712c Mon Sep 17 00:00:00 2001
From: kailixu <klxu@taosdata.com>
Date: Fri, 16 Dec 2022 16:55:12 +0800
Subject: [PATCH 06/13] fix(tsdb): default value of tsdbPageSize

---
 source/dnode/vnode/src/vnd/vnodeCfg.c | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/source/dnode/vnode/src/vnd/vnodeCfg.c b/source/dnode/vnode/src/vnd/vnodeCfg.c
index 5adb2eb359..1461d54d7c 100644
--- a/source/dnode/vnode/src/vnd/vnodeCfg.c
+++ b/source/dnode/vnode/src/vnd/vnodeCfg.c
@@ -115,7 +115,6 @@ int vnodeEncodeConfig(const void *pObj, SJson *pJson) {
   if (tjsonAddIntegerToObject(pJson, "hashMethod", pCfg->hashMethod) < 0) return -1;
   if (tjsonAddIntegerToObject(pJson, "hashPrefix", pCfg->hashPrefix) < 0) return -1;
   if (tjsonAddIntegerToObject(pJson, "hashSuffix", pCfg->hashSuffix) < 0) return -1;
-  if (tjsonAddIntegerToObject(pJson, "tsdbPageSize", pCfg->tsdbPageSize) < 0) return -1;
 
   if (tjsonAddIntegerToObject(pJson, "syncCfg.replicaNum", pCfg->syncCfg.replicaNum) < 0) return -1;
   if (tjsonAddIntegerToObject(pJson, "syncCfg.myIndex", pCfg->syncCfg.myIndex) < 0) return -1;
@@ -256,7 +255,9 @@ int vnodeDecodeConfig(const SJson *pJson, void *pObj) {
   }
 
   tjsonGetNumberValue(pJson, "tsdbPageSize", pCfg->tsdbPageSize, code);
-  if (code < 0) pCfg->tsdbPageSize = TSDB_DEFAULT_TSDB_PAGESIZE * 1024;
+  if (code < 0 || pCfg->tsdbPageSize < TSDB_MIN_PAGESIZE_PER_VNODE) {
+    pCfg->tsdbPageSize = TSDB_DEFAULT_TSDB_PAGESIZE * 1024;
+  }
 
   return 0;
 }

From 9b98f547d1d0fb5f9e9b74096552dfab69e2140f Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Fri, 16 Dec 2022 17:08:16 +0800
Subject: [PATCH 07/13] fix: reset errorcode after check leader

---
 source/dnode/mnode/impl/src/mndMain.c  | 2 ++
 source/dnode/mnode/impl/src/mndTrans.c | 4 +++-
 2 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c
index 117a9f5e67..854535c82f 100644
--- a/source/dnode/mnode/impl/src/mndMain.c
+++ b/source/dnode/mnode/impl/src/mndMain.c
@@ -663,7 +663,9 @@ _OVER:
 
   const STraceId *trace = &pMsg->info.traceId;
   SEpSet          epSet = {0};
+  int32_t         tmpCode = terrno;
   mndGetMnodeEpSet(pMnode, &epSet);
+  terrno = tmpCode;
 
   mGDebug(
       "msg:%p, type:%s failed to process since %s, mnode restored:%d stopped:%d, sync restored:%d "
diff --git a/source/dnode/mnode/impl/src/mndTrans.c b/source/dnode/mnode/impl/src/mndTrans.c
index 42037304b9..d2fc2dc9b1 100644
--- a/source/dnode/mnode/impl/src/mndTrans.c
+++ b/source/dnode/mnode/impl/src/mndTrans.c
@@ -927,7 +927,8 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
     }
   } else {
     if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
-      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING) {
+      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING ||
+          code == TSDB_CODE_SYN_PROPOSE_NOT_READY) {
         if (pTrans->failedTimes > 60) sendRsp = true;
       } else {
         if (pTrans->failedTimes > 6) sendRsp = true;
@@ -1336,6 +1337,7 @@ static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans) {
   }
 
   if (mndCannotExecuteTransAction(pMnode)) return false;
+  terrno = code;
 
   if (code == 0) {
     pTrans->code = 0;

From 96bfe146d54b6c6259f082010c477dc5b49a001b Mon Sep 17 00:00:00 2001
From: kailixu <klxu@taosdata.com>
Date: Fri, 16 Dec 2022 18:28:10 +0800
Subject: [PATCH 08/13] fix: default value of tsdbPageSize

---
 source/dnode/vnode/src/vnd/vnodeCfg.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/dnode/vnode/src/vnd/vnodeCfg.c b/source/dnode/vnode/src/vnd/vnodeCfg.c
index 1461d54d7c..c96d01f93a 100644
--- a/source/dnode/vnode/src/vnd/vnodeCfg.c
+++ b/source/dnode/vnode/src/vnd/vnodeCfg.c
@@ -255,7 +255,7 @@ int vnodeDecodeConfig(const SJson *pJson, void *pObj) {
   }
 
   tjsonGetNumberValue(pJson, "tsdbPageSize", pCfg->tsdbPageSize, code);
-  if (code < 0 || pCfg->tsdbPageSize < TSDB_MIN_PAGESIZE_PER_VNODE) {
+  if (code < 0 || pCfg->tsdbPageSize < TSDB_MIN_PAGESIZE_PER_VNODE * 1024) {
     pCfg->tsdbPageSize = TSDB_DEFAULT_TSDB_PAGESIZE * 1024;
   }
 

From 8aa20b531078e88bc9dd9b3141cad0bb2690eb36 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Sat, 17 Dec 2022 17:30:43 +0800
Subject: [PATCH 09/13] docs(stream)

---
 docs/en/12-taos-sql/14-stream.md | 30 +++++++++++++++++++++++++++++-
 docs/zh/12-taos-sql/14-stream.md | 29 +++++++++++++++++++++++++----
 2 files changed, 54 insertions(+), 5 deletions(-)

diff --git a/docs/en/12-taos-sql/14-stream.md b/docs/en/12-taos-sql/14-stream.md
index 17e4e4d1b0..8c81dcaeef 100644
--- a/docs/en/12-taos-sql/14-stream.md
+++ b/docs/en/12-taos-sql/14-stream.md
@@ -10,7 +10,7 @@ Because stream processing is built in to TDengine, you are no longer reliant on
 ## Create a Stream
 
 ```sql
-CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name AS subquery
+CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name SUBTABLE(expression) AS subquery
 stream_options: {
  TRIGGER    [AT_ONCE | WINDOW_CLOSE | MAX_DELAY time]
  WATERMARK   time
@@ -30,6 +30,8 @@ subquery: SELECT [DISTINCT] select_list
 
 Session windows, state windows, and sliding windows are supported. When you configure a session or state window for a supertable, you must use PARTITION BY TBNAME.
 
+Subtable Clause defines the naming rules of auto-created subtable, you can see more details in below part: Partitions of Stream.
+
 ```sql
 window_clause: {
     SESSION(ts_col, tol_val)
@@ -47,6 +49,32 @@ CREATE STREAM avg_vol_s INTO avg_vol AS
 SELECT _wstart, count(*), avg(voltage) FROM meters PARTITION BY tbname INTERVAL(1m) SLIDING(30s);
 ```
 
+## Partitions of Stream
+
+A Stream can process data in multiple partitions. Partition rules can be defined by PARTITION BY clause in stream processing. Each partition will have different timelines and windows, and will be processed separately and be written into different subtables of target supertable.
+
+If a stream is created without PARTITION BY clause, all data will be written into one subtable.
+
+If a stream is created with PARTITION BY clause without SUBTABLE clause, each partition will be given a random name. 
+
+If a stream is created with PARTITION BY clause and SUBTABLE clause, the name of each partition will be calculated according to SUBTABLE clause. For example:
+
+```sql
+CREATE STREAM avg_vol_s INTO avg_vol SUBTABLE(CONCAT('new-', tname)) AS SELECT _wstart, count(*), avg(voltage) FROM meters PARTITION BY tbname tname INTERVAL(1m);
+```
+
+IN PARTITION clause, 'tbname', representing each subtable name of source supertable, is given alias 'tname'. And 'tname' is used in SUBTABLE clause. In SUBTABLE clause, each auto created subtable will concat 'new-' and source subtable name as their name. Other expressions are also allowed in SUBTABLE clause, but the output type must be varchar.
+
+If the output length exceeds the limitation of TDengine(192), the name will be truncated. If the generated name is occupied by some other table, the creation and writing of the new subtable will be failed.
+
+## Filling history data
+
+Normally a stream does not process data already or being written into source table when it's being creating. But adding FILL_HISTORY 1 as a stream option when creating the stream will allow it to process data written before and while creating the stream. For example:
+
+```sql
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 interval(10s)
+```
+
 ## Delete a Stream
 
 ```sql
diff --git a/docs/zh/12-taos-sql/14-stream.md b/docs/zh/12-taos-sql/14-stream.md
index 932ad30b1a..1e6be13639 100644
--- a/docs/zh/12-taos-sql/14-stream.md
+++ b/docs/zh/12-taos-sql/14-stream.md
@@ -8,7 +8,7 @@ description: 流式计算的相关 SQL 的详细语法
 ## 创建流式计算
 
 ```sql
-CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name AS subquery
+CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name SUBTABLE(expression) AS subquery
 stream_options: {
  TRIGGER    [AT_ONCE | WINDOW_CLOSE | MAX_DELAY time]
  WATERMARK   time
@@ -28,6 +28,9 @@ subquery: SELECT select_list
 
 支持会话窗口、状态窗口与滑动窗口，其中，会话窗口与状态窗口搭配超级表时必须与partition by tbname一起使用
 
+
+subtable 子句定义了流式计算中创建的子表的命名规则，详见 流式计算的 partition 部分。
+
 ```sql
 window_clause: {
     SESSION(ts_col, tol_val)
@@ -49,11 +52,29 @@ SELECT _wstart, count(*), avg(voltage) FROM meters PARTITION BY tbname INTERVAL(
 
 ## 流式计算的 partition
 
-可以使用 PARTITION BY TBNAME 或 PARTITION BY tag，对一个流进行多分区的计算，每个分区的时间线与时间窗口是独立的，会各自聚合，并写入到目的表中的不同子表。
+可以使用 PARTITION BY TBNAME，tag，普通列或者表达式，对一个流进行多分区的计算，每个分区的时间线与时间窗口是独立的，会各自聚合，并写入到目的表中的不同子表。
 
-不带 PARTITION BY 选项时，所有的数据将写入到一张子表。
+不带 PARTITION BY 子句时，所有的数据将写入到一张子表。
 
-流式计算创建的超级表有唯一的 tag 列 groupId，每个 partition 会被分配唯一 groupId。与 schemaless 写入一致，我们通过 MD5 计算子表名，并自动创建它。
+在创建流时不使用 SUBTABLE 子句时，流式计算创建的超级表有唯一的 tag 列 groupId，每个 partition 会被分配唯一 groupId。与 schemaless 写入一致，我们通过 MD5 计算子表名，并自动创建它。
+
+若创建流的语句中包含 SUBTABLE 子句，用户可以为每个 partition 对应的子表生成自定义的表名，例如：
+
+```sql
+CREATE STREAM avg_vol_s INTO avg_vol SUBTABLE(CONCAT('new-', tname)) AS SELECT _wstart, count(*), avg(voltage) FROM meters PARTITION BY tbname tname INTERVAL(1m);
+```
+
+PARTITION 子句中，为 tbname 定义了一个别名 tname, 在PARTITION 子句中的别名可以用于 SUBTABLE 子句中的表达式计算，在上述示例中，流新创建的子表将以前缀 'new-' 连接原表名作为表名。
+
+注意，子表名的长度若超过 TDengine 的限制，将被截断。若要生成的子表名已经存在于另一超级表，由于 TDengine 的子表名是唯一的，因此对应新子表的创建以及数据的写入将会失败。
+
+## 流式计算读取历史数据
+
+正常情况下，流式计算不会处理创建前已经写入源表中的数据，若要处理已经写入的数据，可以在创建流时设置 fill_history 1 选项，这样创建的流式计算会自动处理创建前、创建中、创建后写入的数据。例如：
+
+```sql
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 interval(10s)
+```
 
 ## 删除流式计算
 

From 191c43654dfe0575173b0dd88bc1159c56aefc1c Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Sat, 17 Dec 2022 18:08:01 +0800
Subject: [PATCH 10/13] docs(stream) (#18988)

---
 docs/en/12-taos-sql/14-stream.md | 30 +++++++++++++++++++++++++++++-
 docs/zh/12-taos-sql/14-stream.md | 29 +++++++++++++++++++++++++----
 2 files changed, 54 insertions(+), 5 deletions(-)

diff --git a/docs/en/12-taos-sql/14-stream.md b/docs/en/12-taos-sql/14-stream.md
index 17e4e4d1b0..8c81dcaeef 100644
--- a/docs/en/12-taos-sql/14-stream.md
+++ b/docs/en/12-taos-sql/14-stream.md
@@ -10,7 +10,7 @@ Because stream processing is built in to TDengine, you are no longer reliant on
 ## Create a Stream
 
 ```sql
-CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name AS subquery
+CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name SUBTABLE(expression) AS subquery
 stream_options: {
  TRIGGER    [AT_ONCE | WINDOW_CLOSE | MAX_DELAY time]
  WATERMARK   time
@@ -30,6 +30,8 @@ subquery: SELECT [DISTINCT] select_list
 
 Session windows, state windows, and sliding windows are supported. When you configure a session or state window for a supertable, you must use PARTITION BY TBNAME.
 
+Subtable Clause defines the naming rules of auto-created subtable, you can see more details in below part: Partitions of Stream.
+
 ```sql
 window_clause: {
     SESSION(ts_col, tol_val)
@@ -47,6 +49,32 @@ CREATE STREAM avg_vol_s INTO avg_vol AS
 SELECT _wstart, count(*), avg(voltage) FROM meters PARTITION BY tbname INTERVAL(1m) SLIDING(30s);
 ```
 
+## Partitions of Stream
+
+A Stream can process data in multiple partitions. Partition rules can be defined by PARTITION BY clause in stream processing. Each partition will have different timelines and windows, and will be processed separately and be written into different subtables of target supertable.
+
+If a stream is created without PARTITION BY clause, all data will be written into one subtable.
+
+If a stream is created with PARTITION BY clause without SUBTABLE clause, each partition will be given a random name. 
+
+If a stream is created with PARTITION BY clause and SUBTABLE clause, the name of each partition will be calculated according to SUBTABLE clause. For example:
+
+```sql
+CREATE STREAM avg_vol_s INTO avg_vol SUBTABLE(CONCAT('new-', tname)) AS SELECT _wstart, count(*), avg(voltage) FROM meters PARTITION BY tbname tname INTERVAL(1m);
+```
+
+IN PARTITION clause, 'tbname', representing each subtable name of source supertable, is given alias 'tname'. And 'tname' is used in SUBTABLE clause. In SUBTABLE clause, each auto created subtable will concat 'new-' and source subtable name as their name. Other expressions are also allowed in SUBTABLE clause, but the output type must be varchar.
+
+If the output length exceeds the limitation of TDengine(192), the name will be truncated. If the generated name is occupied by some other table, the creation and writing of the new subtable will be failed.
+
+## Filling history data
+
+Normally a stream does not process data already or being written into source table when it's being creating. But adding FILL_HISTORY 1 as a stream option when creating the stream will allow it to process data written before and while creating the stream. For example:
+
+```sql
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 interval(10s)
+```
+
 ## Delete a Stream
 
 ```sql
diff --git a/docs/zh/12-taos-sql/14-stream.md b/docs/zh/12-taos-sql/14-stream.md
index 932ad30b1a..1e6be13639 100644
--- a/docs/zh/12-taos-sql/14-stream.md
+++ b/docs/zh/12-taos-sql/14-stream.md
@@ -8,7 +8,7 @@ description: 流式计算的相关 SQL 的详细语法
 ## 创建流式计算
 
 ```sql
-CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name AS subquery
+CREATE STREAM [IF NOT EXISTS] stream_name [stream_options] INTO stb_name SUBTABLE(expression) AS subquery
 stream_options: {
  TRIGGER    [AT_ONCE | WINDOW_CLOSE | MAX_DELAY time]
  WATERMARK   time
@@ -28,6 +28,9 @@ subquery: SELECT select_list
 
 支持会话窗口、状态窗口与滑动窗口，其中，会话窗口与状态窗口搭配超级表时必须与partition by tbname一起使用
 
+
+subtable 子句定义了流式计算中创建的子表的命名规则，详见 流式计算的 partition 部分。
+
 ```sql
 window_clause: {
     SESSION(ts_col, tol_val)
@@ -49,11 +52,29 @@ SELECT _wstart, count(*), avg(voltage) FROM meters PARTITION BY tbname INTERVAL(
 
 ## 流式计算的 partition
 
-可以使用 PARTITION BY TBNAME 或 PARTITION BY tag，对一个流进行多分区的计算，每个分区的时间线与时间窗口是独立的，会各自聚合，并写入到目的表中的不同子表。
+可以使用 PARTITION BY TBNAME，tag，普通列或者表达式，对一个流进行多分区的计算，每个分区的时间线与时间窗口是独立的，会各自聚合，并写入到目的表中的不同子表。
 
-不带 PARTITION BY 选项时，所有的数据将写入到一张子表。
+不带 PARTITION BY 子句时，所有的数据将写入到一张子表。
 
-流式计算创建的超级表有唯一的 tag 列 groupId，每个 partition 会被分配唯一 groupId。与 schemaless 写入一致，我们通过 MD5 计算子表名，并自动创建它。
+在创建流时不使用 SUBTABLE 子句时，流式计算创建的超级表有唯一的 tag 列 groupId，每个 partition 会被分配唯一 groupId。与 schemaless 写入一致，我们通过 MD5 计算子表名，并自动创建它。
+
+若创建流的语句中包含 SUBTABLE 子句，用户可以为每个 partition 对应的子表生成自定义的表名，例如：
+
+```sql
+CREATE STREAM avg_vol_s INTO avg_vol SUBTABLE(CONCAT('new-', tname)) AS SELECT _wstart, count(*), avg(voltage) FROM meters PARTITION BY tbname tname INTERVAL(1m);
+```
+
+PARTITION 子句中，为 tbname 定义了一个别名 tname, 在PARTITION 子句中的别名可以用于 SUBTABLE 子句中的表达式计算，在上述示例中，流新创建的子表将以前缀 'new-' 连接原表名作为表名。
+
+注意，子表名的长度若超过 TDengine 的限制，将被截断。若要生成的子表名已经存在于另一超级表，由于 TDengine 的子表名是唯一的，因此对应新子表的创建以及数据的写入将会失败。
+
+## 流式计算读取历史数据
+
+正常情况下，流式计算不会处理创建前已经写入源表中的数据，若要处理已经写入的数据，可以在创建流时设置 fill_history 1 选项，这样创建的流式计算会自动处理创建前、创建中、创建后写入的数据。例如：
+
+```sql
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 interval(10s)
+```
 
 ## 删除流式计算
 

From 726951fef05bfe737353b53faf34996018e27938 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Sat, 17 Dec 2022 18:27:23 +0800
Subject: [PATCH 11/13] docs(stream)

---
 docs/en/12-taos-sql/14-stream.md | 14 ++++++++++++++
 docs/zh/12-taos-sql/14-stream.md | 14 ++++++++++++++
 2 files changed, 28 insertions(+)

diff --git a/docs/en/12-taos-sql/14-stream.md b/docs/en/12-taos-sql/14-stream.md
index 8c81dcaeef..bd1858d93f 100644
--- a/docs/en/12-taos-sql/14-stream.md
+++ b/docs/en/12-taos-sql/14-stream.md
@@ -75,6 +75,20 @@ Normally a stream does not process data already or being written into source tab
 create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 interval(10s)
 ```
 
+Combining fill_history option and where clause, stream can processing data of specific time range. For example, only process data after a past time. (In this case, 2020-01-30)
+
+```sql
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > timestamp '2020-01-30' interval(10s)
+```
+
+As another example, only processing data starting from some past time, and ending at some future time.
+
+```sql
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > timestamp '2020-01-30' and ts < timestamp '2023-01-01 '> interval(10s)
+```
+
+If some streams are totally outdated, and you do not want it to monitor or process anymore, those streams can be manually dropped and output data will be still kept.
+
 ## Delete a Stream
 
 ```sql
diff --git a/docs/zh/12-taos-sql/14-stream.md b/docs/zh/12-taos-sql/14-stream.md
index 1e6be13639..6c18c41f0c 100644
--- a/docs/zh/12-taos-sql/14-stream.md
+++ b/docs/zh/12-taos-sql/14-stream.md
@@ -76,6 +76,20 @@ PARTITION 子句中，为 tbname 定义了一个别名 tname, 在PARTITION 子
 create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 interval(10s)
 ```
 
+结合 fill_history 1 选项，可以实现只处理特定历史时间范围的数据，例如：只处理某历史时刻（2020年1月30日）之后的数据
+
+```sql
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > timestamp '2020-01-30' interval(10s)
+```
+
+再如，仅处理某时间段内的数据，结束时间可以是未来时间
+
+```sql
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > timestamp '2020-01-30' and ts < timestamp '2023-01-01 '> interval(10s)
+```
+
+如果该流任务已经彻底过期，并且您不再想让它检测或处理数据，您可以手动删除它，被计算出的数据仍会被保留。
+
 ## 删除流式计算
 
 ```sql

From ead5d8edf21317228cbdab4e18942c269759e2eb Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Sat, 17 Dec 2022 18:34:29 +0800
Subject: [PATCH 12/13] fix: reduce drop dnode speed makes data more secure

---
 include/common/tmsg.h                   |  2 +-
 source/dnode/mnode/impl/src/mndVgroup.c | 10 ++++++++-
 source/libs/sync/src/syncMain.c         |  2 +-
 source/libs/sync/src/syncRespMgr.c      | 30 ++++++++++++++++++-------
 4 files changed, 33 insertions(+), 11 deletions(-)

diff --git a/include/common/tmsg.h b/include/common/tmsg.h
index bb1addf1b6..ad6077db09 100644
--- a/include/common/tmsg.h
+++ b/include/common/tmsg.h
@@ -68,7 +68,7 @@ typedef uint16_t tmsg_t;
 
 static inline bool vnodeIsMsgBlock(tmsg_t type) {
   return (type == TDMT_VND_CREATE_TABLE) || (type == TDMT_VND_ALTER_TABLE) || (type == TDMT_VND_DROP_TABLE) ||
-         (type == TDMT_VND_UPDATE_TAG_VAL);
+         (type == TDMT_VND_UPDATE_TAG_VAL) || (type == TDMT_VND_ALTER_CONFIRM);
 }
 
 static inline bool syncUtilUserCommit(tmsg_t msgType) {
diff --git a/source/dnode/mnode/impl/src/mndVgroup.c b/source/dnode/mnode/impl/src/mndVgroup.c
index 4a6f0d14da..31ab1f3259 100644
--- a/source/dnode/mnode/impl/src/mndVgroup.c
+++ b/source/dnode/mnode/impl/src/mndVgroup.c
@@ -1126,8 +1126,12 @@ int32_t mndSetMoveVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb,
   }
 
   if (!force) {
+#if 1
+    {
+#else
     if (newVg.replica == 1) {
-      mInfo("vgId:%d, will add 1 vnode, replca:1", pVgroup->vgId);
+#endif
+      mInfo("vgId:%d, will add 1 vnode, replca:%d", pVgroup->vgId, newVg.replica);
       if (mndAddVnodeToVgroup(pMnode, pTrans, &newVg, pArray) != 0) return -1;
       for (int32_t i = 0; i < newVg.replica - 1; ++i) {
         if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg, newVg.vnodeGid[i].dnodeId) != 0) return -1;
@@ -1155,6 +1159,9 @@ int32_t mndSetMoveVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb,
         if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg, newVg.vnodeGid[i].dnodeId) != 0) return -1;
       }
       if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVg) != 0) return -1;
+#if 1
+    }
+#else
     } else {  // new replica == 3
       mInfo("vgId:%d, will add 1 vnode, replca:3", pVgroup->vgId);
       if (mndAddVnodeToVgroup(pMnode, pTrans, &newVg, pArray) != 0) return -1;
@@ -1181,6 +1188,7 @@ int32_t mndSetMoveVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb,
       if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVg, &newVg.vnodeGid[vnIndex]) != 0) return -1;
       if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVg) != 0) return -1;
     }
+#endif
   } else {
     mInfo("vgId:%d, will add 1 vnode and force remove 1 vnode", pVgroup->vgId);
     if (mndAddVnodeToVgroup(pMnode, pTrans, &newVg, pArray) != 0) return -1;
diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c
index d6ce77193a..6a545424fc 100644
--- a/source/libs/sync/src/syncMain.c
+++ b/source/libs/sync/src/syncMain.c
@@ -233,7 +233,7 @@ int32_t syncSendTimeoutRsp(int64_t rid, int64_t seq) {
     rpcSendResponse(&rpcMsg);
     return 0;
   } else {
-    sInfo("no rpcinfo to send timeout response, seq:%" PRId64, seq);
+    sError("no message handle to send timeout response, seq:%" PRId64, seq);
     return -1;
   }
 }
diff --git a/source/libs/sync/src/syncRespMgr.c b/source/libs/sync/src/syncRespMgr.c
index 049b02d73e..79a38cad7a 100644
--- a/source/libs/sync/src/syncRespMgr.c
+++ b/source/libs/sync/src/syncRespMgr.c
@@ -35,11 +35,16 @@ SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl) {
   pObj->seqNum = 0;
   taosThreadMutexInit(&(pObj->mutex), NULL);
 
+  SSyncNode *pNode = pObj->data;
+  sTrace("vgId:%d, create resp manager", pNode->vgId);
   return pObj;
 }
 
 void syncRespMgrDestroy(SSyncRespMgr *pObj) {
   if (pObj != NULL) {
+    SSyncNode *pNode = pObj->data;
+    sTrace("vgId:%d, destroy resp manager", pNode->vgId);
+
     taosThreadMutexLock(&pObj->mutex);
     taosHashCleanup(pObj->pRespHash);
     taosThreadMutexUnlock(&pObj->mutex);
@@ -81,6 +86,8 @@ int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t seq, SRespStub *pStub) {
 
     taosThreadMutexUnlock(&pObj->mutex);
     return 1;  // get one object
+  } else {
+    sNError(pObj->data, "get message handle, no object of seq:%" PRIu64, seq);
   }
 
   taosThreadMutexUnlock(&pObj->mutex);
@@ -99,6 +106,8 @@ int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t seq, SRpcHandleInfo *p
 
     taosThreadMutexUnlock(&pObj->mutex);
     return 1;  // get one object
+  } else {
+    sNError(pObj->data, "get-and-del message handle, no object of seq:%" PRIu64, seq);
   }
 
   taosThreadMutexUnlock(&pObj->mutex);
@@ -114,7 +123,7 @@ static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
   SArray *delIndexArray = taosArrayInit(4, sizeof(uint64_t));
   if (delIndexArray == NULL) return;
 
-  sDebug("vgId:%d, resp mgr begin clean by ttl", pSyncNode->vgId);
+  sDebug("vgId:%d, resp manager begin clean by ttl", pSyncNode->vgId);
   while (pStub) {
     size_t    len;
     void     *key = taosHashGetKey(pStub, &len);
@@ -143,34 +152,39 @@ static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
 
       // TODO: and make rpcMsg body, call commit cb
       // pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &pStub->rpcMsg, cbMeta);
-
-      pStub->rpcMsg.code = TSDB_CODE_SYN_NOT_LEADER;
-      if (pStub->rpcMsg.info.handle != NULL) {
-        tmsgSendRsp(&pStub->rpcMsg);
-      }
+      SRpcMsg rpcMsg = {.info = pStub->rpcMsg.info, .code = TSDB_CODE_SYN_TIMEOUT};
+      sInfo("vgId:%d, message handle:%p expired, type:%s ahandle:%p", pSyncNode->vgId, rpcMsg.info.handle,
+            TMSG_INFO(pStub->rpcMsg.msgType), rpcMsg.info.ahandle);
+      rpcSendResponse(&rpcMsg);
     }
 
     pStub = taosHashIterate(pObj->pRespHash, pStub);
   }
 
   int32_t arraySize = taosArrayGetSize(delIndexArray);
-  sDebug("vgId:%d, resp mgr end clean by ttl, sum:%d, cnt:%d, array-size:%d", pSyncNode->vgId, sum, cnt, arraySize);
+  sDebug("vgId:%d, resp manager end clean by ttl, sum:%d, cnt:%d, array-size:%d", pSyncNode->vgId, sum, cnt, arraySize);
 
   for (int32_t i = 0; i < arraySize; ++i) {
     uint64_t *pSeqNum = taosArrayGet(delIndexArray, i);
     taosHashRemove(pObj->pRespHash, pSeqNum, sizeof(uint64_t));
-    sDebug("vgId:%d, resp mgr clean by ttl, seq:%" PRId64 "", pSyncNode->vgId, *pSeqNum);
+    sDebug("vgId:%d, resp manager clean by ttl, seq:%" PRId64, pSyncNode->vgId, *pSeqNum);
   }
   taosArrayDestroy(delIndexArray);
 }
 
 void syncRespCleanRsp(SSyncRespMgr *pObj) {
+  SSyncNode *pNode = pObj->data;
+  sTrace("vgId:%d, clean all rsp", pNode->vgId);
+
   taosThreadMutexLock(&pObj->mutex);
   syncRespCleanByTTL(pObj, -1, true);
   taosThreadMutexUnlock(&pObj->mutex);
 }
 
 void syncRespClean(SSyncRespMgr *pObj) {
+  SSyncNode *pNode = pObj->data;
+  sTrace("vgId:%d, clean rsp by ttl", pNode->vgId);
+
   taosThreadMutexLock(&pObj->mutex);
   syncRespCleanByTTL(pObj, pObj->ttl, false);
   taosThreadMutexUnlock(&pObj->mutex);

From ab9c07c4200d29d16af765974562652ea88df8f7 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Sat, 17 Dec 2022 19:11:43 +0800
Subject: [PATCH 13/13] docs(stream)

---
 docs/en/12-taos-sql/14-stream.md | 4 ++--
 docs/zh/12-taos-sql/14-stream.md | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/docs/en/12-taos-sql/14-stream.md b/docs/en/12-taos-sql/14-stream.md
index 07db5f4877..c47d2da0eb 100644
--- a/docs/en/12-taos-sql/14-stream.md
+++ b/docs/en/12-taos-sql/14-stream.md
@@ -78,13 +78,13 @@ create stream if not exists s1 fill_history 1 into st1  as select count(*) from
 Combining fill_history option and where clause, stream can processing data of specific time range. For example, only process data after a past time. (In this case, 2020-01-30)
 
 ```sql
-create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > timestamp '2020-01-30' interval(10s)
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > '2020-01-30' interval(10s)
 ```
 
 As another example, only processing data starting from some past time, and ending at some future time.
 
 ```sql
-create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > timestamp '2020-01-30' and ts < timestamp '2023-01-01 '> interval(10s)
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > '2020-01-30' and ts < '2023-01-01' interval(10s)
 ```
 
 If some streams are totally outdated, and you do not want it to monitor or process anymore, those streams can be manually dropped and output data will be still kept.
diff --git a/docs/zh/12-taos-sql/14-stream.md b/docs/zh/12-taos-sql/14-stream.md
index 6c18c41f0c..a70d559a86 100644
--- a/docs/zh/12-taos-sql/14-stream.md
+++ b/docs/zh/12-taos-sql/14-stream.md
@@ -79,13 +79,13 @@ create stream if not exists s1 fill_history 1 into st1  as select count(*) from
 结合 fill_history 1 选项，可以实现只处理特定历史时间范围的数据，例如：只处理某历史时刻（2020年1月30日）之后的数据
 
 ```sql
-create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > timestamp '2020-01-30' interval(10s)
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > '2020-01-30' interval(10s)
 ```
 
 再如，仅处理某时间段内的数据，结束时间可以是未来时间
 
 ```sql
-create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > timestamp '2020-01-30' and ts < timestamp '2023-01-01 '> interval(10s)
+create stream if not exists s1 fill_history 1 into st1  as select count(*) from t1 where ts > '2020-01-30' and ts < '2023-01-01' interval(10s)
 ```
 
 如果该流任务已经彻底过期，并且您不再想让它检测或处理数据，您可以手动删除它，被计算出的数据仍会被保留。