fix: merge with main

99fd578e · sangshuduo · 769c4b08 · dace3393 · 99fd578e · 99fd578e
264 changed file
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -10,6 +10,8 @@ if (NOT DEFINED TD_SOURCE_DIR)
  set( TD_SOURCE_DIR ${PROJECT_SOURCE_DIR} )
 endif()

+SET(TD_COMMUNITY_DIR ${PROJECT_SOURCE_DIR})
+
 set(TD_SUPPORT_DIR "${TD_SOURCE_DIR}/cmake")
 set(TD_CONTRIB_DIR "${TD_SOURCE_DIR}/contrib")


--- a/Jenkinsfile2
+++ b/Jenkinsfile2
@@ -460,7 +460,7 @@ pipeline {
                                        cd ${WKC}/tests/parallel_test
                                        export DEFAULT_RETRY_TIME=2
                                        date
-                                        ''' + timeout_cmd + ''' time ./run.sh -e -m /home/m.json -t cases.task -b ${BRANCH_NAME}_${BUILD_ID} -l ${WKDIR}/log -o 480 ''' + extra_param + '''
+                                        ''' + timeout_cmd + ''' time ./run.sh -e -m /home/m.json -t cases.task -b ${BRANCH_NAME}_${BUILD_ID} -l ${WKDIR}/log -o 600 ''' + extra_param + '''
                                    '''
                                }
                            }

--- a/cmake/cmake.version
+++ b/cmake/cmake.version
@@ -2,7 +2,7 @@
 IF (DEFINED VERNUMBER)
  SET(TD_VER_NUMBER ${VERNUMBER})
 ELSE ()
-  SET(TD_VER_NUMBER "3.0.3.0")
+  SET(TD_VER_NUMBER "3.0.3.1")
 ENDIF ()

 IF (DEFINED VERCOMPATIBLE)
@@ -16,7 +16,7 @@ find_program(HAVE_GIT NAMES git)
 IF (DEFINED GITINFO)
  SET(TD_VER_GIT ${GITINFO})
 ELSEIF (HAVE_GIT)
-  execute_process(COMMAND git log -1 --format=%H WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR} OUTPUT_VARIABLE GIT_COMMITID)
+  execute_process(COMMAND git log -1 --format=%H WORKING_DIRECTORY ${TD_COMMUNITY_DIR} OUTPUT_VARIABLE GIT_COMMITID)
  #message(STATUS "git log result:${GIT_COMMITID}")
  IF (GIT_COMMITID)
    string (REGEX REPLACE "[\n\t\r]" "" GIT_COMMITID ${GIT_COMMITID})
@@ -30,6 +30,23 @@ ELSE ()
  SET(TD_VER_GIT "no git commit id")
 ENDIF ()

+IF (DEFINED GITINFOI)
+  SET(TD_VER_GIT_INTERNAL ${GITINFOI})
+ELSEIF (HAVE_GIT)
+  execute_process(COMMAND git log -1 --format=%H WORKING_DIRECTORY ${PROJECT_SOURCE_DIR} OUTPUT_VARIABLE GIT_COMMITID)
+  message(STATUS "git log result:${GIT_COMMITID}")
+  IF (GIT_COMMITID)
+    string (REGEX REPLACE "[\n\t\r]" "" GIT_COMMITID ${GIT_COMMITID})
+    SET(TD_VER_GIT_INTERNAL ${GIT_COMMITID})
+  ELSE ()
+    message(STATUS "not a git repository")
+    SET(TD_VER_GIT "no git commit id")
+  ENDIF ()
+ELSE ()
+  message(STATUS "no git cmd")
+  SET(TD_VER_GIT_INTERNAL "no git commit id")
+ENDIF ()
+
 IF (DEFINED VERDATE)
  SET(TD_VER_DATE ${VERDATE})
 ELSE ()

--- a/cmake/taosadapter_CMakeLists.txt.in
+++ b/cmake/taosadapter_CMakeLists.txt.in
@@ -2,7 +2,7 @@
 # taosadapter
 ExternalProject_Add(taosadapter
        GIT_REPOSITORY https://github.com/taosdata/taosadapter.git
-        GIT_TAG d8059ff
+        GIT_TAG cb1e89c
        SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosadapter"
        BINARY_DIR ""
        #BUILD_IN_SOURCE TRUE

--- a/cmake/taostools_CMakeLists.txt.in
+++ b/cmake/taostools_CMakeLists.txt.in
@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
        GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG 1a69bd0
+        GIT_TAG 53d14fa
        SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
        BINARY_DIR ""
        #BUILD_IN_SOURCE TRUE

--- a/docs/en/07-develop/07-tmq.mdx
+++ b/docs/en/07-develop/07-tmq.mdx
@@ -293,7 +293,6 @@ You configure the following parameters when creating a consumer:
 |      `auto.offset.reset`       |  enum   | Initial offset for the consumer group                                     | Specify `earliest`, `latest`, or `none`(default) |
 |      `enable.auto.commit`      | boolean | Commit automatically                                             | Specify `true` or `false`.                   |
 |   `auto.commit.interval.ms`    | integer | Interval for automatic commits, in milliseconds                           |
-| `enable.heartbeat.background`  | boolean | Backend heartbeat; if enabled, the consumer does not go offline even if it has not polled for a long time |                                             |
 | `experimental.snapshot.enable` | boolean | Specify whether to consume messages from the WAL or from TSBS                    |                                             |
 |     `msg.with.table.name`      | boolean | Specify whether to deserialize table names from messages                                 |

@@ -368,7 +367,6 @@ conf := &tmq.ConfigMap{
 "td.connect.port":              "6030",
 "client.id":                    "test_tmq_c",
 "enable.auto.commit":           "false",
- "enable.heartbeat.background":  "true",
 "experimental.snapshot.enable": "true",
 "msg.with.table.name":          "true",
 }
@@ -418,7 +416,6 @@ Python programs use the following parameters:
 | `auto.commit.interval.ms` | string | Interval for automatic commits, in milliseconds | |
 | `auto.offset.reset` | string | Initial offset for the consumer group | Specify `earliest`, `latest`, or `none`(default) |
 | `experimental.snapshot.enable` | string | Specify whether to consume messages from the WAL or from TSDB | Specify `true` or `false` |
-| `enable.heartbeat.background` | string | Backend heartbeat; if enabled, the consumer does not go offline even if it has not polled for a long time | Specify `true` or `false` |

 </TabItem>


--- a/docs/examples/go/sub/main.go
+++ b/docs/examples/go/sub/main.go
@@ -35,7 +35,6 @@ func main() {
 		"td.connect.port":              "6030",
 		"client.id":                    "test_tmq_client",
 		"enable.auto.commit":           "false",
-		"enable.heartbeat.background":  "true",
 		"experimental.snapshot.enable": "true",
 		"msg.with.table.name":          "true",
 	})

--- a/docs/zh/05-get-started/index.md
+++ b/docs/zh/05-get-started/index.md
@@ -41,7 +41,7 @@ TDengine 知识地图中涵盖了 TDengine 的各种知识点，揭示了各概
 <td style={{padding:'1em 3em',border:0}}><img src={official_account} alt="TDengine 微信公众号" width="200" /></td>
 </tr>
 <tr align="center">
-<td style={{padding:'1em 3em',border:0}}>加入“物联网大数据技术群”<br/>与大家进行技术交流</td>
+<td style={{padding:'1em 3em',border:0}}>加入 TDengine 微信群<br/>了解学习最新物联网技术</td>
 <td style={{padding:'1em 3em',border:0}}>关注 TDengine 视频号<br/>收看技术直播与教学视频</td>
 <td style={{padding:'1em 3em',border:0}}>关注 TDengine 公众号<br/>阅读技术文章与行业案例</td>
 </tr>

--- a/docs/zh/07-develop/07-tmq.mdx
+++ b/docs/zh/07-develop/07-tmq.mdx
@@ -291,7 +291,6 @@ CREATE TOPIC topic_name AS DATABASE db_name;
 |      `auto.offset.reset`       |  enum   | 消费组订阅的初始位置                                     | 可选：`earliest`(default), `latest`, `none` |
 |      `enable.auto.commit`      | boolean | 是否启用消费位点自动提交                                 | 合法值：`true`, `false`。                   |
 |   `auto.commit.interval.ms`    | integer | 以毫秒为单位的消费记录自动提交消费位点时间间             | 默认 5000 m                                 |
-| `enable.heartbeat.background`  | boolean | 启用后台心跳，启用后即使长时间不 poll 消息也不会造成离线 | 默认开启                                    |
 | `experimental.snapshot.enable` | boolean | 是否允许从 TSDB 消费数据                                 | 实验功能，默认关闭                          |
 |     `msg.with.table.name`      | boolean | 是否允许从消息中解析表名, 不适用于列订阅（列订阅时可将 tbname 作为列写入 subquery 语句）               | |

@@ -366,7 +365,6 @@ conf := &tmq.ConfigMap{
 "td.connect.port":              "6030",
 "client.id":                    "test_tmq_c",
 "enable.auto.commit":           "false",
- "enable.heartbeat.background":  "true",
 "experimental.snapshot.enable": "true",
 "msg.with.table.name":          "true",
 }
@@ -418,7 +416,6 @@ consumer = Consumer({"group.id": "local", "td.connect.ip": "127.0.0.1"})
 | `auto.commit.interval.ms` | string | 以毫秒为单位的自动提交时间间隔 | 默认值：5000 ms |
 | `auto.offset.reset` | string | 消费组订阅的初始位置 | 可选：`earliest`(default), `latest`, `none` |
 | `experimental.snapshot.enable` | string | 是否允许从 TSDB 消费数据 | 合法值：`true`, `false` |
-| `enable.heartbeat.background` | string | 启用后台心跳，启用后即使长时间不 poll 消息也不会造成离线 | 合法值：`true`, `false` |

 </TabItem>


--- a/examples/JDBC/taosdemo/pom.xml
+++ b/examples/JDBC/taosdemo/pom.xml
@@ -10,7 +10,7 @@
    <description>Demo project for TDengine</description>

    <properties>
-        <spring.version>5.3.20</spring.version>
+        <spring.version>5.3.26</spring.version>
    </properties>

    <dependencies>

--- a/examples/rust/wrapper.h
+++ b/examples/rust/wrapper.h
+#include<taos.h>
--- a/include/common/tcommon.h
+++ b/include/common/tcommon.h
@@ -25,13 +25,6 @@
 extern "C" {
 #endif

-// TODO remove it
-enum {
-  TMQ_CONF__RESET_OFFSET__NONE = -3,
-  TMQ_CONF__RESET_OFFSET__EARLIEAST = -2,
-  TMQ_CONF__RESET_OFFSET__LATEST = -1,
-};
-
 // clang-format off
 #define IS_META_MSG(x) ( \
     x == TDMT_VND_CREATE_STB     \
@@ -192,7 +185,7 @@ typedef struct SBlockID {
 typedef struct SDataBlockInfo {
  STimeWindow window;
  int32_t     rowSize;
-  int32_t     rows;  // todo hide this attribute
+  int64_t     rows;  // todo hide this attribute
  uint32_t    capacity;
  SBlockID    id;
  int16_t     hasVarCol;

--- a/include/common/tdatablock.h
+++ b/include/common/tdatablock.h
@@ -178,7 +178,7 @@ int32_t getJsonValueLen(const char* data);

 int32_t colDataSetVal(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, bool isNull);
 int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, bool isNull);
-int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, uint32_t numOfRows);
+int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, uint32_t numOfRows, bool trimValue);
 int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, int32_t* capacity,
                        const SColumnInfoData* pSource, int32_t numOfRow2);
 int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* pSource, int32_t numOfRows,

--- a/include/common/tglobal.h
+++ b/include/common/tglobal.h
@@ -98,7 +98,8 @@ extern char    *tsSvrCrashReportUri;

 // query buffer management
 extern int32_t tsQueryBufferSize;  // maximum allowed usage buffer size in MB for each data node during query processing
-extern int64_t tsQueryBufferSizeBytes;  // maximum allowed usage buffer size in byte for each data node
+extern int64_t tsQueryBufferSizeBytes;    // maximum allowed usage buffer size in byte for each data node
+extern int32_t tsCacheLazyLoadThreshold;  // cost threshold for last/last_row loading cache as much as possible

 // query client
 extern int32_t tsQueryPolicy;
@@ -145,10 +146,10 @@ extern char tsUdfdResFuncs[];
 extern char tsUdfdLdLibPath[];

 // schemaless
-extern char    tsSmlChildTableName[];
-extern char    tsSmlTagName[];
-//extern bool    tsSmlDataFormat;
-//extern int32_t tsSmlBatchSize;
+extern char tsSmlChildTableName[];
+extern char tsSmlTagName[];
+// extern bool    tsSmlDataFormat;
+// extern int32_t tsSmlBatchSize;

 // wal
 extern int64_t tsWalFsyncDataSizeLimit;

--- a/include/common/tmsg.h
+++ b/include/common/tmsg.h
@@ -1136,6 +1136,7 @@ typedef struct {
  int64_t numOfInsertSuccessReqs;
  int64_t numOfBatchInsertReqs;
  int64_t numOfBatchInsertSuccessReqs;
+  int32_t numOfCachedTables;
 } SVnodeLoad;

 typedef struct {
@@ -3185,6 +3186,7 @@ typedef struct {
  SArray*      blockData;
  SArray*      blockTbName;
  SArray*      blockSchema;
+  // the following attributes are extended from SMqDataRsp
  int32_t      createTableNum;
  SArray*      createTableLen;
  SArray*      createTableReq;

--- a/include/common/tmsgcb.h
+++ b/include/common/tmsgcb.h
@@ -34,7 +34,7 @@ typedef enum {
  WRITE_QUEUE,
  APPLY_QUEUE,
  SYNC_QUEUE,
-  SYNC_CTRL_QUEUE,
+  SYNC_RD_QUEUE,
  STREAM_QUEUE,
  QUEUE_MAX,
 } EQueueType;

--- a/include/common/tmsgdef.h
+++ b/include/common/tmsgdef.h
@@ -259,7 +259,7 @@ enum {

  TD_NEW_MSG_SEG(TDMT_SYNC_MSG)
  TD_DEF_MSG_TYPE(TDMT_SYNC_TIMEOUT, "sync-timer", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_SYNC_PING, "sync-ping", NULL, NULL)             // no longer used
+  TD_DEF_MSG_TYPE(TDMT_SYNC_TIMEOUT_ELECTION, "sync-elect", NULL, NULL)
  TD_DEF_MSG_TYPE(TDMT_SYNC_PING_REPLY, "sync-ping-reply", NULL, NULL) // no longer used
  TD_DEF_MSG_TYPE(TDMT_SYNC_CLIENT_REQUEST, "sync-client-request", NULL, NULL)
  TD_DEF_MSG_TYPE(TDMT_SYNC_CLIENT_REQUEST_BATCH, "sync-client-request-batch", NULL, NULL)

--- a/include/libs/executor/executor.h
+++ b/include/libs/executor/executor.h
@@ -67,9 +67,10 @@ typedef enum {
 * Create the exec task for stream mode
 * @param pMsg
 * @param SReadHandle
+ * @param vgId
 * @return
 */
-qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers);
+qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers, int32_t vgId);

 /**
 * Create the exec task for queue mode
@@ -77,9 +78,18 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers);
 * @param SReadHandle
 * @return
 */
-qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, SSchemaWrapper** pSchema);
+qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int32_t vgId, int32_t* numOfCols, uint64_t id);
+
+/**
+ * set the task Id, usually used by message queue process
+ * @param tinfo
+ * @param taskId
+ * @param queryId
+ */
+void qSetTaskId(qTaskInfo_t tinfo, uint64_t taskId, uint64_t queryId);

 int32_t qSetStreamOpOpen(qTaskInfo_t tinfo);
+
 /**
 * Set multiple input data blocks for the stream scan.
 * @param tinfo
@@ -140,7 +150,6 @@ int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* table
 * @param handle
 * @return
 */
-
 int32_t qExecTaskOpt(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds, bool* hasMore, SLocalFetch* pLocal);

 int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pBlock, uint64_t* useconds);
@@ -153,6 +162,7 @@ void qCleanExecTaskBlockBuf(qTaskInfo_t tinfo);
 * @return
 */
 int32_t qAsyncKillTask(qTaskInfo_t tinfo, int32_t rspCode);
+int32_t qKillTask(qTaskInfo_t tinfo, int32_t rspCode);

 bool qTaskIsExecuting(qTaskInfo_t qinfo);

@@ -162,14 +172,6 @@ bool qTaskIsExecuting(qTaskInfo_t qinfo);
 */
 void qDestroyTask(qTaskInfo_t tinfo);

-/**
- * Extract the qualified table id list, and than pass them to the TSDB driver to load the required table data blocks.
- *
- * @param iter  the table iterator to traverse all tables belongs to a super table, or an invert index
- * @return
- */
-int32_t qGetQualifiedTableIdList(void* pTableList, const char* tagCond, int32_t tagCondLen, SArray* pTableIdList);
-
 void qProcessRspMsg(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet);

 int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, SArray* pExecInfoList /*,int32_t* resNum, SExplainExecInfo** pRes*/);

--- a/include/libs/function/function.h
+++ b/include/libs/function/function.h
@@ -112,7 +112,7 @@ typedef struct SResultDataInfo {
 typedef struct SInputColumnInfoData {
  int32_t           totalRows;        // total rows in current columnar data
  int32_t           startRowIndex;    // handle started row index
-  int32_t           numOfRows;        // the number of rows needs to be handled
+  int64_t           numOfRows;        // the number of rows needs to be handled
  int32_t           numOfInputCols;   // PTS is not included
  bool              colDataSMAIsSet;  // if agg is set or not
  SColumnInfoData  *pPTS;             // primary timestamp column

--- a/include/libs/function/functionMgt.h
+++ b/include/libs/function/functionMgt.h
@@ -241,6 +241,7 @@ int32_t fmGetUdafExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet);
 int32_t fmSetInvertFunc(int32_t funcId, SFuncExecFuncs* pFpSet);
 int32_t fmSetNormalFunc(int32_t funcId, SFuncExecFuncs* pFpSet);
 bool    fmIsInvertible(int32_t funcId);
+char*   fmGetFuncName(int32_t funcId);

 #ifdef __cplusplus
 }

--- a/include/libs/monitor/monitor.h
+++ b/include/libs/monitor/monitor.h
@@ -147,9 +147,9 @@ typedef struct {
 } SMonStbInfo;

 typedef struct {
-  int32_t expire_time;
-  int64_t timeseries_used;
-  int64_t timeseries_total;
+  uint32_t expire_time;
+  int64_t  timeseries_used;
+  int64_t  timeseries_total;
 } SMonGrantInfo;

 typedef struct {

--- a/include/libs/qcom/query.h
+++ b/include/libs/qcom/query.h
@@ -33,6 +33,7 @@ typedef enum {
  JOB_TASK_STATUS_INIT,
  JOB_TASK_STATUS_EXEC,
  JOB_TASK_STATUS_PART_SUCC,
+  JOB_TASK_STATUS_FETCH,
  JOB_TASK_STATUS_SUCC,
  JOB_TASK_STATUS_FAIL,
  JOB_TASK_STATUS_DROP,

--- a/include/libs/qworker/qworker.h
+++ b/include/libs/qworker/qworker.h
@@ -106,6 +106,8 @@ int32_t qWorkerProcessLocalQuery(void *pMgmt, uint64_t sId, uint64_t qId, uint64
 int32_t qWorkerProcessLocalFetch(void *pMgmt, uint64_t sId, uint64_t qId, uint64_t tId, int64_t rId, int32_t eId,
                                 void **pRsp, SArray *explainRes);

+int32_t qWorkerDbgEnableDebug(char *option);
+
 #ifdef __cplusplus
 }
 #endif

--- a/include/libs/stream/streamState.h
+++ b/include/libs/stream/streamState.h
@@ -23,20 +23,19 @@ extern "C" {
 #ifndef _STREAM_STATE_H_
 #define _STREAM_STATE_H_

-typedef struct SStreamTask SStreamTask;
-
 typedef bool (*state_key_cmpr_fn)(void* pKey1, void* pKey2);

 typedef struct STdbState {
-  SStreamTask* pOwner;
-  TDB*         db;
-  TTB*         pStateDb;
-  TTB*         pFuncStateDb;
-  TTB*         pFillStateDb;  // todo refactor
-  TTB*         pSessionStateDb;
-  TTB*         pParNameDb;
-  TTB*         pParTagDb;
-  TXN*         txn;
+  struct SStreamTask* pOwner;
+
+  TDB* db;
+  TTB* pStateDb;
+  TTB* pFuncStateDb;
+  TTB* pFillStateDb;  // todo refactor
+  TTB* pSessionStateDb;
+  TTB* pParNameDb;
+  TTB* pParTagDb;
+  TXN* txn;
 } STdbState;

 // incremental state storage
@@ -45,7 +44,7 @@ typedef struct {
  int32_t    number;
 } SStreamState;

-SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath, int32_t szPage, int32_t pages);
+SStreamState* streamStateOpen(char* path, struct SStreamTask* pTask, bool specPath, int32_t szPage, int32_t pages);
 void          streamStateClose(SStreamState* pState);
 int32_t       streamStateBegin(SStreamState* pState);
 int32_t       streamStateCommit(SStreamState* pState);

--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -295,7 +295,7 @@ typedef struct {
  SEpSet  epSet;
 } SStreamChildEpInfo;

-typedef struct SStreamTask {
+struct SStreamTask {
  int64_t streamId;
  int32_t taskId;
  int32_t totalLevel;
@@ -362,8 +362,7 @@ typedef struct SStreamTask {

  int64_t checkpointingId;
  int32_t checkpointAlignCnt;
-
-} SStreamTask;
+};

 int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo);
 int32_t tDecodeStreamEpInfo(SDecoder* pDecoder, SStreamChildEpInfo* pInfo);
@@ -587,7 +586,7 @@ int32_t streamProcessDispatchReq(SStreamTask* pTask, SStreamDispatchReq* pReq, S
 int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, int32_t code);

 int32_t streamProcessRetrieveReq(SStreamTask* pTask, SStreamRetrieveReq* pReq, SRpcMsg* pMsg);
-int32_t streamProcessRetrieveRsp(SStreamTask* pTask, SStreamRetrieveRsp* pRsp);
+// int32_t streamProcessRetrieveRsp(SStreamTask* pTask, SStreamRetrieveRsp* pRsp);

 int32_t streamTryExec(SStreamTask* pTask);
 int32_t streamSchedExec(SStreamTask* pTask);

--- a/include/libs/sync/sync.h
+++ b/include/libs/sync/sync.h
@@ -48,7 +48,7 @@ extern "C" {
 #define SYNC_HEARTBEAT_REPLY_SLOW_MS 1500
 #define SYNC_SNAP_RESEND_MS          1000 * 60

-#define SYNC_VND_COMMIT_MIN_MS 1000
+#define SYNC_VND_COMMIT_MIN_MS 3000

 #define SYNC_MAX_BATCH_SIZE 1
 #define SYNC_INDEX_BEGIN    0
@@ -143,10 +143,11 @@ typedef struct SSyncFSM {
  void* data;

  int32_t (*FpCommitCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta);
+  SyncIndex (*FpAppliedIndexCb)(const struct SSyncFSM* pFsm);
  int32_t (*FpPreCommitCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta);
  void (*FpRollBackCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta);

-  void (*FpRestoreFinishCb)(const struct SSyncFSM* pFsm);
+  void (*FpRestoreFinishCb)(const struct SSyncFSM* pFsm, const SyncIndex commitIdx);
  void (*FpReConfigCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SReConfigCbMeta* pMeta);
  void (*FpLeaderTransferCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta);
  bool (*FpApplyQueueEmptyCb)(const struct SSyncFSM* pFsm);

--- a/include/util/taoserror.h
+++ b/include/util/taoserror.h
@@ -542,7 +542,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_SYN_BATCH_ERROR               TAOS_DEF_ERROR_CODE(0, 0x0913)
 #define TSDB_CODE_SYN_RESTORING                 TAOS_DEF_ERROR_CODE(0, 0x0914)
 #define TSDB_CODE_SYN_INVALID_SNAPSHOT_MSG      TAOS_DEF_ERROR_CODE(0, 0x0915)  // internal
-#define TSDB_CODE_SYN_BUFFER_FULL               TAOS_DEF_ERROR_CODE(0, 0x0916)  //
+#define TSDB_CODE_SYN_BUFFER_FULL               TAOS_DEF_ERROR_CODE(0, 0x0916)
+#define TSDB_CODE_SYN_WRITE_STALL               TAOS_DEF_ERROR_CODE(0, 0x0917)
 #define TSDB_CODE_SYN_INTERNAL_ERROR            TAOS_DEF_ERROR_CODE(0, 0x09FF)

 // tq
@@ -687,6 +688,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_NOT_SUPPORT_JOIN          TAOS_DEF_ERROR_CODE(0, 0x2664)
 #define TSDB_CODE_PAR_INVALID_TAGS_PC           TAOS_DEF_ERROR_CODE(0, 0x2665)
 #define TSDB_CODE_PAR_INVALID_TIMELINE_QUERY    TAOS_DEF_ERROR_CODE(0, 0x2666)
+#define TSDB_CODE_PAR_INVALID_OPTR_USAGE        TAOS_DEF_ERROR_CODE(0, 0x2667)
 #define TSDB_CODE_PAR_INTERNAL_ERROR            TAOS_DEF_ERROR_CODE(0, 0x26FF)

 //planner

--- a/include/util/tarray.h
+++ b/include/util/tarray.h
@@ -69,14 +69,6 @@ void* taosArrayAddBatch(SArray* pArray, const void* pData, int32_t nEles);
 */
 void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp)(void*));

-/**
- *
- * @param pArray
- * @param comparFn
- * @param fp
- */
-void taosArrayRemoveDuplicateP(SArray* pArray, __compar_fn_t comparFn, void (*fp)(void*));
-
 /**
 *  add all element from the source array list into the destination
 * @param pArray
@@ -216,6 +208,8 @@ void  taosArrayDestroyP(SArray* pArray, FDelete fp);

 void  taosArrayDestroyEx(SArray* pArray, FDelete fp);

+void taosArraySwap(SArray* a, SArray* b);
+
 /**
 * sort the array
 * @param pArray
@@ -252,14 +246,6 @@ void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* par
 int32_t taosEncodeArray(void** buf, const SArray* pArray, FEncode encode);
 void*   taosDecodeArray(const void* buf, SArray** pArray, FDecode decode, int32_t dataSz);

-/**
- * swap array
- * @param a
- * @param b
- * @return
- */
-void taosArraySwap(SArray* a, SArray* b);
-
 #ifdef __cplusplus
 }
 #endif

--- a/include/util/tdef.h
+++ b/include/util/tdef.h
@@ -248,7 +248,7 @@ typedef enum ELogicConditionType {
 #define TSDB_AUTH_LEN          16
 #define TSDB_PASSWORD_LEN      32
 #define TSDB_USET_PASSWORD_LEN 129
-#define TSDB_VERSION_LEN       12
+#define TSDB_VERSION_LEN       32
 #define TSDB_LABEL_LEN         8
 #define TSDB_JOB_STATUS_LEN    32

@@ -316,10 +316,10 @@ typedef enum ELogicConditionType {
 #define TSDB_MAX_KEEP_NS                (365 * 292 * 1440)  // data in db to be reserved.
 #define TSDB_DEFAULT_KEEP               (3650 * 1440)       // ten years
 #define TSDB_MIN_MINROWS_FBLOCK         10
-#define TSDB_MAX_MINROWS_FBLOCK         1000
+#define TSDB_MAX_MINROWS_FBLOCK         1000000
 #define TSDB_DEFAULT_MINROWS_FBLOCK     100
 #define TSDB_MIN_MAXROWS_FBLOCK         200
-#define TSDB_MAX_MAXROWS_FBLOCK         10000
+#define TSDB_MAX_MAXROWS_FBLOCK         10000000
 #define TSDB_DEFAULT_MAXROWS_FBLOCK     4096
 #define TSDB_MIN_FSYNC_PERIOD           0
 #define TSDB_MAX_FSYNC_PERIOD           180000  // millisecond

--- a/include/util/tlrucache.h
+++ b/include/util/tlrucache.h
@@ -55,6 +55,8 @@ void *taosLRUCacheValue(SLRUCache *cache, LRUHandle *handle);
 size_t taosLRUCacheGetUsage(SLRUCache *cache);
 size_t taosLRUCacheGetPinnedUsage(SLRUCache *cache);

+int32_t taosLRUCacheGetElems(SLRUCache *cache);
+
 void   taosLRUCacheSetCapacity(SLRUCache *cache, size_t capacity);
 size_t taosLRUCacheGetCapacity(SLRUCache *cache);


--- a/include/util/tsimplehash.h
+++ b/include/util/tsimplehash.h
@@ -48,6 +48,13 @@ SSHashObj *tSimpleHashInit(size_t capacity, _hash_fn_t fn);
 */
 int32_t tSimpleHashGetSize(const SSHashObj *pHashObj);

+/**
+ * set the free function pointer
+ * @param pHashObj
+ * @param freeFp
+ */
+void tSimpleHashSetFreeFp(SSHashObj* pHashObj, _hash_free_fn_t freeFp);
+
 int32_t tSimpleHashPrint(const SSHashObj *pHashObj);

 /**

--- a/include/util/version.h
+++ b/include/util/version.h
@@ -23,6 +23,7 @@ extern "C" {
 extern char version[];
 extern char compatible_version[];
 extern char gitinfo[];
+extern char gitinfoOfInternal[];
 extern char buildinfo[];

 #ifdef __cplusplus

--- a/packaging/rpm/tdengine.spec
+++ b/packaging/rpm/tdengine.spec
@@ -90,45 +90,33 @@ cp %{_compiledir}/../include/libs/function/taosudf.h       %{buildroot}%{homepat
 cp -r %{_compiledir}/../examples/*                  %{buildroot}%{homepath}/examples

 if [ -f %{_compiledir}/build/bin/jemalloc-config ]; then
-    mkdir -p %{buildroot}%{userlocalpath}/bin
-    mkdir -p %{buildroot}%{userlocalpath}/lib
-    mkdir -p %{buildroot}%{userlocalpath}/lib/pkgconfig
-    mkdir -p %{buildroot}%{userlocalpath}/include
-    mkdir -p %{buildroot}%{userlocalpath}/include/jemalloc
-    mkdir -p %{buildroot}%{userlocalpath}/share
-    mkdir -p %{buildroot}%{userlocalpath}/share/doc
-    mkdir -p %{buildroot}%{userlocalpath}/share/doc/jemalloc
-    mkdir -p %{buildroot}%{userlocalpath}/share/man
-    mkdir -p %{buildroot}%{userlocalpath}/share/man/man3
-
-    cp %{_compiledir}/build/bin/jemalloc-config %{buildroot}%{userlocalpath}/bin/
+    mkdir -p %{buildroot}%{homepath}/jemalloc/ ||:
+    mkdir -p %{buildroot}%{homepath}/jemalloc/include/jemalloc/ ||:
+    mkdir -p %{buildroot}%{homepath}/jemalloc/lib/ ||:
+    mkdir -p %{buildroot}%{homepath}/jemalloc/lib/pkgconfig ||:
+
+    cp %{_compiledir}/build/bin/jemalloc-config %{buildroot}%{homepath}/jemalloc/bin
    if [ -f %{_compiledir}/build/bin/jemalloc.sh ]; then
-        cp %{_compiledir}/build/bin/jemalloc.sh %{buildroot}%{userlocalpath}/bin/
+        cp %{_compiledir}/build/bin/jemalloc.sh %{buildroot}%{homepath}/jemalloc/bin
    fi
    if [ -f %{_compiledir}/build/bin/jeprof ]; then
-        cp %{_compiledir}/build/bin/jeprof %{buildroot}%{userlocalpath}/bin/
+        cp %{_compiledir}/build/bin/jeprof %{buildroot}%{homepath}/jemalloc/bin
    fi
    if [ -f %{_compiledir}/build/include/jemalloc/jemalloc.h ]; then
-        cp %{_compiledir}/build/include/jemalloc/jemalloc.h %{buildroot}%{userlocalpath}/include/jemalloc/
+        cp %{_compiledir}/build/include/jemalloc/jemalloc.h %{buildroot}%{homepath}/jemalloc/include/jemalloc/
    fi
    if [ -f %{_compiledir}/build/lib/libjemalloc.so.2 ]; then
-        cp %{_compiledir}/build/lib/libjemalloc.so.2 %{buildroot}%{userlocalpath}/lib/
-        ln -sf libjemalloc.so.2 %{buildroot}%{userlocalpath}/lib/libjemalloc.so
+        cp %{_compiledir}/build/lib/libjemalloc.so.2 %{buildroot}%{homepath}/jemalloc/lib
+        ln -sf libjemalloc.so.2 %{buildroot}%{homepath}/jemalloc/lib/libjemalloc.so
    fi
    if [ -f %{_compiledir}/build/lib/libjemalloc.a ]; then
-        cp %{_compiledir}/build/lib/libjemalloc.a %{buildroot}%{userlocalpath}/lib/
+        cp %{_compiledir}/build/lib/libjemalloc.a %{buildroot}%{homepath}/jemalloc/lib
    fi
    if [ -f %{_compiledir}/build/lib/libjemalloc_pic.a ]; then
-        cp %{_compiledir}/build/lib/libjemalloc_pic.a %{buildroot}%{userlocalpath}/lib/
+        cp %{_compiledir}/build/lib/libjemalloc_pic.a %{buildroot}%{homepath}/jemalloc/lib
    fi
    if [ -f %{_compiledir}/build/lib/pkgconfig/jemalloc.pc ]; then
-        cp %{_compiledir}/build/lib/pkgconfig/jemalloc.pc %{buildroot}%{userlocalpath}/lib/pkgconfig/
-    fi
-    if [ -f %{_compiledir}/build/share/doc/jemalloc/jemalloc.html ]; then
-        cp %{_compiledir}/build/share/doc/jemalloc/jemalloc.html %{buildroot}%{userlocalpath}/share/doc/jemalloc/
-    fi
-    if [ -f %{_compiledir}/build/share/man/man3/jemalloc.3 ]; then
-        cp %{_compiledir}/build/share/man/man3/jemalloc.3 %{buildroot}%{userlocalpath}/share/man/man3/
+        cp %{_compiledir}/build/lib/pkgconfig/jemalloc.pc %{buildroot}%{homepath}/jemalloc/lib/pkgconfig
    fi
 fi


--- a/packaging/tools/com.taosdata.taoskeeper.plist
+++ b/packaging/tools/com.taosdata.taoskeeper.plist
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+<dict>
+    <key>Label</key>
+    <string>com.tdengine.taoskeeper</string>
+    <key>ProgramArguments</key>
+    <array>
+        <string>/usr/local/bin/taoskeeper</string>
+    </array>
+    <key>ProcessType</key>
+    <string>Interactive</string>
+    <key>Disabled</key>
+    <false/>
+    <key>RunAtLoad</key>
+    <false/>
+    <key>LaunchOnlyOnce</key>
+    <false/>
+    <key>SessionCreate</key>
+    <true/>
+    <key>ExitTimeOut</key>
+    <integer>600</integer>
+    <key>KeepAlive</key>
+    <dict>
+        <key>SuccessfulExit</key>
+        <false/>
+        <key>AfterInitialDemand</key>
+        <true/>
+    </dict>
+    <key>Program</key>
+    <string>/usr/local/bin/taoskeeper</string>
+</dict>
+</plist>
\ No newline at end of file
--- a/packaging/tools/makepkg.sh
+++ b/packaging/tools/makepkg.sh
@@ -51,9 +51,9 @@ fi

 if [ -d ${top_dir}/tools/taos-tools/packaging/deb ]; then
  cd ${top_dir}/tools/taos-tools/packaging/deb
+  
+  taostools_ver=$(git for-each-ref --sort=taggerdate --format '%(tag)' refs/tags|grep -v taos | tail -1)
  [ -z "$taos_tools_ver" ] && taos_tools_ver="0.1.0"
-
-  taostools_ver=$(git tag |grep -v taos | sort | tail -1)
  taostools_install_dir="${release_dir}/${clientName2}Tools-${taostools_ver}"

  cd ${curr_dir}

--- a/packaging/tools/post.sh
+++ b/packaging/tools/post.sh
@@ -145,6 +145,59 @@ function install_include() {
    log_print "install include success"
 }

+function install_jemalloc() {
+  jemalloc_dir=${script_dir}/../jemalloc
+
+  if [ -d ${jemalloc_dir} ]; then
+    ${csudo}/usr/bin/install -c -d /usr/local/bin
+
+    if [ -f ${jemalloc_dir}/bin/jemalloc-config ]; then
+      ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/bin/jemalloc-config /usr/local/bin
+    fi
+    if [ -f ${jemalloc_dir}/bin/jemalloc.sh ]; then
+      ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/bin/jemalloc.sh /usr/local/bin
+    fi
+    if [ -f ${jemalloc_dir}/bin/jeprof ]; then
+      ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/bin/jeprof /usr/local/bin
+    fi
+    if [ -f ${jemalloc_dir}/include/jemalloc/jemalloc.h ]; then
+      ${csudo}/usr/bin/install -c -d /usr/local/include/jemalloc
+      ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/include/jemalloc/jemalloc.h /usr/local/include/jemalloc
+    fi
+    if [ -f ${jemalloc_dir}/lib/libjemalloc.so.2 ]; then
+      ${csudo}/usr/bin/install -c -d /usr/local/lib
+      ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.so.2 /usr/local/lib
+      ${csudo}ln -sf libjemalloc.so.2 /usr/local/lib/libjemalloc.so
+      ${csudo}/usr/bin/install -c -d /usr/local/lib
+      if [ -f ${jemalloc_dir}/lib/libjemalloc.a ]; then
+        ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.a /usr/local/lib
+      fi
+      if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
+        ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc_pic.a /usr/local/lib
+      fi
+      if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
+        ${csudo}/usr/bin/install -c -d /usr/local/lib/pkgconfig
+        ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/lib/pkgconfig/jemalloc.pc /usr/local/lib/pkgconfig
+      fi
+    fi
+    if [ -f ${jemalloc_dir}/share/doc/jemalloc/jemalloc.html ]; then
+      ${csudo}/usr/bin/install -c -d /usr/local/share/doc/jemalloc
+      ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/share/doc/jemalloc/jemalloc.html /usr/local/share/doc/jemalloc
+    fi
+    if [ -f ${jemalloc_dir}/share/man/man3/jemalloc.3 ]; then
+      ${csudo}/usr/bin/install -c -d /usr/local/share/man/man3
+      ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/share/man/man3/jemalloc.3 /usr/local/share/man/man3
+    fi
+
+    if [ -d /etc/ld.so.conf.d ]; then
+      echo "/usr/local/lib" | ${csudo}tee /etc/ld.so.conf.d/jemalloc.conf >/dev/null || echo -e "failed to write /etc/ld.so.conf.d/jemalloc.conf"
+      ${csudo}ldconfig
+    else
+      echo "/etc/ld.so.conf.d not found!"
+    fi
+  fi
+}
+
 function install_lib() {
    log_print "start install lib from ${lib_dir} to ${lib_link_dir}"
    ${csudo}rm -f ${lib_link_dir}/libtaos* || :
@@ -582,6 +635,11 @@ function install_service_on_launchctl() {
    ${csudo}cp ${install_main_dir}/service/com.taosdata.taosadapter.plist /Library/LaunchDaemons/com.taosdata.taosadapter.plist || :
    ${csudo}launchctl load -w /Library/LaunchDaemons/com.taosdata.taosadapter.plist || :
  fi
+  if [ -f ${install_main_dir}/service/com.taosdata.taoskeeper.plist ]; then
+    ${csudo}launchctl unload -w /Library/LaunchDaemons/com.taosdata.taoskeeper.plist > /dev/null 2>&1 || :
+    ${csudo}cp ${install_main_dir}/service/com.taosdata.taoskeeper.plist /Library/LaunchDaemons/com.taosdata.taoskeeper.plist || :
+    ${csudo}launchctl load -w /Library/LaunchDaemons/com.taosdata.taoskeeper.plist || :
+  fi
 }

 function install_taosadapter_service() {
@@ -658,6 +716,7 @@ function install_TDengine() {
    # Install include, lib, binary and service
    install_include &&
    install_lib &&
+    install_jemalloc
    install_bin

    if [[ "$?" != 0 ]];then

--- a/packaging/tools/tdengine.iss
+++ b/packaging/tools/tdengine.iss
@@ -21,7 +21,7 @@
 [Setup]
 VersionInfoVersion={#MyAppVersion}
 AppId={{A0F7A93C-79C4-485D-B2B8-F0D03DF42FAB}
-AppName={#MyAppName}
+AppName={#CusName}
 AppVersion={#MyAppVersion}
 ;AppVerName={#MyAppName} {#MyAppVersion}
 AppPublisher={#MyAppPublisher}
@@ -64,8 +64,8 @@ Source: {#MyAppSourceDir}\taosdump.exe; DestDir: "{app}"; DestName: "{#CusPrompt


 [run]
-Filename: {sys}\sc.exe; Parameters: "create taosd start= DEMAND binPath= ""C:\\{#CusName}\\taosd.exe --win_service""" ; Flags: runhidden
-Filename: {sys}\sc.exe; Parameters: "create taosadapter start= DEMAND binPath= ""C:\\{#CusName}\\taosadapter.exe""" ; Flags: runhidden
+Filename: {sys}\sc.exe; Parameters: "create taosd start= DEMAND binPath= ""C:\\TDengine\\taosd.exe --win_service""" ; Flags: runhidden
+Filename: {sys}\sc.exe; Parameters: "create taosadapter start= DEMAND binPath= ""C:\\TDengine\\taosadapter.exe""" ; Flags: runhidden

 [UninstallRun]
 RunOnceId: "stoptaosd"; Filename: {sys}\sc.exe; Parameters: "stop taosd" ; Flags: runhidden

--- a/packaging/tools/windows_before_install.txt
+++ b/packaging/tools/windows_before_install.txt
-TDengine is a high-efficient, scalable, high-available distributed time-series database, which makes a lot of optimizations on inserting and querying data, which is far more efficient than normal regular databases. So TDengine can meet the high requirements of IOT and other areas on storing and querying a large amount of data. 
+TDengine is an open-source, cloud-native time-series database optimized for Internet of Things (IoT), Connected Cars, and Industrial IoT. With its built-in caching, stream processing, and data subscription capabilities, TDengine offers a simplified solution for time-series data processing.

 TDengine will be installed under C:\TDengine, users can modify configuration file C:\TDengine\cfg\taos.cfg, set the log file path or other parameters.
 To start/stop TDengine with administrator privileges:  sc start/stop taosd

--- a/source/client/inc/clientInt.h
+++ b/source/client/inc/clientInt.h
@@ -287,22 +287,25 @@ static FORCE_INLINE SReqResultInfo* tmqGetCurResInfo(TAOS_RES* res) {
 }

 static FORCE_INLINE SReqResultInfo* tmqGetNextResInfo(TAOS_RES* res, bool convertUcs4) {
-  SMqRspObj* msg = (SMqRspObj*)res;
-  msg->resIter++;
-  if (msg->resIter < msg->rsp.blockNum) {
-    SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(msg->rsp.blockData, msg->resIter);
-    if (msg->rsp.withSchema) {
-      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(msg->rsp.blockSchema, msg->resIter);
-      setResSchemaInfo(&msg->resInfo, pSW->pSchema, pSW->nCols);
-      taosMemoryFreeClear(msg->resInfo.row);
-      taosMemoryFreeClear(msg->resInfo.pCol);
-      taosMemoryFreeClear(msg->resInfo.length);
-      taosMemoryFreeClear(msg->resInfo.convertBuf);
-      taosMemoryFreeClear(msg->resInfo.convertJson);
+  SMqRspObj* pRspObj = (SMqRspObj*)res;
+  pRspObj->resIter++;
+
+  if (pRspObj->resIter < pRspObj->rsp.blockNum) {
+    SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(pRspObj->rsp.blockData, pRspObj->resIter);
+    if (pRspObj->rsp.withSchema) {
+      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(pRspObj->rsp.blockSchema, pRspObj->resIter);
+      setResSchemaInfo(&pRspObj->resInfo, pSW->pSchema, pSW->nCols);
+      taosMemoryFreeClear(pRspObj->resInfo.row);
+      taosMemoryFreeClear(pRspObj->resInfo.pCol);
+      taosMemoryFreeClear(pRspObj->resInfo.length);
+      taosMemoryFreeClear(pRspObj->resInfo.convertBuf);
+      taosMemoryFreeClear(pRspObj->resInfo.convertJson);
    }
-    setQueryResultFromRsp(&msg->resInfo, pRetrieve, convertUcs4, false);
-    return &msg->resInfo;
+
+    setQueryResultFromRsp(&pRspObj->resInfo, pRetrieve, convertUcs4, false);
+    return &pRspObj->resInfo;
  }
+
  return NULL;
 }


--- a/source/client/inc/clientLog.h
+++ b/source/client/inc/clientLog.h
@@ -26,6 +26,7 @@ extern "C" {
 #define tscFatal(...)  do { if (cDebugFlag & DEBUG_FATAL) { taosPrintLog("TSC FATAL ", DEBUG_FATAL, cDebugFlag, __VA_ARGS__); }} while(0)
 #define tscError(...)  do { if (cDebugFlag & DEBUG_ERROR) { taosPrintLog("TSC ERROR ", DEBUG_ERROR, cDebugFlag, __VA_ARGS__); }} while(0)
 #define tscWarn(...)   do { if (cDebugFlag & DEBUG_WARN)  { taosPrintLog("TSC WARN ", DEBUG_WARN, cDebugFlag, __VA_ARGS__); }}  while(0)
+#define tscWarnL(...)  do { if (cDebugFlag & DEBUG_WARN)  { taosPrintLongString("TSC WARN ", DEBUG_WARN, cDebugFlag, __VA_ARGS__); }}  while(0)
 #define tscInfo(...)   do { if (cDebugFlag & DEBUG_INFO)  { taosPrintLog("TSC ", DEBUG_INFO, cDebugFlag, __VA_ARGS__); }} while(0)
 #define tscDebug(...)  do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLog("TSC ", DEBUG_DEBUG, cDebugFlag, __VA_ARGS__); }} while(0)
 #define tscTrace(...)  do { if (cDebugFlag & DEBUG_TRACE) { taosPrintLog("TSC ", DEBUG_TRACE, cDebugFlag, __VA_ARGS__); }} while(0)

--- a/source/client/inc/clientSml.h
+++ b/source/client/inc/clientSml.h
@@ -70,7 +70,7 @@ extern "C" {
 #define VALUE_LEN 6

 #define OTD_JSON_FIELDS_NUM     4
-#define MAX_RETRY_TIMES 5
+#define MAX_RETRY_TIMES 100
 typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;

 typedef enum {

--- a/source/client/inc/clientStmt.h
+++ b/source/client/inc/clientStmt.h
@@ -102,6 +102,7 @@ typedef struct STscStmt {
  SStmtBindInfo bInfo;

  int64_t reqid;
+  int32_t errCode;
 } STscStmt;

 extern char *gStmtStatusStr[];
@@ -121,6 +122,7 @@ extern char *gStmtStatusStr[];
    int32_t _code = c;                \
    if (_code != TSDB_CODE_SUCCESS) { \
      terrno = _code;                 \
+      pStmt->errCode = _code;         \
      return _code;                   \
    }                                 \
  } while (0)
@@ -129,6 +131,7 @@ extern char *gStmtStatusStr[];
    int32_t _code = c;                \
    if (_code != TSDB_CODE_SUCCESS) { \
      terrno = _code;                 \
+      pStmt->errCode = _code;         \
    }                                 \
    return _code;                     \
  } while (0)
@@ -137,6 +140,7 @@ extern char *gStmtStatusStr[];
    code = c;                        \
    if (code != TSDB_CODE_SUCCESS) { \
      terrno = code;                 \
+      pStmt->errCode = code;         \
      goto _return;                  \
    }                                \
  } while (0)

--- a/source/client/src/clientEnv.c
+++ b/source/client/src/clientEnv.c
@@ -103,6 +103,7 @@ static void deregisterRequest(SRequestObj *pRequest) {

  if (duration >= SLOW_QUERY_INTERVAL) {
    atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1);
+    tscWarnL("slow query: %s, duration:%" PRId64, pRequest->sqlstr, duration);
  }

  releaseTscObj(pTscObj->id);
@@ -543,7 +544,7 @@ void taos_init_imp(void) {

  if (taosCreateLog("taoslog", 10, configDir, NULL, NULL, NULL, NULL, 1) != 0) {
    // ignore create log failed, only print
-    printf(" WARING: Create taoslog failed. configDir=%s\n", configDir);
+    printf(" WARING: Create taoslog failed:%s. configDir=%s\n", strerror(errno), configDir);
  }

  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1) != 0) {

--- a/source/client/src/clientImpl.c
+++ b/source/client/src/clientImpl.c
@@ -192,7 +192,7 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param,
  (*pRequest)->sqlLen = sqlLen;
  (*pRequest)->validateOnly = validateSql;

-  SSyncQueryParam* newpParam;
+  SSyncQueryParam* newpParam = NULL;
  if (param == NULL) {
    newpParam = taosMemoryCalloc(1, sizeof(SSyncQueryParam));
    if (newpParam == NULL) {
@@ -1085,6 +1085,10 @@ static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaDat
    tscDebug("0x%" PRIx64 " plan not executed, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
             pRequest->requestId);
    destorySqlCallbackWrapper(pWrapper);
+    if (TSDB_CODE_SUCCESS != code) {
+      pRequest->code = terrno;
+    }
+
    pRequest->body.queryFp(pRequest->body.param, pRequest, code);
  }

@@ -1132,11 +1136,6 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
      pRequest->body.queryFp(pRequest->body.param, pRequest, -1);
      break;
  }
-
-  // TODO weired responding code?
-  if (TSDB_CODE_SUCCESS != code) {
-    pRequest->code = terrno;
-  }
 }

 int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest) {

--- a/source/client/src/clientJniConnector.c
+++ b/source/client/src/clientJniConnector.c
@@ -581,8 +581,8 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_fetchBlockImp(JNI
    return JNI_RESULT_SET_NULL;
  }

-  void   *data;
-  int32_t numOfRows;
+  void   *data = NULL;
+  int32_t numOfRows = 0;
  int     error_code = taos_fetch_raw_block(tres, &numOfRows, &data);
  if (numOfRows == 0) {
    if (error_code == JNI_SUCCESS) {

--- a/source/client/src/clientMain.c
+++ b/source/client/src/clientMain.c
@@ -271,8 +271,6 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
    SReqResultInfo *pResultInfo;
    if (msg->resIter == -1) {
      pResultInfo = tmqGetNextResInfo(res, true);
-      tscDebug("consumer:0x%" PRIx64 ", vgId:%d, numOfRows:%" PRId64 ", total rows:%" PRId64, msg->rsp.head.consumerId,
-               msg->vgId, pResultInfo->numOfRows, pResultInfo->totalRows);
    } else {
      pResultInfo = tmqGetCurResInfo(res);
    }
@@ -287,9 +285,6 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
        return NULL;
      }

-      tscDebug("consumer:0x%" PRIx64 " vgId:%d, numOfRows:%" PRId64 ", total rows:%" PRId64, msg->rsp.head.consumerId,
-               msg->vgId, pResultInfo->numOfRows, pResultInfo->totalRows);
-
      doSetOneRowPtr(pResultInfo);
      pResultInfo->current += 1;
      return pResultInfo->row;
@@ -611,6 +606,9 @@ int taos_fetch_block_s(TAOS_RES *res, int *numOfRows, TAOS_ROW *rows) {
 }

 int taos_fetch_raw_block(TAOS_RES *res, int *numOfRows, void **pData) {
+  *numOfRows = 0;
+  *pData = NULL;
+
  if (res == NULL || TD_RES_TMQ_META(res)) {
    return 0;
  }

--- a/source/client/src/clientMsgHandler.c
+++ b/source/client/src/clientMsgHandler.c
@@ -76,7 +76,7 @@ int32_t processConnectRsp(void* param, SDataBuf* pMsg, int32_t code) {
    goto End;
  }

-  if ((code = taosCheckVersionCompatibleFromStr(version, connectRsp.sVer, 2)) != 0) {
+  if ((code = taosCheckVersionCompatibleFromStr(version, connectRsp.sVer, 3)) != 0) {
    setErrno(pRequest, code);
    tsem_post(&pRequest->body.rspSem);
    goto End;
@@ -506,6 +506,9 @@ int32_t processShowVariablesRsp(void* param, SDataBuf* pMsg, int32_t code) {
      code = setQueryResultFromRsp(&pRequest->body.resInfo, pRes, false, true);
    }

+    if (code != 0) {
+      taosMemoryFree(pRes);
+    }
    tFreeSShowVariablesRsp(&rsp);
  }


--- a/source/client/src/clientRawBlockWrite.c
+++ b/source/client/src/clientRawBlockWrite.c
--- a/source/client/src/clientSml.c
+++ b/source/client/src/clientSml.c
--- a/source/client/src/clientSmlJson.c
+++ b/source/client/src/clientSmlJson.c
@@ -1237,10 +1237,12 @@ int32_t smlParseJSON(SSmlHandle *info, char *payload) {
      if (cnt >= payloadNum) {
        payloadNum = payloadNum << 1;
        void *tmp = taosMemoryRealloc(info->lines, payloadNum * sizeof(SSmlLineInfo));
-        if (tmp != NULL) {
-          info->lines = (SSmlLineInfo *)tmp;
-          memset(info->lines + cnt, 0, (payloadNum - cnt) * sizeof(SSmlLineInfo));
+        if (tmp == NULL) {
+          ret = TSDB_CODE_OUT_OF_MEMORY;
+          return ret;
        }
+        info->lines = (SSmlLineInfo *)tmp;
+        memset(info->lines + cnt, 0, (payloadNum - cnt) * sizeof(SSmlLineInfo));
      }
      ret = smlParseJSONString(info, &dataPointStart, info->lines + cnt);
      if ((info->lines + cnt)->measure == NULL) break;

--- a/source/client/src/clientSmlLine.c
+++ b/source/client/src/clientSmlLine.c
@@ -292,6 +292,7 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
    info->currSTableMeta->uid = tinfo->uid;
    tinfo->tableDataCtx = smlInitTableDataCtx(info->pQuery, info->currSTableMeta);
    if (tinfo->tableDataCtx == NULL) {
+      smlDestroyTableInfo(info, tinfo);
      smlBuildInvalidDataMsg(&info->msgBuf, "smlInitTableDataCtx error", NULL);
      return TSDB_CODE_SML_INVALID_DATA;
    }
@@ -582,12 +583,14 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
               .i = ts,
               .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes};
  if (info->dataFormat) {
+    uDebug("SML:0x%" PRIx64 " smlParseInfluxString format true, ts:%" PRId64, info->id, ts);
    ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, 0);
    if(ret != TSDB_CODE_SUCCESS){return ret;}
    ret = smlBuildRow(info->currTableDataCtx);
    if(ret != TSDB_CODE_SUCCESS){return ret;}
    clearColValArray(info->currTableDataCtx->pValues);
  } else {
+    uDebug("SML:0x%" PRIx64 " smlParseInfluxString format false, ts:%" PRId64, info->id, ts);
    taosArraySet(elements->colArray, 0, &kv);
  }
  info->preLine = *elements;

--- a/source/client/src/clientSmlTelnet.c
+++ b/source/client/src/clientSmlTelnet.c
@@ -292,7 +292,7 @@ int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
    return TSDB_CODE_SUCCESS;
  }

-  if (info->dataFormat) {
+  if (info->dataFormat && info->currSTableMeta != NULL) {
    if (needConverTime) {
      kvTs.i = convertTimePrecision(kvTs.i, TSDB_TIME_PRECISION_NANO, info->currSTableMeta->tableInfo.precision);
    }

--- a/source/client/src/clientStmt.c
+++ b/source/client/src/clientStmt.c
@@ -32,8 +32,14 @@ int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
    STMT_LOG_SEQ(newStatus);
  }

+  if (pStmt->errCode && newStatus != STMT_PREPARE) {
+    STMT_DLOG("stmt already failed with err: %s", tstrerror(pStmt->errCode));
+    return pStmt->errCode;
+  }
+
  switch (newStatus) {
    case STMT_PREPARE:
+      pStmt->errCode = 0;    
      break;
    case STMT_SETTBNAME:
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL)) {
@@ -197,7 +203,10 @@ int32_t stmtGetExecInfo(TAOS_STMT* stmt, SHashObj** pVgHash, SHashObj** pBlockHa
  STscStmt* pStmt = (STscStmt*)stmt;

  *pVgHash = pStmt->sql.pVgHash;
+  pStmt->sql.pVgHash = NULL;
+  
  *pBlockHash = pStmt->exec.pBlockHash;
+  pStmt->exec.pBlockHash = NULL;

  return TSDB_CODE_SUCCESS;
 }
@@ -325,6 +334,8 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
 }

 int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
+  STMT_DLOG_E("start to free SQL info");
+  
  taosMemoryFree(pStmt->sql.queryRes.fields);
  taosMemoryFree(pStmt->sql.queryRes.userFields);
  taosMemoryFree(pStmt->sql.sqlStr);
@@ -351,6 +362,8 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {

  memset(&pStmt->sql, 0, sizeof(pStmt->sql));

+  STMT_DLOG_E("end to free SQL info");
+
  return TSDB_CODE_SUCCESS;
 }

@@ -441,11 +454,10 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
-    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
-
    tscDebug("tb %s not exist", pStmt->bInfo.tbFName);
+    stmtCleanBindInfo(pStmt);

-    return TSDB_CODE_SUCCESS;
+    STMT_ERR_RET(code);
  }

  STMT_ERR_RET(code);
@@ -922,9 +934,13 @@ _return:
 int stmtClose(TAOS_STMT* stmt) {
  STscStmt* pStmt = (STscStmt*)stmt;

+  STMT_DLOG_E("start to free stmt");
+
  stmtCleanSQLInfo(pStmt);
  taosMemoryFree(stmt);

+  STMT_DLOG_E("stmt freed");
+
  return TSDB_CODE_SUCCESS;
 }


--- a/source/client/src/clientTmq.c
+++ b/source/client/src/clientTmq.c
--- a/source/client/src/clientTmqConnector.c
+++ b/source/client/src/clientTmqConnector.c
@@ -361,8 +361,8 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_fetchRawBlockImp(

  TAOS_RES *tres = (TAOS_RES *)res;

-  void   *data;
-  int32_t numOfRows;
+  void   *data = NULL;
+  int32_t numOfRows = 0;
  int     error_code = taos_fetch_raw_block(tres, &numOfRows, &data);
  if (numOfRows == 0) {
    if (error_code == JNI_SUCCESS) {

--- a/source/client/test/clientTests.cpp
+++ b/source/client/test/clientTests.cpp
@@ -112,7 +112,7 @@ void createNewTable(TAOS* pConn, int32_t index) {
  }
  taos_free_result(pRes);

-  for(int32_t i = 0; i < 2000; i += 20) {
+  for(int32_t i = 0; i < 100; i += 20) {
    char sql[1024] = {0};
    sprintf(sql,
            "insert into tu%d values(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)"
@@ -167,6 +167,80 @@ void tmq_commit_cb_print(tmq_t *pTmq, int32_t code, void *param) {
  printf("success, code:%d\n", code);
 }

+void* doConsumeData(void* param) {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+
+  tmq_conf_t* conf = tmq_conf_new();
+  tmq_conf_set(conf, "enable.auto.commit", "true");
+  tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
+  tmq_conf_set(conf, "group.id", "cgrpName12");
+  tmq_conf_set(conf, "td.connect.user", "root");
+  tmq_conf_set(conf, "td.connect.pass", "taosdata");
+  tmq_conf_set(conf, "auto.offset.reset", "earliest");
+  tmq_conf_set(conf, "experimental.snapshot.enable", "true");
+  tmq_conf_set(conf, "msg.with.table.name", "true");
+  tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
+
+  tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
+  tmq_conf_destroy(conf);
+
+  // 创建订阅 topics 列表
+  tmq_list_t* topicList = tmq_list_new();
+  tmq_list_append(topicList, "topic_t2");
+
+  // 启动订阅
+  tmq_subscribe(tmq, topicList);
+
+  tmq_list_destroy(topicList);
+
+  TAOS_FIELD* fields = NULL;
+  int32_t     numOfFields = 0;
+  int32_t     precision = 0;
+  int32_t     totalRows = 0;
+  int32_t     msgCnt = 0;
+  int32_t     timeout = 25000;
+
+  int32_t count = 0;
+
+  while (1) {
+    TAOS_RES* pRes = tmq_consumer_poll(tmq, timeout);
+    if (pRes) {
+      char    buf[1024];
+
+      const char* topicName = tmq_get_topic_name(pRes);
+      const char* dbName = tmq_get_db_name(pRes);
+      int32_t     vgroupId = tmq_get_vgroup_id(pRes);
+
+      printf("topic: %s\n", topicName);
+      printf("db: %s\n", dbName);
+      printf("vgroup id: %d\n", vgroupId);
+
+      while (1) {
+        TAOS_ROW row = taos_fetch_row(pRes);
+        if (row == NULL) {
+          break;
+        }
+
+        fields = taos_fetch_fields(pRes);
+        numOfFields = taos_field_count(pRes);
+        precision = taos_result_precision(pRes);
+        taos_print_row(buf, row, fields, numOfFields);
+        totalRows += 1;
+//        printf("precision: %d, row content: %s\n", precision, buf);
+      }
+
+      taos_free_result(pRes);
+    } else {
+      break;
+    }
+  }
+
+  tmq_consumer_close(tmq);
+  taos_close(pConn);
+  fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+  return NULL;
+}
+
 }  // namespace

 int main(int argc, char** argv) {
@@ -188,7 +262,6 @@ TEST(clientCase, driverInit_Test) {

 TEST(clientCase, connect_Test) {
  taos_options(TSDB_OPTION_CONFIGDIR, "~/first/cfg");
-
  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
  if (pConn == NULL) {
    printf("failed to connect to server, reason:%s\n", taos_errstr(NULL));
@@ -708,7 +781,7 @@ TEST(clientCase, projection_query_tables) {
  //  }
  //  taos_free_result(pRes);

-  TAOS_RES* pRes = taos_query(pConn, "use abc2");
+  TAOS_RES* pRes = taos_query(pConn, "use abc1");
  taos_free_result(pRes);

  pRes = taos_query(pConn, "create stable st1 (ts timestamp, k int) tags(a int)");
@@ -730,7 +803,7 @@ TEST(clientCase, projection_query_tables) {
  }
  taos_free_result(pRes);

-  for (int32_t i = 0; i < 2; ++i) {
+  for (int32_t i = 0; i < 10000; ++i) {
    printf("create table :%d\n", i);
    createNewTable(pConn, i);
  }
@@ -898,7 +971,7 @@ TEST(clientCase, update_test) {
  }
 }

-TEST(clientCase, subscription_test) {
+TEST(clientCase, sub_db_test) {
  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
  ASSERT_NE(pConn, nullptr);

@@ -912,7 +985,7 @@ TEST(clientCase, subscription_test) {
  tmq_conf_t* conf = tmq_conf_new();
  tmq_conf_set(conf, "enable.auto.commit", "true");
  tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
-  tmq_conf_set(conf, "group.id", "cgrpName");
+  tmq_conf_set(conf, "group.id", "cgrpNamedb");
  tmq_conf_set(conf, "td.connect.user", "root");
  tmq_conf_set(conf, "td.connect.pass", "taosdata");
  tmq_conf_set(conf, "auto.offset.reset", "earliest");
@@ -925,7 +998,8 @@ TEST(clientCase, subscription_test) {

  // 创建订阅 topics 列表
  tmq_list_t* topicList = tmq_list_new();
-//  tmq_list_append(topicList, "topic_t1");
+  tmq_list_append(topicList, "topic_t1");
+  tmq_list_append(topicList, "topic_s2");

  // 启动订阅
  tmq_subscribe(tmq, topicList);
@@ -954,7 +1028,7 @@ TEST(clientCase, subscription_test) {
      printf("db: %s\n", dbName);
      printf("vgroup id: %d\n", vgroupId);

-      if (count ++ > 20) {
+      if (count ++ > 200) {
        tmq_unsubscribe(tmq);
        break;
      }
@@ -970,11 +1044,103 @@ TEST(clientCase, subscription_test) {
        taos_print_row(buf, row, fields, numOfFields);
        printf("precision: %d, row content: %s\n", precision, buf);
      }
+      taos_free_result(pRes);
    }
-//      return rows;
  }

  fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
 }

+TEST(clientCase, sub_tb_test) {
+  taos_options(TSDB_OPTION_CONFIGDIR, "~/first/cfg");
+
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(pConn, nullptr);
+
+  tmq_conf_t* conf = tmq_conf_new();
+  tmq_conf_set(conf, "enable.auto.commit", "true");
+  tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
+  tmq_conf_set(conf, "group.id", "cgrpName27");
+  tmq_conf_set(conf, "td.connect.user", "root");
+  tmq_conf_set(conf, "td.connect.pass", "taosdata");
+  tmq_conf_set(conf, "auto.offset.reset", "earliest");
+  tmq_conf_set(conf, "experimental.snapshot.enable", "true");
+  tmq_conf_set(conf, "msg.with.table.name", "true");
+  tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
+
+  tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
+  tmq_conf_destroy(conf);
+
+  // 创建订阅 topics 列表
+  tmq_list_t* topicList = tmq_list_new();
+  tmq_list_append(topicList, "topic_t2");
+
+  // 启动订阅
+  tmq_subscribe(tmq, topicList);
+
+  tmq_list_destroy(topicList);
+
+  TAOS_FIELD* fields = NULL;
+  int32_t     numOfFields = 0;
+  int32_t     precision = 0;
+  int32_t     totalRows = 0;
+  int32_t     msgCnt = 0;
+  int32_t     timeout = 25000;
+
+  int32_t count = 0;
+
+  while (1) {
+    TAOS_RES* pRes = tmq_consumer_poll(tmq, timeout);
+    if (pRes) {
+      char    buf[1024];
+
+      const char* topicName = tmq_get_topic_name(pRes);
+      const char* dbName = tmq_get_db_name(pRes);
+      int32_t     vgroupId = tmq_get_vgroup_id(pRes);
+
+      printf("topic: %s\n", topicName);
+      printf("db: %s\n", dbName);
+      printf("vgroup id: %d\n", vgroupId);
+
+      while (1) {
+        TAOS_ROW row = taos_fetch_row(pRes);
+        if (row == NULL) {
+          break;
+        }
+
+        fields = taos_fetch_fields(pRes);
+        numOfFields = taos_field_count(pRes);
+        precision = taos_result_precision(pRes);
+        taos_print_row(buf, row, fields, numOfFields);
+        totalRows += 1;
+        printf("precision: %d, row content: %s\n", precision, buf);
+      }
+
+      taos_free_result(pRes);
+//      if ((++count) > 1) {
+//        break;
+//      }
+    } else {
+      break;
+    }
+  }
+
+  tmq_consumer_close(tmq);
+  taos_close(pConn);
+  fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+}
+
+TEST(clientCase, sub_tb_mt_test) {
+  taos_options(TSDB_OPTION_CONFIGDIR, "~/first/cfg");
+  TdThread qid[20] = {0};
+
+  for(int32_t i = 0; i < 1; ++i) {
+    taosThreadCreate(&qid[i], NULL, doConsumeData, NULL);
+  }
+
+  for(int32_t i = 0; i < 4; ++i) {
+    taosThreadJoin(qid[i], NULL);
+  }
+}
+
 #pragma GCC diagnostic pop
--- a/source/common/src/systable.c
+++ b/source/common/src/systable.c
@@ -233,6 +233,7 @@ static const SSysDbTableSchema vgroupsSchema[] = {
    {.name = "v4_dnode", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT, .sysInfo = true},
    {.name = "v4_status", .bytes = 9 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
    {.name = "cacheload", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
+    {.name = "cacheelements", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
    {.name = "tsma", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT, .sysInfo = true},
    // {.name = "compact_start_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = false},
 };

--- a/source/common/src/tdatablock.c
+++ b/source/common/src/tdatablock.c
@@ -147,9 +147,17 @@ int32_t colDataReserve(SColumnInfoData* pColumnInfoData, size_t newSize) {
  return TSDB_CODE_SUCCESS;
 }

-static void doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t currentRow, const char* pData,
-                         int32_t itemLen, int32_t numOfRows) {
-  ASSERT(pColumnInfoData->info.bytes >= itemLen);
+static int32_t doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t currentRow, const char* pData,
+                         int32_t itemLen, int32_t numOfRows, bool trimValue) {
+  if (pColumnInfoData->info.bytes < itemLen) {
+    uWarn("column/tag actual data len %d is bigger than schema len %d, trim it:%d", itemLen, pColumnInfoData->info.bytes, trimValue);
+    if (trimValue) {
+      itemLen = pColumnInfoData->info.bytes;
+    } else {
+      return TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
+    }
+  }
+  
  size_t start = 1;

  // the first item
@@ -178,10 +186,12 @@ static void doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t curren

    pColumnInfoData->varmeta.length += numOfRows * itemLen;
  }
+
+  return TSDB_CODE_SUCCESS;
 }

 int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData,
-                            uint32_t numOfRows) {
+                            uint32_t numOfRows, bool trimValue) {
  int32_t len = pColumnInfoData->info.bytes;
  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
    len = varDataTLen(pData);
@@ -193,8 +203,7 @@ int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow,
    }
  }

-  doCopyNItems(pColumnInfoData, currentRow, pData, len, numOfRows);
-  return TSDB_CODE_SUCCESS;
+  return doCopyNItems(pColumnInfoData, currentRow, pData, len, numOfRows, trimValue);
 }

 static void doBitmapMerge(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, const SColumnInfoData* pSource,
@@ -973,7 +982,7 @@ int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo) {
        taosSort(pColInfoData->pData, pDataBlock->info.rows, pColInfoData->info.bytes, fn);

        int64_t p1 = taosGetTimestampUs();
-        uDebug("blockDataSort easy cost:%" PRId64 ", rows:%d\n", p1 - p0, pDataBlock->info.rows);
+        uDebug("blockDataSort easy cost:%" PRId64 ", rows:%" PRId64 "\n", p1 - p0, pDataBlock->info.rows);

        return TSDB_CODE_SUCCESS;
      } else {  // var data type
@@ -1026,6 +1035,7 @@ int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo) {
  return TSDB_CODE_SUCCESS;
 }

+#if 0
 typedef struct SHelper {
  int32_t index;
  union {
@@ -1074,59 +1084,20 @@ SHelper* createTupleIndex_rv(int32_t numOfRows, SArray* pOrderInfo, SSDataBlock*
 int32_t dataBlockCompar_rv(const void* p1, const void* p2, const void* param) {
  const SSDataBlockSortHelper* pHelper = (const SSDataBlockSortHelper*)param;

-  //  SSDataBlock* pDataBlock = pHelper->pDataBlock;
-
  SHelper* left = (SHelper*)p1;
  SHelper* right = (SHelper*)p2;

  SArray* pInfo = pHelper->orderInfo;

  int32_t offset = 0;
-  //  for(int32_t i = 0; i < pInfo->size; ++i) {
-  //    SBlockOrderInfo* pOrder = TARRAY_GET_ELEM(pInfo, 0);
-  //    SColumnInfoData* pColInfoData = pOrder->pColData;//TARRAY_GET_ELEM(pDataBlock->pDataBlock, pOrder->colIndex);
-
-  //    if (pColInfoData->hasNull) {
-  //      bool leftNull  = colDataIsNull(pColInfoData, pDataBlock->info.rows, left, pDataBlock->pBlockAgg);
-  //      bool rightNull = colDataIsNull(pColInfoData, pDataBlock->info.rows, right, pDataBlock->pBlockAgg);
-  //      if (leftNull && rightNull) {
-  //        continue; // continue to next slot
-  //      }
-  //
-  //      if (rightNull) {
-  //        return pHelper->nullFirst? 1:-1;
-  //      }
-  //
-  //      if (leftNull) {
-  //        return pHelper->nullFirst? -1:1;
-  //      }
-  //    }
-
-  //    void* left1  = colDataGetData(pColInfoData, left);
-  //    void* right1 = colDataGetData(pColInfoData, right);
-
-  //    switch(pColInfoData->info.type) {
-  //      case TSDB_DATA_TYPE_INT: {
  int32_t leftx = *(int32_t*)left->pData;    //*(int32_t*)(left->pData + offset);
  int32_t rightx = *(int32_t*)right->pData;  //*(int32_t*)(right->pData + offset);

-  //        offset += pColInfoData->info.bytes;
  if (leftx == rightx) {
-    //          break;
    return 0;
  } else {
-    //          if (pOrder->order == TSDB_ORDER_ASC) {
    return (leftx < rightx) ? -1 : 1;
-    //          } else {
-    //            return (leftx < rightx)? 1:-1;
-    //          }
-  }
-  //      }
-  //      default:
-  //        assert(0);
-  //    }
-  //  }
-
+  }
  return 0;
 }

@@ -1170,6 +1141,7 @@ int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullF
  //  destroyTupleIndex(index);
  return 0;
 }
+#endif

 void blockDataCleanup(SSDataBlock* pDataBlock) {
  blockDataEmpty(pDataBlock);
@@ -1180,7 +1152,7 @@ void blockDataCleanup(SSDataBlock* pDataBlock) {

 void blockDataEmpty(SSDataBlock* pDataBlock) {
  SDataBlockInfo* pInfo = &pDataBlock->info;
-  if (pInfo->capacity == 0 || pInfo->rows > pDataBlock->info.capacity) {
+  if (pInfo->capacity == 0) {
    return;
  }

@@ -1739,14 +1711,14 @@ int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {
  int64_t tbUid = pBlock->info.id.uid;
  int16_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
  int16_t hasVarCol = pBlock->info.hasVarCol;
-  int32_t rows = pBlock->info.rows;
+  int64_t rows = pBlock->info.rows;
  int32_t sz = taosArrayGetSize(pBlock->pDataBlock);

  int32_t tlen = 0;
  tlen += taosEncodeFixedI64(buf, tbUid);
  tlen += taosEncodeFixedI16(buf, numOfCols);
  tlen += taosEncodeFixedI16(buf, hasVarCol);
-  tlen += taosEncodeFixedI32(buf, rows);
+  tlen += taosEncodeFixedI64(buf, rows);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SColumnInfoData* pColData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
@@ -1777,7 +1749,7 @@ void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock) {
  buf = taosDecodeFixedU64(buf, &pBlock->info.id.uid);
  buf = taosDecodeFixedI16(buf, &numOfCols);
  buf = taosDecodeFixedI16(buf, &pBlock->info.hasVarCol);
-  buf = taosDecodeFixedI32(buf, &pBlock->info.rows);
+  buf = taosDecodeFixedI64(buf, &pBlock->info.rows);
  buf = taosDecodeFixedI32(buf, &sz);
  pBlock->pDataBlock = taosArrayInit(sz, sizeof(SColumnInfoData));
  for (int32_t i = 0; i < sz; i++) {
@@ -1878,6 +1850,7 @@ static char* formatTimestamp(char* buf, int64_t val, int precision) {
  return buf;
 }

+#if 0
 void blockDebugShowDataBlock(SSDataBlock* pBlock, const char* flag) {
  SArray* dataBlocks = taosArrayInit(1, sizeof(SSDataBlock*));
  taosArrayPush(dataBlocks, &pBlock);
@@ -1970,6 +1943,8 @@ void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag) {
  }
 }

+#endif
+
 // for debug
 char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf) {
  int32_t size = 2048;
@@ -1981,7 +1956,7 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
  int32_t len = 0;
  len += snprintf(dumpBuf + len, size - len,
                  "===stream===%s|block type %d|child id %d|group id:%" PRIu64 "|uid:%" PRId64
-                  "|rows:%d|version:%" PRIu64 "|cal start:%" PRIu64 "|cal end:%" PRIu64 "|tbl:%s\n",
+                  "|rows:%" PRId64 "|version:%" PRIu64 "|cal start:%" PRIu64 "|cal end:%" PRIu64 "|tbl:%s\n",
                  flag, (int32_t)pDataBlock->info.type, pDataBlock->info.childId, pDataBlock->info.id.groupId,
                  pDataBlock->info.id.uid, pDataBlock->info.rows, pDataBlock->info.version,
                  pDataBlock->info.calWin.skey, pDataBlock->info.calWin.ekey, pDataBlock->info.parTbName);

--- a/source/common/src/tdataformat.c
+++ b/source/common/src/tdataformat.c
@@ -2439,6 +2439,12 @@ _exit:
 int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes, int32_t nRows, char *lengthOrbitmap,
                                    char *data) {
  int32_t code = 0;
+  if(data == NULL){
+    for (int32_t i = 0; i < nRows; ++i) {
+      code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NONE](pColData, NULL, 0);
+    }
+    goto _exit;
+  }

  if (IS_VAR_DATA_TYPE(type)) {  // var-length data type
    for (int32_t i = 0; i < nRows; ++i) {
@@ -2447,6 +2453,11 @@ int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t byt
        code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NULL](pColData, NULL, 0);
        if (code) goto _exit;
      } else {
+        if(ASSERT(varDataTLen(data + offset) <= bytes)){
+          uError("var data length invalid, varDataTLen(data + offset):%d <= bytes:%d", (int)varDataTLen(data + offset), bytes);
+          code = TSDB_CODE_INVALID_PARA;
+          goto _exit;
+        }
        code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_VALUE](pColData, (uint8_t *)varDataVal(data + offset),
                                                                      varDataLen(data + offset));
      }

--- a/source/common/src/tglobal.c
+++ b/source/common/src/tglobal.c
@@ -154,6 +154,7 @@ char    tsTagFilterCache = 0;
 // positive value (in MB)
 int32_t tsQueryBufferSize = -1;
 int64_t tsQueryBufferSizeBytes = -1;
+int32_t tsCacheLazyLoadThreshold = 500;

 int32_t  tsDiskCfgNum = 0;
 SDiskCfg tsDiskCfg[TFS_MAX_DISKS] = {0};
@@ -497,6 +498,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {

  if (cfgAddBool(pCfg, "disableStream", tsDisableStream, 0) != 0) return -1;

+  if (cfgAddInt32(pCfg, "cacheLazyLoadThreshold", tsCacheLazyLoadThreshold, 0, 100000, 0) != 0) return -1;
+
  GRANT_CFG_ADD;
  return 0;
 }
@@ -824,6 +827,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
    tsQueryBufferSizeBytes = tsQueryBufferSize * 1048576UL;
  }

+  tsCacheLazyLoadThreshold = cfgGetItem(pCfg, "cacheLazyLoadThreshold")->i32;
+
  tsDisableStream = cfgGetItem(pCfg, "disableStream")->bval;

  GRANT_CFG_GET;
@@ -1233,13 +1238,13 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
  }

  if (taosLoadCfg(pCfg, envCmd, cfgDir, envFile, apolloUrl) != 0) {
-    uError("failed to load cfg since %s", terrstr());
+    printf("failed to load cfg since %s", terrstr());
    cfgCleanup(pCfg);
    return -1;
  }

  if (cfgLoadFromArray(pCfg, pArgs) != 0) {
-    uError("failed to load cfg from array since %s", terrstr());
+    printf("failed to load cfg from array since %s", terrstr());
    cfgCleanup(pCfg);
    return -1;
  }
@@ -1255,13 +1260,13 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi

  if (taosMulModeMkDir(tsLogDir, 0777) != 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
-    uError("failed to create dir:%s since %s", tsLogDir, terrstr());
+    printf("failed to create dir:%s since %s", tsLogDir, terrstr());
    cfgCleanup(pCfg);
    return -1;
  }

  if (taosInitLog(logname, logFileNum) != 0) {
-    uError("failed to init log file since %s", terrstr());
+    printf("failed to init log file since %s", terrstr());
    cfgCleanup(pCfg);
    return -1;
  }

--- a/source/common/src/tmsg.c
+++ b/source/common/src/tmsg.c
@@ -1070,7 +1070,8 @@ int32_t tSerializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
    if (tEncodeI64(&encoder, pload->totalStorage) < 0) return -1;
    if (tEncodeI64(&encoder, pload->compStorage) < 0) return -1;
    if (tEncodeI64(&encoder, pload->pointsWritten) < 0) return -1;
-    if (tEncodeI64(&encoder, reserved) < 0) return -1;
+    if (tEncodeI32(&encoder, pload->numOfCachedTables) < 0) return -1;
+    if (tEncodeI32(&encoder, reserved) < 0) return -1;
    if (tEncodeI64(&encoder, reserved) < 0) return -1;
    if (tEncodeI64(&encoder, reserved) < 0) return -1;
  }
@@ -1148,7 +1149,8 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
    if (tDecodeI64(&decoder, &vload.totalStorage) < 0) return -1;
    if (tDecodeI64(&decoder, &vload.compStorage) < 0) return -1;
    if (tDecodeI64(&decoder, &vload.pointsWritten) < 0) return -1;
-    if (tDecodeI64(&decoder, &reserved) < 0) return -1;
+    if (tDecodeI32(&decoder, &vload.numOfCachedTables) < 0) return -1;
+    if (tDecodeI32(&decoder, (int32_t*)&reserved) < 0) return -1;
    if (tDecodeI64(&decoder, &reserved) < 0) return -1;
    if (tDecodeI64(&decoder, &reserved) < 0) return -1;
    if (taosArrayPush(pReq->pVloads, &vload) == NULL) {
@@ -6637,8 +6639,9 @@ int32_t tFormatOffset(char *buf, int32_t maxLen, const STqOffsetVal *pVal) {
  } else if (pVal->type == TMQ_OFFSET__SNAPSHOT_DATA || pVal->type == TMQ_OFFSET__SNAPSHOT_META) {
    snprintf(buf, maxLen, "offset(snapshot) uid:%" PRId64 " ts:%" PRId64, pVal->uid, pVal->ts);
  } else {
-    ASSERT(0);
+    return TSDB_CODE_INVALID_PARA;
  }
+
  return 0;
 }

@@ -6821,8 +6824,7 @@ int32_t tDecodeSMqDataRsp(SDecoder *pDecoder, SMqDataRsp *pRsp) {
 }

 void tDeleteSMqDataRsp(SMqDataRsp *pRsp) {
-  taosArrayDestroy(pRsp->blockDataLen);
-  pRsp->blockDataLen = NULL;
+  pRsp->blockDataLen = taosArrayDestroy(pRsp->blockDataLen);;
  taosArrayDestroyP(pRsp->blockData, (FDelete)taosMemoryFree);
  pRsp->blockData = NULL;
  taosArrayDestroyP(pRsp->blockSchema, (FDelete)tDeleteSSchemaWrapper);

--- a/source/dnode/mgmt/exe/dmMain.c
+++ b/source/dnode/mgmt/exe/dmMain.c
@@ -18,6 +18,7 @@
 #include "mnode.h"
 #include "tconfig.h"
 #include "tglobal.h"
+#include "version.h"

 // clang-format off
 #define DM_APOLLO_URL    "The apollo string to use when configuring the server, such as: -a 'jsonFile:./tests/cfg.json', cfg.json text can be '{\"fqdn\":\"td1\"}'."
@@ -76,28 +77,28 @@ void dmLogCrash(int signum, void *sigInfo, void *context) {
  taosIgnSignal(SIGINT);
  taosIgnSignal(SIGBREAK);

-#ifndef WINDOWS  
+#ifndef WINDOWS
  taosIgnSignal(SIGBUS);
 #endif
  taosIgnSignal(SIGABRT);
  taosIgnSignal(SIGFPE);
  taosIgnSignal(SIGSEGV);

-  char *pMsg = NULL;
+  char       *pMsg = NULL;
  const char *flags = "UTL FATAL ";
  ELogLevel   level = DEBUG_FATAL;
  int32_t     dflag = 255;
-  int64_t     msgLen= -1;
-  
+  int64_t     msgLen = -1;
+
  if (tsEnableCrashReport) {
    if (taosGenCrashJsonMsg(signum, &pMsg, dmGetClusterId(), global.startTime)) {
      taosPrintLog(flags, level, dflag, "failed to generate crash json msg");
      goto _return;
    } else {
-      msgLen = strlen(pMsg);  
+      msgLen = strlen(pMsg);
    }
  }
-  
+
 _return:

  taosLogCrashInfo("taosd", pMsg, msgLen, signum, sigInfo);
@@ -123,7 +124,7 @@ static void dmSetSignalHandle() {

 #ifndef WINDOWS
  taosSetSignal(SIGBUS, dmLogCrash);
-#endif  
+#endif
  taosSetSignal(SIGABRT, dmLogCrash);
  taosSetSignal(SIGFPE, dmLogCrash);
  taosSetSignal(SIGSEGV, dmLogCrash);
@@ -134,7 +135,7 @@ static int32_t dmParseArgs(int32_t argc, char const *argv[]) {

  int32_t cmdEnvIndex = 0;
  if (argc < 2) return 0;
-  
+
  global.envCmd = taosMemoryMalloc((argc - 1) * sizeof(char *));
  memset(global.envCmd, 0, (argc - 1) * sizeof(char *));
  for (int32_t i = 1; i < argc; ++i) {
@@ -203,6 +204,9 @@ static void dmPrintVersion() {
 #endif
  printf("%s version: %s compatible_version: %s\n", releaseName, version, compatible_version);
  printf("gitinfo: %s\n", gitinfo);
+#ifdef TD_ENTERPRISE
+  printf("gitinfoOfInternal: %s\n", gitinfoOfInternal);
+#endif
  printf("buildInfo: %s\n", buildinfo);
 }

@@ -284,7 +288,7 @@ int mainWindows(int argc, char **argv) {
      printf("failed to init memory dbg, error:%s\n", tstrerror(code));
      return code;
    }
-    tsAsyncLog = false;    
+    tsAsyncLog = false;
    printf("memory dbg enabled\n");
  }
 #endif

--- a/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h
+++ b/source/dnode/mgmt/mgmt_mnode/inc/mmInt.h
@@ -34,7 +34,7 @@ typedef struct SMnodeMgmt {
  SSingleWorker  readWorker;
  SSingleWorker  writeWorker;
  SSingleWorker  syncWorker;
-  SSingleWorker  syncCtrlWorker;
+  SSingleWorker  syncRdWorker;
  bool           stopped;
  int32_t        refCount;
  TdThreadRwlock lock;
@@ -54,7 +54,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt);
 void    mmStopWorker(SMnodeMgmt *pMgmt);
 int32_t mmPutMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t mmPutMsgToSyncCtrlQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t mmPutMsgToSyncRdQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutMsgToFetchQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);

--- a/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
+++ b/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
@@ -188,7 +188,8 @@ SArray *mmGetMsgHandles() {
  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_INDEX_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_DISABLE_WRITE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT_ELECTION, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT, mmPutMsgToSyncRdQueue, 1) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_BATCH, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_REPLY, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
@@ -198,11 +199,12 @@ SArray *mmGetMsgHandles() {
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_BATCH, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_REPLY, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_SEND, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT_REPLY, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT, mmPutMsgToSyncCtrlQueue, 1) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT_REPLY, mmPutMsgToSyncCtrlQueue, 1) == NULL) goto _OVER;
+
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, mmPutMsgToSyncRdQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT_REPLY, mmPutMsgToSyncRdQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, mmPutMsgToSyncRdQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT_REPLY, mmPutMsgToSyncRdQueue, 1) == NULL) goto _OVER;

  code = 0;


--- a/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
+++ b/source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
@@ -111,8 +111,8 @@ int32_t mmPutMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
  return mmPutMsgToWorker(pMgmt, &pMgmt->syncWorker, pMsg);
 }

-int32_t mmPutMsgToSyncCtrlQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return mmPutMsgToWorker(pMgmt, &pMgmt->syncCtrlWorker, pMsg);
+int32_t mmPutMsgToSyncRdQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
+  return mmPutMsgToWorker(pMgmt, &pMgmt->syncRdWorker, pMsg);
 }

 int32_t mmPutMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
@@ -151,8 +151,8 @@ int32_t mmPutMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
    case SYNC_QUEUE:
      pWorker = &pMgmt->syncWorker;
      break;
-    case SYNC_CTRL_QUEUE:
-      pWorker = &pMgmt->syncCtrlWorker;
+    case SYNC_RD_QUEUE:
+      pWorker = &pMgmt->syncRdWorker;
      break;
    default:
      terrno = TSDB_CODE_INVALID_PARA;
@@ -238,12 +238,12 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
  SSingleWorkerCfg scCfg = {
      .min = 1,
      .max = 1,
-      .name = "mnode-sync-ctrl",
+      .name = "mnode-sync-rd",
      .fp = (FItem)mmProcessSyncMsg,
      .param = pMgmt,
  };
-  if (tSingleWorkerInit(&pMgmt->syncCtrlWorker, &scCfg) != 0) {
-    dError("failed to start mnode mnode-sync-ctrl worker since %s", terrstr());
+  if (tSingleWorkerInit(&pMgmt->syncRdWorker, &scCfg) != 0) {
+    dError("failed to start mnode mnode-sync-rd worker since %s", terrstr());
    return -1;
  }

@@ -259,6 +259,6 @@ void mmStopWorker(SMnodeMgmt *pMgmt) {
  tSingleWorkerCleanup(&pMgmt->readWorker);
  tSingleWorkerCleanup(&pMgmt->writeWorker);
  tSingleWorkerCleanup(&pMgmt->syncWorker);
-  tSingleWorkerCleanup(&pMgmt->syncCtrlWorker);
+  tSingleWorkerCleanup(&pMgmt->syncRdWorker);
  dDebug("mnode workers are closed");
 }
--- a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
+++ b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
@@ -59,7 +59,7 @@ typedef struct {
  SVnode       *pImpl;
  SMultiWorker  pWriteW;
  SMultiWorker  pSyncW;
-  SMultiWorker  pSyncCtrlW;
+  SMultiWorker  pSyncRdW;
  SMultiWorker  pApplyW;
  STaosQueue   *pQueryQ;
  STaosQueue   *pStreamQ;
@@ -107,7 +107,7 @@ int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc);

 int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutMsgToSyncCtrlQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmPutMsgToSyncRdQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmPutMsgToStreamQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);

--- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
@@ -549,7 +549,8 @@ SArray *vmGetMsgHandles() {
  if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;

-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT_ELECTION, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_BATCH, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
@@ -559,12 +560,12 @@ SArray *vmGetMsgHandles() {
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_BATCH, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_SEND, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;

-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT, vmPutMsgToSyncCtrlQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT_REPLY, vmPutMsgToSyncCtrlQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT_REPLY, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT_REPLY, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;

  code = 0;


--- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
@@ -98,9 +98,9 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal)
        pVnode->pSyncW.queue->threadId);
  tMultiWorkerCleanup(&pVnode->pSyncW);

-  dInfo("vgId:%d, wait for vnode sync ctrl queue:%p is empty, thread:%08" PRId64, pVnode->vgId,
-        pVnode->pSyncCtrlW.queue, pVnode->pSyncCtrlW.queue->threadId);
-  tMultiWorkerCleanup(&pVnode->pSyncCtrlW);
+  dInfo("vgId:%d, wait for vnode sync rd queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncRdW.queue,
+        pVnode->pSyncRdW.queue->threadId);
+  tMultiWorkerCleanup(&pVnode->pSyncRdW);

  dInfo("vgId:%d, wait for vnode apply queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue,
        pVnode->pApplyW.queue->threadId);

--- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
@@ -216,9 +216,9 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
      dGTrace("vgId:%d, msg:%p put into vnode-sync queue", pVnode->vgId, pMsg);
      taosWriteQitem(pVnode->pSyncW.queue, pMsg);
      break;
-    case SYNC_CTRL_QUEUE:
-      dGTrace("vgId:%d, msg:%p put into vnode-sync-ctrl queue", pVnode->vgId, pMsg);
-      taosWriteQitem(pVnode->pSyncCtrlW.queue, pMsg);
+    case SYNC_RD_QUEUE:
+      dGTrace("vgId:%d, msg:%p put into vnode-sync-rd queue", pVnode->vgId, pMsg);
+      taosWriteQitem(pVnode->pSyncRdW.queue, pMsg);
      break;
    case APPLY_QUEUE:
      dGTrace("vgId:%d, msg:%p put into vnode-apply queue", pVnode->vgId, pMsg);
@@ -234,9 +234,7 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
  return code;
 }

-int32_t vmPutMsgToSyncCtrlQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return vmPutMsgToQueue(pMgmt, pMsg, SYNC_CTRL_QUEUE);
-}
+int32_t vmPutMsgToSyncRdQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, SYNC_RD_QUEUE); }

 int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, SYNC_QUEUE); }

@@ -327,18 +325,18 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {
 int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
  SMultiWorkerCfg wcfg = {.max = 1, .name = "vnode-write", .fp = (FItems)vnodeProposeWriteMsg, .param = pVnode->pImpl};
  SMultiWorkerCfg scfg = {.max = 1, .name = "vnode-sync", .fp = (FItems)vmProcessSyncQueue, .param = pVnode};
-  SMultiWorkerCfg sccfg = {.max = 1, .name = "vnode-sync-ctrl", .fp = (FItems)vmProcessSyncQueue, .param = pVnode};
+  SMultiWorkerCfg sccfg = {.max = 1, .name = "vnode-sync-rd", .fp = (FItems)vmProcessSyncQueue, .param = pVnode};
  SMultiWorkerCfg acfg = {.max = 1, .name = "vnode-apply", .fp = (FItems)vnodeApplyWriteMsg, .param = pVnode->pImpl};
  (void)tMultiWorkerInit(&pVnode->pWriteW, &wcfg);
  (void)tMultiWorkerInit(&pVnode->pSyncW, &scfg);
-  (void)tMultiWorkerInit(&pVnode->pSyncCtrlW, &sccfg);
+  (void)tMultiWorkerInit(&pVnode->pSyncRdW, &sccfg);
  (void)tMultiWorkerInit(&pVnode->pApplyW, &acfg);

  pVnode->pQueryQ = tQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)vmProcessQueryQueue);
  pVnode->pStreamQ = tAutoQWorkerAllocQueue(&pMgmt->streamPool, pVnode, (FItem)vmProcessStreamQueue);
  pVnode->pFetchQ = tWWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItems)vmProcessFetchQueue);

-  if (pVnode->pWriteW.queue == NULL || pVnode->pSyncW.queue == NULL || pVnode->pSyncCtrlW.queue == NULL ||
+  if (pVnode->pWriteW.queue == NULL || pVnode->pSyncW.queue == NULL || pVnode->pSyncRdW.queue == NULL ||
      pVnode->pApplyW.queue == NULL || pVnode->pQueryQ == NULL || pVnode->pStreamQ == NULL || pVnode->pFetchQ == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
@@ -348,8 +346,8 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
        pVnode->pWriteW.queue->threadId);
  dInfo("vgId:%d, sync-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncW.queue,
        pVnode->pSyncW.queue->threadId);
-  dInfo("vgId:%d, sync-ctrl-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncCtrlW.queue,
-        pVnode->pSyncCtrlW.queue->threadId);
+  dInfo("vgId:%d, sync-rd-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncRdW.queue,
+        pVnode->pSyncRdW.queue->threadId);
  dInfo("vgId:%d, apply-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue,
        pVnode->pApplyW.queue->threadId);
  dInfo("vgId:%d, query-queue:%p is alloced", pVnode->vgId, pVnode->pQueryQ);

--- a/source/dnode/mnode/impl/inc/mndConsumer.h
+++ b/source/dnode/mnode/impl/inc/mndConsumer.h
@@ -24,10 +24,10 @@ extern "C" {

 enum {
  MQ_CONSUMER_STATUS__MODIFY = 1,
-  MQ_CONSUMER_STATUS__MODIFY_IN_REB,
+  MQ_CONSUMER_STATUS__MODIFY_IN_REB,     // this value is not used anymore
  MQ_CONSUMER_STATUS__READY,
  MQ_CONSUMER_STATUS__LOST,
-  MQ_CONSUMER_STATUS__LOST_IN_REB,
+  MQ_CONSUMER_STATUS__LOST_IN_REB,       // this value is not used anymore
  MQ_CONSUMER_STATUS__LOST_REBD,
  MQ_CONSUMER_STATUS__REMOVED,
 };

--- a/source/dnode/mnode/impl/inc/mndDef.h
+++ b/source/dnode/mnode/impl/inc/mndDef.h
--- a/source/dnode/mnode/impl/inc/mndInt.h
+++ b/source/dnode/mnode/impl/inc/mndInt.h
--- a/source/dnode/mnode/impl/src/mndConsumer.c
+++ b/source/dnode/mnode/impl/src/mndConsumer.c
--- a/source/dnode/mnode/impl/src/mndDb.c
+++ b/source/dnode/mnode/impl/src/mndDb.c
--- a/source/dnode/mnode/impl/src/mndDnode.c
+++ b/source/dnode/mnode/impl/src/mndDnode.c
--- a/source/dnode/mnode/impl/src/mndIndex.c
+++ b/source/dnode/mnode/impl/src/mndIndex.c
--- a/source/dnode/mnode/impl/src/mndMain.c
+++ b/source/dnode/mnode/impl/src/mndMain.c
--- a/source/dnode/mnode/impl/src/mndProfile.c
+++ b/source/dnode/mnode/impl/src/mndProfile.c
--- a/source/dnode/mnode/impl/src/mndScheduler.c
+++ b/source/dnode/mnode/impl/src/mndScheduler.c
--- a/source/dnode/mnode/impl/src/mndShow.c
+++ b/source/dnode/mnode/impl/src/mndShow.c
--- a/source/dnode/mnode/impl/src/mndStb.c
+++ b/source/dnode/mnode/impl/src/mndStb.c
--- a/source/dnode/mnode/impl/src/mndSubscribe.c
+++ b/source/dnode/mnode/impl/src/mndSubscribe.c
--- a/source/dnode/mnode/impl/src/mndSync.c
+++ b/source/dnode/mnode/impl/src/mndSync.c
--- a/source/dnode/mnode/impl/src/mndTrans.c
+++ b/source/dnode/mnode/impl/src/mndTrans.c
--- a/source/dnode/mnode/impl/src/mndVgroup.c
+++ b/source/dnode/mnode/impl/src/mndVgroup.c
--- a/source/dnode/snode/src/snode.c
+++ b/source/dnode/snode/src/snode.c
--- a/source/dnode/vnode/CMakeLists.txt
+++ b/source/dnode/vnode/CMakeLists.txt
--- a/source/dnode/vnode/inc/vnode.h
+++ b/source/dnode/vnode/inc/vnode.h
--- a/source/dnode/vnode/src/inc/tq.h
+++ b/source/dnode/vnode/src/inc/tq.h
--- a/source/dnode/vnode/src/inc/tsdb.h
+++ b/source/dnode/vnode/src/inc/tsdb.h
--- a/source/dnode/vnode/src/inc/vnodeInt.h
+++ b/source/dnode/vnode/src/inc/vnodeInt.h
--- a/source/dnode/vnode/src/meta/metaCache.c
+++ b/source/dnode/vnode/src/meta/metaCache.c
--- a/source/dnode/vnode/src/meta/metaQuery.c
+++ b/source/dnode/vnode/src/meta/metaQuery.c
--- a/source/dnode/vnode/src/sma/smaRollup.c
+++ b/source/dnode/vnode/src/sma/smaRollup.c
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
--- a/source/dnode/vnode/src/tq/tqMeta.c
+++ b/source/dnode/vnode/src/tq/tqMeta.c
--- a/source/dnode/vnode/src/tq/tqPush.c
+++ b/source/dnode/vnode/src/tq/tqPush.c
--- a/source/dnode/vnode/src/tq/tqRead.c
+++ b/source/dnode/vnode/src/tq/tqRead.c
--- a/source/dnode/vnode/src/tq/tqExec.c
+++ b/source/dnode/vnode/src/tq/tqExec.c
--- a/source/dnode/vnode/src/tq/tqSink.c
+++ b/source/dnode/vnode/src/tq/tqSink.c
--- a/source/dnode/vnode/src/tsdb/tsdbCache.c
+++ b/source/dnode/vnode/src/tsdb/tsdbCache.c
--- a/source/dnode/vnode/src/tsdb/tsdbCacheRead.c
+++ b/source/dnode/vnode/src/tsdb/tsdbCacheRead.c
--- a/source/dnode/vnode/src/tsdb/tsdbMemTable.c
+++ b/source/dnode/vnode/src/tsdb/tsdbMemTable.c
--- a/source/dnode/vnode/src/tsdb/tsdbMergeTree.c
+++ b/source/dnode/vnode/src/tsdb/tsdbMergeTree.c
--- a/source/dnode/vnode/src/tsdb/tsdbRead.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRead.c
--- a/source/dnode/vnode/src/vnd/vnodeCommit.c
+++ b/source/dnode/vnode/src/vnd/vnodeCommit.c
--- a/source/dnode/vnode/src/vnd/vnodeQuery.c
+++ b/source/dnode/vnode/src/vnd/vnodeQuery.c
--- a/source/dnode/vnode/src/vnd/vnodeSnapshot.c
+++ b/source/dnode/vnode/src/vnd/vnodeSnapshot.c
--- a/source/dnode/vnode/src/vnd/vnodeSvr.c
+++ b/source/dnode/vnode/src/vnd/vnodeSvr.c
--- a/source/dnode/vnode/src/vnd/vnodeSync.c
+++ b/source/dnode/vnode/src/vnd/vnodeSync.c
--- a/source/libs/catalog/src/ctgDbg.c
+++ b/source/libs/catalog/src/ctgDbg.c
--- a/source/libs/catalog/src/ctgUtil.c
+++ b/source/libs/catalog/src/ctgUtil.c
--- a/source/libs/command/src/command.c
+++ b/source/libs/command/src/command.c
--- a/source/libs/executor/inc/executil.h
+++ b/source/libs/executor/inc/executil.h
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
--- a/source/libs/executor/inc/tfill.h
+++ b/source/libs/executor/inc/tfill.h
--- a/source/libs/executor/inc/tsort.h
+++ b/source/libs/executor/inc/tsort.h
--- a/source/libs/executor/src/aggregateoperator.c
+++ b/source/libs/executor/src/aggregateoperator.c
--- a/source/libs/executor/src/cachescanoperator.c
+++ b/source/libs/executor/src/cachescanoperator.c
--- a/source/libs/executor/src/exchangeoperator.c
+++ b/source/libs/executor/src/exchangeoperator.c
--- a/source/libs/executor/src/executil.c
+++ b/source/libs/executor/src/executil.c
--- a/source/libs/executor/src/executor.c
+++ b/source/libs/executor/src/executor.c
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
--- a/source/libs/executor/src/filloperator.c
+++ b/source/libs/executor/src/filloperator.c
--- a/source/libs/executor/src/groupoperator.c
+++ b/source/libs/executor/src/groupoperator.c
--- a/source/libs/executor/src/projectoperator.c
+++ b/source/libs/executor/src/projectoperator.c
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
--- a/source/libs/executor/src/sortoperator.c
+++ b/source/libs/executor/src/sortoperator.c
--- a/source/libs/executor/src/sysscanoperator.c
+++ b/source/libs/executor/src/sysscanoperator.c
--- a/source/libs/executor/src/timewindowoperator.c
+++ b/source/libs/executor/src/timewindowoperator.c
--- a/source/libs/executor/src/tsort.c
+++ b/source/libs/executor/src/tsort.c
--- a/source/libs/function/inc/builtinsimpl.h
+++ b/source/libs/function/inc/builtinsimpl.h
--- a/source/libs/function/src/builtins.c
+++ b/source/libs/function/src/builtins.c
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
--- a/source/libs/function/src/functionMgt.c
+++ b/source/libs/function/src/functionMgt.c
--- a/source/libs/function/src/udfd.c
+++ b/source/libs/function/src/udfd.c
--- a/source/libs/index/src/indexFilter.c
+++ b/source/libs/index/src/indexFilter.c
--- a/source/libs/index/src/indexFstFile.c
+++ b/source/libs/index/src/indexFstFile.c
--- a/source/libs/parser/src/parCalcConst.c
+++ b/source/libs/parser/src/parCalcConst.c
--- a/source/libs/parser/src/parInsertSml.c
+++ b/source/libs/parser/src/parInsertSml.c
--- a/source/libs/parser/src/parInsertUtil.c
+++ b/source/libs/parser/src/parInsertUtil.c
--- a/source/libs/parser/src/parTranslater.c
+++ b/source/libs/parser/src/parTranslater.c
--- a/source/libs/parser/src/parUtil.c
+++ b/source/libs/parser/src/parUtil.c
--- a/source/libs/parser/test/parSelectTest.cpp
+++ b/source/libs/parser/test/parSelectTest.cpp
--- a/source/libs/planner/src/planOptimizer.c
+++ b/source/libs/planner/src/planOptimizer.c
--- a/source/libs/planner/src/planPhysiCreater.c
+++ b/source/libs/planner/src/planPhysiCreater.c
--- a/source/libs/planner/src/planSpliter.c
+++ b/source/libs/planner/src/planSpliter.c
--- a/source/libs/qcom/src/queryUtil.c
+++ b/source/libs/qcom/src/queryUtil.c
--- a/source/libs/qworker/src/qwDbg.c
+++ b/source/libs/qworker/src/qwDbg.c
--- a/source/libs/qworker/src/qworker.c
+++ b/source/libs/qworker/src/qworker.c
--- a/source/libs/scalar/src/sclfunc.c
+++ b/source/libs/scalar/src/sclfunc.c
--- a/source/libs/scheduler/inc/schInt.h
+++ b/source/libs/scheduler/inc/schInt.h
--- a/source/libs/scheduler/src/schFlowCtrl.c
+++ b/source/libs/scheduler/src/schFlowCtrl.c
--- a/source/libs/scheduler/src/schJob.c
+++ b/source/libs/scheduler/src/schJob.c
--- a/source/libs/scheduler/src/schRemote.c
+++ b/source/libs/scheduler/src/schRemote.c
--- a/source/libs/scheduler/src/schStatus.c
+++ b/source/libs/scheduler/src/schStatus.c
--- a/source/libs/scheduler/src/schTask.c
+++ b/source/libs/scheduler/src/schTask.c
--- a/source/libs/scheduler/src/scheduler.c
+++ b/source/libs/scheduler/src/scheduler.c
--- a/source/libs/stream/src/stream.c
+++ b/source/libs/stream/src/stream.c
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
--- a/source/libs/stream/src/streamState.c
+++ b/source/libs/stream/src/streamState.c
--- a/source/libs/sync/src/syncAppendEntries.c
+++ b/source/libs/sync/src/syncAppendEntries.c
--- a/source/libs/sync/src/syncElection.c
+++ b/source/libs/sync/src/syncElection.c
--- a/source/libs/sync/src/syncMain.c
+++ b/source/libs/sync/src/syncMain.c
--- a/source/libs/sync/src/syncMessage.c
+++ b/source/libs/sync/src/syncMessage.c
--- a/source/libs/sync/src/syncPipeline.c
+++ b/source/libs/sync/src/syncPipeline.c
--- a/source/libs/sync/src/syncRequestVote.c
+++ b/source/libs/sync/src/syncRequestVote.c
--- a/source/libs/sync/src/syncSnapshot.c
+++ b/source/libs/sync/src/syncSnapshot.c
--- a/source/libs/sync/src/syncTimeout.c
+++ b/source/libs/sync/src/syncTimeout.c
--- a/source/libs/tdb/src/db/tdbPager.c
+++ b/source/libs/tdb/src/db/tdbPager.c
--- a/source/libs/transport/src/transCli.c
+++ b/source/libs/transport/src/transCli.c
--- a/source/util/src/tarray.c
+++ b/source/util/src/tarray.c
--- a/source/util/src/tconfig.c
+++ b/source/util/src/tconfig.c
--- a/source/util/src/terror.c
+++ b/source/util/src/terror.c
--- a/source/util/src/tlog.c
+++ b/source/util/src/tlog.c
--- a/source/util/src/tlrucache.c
+++ b/source/util/src/tlrucache.c
--- a/source/util/src/tsimplehash.c
+++ b/source/util/src/tsimplehash.c
--- a/source/util/src/tversion.c
+++ b/source/util/src/tversion.c
--- a/source/util/src/version.c.in
+++ b/source/util/src/version.c.in
--- a/tests/develop-test/5-taos-tools/taosbenchmark/json/sml_telnet_tcp.json
+++ b/tests/develop-test/5-taos-tools/taosbenchmark/json/sml_telnet_tcp.json
--- a/tests/parallel_test/cases.task
+++ b/tests/parallel_test/cases.task
--- a/tests/parallel_test/container_build.sh
+++ b/tests/parallel_test/container_build.sh
--- a/tests/parallel_test/run.sh
+++ b/tests/parallel_test/run.sh
--- a/tests/pytest/util/autogen.py
+++ b/tests/pytest/util/autogen.py
--- a/tests/pytest/util/common.py
+++ b/tests/pytest/util/common.py
--- a/tests/pytest/util/sql.py
+++ b/tests/pytest/util/sql.py
--- a/tests/requirements.txt
+++ b/tests/requirements.txt
--- a/tests/script/api/batchprepare.c
+++ b/tests/script/api/batchprepare.c
--- a/tests/script/coverage_test.sh
+++ b/tests/script/coverage_test.sh
--- a/tests/script/tsim/db/alter_option.sim
+++ b/tests/script/tsim/db/alter_option.sim
--- a/tests/script/tsim/db/create_all_options.sim
+++ b/tests/script/tsim/db/create_all_options.sim
--- a/tests/script/tsim/parser/col_arithmetic_operation.sim
+++ b/tests/script/tsim/parser/col_arithmetic_operation.sim
--- a/tests/script/tsim/parser/col_arithmetic_query.sim
+++ b/tests/script/tsim/parser/col_arithmetic_query.sim
--- a/tests/script/tsim/parser/commit.sim
+++ b/tests/script/tsim/parser/commit.sim
--- a/tests/script/tsim/parser/groupby.sim
+++ b/tests/script/tsim/parser/groupby.sim
--- a/tests/script/tsim/parser/interp.sim
+++ b/tests/script/tsim/parser/interp.sim
--- a/tests/script/tsim/parser/join_manyblocks.sim
+++ b/tests/script/tsim/parser/join_manyblocks.sim
--- a/tests/script/tsim/parser/limit1.sim
+++ b/tests/script/tsim/parser/limit1.sim
--- a/tests/script/tsim/parser/limit1_stb.sim
+++ b/tests/script/tsim/parser/limit1_stb.sim
--- a/tests/script/tsim/parser/limit1_tb.sim
+++ b/tests/script/tsim/parser/limit1_tb.sim
--- a/tests/script/tsim/parser/nestquery.sim
+++ b/tests/script/tsim/parser/nestquery.sim
--- a/tests/script/tsim/parser/projection_limit_offset.sim
+++ b/tests/script/tsim/parser/projection_limit_offset.sim
--- a/tests/script/tsim/parser/selectResNum.sim
+++ b/tests/script/tsim/parser/selectResNum.sim
--- a/tests/script/tsim/parser/sliding.sim
+++ b/tests/script/tsim/parser/sliding.sim
--- a/tests/script/tsim/parser/slimit1_query.sim
+++ b/tests/script/tsim/parser/slimit1_query.sim
--- a/tests/script/tsim/parser/table_merge_limit.sim
+++ b/tests/script/tsim/parser/table_merge_limit.sim
--- a/tests/script/tsim/parser/union.sim
+++ b/tests/script/tsim/parser/union.sim
--- a/tests/script/tsim/parser/where.sim
+++ b/tests/script/tsim/parser/where.sim
--- a/tests/script/tsim/query/explain.sim
+++ b/tests/script/tsim/query/explain.sim
--- a/tests/script/tsim/query/interval-offset.sim
+++ b/tests/script/tsim/query/interval-offset.sim
--- a/tests/script/tsim/query/join_interval.sim
+++ b/tests/script/tsim/query/join_interval.sim
--- a/tests/script/tsim/query/nullColSma.sim
+++ b/tests/script/tsim/query/nullColSma.sim
--- a/tests/script/tsim/query/sys_tbname.sim
+++ b/tests/script/tsim/query/sys_tbname.sim
--- a/tests/script/tsim/query/tableCount.sim
+++ b/tests/script/tsim/query/tableCount.sim
--- a/tests/script/tsim/query/unionall_as_table.sim
+++ b/tests/script/tsim/query/unionall_as_table.sim
--- a/tests/script/tsim/show/basic.sim
+++ b/tests/script/tsim/show/basic.sim
--- a/tests/script/tsim/stream/deleteState.sim
+++ b/tests/script/tsim/stream/deleteState.sim
--- a/tests/script/tsim/stream/distributeInterval0.sim
+++ b/tests/script/tsim/stream/distributeInterval0.sim
--- a/tests/script/tsim/stream/distributeIntervalRetrive0.sim
+++ b/tests/script/tsim/stream/distributeIntervalRetrive0.sim
--- a/tests/script/tsim/stream/partitionbyColumnState.sim
+++ b/tests/script/tsim/stream/partitionbyColumnState.sim
--- a/tests/script/tsim/stream/state1.sim
+++ b/tests/script/tsim/stream/state1.sim
--- a/tests/script/tsim/valgrind/checkError7.sim
+++ b/tests/script/tsim/valgrind/checkError7.sim
--- a/tests/system-test/0-others/check_assert.py
+++ b/tests/system-test/0-others/check_assert.py
--- a/tests/system-test/0-others/compatibility.py
+++ b/tests/system-test/0-others/compatibility.py
--- a/tests/system-test/0-others/information_schema.py
+++ b/tests/system-test/0-others/information_schema.py
--- a/tests/system-test/0-others/tag_index_basic.py
+++ b/tests/system-test/0-others/tag_index_basic.py
--- a/tests/system-test/0-others/taosdMonitor.py
+++ b/tests/system-test/0-others/taosdMonitor.py
--- a/tests/system-test/0-others/telemetry.py
+++ b/tests/system-test/0-others/telemetry.py
--- a/tests/system-test/0-others/tmqBasic.json
+++ b/tests/system-test/0-others/tmqBasic.json
--- a/tests/system-test/1-insert/insert_wide_column.py
+++ b/tests/system-test/1-insert/insert_wide_column.py
--- a/tests/system-test/2-query/Today.py
+++ b/tests/system-test/2-query/Today.py
--- a/tests/system-test/2-query/cast.py
+++ b/tests/system-test/2-query/cast.py
--- a/tests/system-test/2-query/columnLenUpdated.py
+++ b/tests/system-test/2-query/columnLenUpdated.py
--- a/tests/system-test/2-query/json_tag.py
+++ b/tests/system-test/2-query/json_tag.py
--- a/tests/system-test/2-query/last_row_interval.py
+++ b/tests/system-test/2-query/last_row_interval.py
--- a/tests/system-test/2-query/limit.py
+++ b/tests/system-test/2-query/limit.py
--- a/tests/system-test/2-query/nestedQuery.py
+++ b/tests/system-test/2-query/nestedQuery.py
--- a/tests/system-test/2-query/nestedQuery_26.py
+++ b/tests/system-test/2-query/nestedQuery_26.py
--- a/tests/system-test/2-query/nestedQuery_math.py
+++ b/tests/system-test/2-query/nestedQuery_math.py
--- a/tests/system-test/2-query/odbc.py
+++ b/tests/system-test/2-query/odbc.py
--- a/tests/system-test/2-query/out_of_order.py
+++ b/tests/system-test/2-query/out_of_order.py
--- a/tests/system-test/2-query/timeerror.py
+++ b/tests/system-test/2-query/timeerror.py
--- a/tests/system-test/2-query/tsbsQuery.py
+++ b/tests/system-test/2-query/tsbsQuery.py
--- a/tests/system-test/6-cluster/5dnode1mnode.py
+++ b/tests/system-test/6-cluster/5dnode1mnode.py
--- a/tests/system-test/7-tmq/raw_block_interface_test.py
+++ b/tests/system-test/7-tmq/raw_block_interface_test.py
--- a/tests/system-test/7-tmq/subscribeDb.py
+++ b/tests/system-test/7-tmq/subscribeDb.py
--- a/tests/system-test/7-tmq/subscribeDb1.py
+++ b/tests/system-test/7-tmq/subscribeDb1.py
--- a/tests/system-test/7-tmq/subscribeStb1.py
+++ b/tests/system-test/7-tmq/subscribeStb1.py
--- a/tests/system-test/7-tmq/tmq3mnodeSwitch.py
+++ b/tests/system-test/7-tmq/tmq3mnodeSwitch.py
--- a/tests/system-test/7-tmq/tmqConsFromTsdb-mutilVg.py
+++ b/tests/system-test/7-tmq/tmqConsFromTsdb-mutilVg.py
--- a/tests/system-test/7-tmq/tmqDelete-1ctb.py
+++ b/tests/system-test/7-tmq/tmqDelete-1ctb.py
--- a/tests/system-test/7-tmq/tmqDnodeRestart.py
+++ b/tests/system-test/7-tmq/tmqDnodeRestart.py
--- a/tests/system-test/7-tmq/tmqDropNtb-snapshot0.py
+++ b/tests/system-test/7-tmq/tmqDropNtb-snapshot0.py
--- a/tests/system-test/7-tmq/tmqMultiConsumer.py
+++ b/tests/system-test/7-tmq/tmqMultiConsumer.py
--- a/tools/shell/inc/shellInt.h
+++ b/tools/shell/inc/shellInt.h
--- a/tools/shell/src/shellArguments.c
+++ b/tools/shell/src/shellArguments.c
--- a/tools/shell/src/shellEngine.c
+++ b/tools/shell/src/shellEngine.c
--- a/tools/shell/src/shellMain.c
+++ b/tools/shell/src/shellMain.c
--- a/tools/shell/src/shellWebsocket.c
+++ b/tools/shell/src/shellWebsocket.c
--- a/utils/test/c/CMakeLists.txt
+++ b/utils/test/c/CMakeLists.txt
--- a/utils/test/c/sml_test.c
+++ b/utils/test/c/sml_test.c
--- a/utils/test/c/tmqSim.c
+++ b/utils/test/c/tmqSim.c
--- a/utils/test/c/tmq_taosx_ci.c
+++ b/utils/test/c/tmq_taosx_ci.c
--- a/utils/test/c/write_raw_block_test.c
+++ b/utils/test/c/write_raw_block_test.c