diff --git a/Jenkinsfile2 b/Jenkinsfile2
index f582461fb25c43b6901db65a088f122ee9e223b5..f618912295e5812b305666677dcd1c6e9d4cac1c 100644
--- a/Jenkinsfile2
+++ b/Jenkinsfile2
@@ -338,6 +338,14 @@ pipeline {
                         changeRequest()
                     }
                     steps {
+                        script {
+                            def linux_node_ip = sh (
+                                script: 'ip addr|grep 192|grep -v virbr|awk "{print \\\$2}"|sed "s/\\/.*//"',
+                                returnStdout: true
+                            ).trim()
+                            echo "${linux_node_ip}"
+                            echo "${WKDIR}/restore.sh -p ${BRANCH_NAME} -n ${BUILD_ID} -c {container name}"
+                        }
                         catchError(buildResult: 'FAILURE', stageResult: 'FAILURE') {
                             timeout(time: 120, unit: 'MINUTES'){
                                 pre_test()
diff --git a/docs/en/21-tdinternal/01-arch.md b/docs/en/21-tdinternal/01-arch.md
index d7d472eb98a22325e850f4f040dccaa34d02bbff..44651c0496481c410640e577aaad5781f846e302 100644
--- a/docs/en/21-tdinternal/01-arch.md
+++ b/docs/en/21-tdinternal/01-arch.md
@@ -22,7 +22,7 @@ A complete TDengine system runs on one or more physical nodes. Logically, it inc
 
 **Virtual node (vnode)**: To better support data sharding, load balancing and prevent data from overheating or skewing, data nodes are virtualized into multiple virtual nodes (vnode, V2, V3, V4, etc. in the figure). Each vnode is a relatively independent work unit, which is the basic unit of time-series data storage and has independent running threads, memory space and persistent storage path. A vnode contains a certain number of tables (data collection points). When a new table is created, the system checks whether a new vnode needs to be created. The number of vnodes that can be created on a data node depends on the capacity of the hardware of the physical node where the data node is located. A vnode belongs to only one DB, but a DB can have multiple vnodes. In addition to the stored time-series data, a vnode also stores the schema and tag values of the included tables. A virtual node is uniquely identified in the system by the EP of the data node and the VGroup ID to which it belongs and is created and managed by the management node.
 
-**Management node (mnode)**: A virtual logical unit responsible for monitoring and maintaining the running status of all data nodes and load balancing among nodes (M in the figure). At the same time, the management node is also responsible for the storage and management of metadata (including users, databases, tables, static tags, etc.), so it is also called Meta Node. Multiple (up to 5) mnodes can be configured in a TDengine cluster, and they are automatically constructed into a virtual management node group (M0, M1, M2 in the figure). The leader/follower mechanism is adopted for the mnode group and the data synchronization is carried out in a strongly consistent way. Any data update operation can only be executed on the leader. The creation of mnode cluster is completed automatically by the system without manual intervention. There is at most one mnode on each dnode, which is uniquely identified by the EP of the data node to which it belongs. Each dnode automatically obtains the EP of the dnode where all mnodes in the whole cluster are located, through internal messaging interaction.
+**Management node (mnode)**: A virtual logical unit responsible for monitoring and maintaining the running status of all data nodes and load balancing among nodes (M in the figure). At the same time, the management node is also responsible for the storage and management of metadata (including users, databases, tables, static tags, etc.), so it is also called Meta Node. Multiple (up to 3) mnodes can be configured in a TDengine cluster, and they are automatically constructed into a virtual management node group (M0, M1, M2 in the figure). The leader/follower mechanism is adopted for the mnode group and the data synchronization is carried out in a strongly consistent way. Any data update operation can only be executed on the leader. The creation of mnode cluster is completed automatically by the system without manual intervention. There is at most one mnode on each dnode, which is uniquely identified by the EP of the data node to which it belongs. Each dnode automatically obtains the EP of the dnode where all mnodes in the whole cluster are located, through internal messaging interaction.
 
 **Virtual node group (VGroup)**: Vnodes on different data nodes can form a virtual node group to ensure the high availability of the system. The virtual node group is managed in a leader/follower mechanism. Write operations can only be performed on the leader vnode, and then replicated to follower vnodes, thus ensuring that one single replica of data is copied on multiple physical nodes. The number of virtual nodes in a vgroup equals the number of data replicas. If the number of replicas of a DB is N, the system must have at least N data nodes. The number of replicas can be specified by the parameter `“replica”` when creating a DB, and the default is 1. Using the multi-replication feature of TDengine, the same high data reliability can be achieved without the need for expensive storage devices such as disk arrays. Virtual node groups are created and managed by the management node, and the management node assigns a system unique ID, aka VGroup ID. If two virtual nodes have the same vnode group ID, it means that they belong to the same group and the data is backed up to each other. The number of virtual nodes in a virtual node group can be dynamically changed, allowing only one, that is, no data replication. VGroup ID is never changed. Even if a virtual node group is deleted, its ID will not be reused.
 
diff --git a/docs/zh/21-tdinternal/01-arch.md b/docs/zh/21-tdinternal/01-arch.md
index 433cb4808b60ce73c639a23beef45fb8e1afb7dd..507ccea6291cd2cdd5d2477e151ee58d2865e0be 100644
--- a/docs/zh/21-tdinternal/01-arch.md
+++ b/docs/zh/21-tdinternal/01-arch.md
@@ -23,7 +23,7 @@ TDengine 分布式架构的逻辑结构图如下：
 
 **虚拟节点（vnode）：** 为更好的支持数据分片、负载均衡，防止数据过热或倾斜，数据节点被虚拟化成多个虚拟节点（vnode，图中 V2，V3，V4 等）。每个 vnode 都是一个相对独立的工作单元，是时序数据存储的基本单元，具有独立的运行线程、内存空间与持久化存储的路径。一个 vnode 包含一定数量的表（数据采集点）。当创建一张新表时，系统会检查是否需要创建新的 vnode。一个数据节点上能创建的 vnode 的数量取决于该数据节点所在物理节点的硬件资源。一个 vnode 只属于一个 DB，但一个 DB 可以有多个 vnode。一个 vnode 除存储的时序数据外，也保存有所包含的表的 schema、标签值等。一个虚拟节点由所属的数据节点的 EP，以及所属的 VGroup ID 在系统内唯一标识，由管理节点创建并管理。
 
-**管理节点（mnode）：** 一个虚拟的逻辑单元，负责所有数据节点运行状态的监控和维护，以及节点之间的负载均衡（图中 M）。同时，管理节点也负责元数据（包括用户、数据库、表、静态标签等）的存储和管理，因此也称为 Meta Node。TDengine 集群中可配置多个（开源版最多不超过 3 个）mnode，它们自动构建成为一个虚拟管理节点组（图中 M0，M1，M2）。mnode 间采用 master/slave 的机制进行管理，而且采取强一致方式进行数据同步，任何数据更新操作只能在 Master 上进行。mnode 集群的创建由系统自动完成，无需人工干预。每个 dnode 上至多有一个 mnode，由所属的数据节点的 EP 来唯一标识。每个 dnode 通过内部消息交互自动获取整个集群中所有 mnode 所在的 dnode 的 EP。
+**管理节点（mnode）：** 一个虚拟的逻辑单元，负责所有数据节点运行状态的监控和维护，以及节点之间的负载均衡（图中 M）。同时，管理节点也负责元数据（包括用户、数据库、表、静态标签等）的存储和管理，因此也称为 Meta Node。TDengine 集群中可配置多个（最多不超过 3 个）mnode，它们自动构建成为一个虚拟管理节点组（图中 M0，M1，M2）。mnode 间采用 master/slave 的机制进行管理，而且采取强一致方式进行数据同步，任何数据更新操作只能在 Master 上进行。mnode 集群的创建由系统自动完成，无需人工干预。每个 dnode 上至多有一个 mnode，由所属的数据节点的 EP 来唯一标识。每个 dnode 通过内部消息交互自动获取整个集群中所有 mnode 所在的 dnode 的 EP。
 
 **虚拟节点组（VGroup）：** 不同数据节点上的 vnode 可以组成一个虚拟节点组（vgroup）来保证系统的高可靠。虚拟节点组内采取 master/slave 的方式进行管理。写操作只能在 master vnode 上进行，系统采用异步复制的方式将数据同步到 slave vnode，这样确保了一份数据在多个物理节点上有拷贝。一个 vgroup 里虚拟节点个数就是数据的副本数。如果一个 DB 的副本数为 N，系统必须有至少 N 数据节点。副本数在创建 DB 时通过参数 replica 可以指定，缺省为 1。使用 TDengine 的多副本特性，可以不再需要昂贵的磁盘阵列等存储设备，就可以获得同样的数据高可靠性。虚拟节点组由管理节点创建、管理，并且由管理节点分配一个系统唯一的 ID，VGroup ID。如果两个虚拟节点的 VGroup ID 相同，说明他们属于同一个组，数据互为备份。虚拟节点组里虚拟节点的个数是可以动态改变的，容许只有一个，也就是没有数据复制。VGroup ID 是永远不变的，即使一个虚拟节点组被删除，它的 ID 也不会被收回重复利用。
 
diff --git a/include/client/taos.h b/include/client/taos.h
index 5e7f12de0a3ebe7c5b6a38c3f0c247e814e5e985..d31d5c582cd128db3b5217698a27dd92d8a3d108 100644
--- a/include/client/taos.h
+++ b/include/client/taos.h
@@ -128,6 +128,7 @@ typedef struct setConfRet {
 
 DLL_EXPORT void       taos_cleanup(void);
 DLL_EXPORT int        taos_options(TSDB_OPTION option, const void *arg, ...);
+DLL_EXPORT setConfRet taos_set_config(const char *config);
 DLL_EXPORT int        taos_init(void);
 DLL_EXPORT TAOS      *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port);
 DLL_EXPORT TAOS *taos_connect_auth(const char *ip, const char *user, const char *auth, const char *db, uint16_t port);
diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h
index 52d9b6f810d8e01f750160a6e212faa0828a38de..5fa7eed40c0b5514324fe138682b70971d01b5ec 100644
--- a/include/libs/sync/sync.h
+++ b/include/libs/sync/sync.h
@@ -26,9 +26,9 @@ extern "C" {
 
 extern bool gRaftDetailLog;
 
-#define SYNC_INDEX_BEGIN   0
+#define SYNC_INDEX_BEGIN 0
 #define SYNC_INDEX_INVALID -1
-#define SYNC_TERM_INVALID  0xFFFFFFFFFFFFFFFF
+#define SYNC_TERM_INVALID 0xFFFFFFFFFFFFFFFF
 
 typedef uint64_t SyncNodeId;
 typedef int32_t  SyncGroupId;
diff --git a/include/libs/sync/syncTools.h b/include/libs/sync/syncTools.h
index e51b20a40435e67152f11b19a566bc298e11911d..5d892352d68638ddace5ba8191c19b6b25a02439 100644
--- a/include/libs/sync/syncTools.h
+++ b/include/libs/sync/syncTools.h
@@ -524,7 +524,7 @@ void syncReconfigFinishLog2(char* s, const SyncReconfigFinish* pMsg);
 int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
 int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);
 int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg);
-int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg);
+int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg, SyncIndex* pRetIndex);
 int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg);
 int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg);
 int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg);
@@ -541,7 +541,7 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode* ths, SyncSnapshotRsp* pMsg);
 // -----------------------------------------
 typedef int32_t (*FpOnPingCb)(SSyncNode* ths, SyncPing* pMsg);
 typedef int32_t (*FpOnPingReplyCb)(SSyncNode* ths, SyncPingReply* pMsg);
-typedef int32_t (*FpOnClientRequestCb)(SSyncNode* ths, SyncClientRequest* pMsg);
+typedef int32_t (*FpOnClientRequestCb)(SSyncNode* ths, SyncClientRequest* pMsg, SyncIndex* pRetIndex);
 typedef int32_t (*FpOnRequestVoteCb)(SSyncNode* ths, SyncRequestVote* pMsg);
 typedef int32_t (*FpOnRequestVoteReplyCb)(SSyncNode* ths, SyncRequestVoteReply* pMsg);
 typedef int32_t (*FpOnAppendEntriesCb)(SSyncNode* ths, SyncAppendEntries* pMsg);
diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c
index bbd477fa3b5b9a7454bcd37b1188c09db22c2878..52574dcc9f7ac5a0862ad295babd7d28cafa1880 100644
--- a/source/client/src/clientMain.c
+++ b/source/client/src/clientMain.c
@@ -81,6 +81,19 @@ void taos_cleanup(void) {
   taosCloseLog();
 }
 
+static setConfRet taos_set_config_imp(const char *config){
+  setConfRet ret = {SET_CONF_RET_SUCC, {0}};
+  // TODO: need re-implementation
+  return ret;
+}
+
+setConfRet taos_set_config(const char *config){
+// TODO  pthread_mutex_lock(&setConfMutex);
+  setConfRet ret = taos_set_config_imp(config);
+//  pthread_mutex_unlock(&setConfMutex);
+  return ret;
+}
+
 TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port) {
   tscDebug("try to connect to %s:%u, user:%s db:%s", ip, port, user, db);
   if (user == NULL) {
diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c
index 4c9974ba1a9e0d3346cd9e5b2f00406837db4b97..b66082dd98700dfea9e16102a9f9b0551c70ec39 100644
--- a/source/dnode/mnode/impl/src/mndMain.c
+++ b/source/dnode/mnode/impl/src/mndMain.c
@@ -442,7 +442,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
       syncPingReplyDestroy(pSyncMsg);
     } else if (pMsg->msgType == TDMT_SYNC_CLIENT_REQUEST) {
       SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pMsg);
-      code = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg);
+      code = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg, NULL);
       syncClientRequestDestroy(pSyncMsg);
     } else if (pMsg->msgType == TDMT_SYNC_REQUEST_VOTE) {
       SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pMsg);
@@ -491,7 +491,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
       syncPingReplyDestroy(pSyncMsg);
     } else if (pMsg->msgType == TDMT_SYNC_CLIENT_REQUEST) {
       SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pMsg);
-      code = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg);
+      code = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg, NULL);
       syncClientRequestDestroy(pSyncMsg);
     } else if (pMsg->msgType == TDMT_SYNC_REQUEST_VOTE) {
       SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pMsg);
diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c
index 4c561809799492ef2ae047073007b927d40f8c12..b998f401951380b15c95a1033a0c8281165dc2e9 100644
--- a/source/dnode/vnode/src/vnd/vnodeSync.c
+++ b/source/dnode/vnode/src/vnd/vnodeSync.c
@@ -137,6 +137,25 @@ void vnodeProposeMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
         vError("vgId:%d, failed to pre-process msg:%p since %s", vgId, pMsg, terrstr());
       } else {
         code = syncPropose(pVnode->sync, pMsg, vnodeIsMsgWeak(pMsg->msgType));
+        if (code == 1) {
+          do {
+            static int32_t cnt = 0;
+            if (cnt++ % 1000 == 1) {
+              vInfo("vgId:%d, msg:%p apply right now, apply index:%ld, msgtype:%s,%d", vgId, pMsg,
+                    pMsg->info.conn.applyIndex, TMSG_INFO(pMsg->msgType), pMsg->msgType);
+            }
+          } while (0);
+
+          SRpcMsg rsp = {.code = pMsg->code, .info = pMsg->info};
+          if (vnodeProcessWriteReq(pVnode, pMsg, pMsg->info.conn.applyIndex, &rsp) < 0) {
+            rsp.code = terrno;
+            vInfo("vgId:%d, msg:%p failed to apply right now since %s", vgId, pMsg, terrstr());
+          }
+
+          if (rsp.info.handle != NULL) {
+            tmsgSendRsp(&rsp);
+          }
+        }
       }
     }
 
@@ -163,10 +182,12 @@ void vnodeProposeMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
       SRpcMsg rsp = {.code = TSDB_CODE_RPC_REDIRECT, .info = pMsg->info};
       tmsgSendRedirectRsp(&rsp, &newEpSet);
     } else {
-      if (terrno != 0) code = terrno;
-      vError("vgId:%d, msg:%p failed to propose since %s, code:0x%x", vgId, pMsg, tstrerror(code), code);
-      SRpcMsg rsp = {.code = code, .info = pMsg->info};
-      tmsgSendRsp(&rsp);
+      if (code != 1) {
+        if (terrno != 0) code = terrno;
+        vError("vgId:%d, msg:%p failed to propose since %s, code:0x%x", vgId, pMsg, tstrerror(code), code);
+        SRpcMsg rsp = {.code = code, .info = pMsg->info};
+        tmsgSendRsp(&rsp);
+      }
     }
 
     vGTrace("vgId:%d, msg:%p is freed, code:0x%x", vgId, pMsg, code);
@@ -260,7 +281,7 @@ int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
       SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg);
       assert(pSyncMsg != NULL);
 
-      ret = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg);
+      ret = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg, NULL);
       syncClientRequestDestroy(pSyncMsg);
 
     } else if (pRpcMsg->msgType == TDMT_SYNC_REQUEST_VOTE) {
@@ -359,34 +380,18 @@ static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta c
   SyncIndex beginIndex = SYNC_INDEX_INVALID;
   char      logBuf[256] = {0};
 
-  if (pFsm->FpGetSnapshotInfo != NULL) {
-    (*pFsm->FpGetSnapshotInfo)(pFsm, &snapshot);
-    beginIndex = snapshot.lastApplyIndex;
-  }
-
-  if (cbMeta.index > beginIndex) {
-    snprintf(
-        logBuf, sizeof(logBuf),
-        "==callback== ==CommitCb== execute, pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s, beginIndex :%ld\n",
-        pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), beginIndex);
-    syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg);
-
-    SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
-    rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
-    memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
-    syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info);
-    rpcMsg.info.conn.applyIndex = cbMeta.index;
-    tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
+  snprintf(logBuf, sizeof(logBuf),
+           "==callback== ==CommitCb== execute, pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s, beginIndex :%ld\n",
+           pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state),
+           beginIndex);
+  syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg);
 
-  } else {
-    char logBuf[256] = {0};
-    snprintf(logBuf, sizeof(logBuf),
-             "==callback== ==CommitCb== do not execute, pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s, "
-             "beginIndex :%ld\n",
-             pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state),
-             beginIndex);
-    syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg);
-  }
+  SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
+  rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
+  memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
+  syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info);
+  rpcMsg.info.conn.applyIndex = cbMeta.index;
+  tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
 }
 
 static void vnodeSyncPreCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h
index eecefe625c78f9bff9a892584b9c5ae73dd25d2d..a139b511a11a4551ef74fedb2e05b9eae22ca2f8 100644
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@@ -403,7 +403,7 @@ typedef struct SIntervalAggOperatorInfo {
   // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
   SOptrBasicInfo     binfo;              // basic info
   SAggSupporter      aggSup;             // aggregate supporter
-
+  SExprSupp          scalarSupp;         // supporter for perform scalar function
   SGroupResInfo      groupResInfo;       // multiple results build supporter
   SInterval          interval;           // interval info
   int32_t            primaryTsIndex;     // primary time stamp slot id from result of downstream operator.
@@ -738,7 +738,7 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t
 
 SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                           SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
-                                          STimeWindowAggSupp *pTwAggSupp, SExecTaskInfo* pTaskInfo, bool isStream);
+                                          STimeWindowAggSupp* pTwAggSupp, SIntervalPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, bool isStream);
 
 SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                           SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index 0720e8d8c448fcdccb111e21ae2a20a1c8ea27a2..0978bf3f8b147de98741af26affc01c55ed4e5cd 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -4224,8 +4224,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
 
     int32_t tsSlotId = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
     bool    isStream = (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type);
-    pOptr =
-        createIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, tsSlotId, &as, pTaskInfo, isStream);
+    pOptr = createIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, tsSlotId, &as, pIntervalPhyNode,
+                                       pTaskInfo, isStream);
 
   } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == type) {
     SMergeAlignedIntervalPhysiNode* pIntervalPhyNode = (SMergeAlignedIntervalPhysiNode*)pPhyNode;
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index ac57d3a9dadefa4a911ebc31c3f033e7d9601f3d..d89e6f88746c75e8797b080b4aff7ef30502d2c4 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -448,7 +448,7 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
       qDebug("%s start to repeat ascending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
       for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
         STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
-        qDebug("%s\t qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
+        qDebug("%s qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
       }
       // do prepare for the next round table scan operation
       tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
@@ -467,7 +467,7 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
     qDebug("%s start to descending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
     for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
       STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
-      qDebug("%s\t qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
+      qDebug("%s qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
     }
 
     while (pTableScanInfo->scanTimes < total) {
@@ -492,7 +492,7 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
                GET_TASKID(pTaskInfo));
         for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
           STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
-          qDebug("%s\t qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
+          qDebug("%s qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
         }
         tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
         pTableScanInfo->curTWinIdx = 0;
diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c
index b7d39207d838cee16cf312dcf36be46f31f2b438..963e714972d3608764f852be7c8d95a8dbd2bf8a 100644
--- a/source/libs/executor/src/timewindowoperator.c
+++ b/source/libs/executor/src/timewindowoperator.c
@@ -969,6 +969,12 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
 
     getTableScanInfo(pOperator, &pInfo->order, &scanFlag);
 
+    if (pInfo->scalarSupp.pExprInfo != NULL) {
+      SExprSupp* pExprSup =& pInfo->scalarSupp;
+      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx,
+          pExprSup->numOfExprs, NULL);
+    }
+
     // the pDataBlock are always the same one, no need to call this again
     setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->order, scanFlag, true);
     hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, scanFlag, NULL);
@@ -1381,6 +1387,11 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
       continue;
     }
 
+    if (pInfo->scalarSupp.pExprInfo != NULL) {
+      SExprSupp* pExprSup = &pInfo->scalarSupp;
+      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
+    }
+
     // The timewindow that overlaps the timestamps of the input pBlock need to be recalculated and return to the
     // caller. Note that all the time window are not close till now.
     // the pDataBlock are always the same one, no need to call this again
@@ -1498,18 +1509,27 @@ void increaseTs(SqlFunctionCtx* pCtx) {
 
 SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                           SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
-                                          STimeWindowAggSupp* pTwAggSupp, SExecTaskInfo* pTaskInfo, bool isStream) {
+                                          STimeWindowAggSupp* pTwAggSupp, SIntervalPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, bool isStream) {
   SIntervalAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
   SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
     goto _error;
   }
 
-  pInfo->win = pTaskInfo->window;
-  pInfo->order = TSDB_ORDER_ASC;
-  pInfo->interval = *pInterval;
+  pInfo->win       = pTaskInfo->window;
+  pInfo->order     = TSDB_ORDER_ASC;
+  pInfo->interval  = *pInterval;
   pInfo->execModel = pTaskInfo->execModel;
-  pInfo->twAggSup = *pTwAggSupp;
+  pInfo->twAggSup  = *pTwAggSupp;
+
+  if (pPhyNode->window.pExprs != NULL) {
+    int32_t    numOfScalar = 0;
+    SExprInfo* pScalarExprInfo = createExprInfo(pPhyNode->window.pExprs, NULL, &numOfScalar);
+    int32_t code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
+    if (code != TSDB_CODE_SUCCESS) {
+      goto _error;
+    }
+  }
 
   pInfo->primaryTsIndex = primaryTsSlotId;
 
@@ -2473,7 +2493,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
       if (IS_FINAL_OP(pInfo)) {
         int32_t                   childIndex = getChildIndex(pBlock);
         SOperatorInfo*            pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
-        SIntervalAggOperatorInfo* pChildInfo = pChildOp->info;
+        SStreamFinalIntervalOperatorInfo* pChildInfo = pChildOp->info;
         SExprSupp*                pChildSup = &pChildOp->exprSupp;
 
         doClearWindows(&pChildInfo->aggSup, pChildSup, &pChildInfo->interval, pChildInfo->primaryTsIndex,
diff --git a/source/libs/sync/inc/syncEnv.h b/source/libs/sync/inc/syncEnv.h
index 2e8e0e1227c81a0faf9c828eb8f403367495b408..beddec64c5b8c6479a04f50b8a35847e8f65a5ee 100644
--- a/source/libs/sync/inc/syncEnv.h
+++ b/source/libs/sync/inc/syncEnv.h
@@ -28,13 +28,13 @@ extern "C" {
 #include "trpc.h"
 #include "ttimer.h"
 
-#define TIMER_MAX_MS         0x7FFFFFFF
-#define ENV_TICK_TIMER_MS    1000
-#define PING_TIMER_MS        1000
-#define ELECT_TIMER_MS_MIN   1300
-#define ELECT_TIMER_MS_MAX   (ELECT_TIMER_MS_MIN * 2)
+#define TIMER_MAX_MS 0x7FFFFFFF
+#define ENV_TICK_TIMER_MS 1000
+#define PING_TIMER_MS 1000
+#define ELECT_TIMER_MS_MIN 1300
+#define ELECT_TIMER_MS_MAX (ELECT_TIMER_MS_MIN * 2)
 #define ELECT_TIMER_MS_RANGE (ELECT_TIMER_MS_MAX - ELECT_TIMER_MS_MIN)
-#define HEARTBEAT_TIMER_MS   900
+#define HEARTBEAT_TIMER_MS 900
 
 #define EMPTY_RAFT_ID ((SRaftId){.addr = 0, .vgId = 0})
 
diff --git a/source/libs/sync/inc/syncIO.h b/source/libs/sync/inc/syncIO.h
index 88d3065375a007c367b5301679cb84a697045922..f3064e668df575d51dbd1efc333e373543522de1 100644
--- a/source/libs/sync/inc/syncIO.h
+++ b/source/libs/sync/inc/syncIO.h
@@ -50,7 +50,7 @@ typedef struct SSyncIO {
   void *pSyncNode;
   int32_t (*FpOnSyncPing)(SSyncNode *pSyncNode, SyncPing *pMsg);
   int32_t (*FpOnSyncPingReply)(SSyncNode *pSyncNode, SyncPingReply *pMsg);
-  int32_t (*FpOnSyncClientRequest)(SSyncNode *pSyncNode, SyncClientRequest *pMsg);
+  int32_t (*FpOnSyncClientRequest)(SSyncNode *pSyncNode, SyncClientRequest *pMsg, SyncIndex *pRetIndex);
   int32_t (*FpOnSyncRequestVote)(SSyncNode *pSyncNode, SyncRequestVote *pMsg);
   int32_t (*FpOnSyncRequestVoteReply)(SSyncNode *pSyncNode, SyncRequestVoteReply *pMsg);
   int32_t (*FpOnSyncAppendEntries)(SSyncNode *pSyncNode, SyncAppendEntries *pMsg);
diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h
index 1bd0671fb5515d901a4271f39d1e9f05131d868d..d351dc50f435867a44cbc73608432c019dc6fa15 100644
--- a/source/libs/sync/inc/syncInt.h
+++ b/source/libs/sync/inc/syncInt.h
@@ -169,7 +169,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo);
 void       syncNodeStart(SSyncNode* pSyncNode);
 void       syncNodeStartStandBy(SSyncNode* pSyncNode);
 void       syncNodeClose(SSyncNode* pSyncNode);
-int32_t    syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak);
+int32_t    syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak);
 
 // option
 bool      syncNodeSnapshotEnable(SSyncNode* pSyncNode);
@@ -233,6 +233,7 @@ SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index);
 SyncTerm  syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index);
 int32_t   syncNodeGetPreIndexTerm(SSyncNode* pSyncNode, SyncIndex index, SyncIndex* pPreIndex, SyncTerm* pPreTerm);
 
+bool    syncNodeIsOptimizedOneReplica(SSyncNode* ths, SRpcMsg* pMsg);
 int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, uint64_t flag);
 
 int32_t syncNodeUpdateNewConfigIndex(SSyncNode* ths, SSyncCfg* pNewCfg);
diff --git a/source/libs/sync/inc/syncRaftCfg.h b/source/libs/sync/inc/syncRaftCfg.h
index efe8a65b776f7c00c9c54f49f7eaadddcebb153c..521ca6068dbcdb0c7d911be0efb0038ffa591af3 100644
--- a/source/libs/sync/inc/syncRaftCfg.h
+++ b/source/libs/sync/inc/syncRaftCfg.h
@@ -49,14 +49,14 @@ int32_t   raftCfgClose(SRaftCfg *pRaftCfg);
 int32_t   raftCfgPersist(SRaftCfg *pRaftCfg);
 int32_t   raftCfgAddConfigIndex(SRaftCfg *pRaftCfg, SyncIndex configIndex);
 
-cJSON  *syncCfg2Json(SSyncCfg *pSyncCfg);
-char   *syncCfg2Str(SSyncCfg *pSyncCfg);
-char   *syncCfg2SimpleStr(SSyncCfg *pSyncCfg);
+cJSON * syncCfg2Json(SSyncCfg *pSyncCfg);
+char *  syncCfg2Str(SSyncCfg *pSyncCfg);
+char *  syncCfg2SimpleStr(SSyncCfg *pSyncCfg);
 int32_t syncCfgFromJson(const cJSON *pRoot, SSyncCfg *pSyncCfg);
 int32_t syncCfgFromStr(const char *s, SSyncCfg *pSyncCfg);
 
-cJSON  *raftCfg2Json(SRaftCfg *pRaftCfg);
-char   *raftCfg2Str(SRaftCfg *pRaftCfg);
+cJSON * raftCfg2Json(SRaftCfg *pRaftCfg);
+char *  raftCfg2Str(SRaftCfg *pRaftCfg);
 int32_t raftCfgFromJson(const cJSON *pRoot, SRaftCfg *pRaftCfg);
 int32_t raftCfgFromStr(const char *s, SRaftCfg *pRaftCfg);
 
diff --git a/source/libs/sync/inc/syncRaftLog.h b/source/libs/sync/inc/syncRaftLog.h
index f124867097c060fc89ffef405bde8ba7f547fb5b..f3ed9e302b5b0bc23397051bf665feab335754f7 100644
--- a/source/libs/sync/inc/syncRaftLog.h
+++ b/source/libs/sync/inc/syncRaftLog.h
@@ -29,9 +29,12 @@ extern "C" {
 #include "wal.h"
 
 typedef struct SSyncLogStoreData {
-  SSyncNode*      pSyncNode;
-  SWal*           pWal;
+  SSyncNode* pSyncNode;
+  SWal*      pWal;
+
+  TdThreadMutex   mutex;
   SWalReadHandle* pWalHandle;
+
   // SyncIndex       beginIndex;  // valid begin index, default 0, may be set beginIndex > 0
 } SSyncLogStoreData;
 
diff --git a/source/libs/sync/src/syncCommit.c b/source/libs/sync/src/syncCommit.c
index c92edae381ae631cacb2b2dfd20c25e8f46eb4c9..982d37826ad1e7f03b3b55f93d24d51ca01c4dbe 100644
--- a/source/libs/sync/src/syncCommit.c
+++ b/source/libs/sync/src/syncCommit.c
@@ -102,6 +102,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
     }
   }
 
+  // maybe execute fsm
   if (newCommitIndex > pSyncNode->commitIndex) {
     SyncIndex beginIndex = pSyncNode->commitIndex + 1;
     SyncIndex endIndex = newCommitIndex;
diff --git a/source/libs/sync/src/syncIO.c b/source/libs/sync/src/syncIO.c
index 038c36c417e5586ff144d173cf42fa7094724a94..663745a7d76e7ece6bf9cf9619bbd504ee5630a8 100644
--- a/source/libs/sync/src/syncIO.c
+++ b/source/libs/sync/src/syncIO.c
@@ -30,7 +30,7 @@ static int32_t  syncIODestroy(SSyncIO *io);
 static int32_t  syncIOStartInternal(SSyncIO *io);
 static int32_t  syncIOStopInternal(SSyncIO *io);
 
-static void   *syncIOConsumerFunc(void *param);
+static void *  syncIOConsumerFunc(void *param);
 static void    syncIOProcessRequest(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
 static void    syncIOProcessReply(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
 static int32_t syncIOAuth(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey);
@@ -242,9 +242,9 @@ static int32_t syncIOStopInternal(SSyncIO *io) {
 }
 
 static void *syncIOConsumerFunc(void *param) {
-  SSyncIO   *io = param;
+  SSyncIO *  io = param;
   STaosQall *qall;
-  SRpcMsg   *pRpcMsg, rpcMsg;
+  SRpcMsg *  pRpcMsg, rpcMsg;
   qall = taosAllocateQall();
 
   while (1) {
@@ -281,7 +281,7 @@ static void *syncIOConsumerFunc(void *param) {
         if (io->FpOnSyncClientRequest != NULL) {
           SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg);
           ASSERT(pSyncMsg != NULL);
-          io->FpOnSyncClientRequest(io->pSyncNode, pSyncMsg);
+          io->FpOnSyncClientRequest(io->pSyncNode, pSyncMsg, NULL);
           syncClientRequestDestroy(pSyncMsg);
         }
 
diff --git a/source/libs/sync/src/syncIndexMgr.c b/source/libs/sync/src/syncIndexMgr.c
index 5b432aeec4e214a7c79106934965012d1b162b02..10f0e0e335008e87d3042fa385fe93be5827a449 100644
--- a/source/libs/sync/src/syncIndexMgr.c
+++ b/source/libs/sync/src/syncIndexMgr.c
@@ -126,7 +126,7 @@ cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) {
 
 char *syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr) {
   cJSON *pJson = syncIndexMgr2Json(pSyncIndexMgr);
-  char  *serialized = cJSON_Print(pJson);
+  char * serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c
index 3100d0525cfdc566be97a71728785027fb62b1d1..3996193037cc992ee2f1a6bf5959ce78b54686a6 100644
--- a/source/libs/sync/src/syncMain.c
+++ b/source/libs/sync/src/syncMain.c
@@ -50,7 +50,7 @@ static int32_t syncNodeAppendNoop(SSyncNode* ths);
 // process message ----
 int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
 int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);
-int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg);
+int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg, SyncIndex* pRetIndex);
 
 // life cycle
 static void syncFreeNode(void* param);
@@ -627,7 +627,7 @@ int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
   return ret;
 }
 
-int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak) {
+int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) {
   int32_t ret = 0;
 
   char eventLog[128];
@@ -664,13 +664,34 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak)
     SRpcMsg            rpcMsg;
     syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
 
-    if (pSyncNode->FpEqMsg != NULL && (*pSyncNode->FpEqMsg)(pSyncNode->msgcb, &rpcMsg) == 0) {
-      ret = 0;
+    // optimized one replica
+    if (syncNodeIsOptimizedOneReplica(pSyncNode, pMsg)) {
+      SyncIndex retIndex;
+      int32_t   code = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg, &retIndex);
+      if (code == 0) {
+        pMsg->info.conn.applyIndex = retIndex;
+        rpcFreeCont(rpcMsg.pCont);
+        syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum);
+        ret = 1;
+        sDebug("vgId:%d optimized index:%ld success, msgtype:%s,%d", pSyncNode->vgId, retIndex,
+               TMSG_INFO(pMsg->msgType), pMsg->msgType);
+      } else {
+        ret = -1;
+        terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+        sError("vgId:%d optimized index:%ld error, msgtype:%s,%d", pSyncNode->vgId, retIndex, TMSG_INFO(pMsg->msgType),
+               pMsg->msgType);
+      }
+
     } else {
-      ret = -1;
-      terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
-      sError("syncPropose pSyncNode->FpEqMsg is NULL");
+      if (pSyncNode->FpEqMsg != NULL && (*pSyncNode->FpEqMsg)(pSyncNode->msgcb, &rpcMsg) == 0) {
+        ret = 0;
+      } else {
+        ret = -1;
+        terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+        sError("enqueue msg error, FpEqMsg is NULL");
+      }
     }
+
     syncClientRequestDestroy(pSyncMsg);
     goto _END;
 
@@ -2377,7 +2398,7 @@ int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg) {
 //     /\ UNCHANGED <<messages, serverVars, candidateVars,
 //                    leaderVars, commitIndex>>
 //
-int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg) {
+int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg, SyncIndex* pRetIndex) {
   int32_t ret = 0;
   syncClientRequestLog2("==syncNodeOnClientRequestCb==", pMsg);
 
@@ -2436,6 +2457,14 @@ int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg) {
     rpcFreeCont(rpcMsg.pCont);
   }
 
+  if (pRetIndex != NULL) {
+    if (ret == 0 && pEntry != NULL) {
+      *pRetIndex = pEntry->index;
+    } else {
+      *pRetIndex = SYNC_INDEX_INVALID;
+    }
+  }
+
   syncEntryDestory(pEntry);
   return ret;
 }
@@ -2600,6 +2629,10 @@ static int32_t syncNodeProposeConfigChangeFinish(SSyncNode* ths, SyncReconfigFin
   return 0;
 }
 
+bool syncNodeIsOptimizedOneReplica(SSyncNode* ths, SRpcMsg* pMsg) {
+  return (ths->replicaNum == 1 && syncUtilUserCommit(pMsg->msgType) && ths->vgId != 1);
+}
+
 int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, uint64_t flag) {
   int32_t    code = 0;
   ESyncState state = flag;
@@ -2621,19 +2654,33 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex,
         syncEntry2OriginalRpc(pEntry, &rpcMsg);
 
         // user commit
-        if (ths->pFsm->FpCommitCb != NULL && syncUtilUserCommit(pEntry->originalRpcType)) {
-          SFsmCbMeta cbMeta = {0};
-          cbMeta.index = pEntry->index;
-          cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index);
-          cbMeta.isWeak = pEntry->isWeak;
-          cbMeta.code = 0;
-          cbMeta.state = ths->state;
-          cbMeta.seqNum = pEntry->seqNum;
-          cbMeta.term = pEntry->term;
-          cbMeta.currentTerm = ths->pRaftStore->currentTerm;
-          cbMeta.flag = flag;
-
-          ths->pFsm->FpCommitCb(ths->pFsm, &rpcMsg, cbMeta);
+        if ((ths->pFsm->FpCommitCb != NULL) && syncUtilUserCommit(pEntry->originalRpcType)) {
+          bool internalExecute = true;
+          if ((ths->replicaNum == 1) && ths->restoreFinish && (ths->vgId != 1)) {
+            internalExecute = false;
+          }
+
+          do {
+            char logBuf[128];
+            snprintf(logBuf, sizeof(logBuf), "index:%ld, internalExecute:%d", i, internalExecute);
+            syncNodeEventLog(ths, logBuf);
+          } while (0);
+
+          // execute fsm in apply thread, or execute outside syncPropose
+          if (internalExecute) {
+            SFsmCbMeta cbMeta = {0};
+            cbMeta.index = pEntry->index;
+            cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index);
+            cbMeta.isWeak = pEntry->isWeak;
+            cbMeta.code = 0;
+            cbMeta.state = ths->state;
+            cbMeta.seqNum = pEntry->seqNum;
+            cbMeta.term = pEntry->term;
+            cbMeta.currentTerm = ths->pRaftStore->currentTerm;
+            cbMeta.flag = flag;
+
+            ths->pFsm->FpCommitCb(ths->pFsm, &rpcMsg, cbMeta);
+          }
         }
 
         // config change
diff --git a/source/libs/sync/src/syncRaftCfg.c b/source/libs/sync/src/syncRaftCfg.c
index f40eff57d9d4304bad850369cf332e68b12a0ec7..9d16bed6c1331a4fef152e82b3e55507c832db2d 100644
--- a/source/libs/sync/src/syncRaftCfg.c
+++ b/source/libs/sync/src/syncRaftCfg.c
@@ -101,7 +101,7 @@ cJSON *syncCfg2Json(SSyncCfg *pSyncCfg) {
 
 char *syncCfg2Str(SSyncCfg *pSyncCfg) {
   cJSON *pJson = syncCfg2Json(pSyncCfg);
-  char  *serialized = cJSON_Print(pJson);
+  char * serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
@@ -109,7 +109,7 @@ char *syncCfg2Str(SSyncCfg *pSyncCfg) {
 char *syncCfg2SimpleStr(SSyncCfg *pSyncCfg) {
   if (pSyncCfg != NULL) {
     int32_t len = 512;
-    char   *s = taosMemoryMalloc(len);
+    char *  s = taosMemoryMalloc(len);
     memset(s, 0, len);
 
     snprintf(s, len, "{replica-num:%d, my-index:%d, ", pSyncCfg->replicaNum, pSyncCfg->myIndex);
@@ -205,7 +205,7 @@ cJSON *raftCfg2Json(SRaftCfg *pRaftCfg) {
 
 char *raftCfg2Str(SRaftCfg *pRaftCfg) {
   cJSON *pJson = raftCfg2Json(pRaftCfg);
-  char  *serialized = cJSON_Print(pJson);
+  char * serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
@@ -271,7 +271,7 @@ int32_t raftCfgFromJson(const cJSON *pRoot, SRaftCfg *pRaftCfg) {
     (pRaftCfg->configIndexArr)[i] = atoll(pIndex->valuestring);
   }
 
-  cJSON  *pJsonSyncCfg = cJSON_GetObjectItem(pJson, "SSyncCfg");
+  cJSON * pJsonSyncCfg = cJSON_GetObjectItem(pJson, "SSyncCfg");
   int32_t code = syncCfgFromJson(pJsonSyncCfg, &(pRaftCfg->cfg));
   ASSERT(code == 0);
 
diff --git a/source/libs/sync/src/syncRaftLog.c b/source/libs/sync/src/syncRaftLog.c
index cc606f4df884bfd19902c9f775e35e98dc99aa0e..d84c576714f056e1d48272f929c27dad9d92ba6f 100644
--- a/source/libs/sync/src/syncRaftLog.c
+++ b/source/libs/sync/src/syncRaftLog.c
@@ -257,6 +257,8 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
     return -1;
   }
 
+  taosThreadMutexLock(&(pData->mutex));
+
   code = walReadWithHandle(pWalHandle, index);
   if (code != 0) {
     int32_t     err = terrno;
@@ -281,6 +283,7 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
         terrno = saveErr;
     */
 
+    taosThreadMutexUnlock(&(pData->mutex));
     return code;
   }
 
@@ -301,6 +304,7 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
     terrno = saveErr;
   */
 
+  taosThreadMutexUnlock(&(pData->mutex));
   return code;
 }
 
@@ -364,6 +368,7 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
   pData->pWal = pSyncNode->pWal;
   ASSERT(pData->pWal != NULL);
 
+  taosThreadMutexInit(&(pData->mutex), NULL);
   pData->pWalHandle = walOpenReadHandle(pData->pWal);
   ASSERT(pData->pWalHandle != NULL);
 
@@ -408,9 +413,14 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
 void logStoreDestory(SSyncLogStore* pLogStore) {
   if (pLogStore != NULL) {
     SSyncLogStoreData* pData = pLogStore->data;
+
+    taosThreadMutexLock(&(pData->mutex));
     if (pData->pWalHandle != NULL) {
       walCloseReadHandle(pData->pWalHandle);
+      pData->pWalHandle = NULL;
     }
+    taosThreadMutexUnlock(&(pData->mutex));
+    taosThreadMutexDestroy(&(pData->mutex));
 
     taosMemoryFree(pLogStore->data);
     taosMemoryFree(pLogStore);
@@ -460,6 +470,8 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
   SWal*              pWal = pData->pWal;
 
   if (index >= SYNC_INDEX_BEGIN && index <= logStoreLastIndex(pLogStore)) {
+    taosThreadMutexLock(&(pData->mutex));
+
     // SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
     SWalReadHandle* pWalHandle = pData->pWalHandle;
     ASSERT(pWalHandle != NULL);
@@ -503,6 +515,7 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
         terrno = saveErr;
     */
 
+    taosThreadMutexUnlock(&(pData->mutex));
     return pEntry;
 
   } else {
diff --git a/source/libs/sync/src/syncRaftStore.c b/source/libs/sync/src/syncRaftStore.c
index a1ab95c00fc3d07b57b86fca79b19c8a1048aef3..9f5cba6c6607d3a3146fa0c1c55870d9713e110d 100644
--- a/source/libs/sync/src/syncRaftStore.c
+++ b/source/libs/sync/src/syncRaftStore.c
@@ -216,7 +216,7 @@ cJSON *raftStore2Json(SRaftStore *pRaftStore) {
 
 char *raftStore2Str(SRaftStore *pRaftStore) {
   cJSON *pJson = raftStore2Json(pRaftStore);
-  char  *serialized = cJSON_Print(pJson);
+  char * serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c
index 4fc01284007ec389888b4dc160f68fc1dc954523..0feaf532bb4798f39b2ea03d96cc725fb2d4a34a 100644
--- a/source/libs/sync/src/syncSnapshot.c
+++ b/source/libs/sync/src/syncSnapshot.c
@@ -314,14 +314,14 @@ cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) {
 
 char *snapshotSender2Str(SSyncSnapshotSender *pSender) {
   cJSON *pJson = snapshotSender2Json(pSender);
-  char  *serialized = cJSON_Print(pJson);
+  char * serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
 
 char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event) {
   int32_t len = 256;
-  char   *s = taosMemoryMalloc(len);
+  char *  s = taosMemoryMalloc(len);
 
   SRaftId  destId = pSender->pSyncNode->replicasId[pSender->replicaIndex];
   char     host[128];
@@ -461,7 +461,7 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) {
     cJSON_AddStringToObject(pFromId, "addr", u64buf);
     {
       uint64_t u64 = pReceiver->fromId.addr;
-      cJSON   *pTmp = pFromId;
+      cJSON *  pTmp = pFromId;
       char     host[128] = {0};
       uint16_t port;
       syncUtilU642Addr(u64, host, sizeof(host), &port);
@@ -494,14 +494,14 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) {
 
 char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver) {
   cJSON *pJson = snapshotReceiver2Json(pReceiver);
-  char  *serialized = cJSON_Print(pJson);
+  char * serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
 
 char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event) {
   int32_t len = 256;
-  char   *s = taosMemoryMalloc(len);
+  char *  s = taosMemoryMalloc(len);
 
   SRaftId  fromId = pReceiver->fromId;
   char     host[128];
diff --git a/source/libs/sync/src/syncVoteMgr.c b/source/libs/sync/src/syncVoteMgr.c
index d6c2cbd34e3d90935bc651fac54a697b2e9bbc4e..2c43312064a868e4d49bfbccb1ed5c0847ffa31b 100644
--- a/source/libs/sync/src/syncVoteMgr.c
+++ b/source/libs/sync/src/syncVoteMgr.c
@@ -127,7 +127,7 @@ cJSON *voteGranted2Json(SVotesGranted *pVotesGranted) {
 
 char *voteGranted2Str(SVotesGranted *pVotesGranted) {
   cJSON *pJson = voteGranted2Json(pVotesGranted);
-  char  *serialized = cJSON_Print(pJson);
+  char * serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
@@ -256,7 +256,7 @@ cJSON *votesRespond2Json(SVotesRespond *pVotesRespond) {
 
 char *votesRespond2Str(SVotesRespond *pVotesRespond) {
   cJSON *pJson = votesRespond2Json(pVotesRespond);
-  char  *serialized = cJSON_Print(pJson);
+  char * serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
diff --git a/source/libs/sync/test/syncRaftLogTest2.cpp b/source/libs/sync/test/syncRaftLogTest2.cpp
index e3ceae061535bdcdb579ba83465803a3eac7b096..9e0c2ecc29c0f649633d5d0ab999b997f88676d5 100644
--- a/source/libs/sync/test/syncRaftLogTest2.cpp
+++ b/source/libs/sync/test/syncRaftLogTest2.cpp
@@ -113,7 +113,7 @@ void test2() {
   pLogStore = logStoreCreate(pSyncNode);
   assert(pLogStore);
   pSyncNode->pLogStore = pLogStore;
-  //pLogStore->syncLogSetBeginIndex(pLogStore, 5);
+  // pLogStore->syncLogSetBeginIndex(pLogStore, 5);
   pLogStore->syncLogRestoreFromSnapshot(pLogStore, 4);
   logStoreLog2((char*)"\n\n\ntest2 ----- ", pLogStore);
 
@@ -229,7 +229,7 @@ void test4() {
   assert(pLogStore);
   pSyncNode->pLogStore = pLogStore;
   logStoreLog2((char*)"\n\n\ntest4 ----- ", pLogStore);
-  //pLogStore->syncLogSetBeginIndex(pLogStore, 5);
+  // pLogStore->syncLogSetBeginIndex(pLogStore, 5);
   pLogStore->syncLogRestoreFromSnapshot(pLogStore, 4);
 
   for (int i = 5; i <= 9; ++i) {
@@ -291,7 +291,7 @@ void test5() {
   assert(pLogStore);
   pSyncNode->pLogStore = pLogStore;
   logStoreLog2((char*)"\n\n\ntest5 ----- ", pLogStore);
-  //pLogStore->syncLogSetBeginIndex(pLogStore, 5);
+  // pLogStore->syncLogSetBeginIndex(pLogStore, 5);
   pLogStore->syncLogRestoreFromSnapshot(pLogStore, 4);
 
   for (int i = 5; i <= 9; ++i) {
@@ -412,26 +412,23 @@ void test6() {
   do {
     SyncIndex firstVer = walGetFirstVer(pWal);
     SyncIndex lastVer = walGetLastVer(pWal);
-    bool isEmpty = walIsEmpty(pWal);
+    bool      isEmpty = walIsEmpty(pWal);
     printf("before -------- firstVer:%ld lastVer:%ld isEmpty:%d \n", firstVer, lastVer, isEmpty);
   } while (0);
 
   logStoreDestory(pLogStore);
   cleanup();
 
-
-
   // restart
   init();
   pLogStore = logStoreCreate(pSyncNode);
   assert(pLogStore);
   pSyncNode->pLogStore = pLogStore;
-  
 
   do {
     SyncIndex firstVer = walGetFirstVer(pWal);
     SyncIndex lastVer = walGetLastVer(pWal);
-    bool isEmpty = walIsEmpty(pWal);
+    bool      isEmpty = walIsEmpty(pWal);
     printf("after -------- firstVer:%ld lastVer:%ld isEmpty:%d \n", firstVer, lastVer, isEmpty);
   } while (0);
 
@@ -461,13 +458,13 @@ int main(int argc, char** argv) {
   }
   sTrace("gAssert : %d", gAssert);
 
-/*
-  test1();
-  test2();
-  test3();
-  test4();
-  test5();
-*/
+  /*
+    test1();
+    test2();
+    test3();
+    test4();
+    test5();
+  */
   test6();
 
   return 0;
diff --git a/source/libs/sync/test/syncRaftLogTest3.cpp b/source/libs/sync/test/syncRaftLogTest3.cpp
index 302e29a091277e085dcd67d0e2a55df00108a0b0..ea1788c545e487f9c2943dc81e826a85a9b0a8db 100644
--- a/source/libs/sync/test/syncRaftLogTest3.cpp
+++ b/source/libs/sync/test/syncRaftLogTest3.cpp
@@ -312,7 +312,7 @@ void test5() {
   pSyncNode->pLogStore = pLogStore;
   logStoreLog2((char*)"\n\n\ntest5 ----- ", pLogStore);
 
-  //pSyncNode->pLogStore->syncLogSetBeginIndex(pSyncNode->pLogStore, 6);
+  // pSyncNode->pLogStore->syncLogSetBeginIndex(pSyncNode->pLogStore, 6);
   pLogStore->syncLogRestoreFromSnapshot(pSyncNode->pLogStore, 5);
   for (int i = 6; i <= 10; ++i) {
     int32_t         dataLen = 10;
diff --git a/tests/system-test/7-tmq/tmqCheckData1.py b/tests/system-test/7-tmq/tmqCheckData1.py
new file mode 100644
index 0000000000000000000000000000000000000000..6cf849d1b989a9b99ea19a7fa77b9483254bd7f8
--- /dev/null
+++ b/tests/system-test/7-tmq/tmqCheckData1.py
@@ -0,0 +1,180 @@
+
+import taos
+import sys
+import time
+import socket
+import os
+import threading
+
+from util.log import *
+from util.sql import *
+from util.cases import *
+from util.dnodes import *
+from util.common import *
+sys.path.append("./7-tmq")
+from tmqCommon import *
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug(f"start to excute {__file__}")
+        tdSql.init(conn.cursor())
+        #tdSql.init(conn.cursor(), logSql)  # output sql.txt file
+
+    def checkFileContent(self, consumerId, queryString):
+        buildPath = tdCom.getBuildPath()
+        cfgPath = tdCom.getClientCfgPath()
+        dstFile = '%s/../log/dstrows_%d.txt'%(cfgPath, consumerId)
+        cmdStr = '%s/build/bin/taos -c %s -s "%s >> %s"'%(buildPath, cfgPath, queryString, dstFile)
+        tdLog.info(cmdStr)
+        os.system(cmdStr)
+        
+        consumeRowsFile = '%s/../log/consumerid_%d.txt'%(cfgPath, consumerId)
+        tdLog.info("rows file: %s, %s"%(consumeRowsFile, dstFile))
+
+        consumeFile = open(consumeRowsFile, mode='r')
+        queryFile = open(dstFile, mode='r')
+        
+        # skip first line for it is schema
+        queryFile.readline()
+
+        while True:
+            dst = queryFile.readline()
+            src = consumeFile.readline()
+            
+            if dst:
+                if dst != src:
+                    tdLog.exit("consumerId %d consume rows is not match the rows by direct query"%consumerId)
+            else:
+                break
+        return 
+
+    def tmqCase1(self):
+        tdLog.printNoPrefix("======== test case 1: ")
+        paraDict = {'dbName':     'db1',
+                    'dropFlag':   1,
+                    'event':      '',
+                    'vgroups':    4,
+                    'stbName':    'stb',
+                    'colPrefix':  'c',
+                    'tagPrefix':  't',
+                    'colSchema':   [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}],
+                    'tagSchema':   [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}],
+                    'ctbPrefix':  'ctb',
+                    'ctbNum':     1,
+                    'rowsPerTbl': 10000,
+                    'batchNum':   10,
+                    'startTs':    1640966400000,  # 2022-01-01 00:00:00.000
+                    'pollDelay':  10,
+                    'showMsg':    1,
+                    'showRow':    1}
+
+        topicNameList = ['topic1', 'topic2', 'topic3']
+        expectRowsList = []
+        tmqCom.initConsumerTable()
+        tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=4,replica=1)
+        tdLog.info("create stb")
+        tdCom.create_stable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"], column_elm_list=paraDict['colSchema'], tag_elm_list=paraDict['tagSchema'])
+        tdLog.info("create ctb")
+        tdCom.create_ctable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"],tag_elm_list=paraDict['tagSchema'],count=paraDict["ctbNum"], default_ctbname_prefix=paraDict['ctbPrefix'])
+        tdLog.info("insert data")
+        tmqCom.insert_data(tdSql,paraDict["dbName"],paraDict["ctbPrefix"],paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"],paraDict["startTs"])
+        
+        tdLog.info("create topics from stb with filter")
+        queryString = "select ts,c1,c2 from %s.%s" %(paraDict['dbName'], paraDict['stbName'])
+        sqlString = "create topic %s as stable %s.%s" %(topicNameList[0], paraDict["dbName"],paraDict["stbName"])
+        tdLog.info("create topic sql: %s"%sqlString)
+        tdSql.execute(sqlString)
+        tdSql.query(queryString)        
+        expectRowsList.append(tdSql.getRows())
+
+        # init consume info, and start tmq_sim, then check consume result
+        tdLog.info("insert consume info to consume processor")
+        consumerId   = 0
+        expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"]
+        topicList    = topicNameList[0]
+        ifcheckdata  = 1
+        ifManualCommit = 1
+        keyList      = 'group.id:cgrp1, enable.auto.commit:false, auto.commit.interval.ms:6000, auto.offset.reset:earliest'
+        tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
+
+        tdLog.info("start consume processor")
+        tmqCom.startTmqSimProcess(paraDict['pollDelay'],paraDict["dbName"],paraDict['showMsg'], paraDict['showRow'])
+
+        tdLog.info("wait the consume result") 
+        expectRows = 1
+        resultList = tmqCom.selectConsumeResult(expectRows)
+        
+        if expectRowsList[0] != resultList[0]:
+            tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[0], resultList[0]))
+            tdLog.exit("0 tmq consume rows error!")
+
+        self.checkFileContent(consumerId, queryString)     
+
+        # reinit consume info, and start tmq_sim, then check consume result
+        tmqCom.initConsumerTable()
+        queryString = "select ts, c1, c2 from %s.%s"%(paraDict['dbName'], paraDict['stbName'])
+        sqlString = "create topic %s as database %s" %(topicNameList[1], paraDict['dbName'])
+        tdLog.info("create topic sql: %s"%sqlString)
+        tdSql.execute(sqlString)
+        tdSql.query(queryString) 
+        expectRowsList.append(tdSql.getRows())
+
+        consumerId   = 1
+        topicList    = topicNameList[1]
+        tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
+
+        tdLog.info("start consume processor")
+        tmqCom.startTmqSimProcess(paraDict['pollDelay'],paraDict["dbName"],paraDict['showMsg'], paraDict['showRow'])
+
+        tdLog.info("wait the consume result") 
+        expectRows = 1
+        resultList = tmqCom.selectConsumeResult(expectRows)
+        if expectRowsList[1] != resultList[0]:
+            tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[1], resultList[0]))
+            tdLog.exit("1 tmq consume rows error!")
+
+        self.checkFileContent(consumerId, queryString)
+
+        # reinit consume info, and start tmq_sim, then check consume result
+        tmqCom.initConsumerTable()
+        queryString = "select * from %s.%s"%(paraDict['dbName'], paraDict['stbName'])
+        sqlString = "create topic %s as %s" %(topicNameList[2], queryString)
+        tdLog.info("create topic sql: %s"%sqlString)
+        tdSql.execute(sqlString)
+        tdSql.query(queryString) 
+        expectRowsList.append(tdSql.getRows())
+
+        consumerId   = 2
+        topicList    = topicNameList[2]
+        tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
+
+        tdLog.info("start consume processor")
+        tmqCom.startTmqSimProcess(paraDict['pollDelay'],paraDict["dbName"],paraDict['showMsg'], paraDict['showRow'])
+
+        tdLog.info("wait the consume result") 
+        expectRows = 1
+        resultList = tmqCom.selectConsumeResult(expectRows)
+        if expectRowsList[2] != resultList[0]:
+            tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[2], resultList[0]))
+            tdLog.exit("2 tmq consume rows error!")
+
+        self.checkFileContent(consumerId, queryString)
+
+        time.sleep(10)        
+        for i in range(len(topicNameList)):
+            tdSql.query("drop topic %s"%topicNameList[i])
+
+        tdLog.printNoPrefix("======== test case 1 end ...... ")
+
+    def run(self):
+        tdSql.prepare()
+        self.tmqCase1()
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success(f"{__file__} successfully executed")
+
+event = threading.Event()
+
+tdCases.addLinux(__file__, TDTestCase())
+tdCases.addWindows(__file__, TDTestCase())
diff --git a/tests/system-test/fulltest.sh b/tests/system-test/fulltest.sh
index 571bb166e1d6ed7a77f956d4a83f1206b75801c1..2e2e390d123c469cfa542e587f99a4c8d6091e3f 100755
--- a/tests/system-test/fulltest.sh
+++ b/tests/system-test/fulltest.sh
@@ -141,6 +141,7 @@ python3 ./test.py -f 7-tmq/tmqError.py
 python3 ./test.py -f 7-tmq/schema.py
 python3 ./test.py -f 7-tmq/stbFilter.py
 python3 ./test.py -f 7-tmq/tmqCheckData.py
+python3 ./test.py -f 7-tmq/tmqCheckData1.py
 python3 ./test.py -f 7-tmq/tmqUdf.py
 #python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 5
 python3 ./test.py -f 7-tmq/tmqConsumerGroup.py