Merge branch '3.0' into feature/3.0_glzhao

cmake/cmake.define

@@ -46,7 +46,7 @@ ENDIF ()
 
 IF (TD_WINDOWS)
     MESSAGE("${Yellow} set compiler flag for Windows! ${ColourReset}")
-    SET(COMMON_FLAGS "/w /D_WIN32")
+    SET(COMMON_FLAGS "/w /D_WIN32 /Zi")
     SET(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} /MANIFEST:NO")
     # IF (MSVC AND (MSVC_VERSION GREATER_EQUAL 1900))
     #  SET(COMMON_FLAGS "${COMMON_FLAGS} /Wv:18")
@@ -71,8 +71,8 @@ ELSE ()
     ENDIF ()
 
     IF (${SANITIZER} MATCHES "true")
-        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS}     -Werror -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -static-libasan -g3")
-        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -static-libasan -g3")
+        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS}     -Werror -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -g3")
+        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -g3")
 	MESSAGE(STATUS "Will compile with Address Sanitizer!")
     ELSE ()
         SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -Werror=return-type -fPIC -gdwarf-2 -g3")

contrib/CMakeLists.txt

@@ -100,8 +100,10 @@ endif(${BUILD_WITH_NURAFT})
 
 # addr2line
 if(${BUILD_ADDR2LINE})
+    if(NOT ${TD_WINDOWS})
         cat("${TD_SUPPORT_DIR}/libdwarf_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
         cat("${TD_SUPPORT_DIR}/addr2line_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+    endif(NOT ${TD_WINDOWS})
 endif(${BUILD_ADDR2LINE})
 
 # download dependencies
@@ -335,6 +337,7 @@ endif(${BUILD_WITH_SQLITE})
 
 # addr2line
 if(${BUILD_ADDR2LINE})
+    if(NOT ${TD_WINDOWS})    
         check_include_file( "sys/types.h"     HAVE_SYS_TYPES_H) 
         check_include_file( "sys/stat.h"      HAVE_SYS_STAT_H ) 
         check_include_file( "inttypes.h"      HAVE_INTTYPES_H   ) 
@@ -374,6 +377,7 @@ if(${BUILD_ADDR2LINE})
         add_library(addr2line STATIC "addr2line/addr2line.c")
         target_link_libraries(addr2line PUBLIC libdwarf dl z)
         target_include_directories(addr2line PUBLIC "libdwarf/src/lib/libdwarf" )
+    endif(NOT ${TD_WINDOWS})
 endif(${BUILD_ADDR2LINE})
 
 

docs-cn/14-reference/03-connector/csharp.mdx

@@ -18,7 +18,7 @@ import CSAsyncQuery from "../../07-develop/04-query-data/_cs_async.mdx"
 
 `TDengine.Connector` 是 TDengine 提供的 C# 语言连接器。C# 开发人员可以通过它开发存取 TDengine 集群数据的 C# 应用软件。
 
-`TDengine.Connector` 连接器支持通过 TDengine 客户端驱动（taosc）建立与 TDengine 运行实例的连接，提供数据写入、查询、订阅、schemaless 数据写入、参数绑定接口数据写入等功能 `TDengine.Connector` 目前暂未提供 REST 连接方式，用户可以参考 [RESTful APIs](https://docs.taosdata.com//reference/restful-api/) 文档自行编写。
+`TDengine.Connector` 连接器支持通过 TDengine 客户端驱动（taosc）建立与 TDengine 运行实例的连接，提供数据写入、查询、订阅、schemaless 数据写入、参数绑定接口数据写入等功能 `TDengine.Connector` 目前暂未提供 REST 连接方式，用户可以参考 [REST API](/reference/rest-api/) 文档自行编写。
 
 本文介绍如何在 Linux 或 Windows 环境中安装 `TDengine.Connector`，并通过 `TDengine.Connector` 连接 TDengine 集群，进行数据写入、查询等基本操作。
 

docs-en/01-index.md

@@ -4,24 +4,24 @@ sidebar_label: Documentation Home
 slug: /
 ---
 
-TDengine is a [high-performance](https://tdengine.com/fast), [scalable](https://tdengine.com/scalable) time series database with [SQL support](https://tdengine.com/sql-support). This document is the TDengine user manual. It introduces the basic concepts, installation, features, SQL, APIs, operation, maintenance, kernel design, etc. It’s written mainly for architects, developers and system administrators.
+TDengine is a [high-performance](https://tdengine.com/fast), [scalable](https://tdengine.com/scalable) time series database with [SQL support](https://tdengine.com/sql-support). This document is the TDengine user manual. It introduces the basic, as well as novel concepts, in TDengine, and also talks in detail about installation, features, SQL, APIs, operation, maintenance, kernel design and other topics. It’s written mainly for architects, developers and system administrators.
 
-To get a global view about TDengine, like feature list, benchmarks, and competitive advantages, please browse through section [Introduction](./intro).
+To get an overview of TDengine, such as a feature list, benchmarks, and competitive advantages, please browse through the [Introduction](./intro) section.
 
-TDengine makes full use of the characteristics of time series data, proposes the concepts of "one table for one data collection point" and "super table", and designs an innovative storage engine, which greatly improves the efficiency of data ingestion, querying and storage. To understand the new concepts and use TDengine in the right way, please read [“Concepts”](./concept) thoroughly.
+TDengine greatly improves the efficiency of data ingestion, querying and storage by exploiting the characteristics of time series data, introducing the novel concepts of "one table for one data collection point" and "super table", and designing an innovative storage engine. To understand the new concepts in TDengine and make full use of the features and capabilities of TDengine, please read [“Concepts”](./concept) thoroughly.
 
-If you are a developer, please read the [“Developer Guide”](./develop) carefully. This section introduces the database connection, data modeling, data ingestion, query, continuous query, cache, data subscription, user-defined function, etc. in detail. Sample code is provided for a variety of programming languages. In most cases, you can just copy and paste the sample code, make a few changes to accommodate your application, and it will work.
+If you are a developer, please read the [“Developer Guide”](./develop) carefully. This section introduces the database connection, data modeling, data ingestion, query, continuous query, cache, data subscription, user-defined functions, and other functionality in detail. Sample code is provided for a variety of programming languages. In most cases, you can just copy and paste the sample code, make a few changes to accommodate your application, and it will work.
 
-We live in the era of big data, and scale-up is unable to meet the growing business needs. Any modern data system must have the ability to scale out, and clustering has become an indispensable feature of big data systems. The TDengine team has not only developed the cluster feature, they also decided to open source this important feature. To learn how to deploy, manage and maintain a TDengine cluster please refer to ["Cluster"](./cluster).
+We live in the era of big data, and scale-up is unable to meet the growing needs of business. Any modern data system must have the ability to scale out, and clustering has become an indispensable feature of big data systems. Not only did the TDengine team develop the cluster feature, but also decided to open source this important feature. To learn how to deploy, manage and maintain a TDengine cluster please refer to ["cluster"](./cluster).
 
-TDengine uses SQL as its query language, which greatly reduces learning costs and migration costs. In addition to the standard SQL, TDengine has extensions to support time series data scenarios better, such as roll up, interpolation, time weighted average, etc. The ["SQL Reference"](./taos-sql) chapter describes the SQL syntax in detail, and lists the various supported commands and functions.
+TDengine uses ubiquitious SQL as its query language, which greatly reduces learning costs and migration costs. In addition to the standard SQL, TDengine has extensions to better support time series data analysis. These extensions include functions such as roll up, interpolation and time weighted average, among many others. The ["SQL Reference"](./taos-sql) chapter describes the SQL syntax in detail, and lists the various supported commands and functions.
 
-If you are a system administrator who cares about installation, upgrade, fault tolerance, disaster recovery, data import, data export, system configuration, how to monitor whether TDengine is running healthily, and how to improve system performance, please refer to the ["Administration"](./operation) thoroughly.
+If you are a system administrator who cares about installation, upgrade, fault tolerance, disaster recovery, data import, data export, system configuration, how to monitor whether TDengine is running healthily, and how to improve system performance, please refer to, and thoroughly read the ["Administration"](./operation) section.
 
-If you want to know more about TDengine tools, REST API, and connectors for various programming languages, please see the ["Reference"](./reference) chapter.
+If you want to know more about TDengine tools, the REST API, and connectors for various programming languages, please see the ["Reference"](./reference) chapter.
 
 If you are very interested in the internal design of TDengine, please read the chapter ["Inside TDengine”](./tdinternal), which introduces the cluster design, data partitioning, sharding, writing, and reading processes in detail. If you want to study TDengine code or even contribute code, please read this chapter carefully.
 
-TDengine is an open source database, you are welcome to be a part of TDengine. If you find any errors in the documentation, or the description is not clear, please click "Edit this page" at the bottom of each page to edit it directly.
+TDengine is an open source database, and we would love for you to be a part of TDengine. If you find any errors in the documentation, or see parts where more clarity or elaboration is needed, please click "Edit this page" at the bottom of each page to edit it directly.
 
 Together, we make a difference.

docs-en/14-reference/03-connector/csharp.mdx

@@ -19,7 +19,7 @@ import CSAsyncQuery from "../../07-develop/04-query-data/_cs_async.mdx"
 
 `TDengine.Connector` is a C# language connector provided by TDengine that allows C# developers to develop C# applications that access TDengine cluster data.
 
-The `TDengine.Connector` connector supports connect to TDengine instances via the TDengine client driver (taosc), providing data writing, querying, subscription, schemaless writing, bind interface, etc. The `TDengine.Connector` currently does not provide a REST connection interface. Developers can write their RESTful application by referring to the [RESTful APIs](https://docs.taosdata.com//reference/restful-api/) documentation.
+The `TDengine.Connector` connector supports connect to TDengine instances via the TDengine client driver (taosc), providing data writing, querying, subscription, schemaless writing, bind interface, etc. The `TDengine.Connector` currently does not provide a REST connection interface. Developers can write their RESTful application by referring to the [REST API](/reference/rest-api/) documentation.
 
 This article describes how to install `TDengine.Connector` in a Linux or Windows environment and connect to TDengine clusters via `TDengine.Connector` to perform basic operations such as data writing and querying.
 

example/src/tstream.c

@@ -82,9 +82,7 @@ int32_t create_stream() {
   /*const char* sql = "select sum(k) from tu1 interval(10m)";*/
   /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/
   pRes = taos_query(
-      pConn,
-      "create stream stream1 trigger at_once into outstb as select _wstartts, min(k), max(k), sum(k) as sum_of_k "
-      "from tu1 interval(10m)");
+      pConn, "create stream stream1 trigger at_once into outstb as select _wstartts, sum(k) from tu1 interval(10m)");
   if (taos_errno(pRes) != 0) {
     printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
     return -1;

include/common/taosdef.h

@@ -96,6 +96,7 @@ extern char *qtypeStr[];
 #define TSDB_PORT_HTTP 11
 
 #undef TD_DEBUG_PRINT_ROW
+#undef TD_DEBUG_PRINT_TSDB_LOAD_DCOLS
 
 #ifdef __cplusplus
 }

include/common/tmsg.h

@@ -244,12 +244,12 @@ typedef struct {
   const void* pMsg;
 } SSubmitMsgIter;
 
-int32_t tInitSubmitMsgIter(const SSubmitReq* pMsg, SSubmitMsgIter* pIter);
+int32_t tInitSubmitMsgIter(SSubmitReq* pMsg, SSubmitMsgIter* pIter);
 int32_t tGetSubmitMsgNext(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
 int32_t tInitSubmitBlkIter(SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
 STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
 // for debug
-int32_t tPrintFixedSchemaSubmitReq(const SSubmitReq* pReq, STSchema* pSchema);
+int32_t tPrintFixedSchemaSubmitReq(SSubmitReq* pReq, STSchema* pSchema);
 
 typedef struct {
   int32_t code;
@@ -1697,7 +1697,7 @@ int32_t tDecodeSRSmaParam(SDecoder* pCoder, SRSmaParam* pRSmaParam);
 
 // TDMT_VND_CREATE_STB ==============
 typedef struct SVCreateStbReq {
-  const char*    name;
+  char*          name;
   tb_uid_t       suid;
   int8_t         rollup;
   SSchemaWrapper schema;
@@ -1710,7 +1710,7 @@ int tDecodeSVCreateStbReq(SDecoder* pCoder, SVCreateStbReq* pReq);
 
 // TDMT_VND_DROP_STB ==============
 typedef struct SVDropStbReq {
-  const char* name;
+  char*    name;
   tb_uid_t suid;
 } SVDropStbReq;
 
@@ -1723,13 +1723,13 @@ typedef struct SVCreateTbReq {
   int32_t  flags;
   tb_uid_t uid;
   int64_t  ctime;
-  const char* name;
+  char*    name;
   int32_t  ttl;
   int8_t   type;
   union {
     struct {
       tb_uid_t suid;
-      const uint8_t* pTag;
+      uint8_t* pTag;
     } ctb;
     struct {
       SSchemaWrapper schema;
@@ -1777,7 +1777,7 @@ int32_t tDeserializeSVCreateTbBatchRsp(void* buf, int32_t bufLen, SVCreateTbBatc
 
 // TDMT_VND_DROP_TABLE =================
 typedef struct {
-  const char* name;
+  char*  name;
   int8_t igNotExists;
 } SVDropTbReq;
 
@@ -1809,9 +1809,9 @@ int32_t tDecodeSVDropTbBatchRsp(SDecoder* pCoder, SVDropTbBatchRsp* pRsp);
 
 // TDMT_VND_ALTER_TABLE =====================
 typedef struct {
-  const char* tbName;
+  char*  tbName;
   int8_t action;
-  const char* colName;
+  char*  colName;
   // TSDB_ALTER_TABLE_ADD_COLUMN
   int8_t  type;
   int8_t  flags;
@@ -1820,17 +1820,17 @@ typedef struct {
   // TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES
   int32_t colModBytes;
   // TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME
-  const char* colNewName;
+  char* colNewName;
   // TSDB_ALTER_TABLE_UPDATE_TAG_VAL
-  const char*    tagName;
+  char*    tagName;
   int8_t   isNull;
   uint32_t nTagVal;
-  const uint8_t* pTagVal;
+  uint8_t* pTagVal;
   // TSDB_ALTER_TABLE_UPDATE_OPTIONS
   int8_t  updateTTL;
   int32_t newTTL;
   int8_t  updateComment;
-  const char* newComment;
+  char*   newComment;
 } SVAlterTbReq;
 
 int32_t tEncodeSVAlterTbReq(SEncoder* pEncoder, const SVAlterTbReq* pReq);
@@ -2020,7 +2020,7 @@ static FORCE_INLINE void tFreeClientHbBatchRsp(void* pRsp) {
 
 int32_t tSerializeSClientHbBatchRsp(void* buf, int32_t bufLen, const SClientHbBatchRsp* pBatchRsp);
 int32_t tDeserializeSClientHbBatchRsp(void* buf, int32_t bufLen, SClientHbBatchRsp* pBatchRsp);
-void tFreeSClientHbBatchRsp(SClientHbBatchRsp *pBatchRsp);
+void    tFreeSClientHbBatchRsp(SClientHbBatchRsp* pBatchRsp);
 
 static FORCE_INLINE int32_t tEncodeSKv(SEncoder* pEncoder, const SKv* pKv) {
   if (tEncodeI32(pEncoder, pKv->key) < 0) return -1;
@@ -2267,8 +2267,8 @@ typedef struct {
   int64_t  interval;
   int64_t  offset;  // use unit by precision of DB
   int64_t  sliding;
-  const char* expr;  // sma expression
-  const char* tagsFilter;
+  char*    expr;  // sma expression
+  char*    tagsFilter;
 } STSma;  // Time-range-wise SMA
 
 typedef STSma SVCreateTSmaReq;
@@ -2600,7 +2600,7 @@ typedef struct {
   int64_t       uid;
   int32_t       sver;
   uint32_t      nData;
-  const uint8_t* pData;
+  uint8_t*      pData;
   SVCreateTbReq cTbReq;
 } SVSubmitBlk;
 

include/common/tmsgdef.h

@@ -158,6 +158,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_DROP_INDEX, "mnode-drop-index", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_GET_DB_CFG, "mnode-get-db-cfg", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_GET_INDEX, "mnode-get-index", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_APPLY_MSG, "mnode-apply-msg", NULL, NULL)
 
   // Requests handled by VNODE
   TD_NEW_MSG_SEG(TDMT_VND_MSG)

include/common/ttypes.h

@@ -30,7 +30,7 @@ typedef uint64_t TDRowVerT;
 typedef int16_t  col_id_t;
 typedef int8_t   col_type_t;
 typedef int32_t  col_bytes_t;
-typedef uint16_t schema_ver_t;
+typedef int32_t schema_ver_t;
 typedef int32_t  func_id_t;
 
 #pragma pack(push, 1)

include/dnode/mnode/mnode.h

@@ -89,6 +89,7 @@ int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad);
  * @return int32_t 0 for success, -1 for failure.
  */
 int32_t mndProcessMsg(SRpcMsg *pMsg);
+int32_t mndProcessSyncMsg(SRpcMsg *pMsg);
 
 /**
  * @brief Generate machine code

include/dnode/mnode/sdb/sdb.h

@@ -304,13 +304,16 @@ int32_t sdbGetMaxId(SSdb *pSdb, ESdbType type);
 int64_t sdbGetTableVer(SSdb *pSdb, ESdbType type);
 
 /**
- * @brief Update the version of sdb
+ * @brief Update the index of sdb
  *
  * @param pSdb The sdb object.
- * @param val The update value of the version.
- * @return int32_t The current version of sdb
+ * @param index The update value of the apply index.
+ * @return int32_t The current index of sdb
  */
-int64_t sdbUpdateVer(SSdb *pSdb, int32_t val);
+void    sdbSetApplyIndex(SSdb *pSdb, int64_t index);
+int64_t sdbGetApplyIndex(SSdb *pSdb);
+void    sdbSetApplyTerm(SSdb *pSdb, int64_t term);
+int64_t sdbGetApplyTerm(SSdb *pSdb);
 
 SSdbRaw *sdbAllocRaw(ESdbType type, int8_t sver, int32_t dataLen);
 void     sdbFreeRaw(SSdbRaw *pRaw);
@@ -339,6 +342,7 @@ typedef struct SSdb {
   char          *tmpDir;
   int64_t        lastCommitVer;
   int64_t        curVer;
+  int64_t        curTerm;
   int64_t        tableVer[SDB_MAX];
   int64_t        maxId[SDB_MAX];
   EKeyType       keyTypes[SDB_MAX];

include/libs/catalog/catalog.h

@@ -46,24 +46,34 @@ typedef enum {
   AUTH_TYPE_OTHER,
 } AUTH_TYPE;
 
+typedef struct SUserAuthInfo {
+  char user[TSDB_USER_LEN]; 
+  char dbFName[TSDB_DB_FNAME_LEN]; 
+  AUTH_TYPE type;
+} SUserAuthInfo;
+
 typedef struct SCatalogReq {
-  SArray *pTableName;     // element is SNAME
-  SArray *pUdf;           // udf name
+  SArray *pTableMeta;     // element is SNAME
+  SArray *pDbVgroup;      // element is db full name
+  SArray *pTableHash;     // element is SNAME
+  SArray *pUdf;           // element is udf name
+  SArray *pDbCfg;         // element is db full name
+  SArray *pIndex;         // element is index name
+  SArray *pUser;          // element is SUserAuthInfo
   bool    qNodeRequired;  // valid qnode
 } SCatalogReq;
 
 typedef struct SMetaData {
-  SArray    *pTableMeta;  // STableMeta array
-  SArray    *pVgroupInfo; // SVgroupInfo list
-  SArray    *pUdfList;    // udf info list
-  SArray    *pQnodeList;  // qnode list, SArray<SQueryNodeAddr>
+  SArray    *pTableMeta;  // SArray<STableMeta>
+  SArray    *pDbVgroup;   // SArray<SArray<SVgroupInfo>*>
+  SArray    *pTableHash;  // SArray<SVgroupInfo>
+  SArray    *pUdfList;    // SArray<SFuncInfo>
+  SArray    *pDbCfg;      // SArray<SDbCfgInfo>
+  SArray    *pIndex;      // SArray<SIndexInfo>
+  SArray    *pUser;       // SArray<bool>
+  SArray    *pQnodeList;  // SArray<SQueryNodeAddr>
 } SMetaData;
 
-typedef struct STbSVersion {
-  char* tbFName;
-  int32_t sver;
-} STbSVersion;
-
 typedef struct SCatalogCfg {
   uint32_t maxTblCacheNum;
   uint32_t maxDBCacheNum;
@@ -88,6 +98,11 @@ typedef struct SDbVgVersion {
   int32_t numOfTable; // unit is TSDB_TABLE_NUM_UNIT
 } SDbVgVersion;
 
+typedef struct STbSVersion {
+  char* tbFName;
+  int32_t sver;
+} STbSVersion;
+
 typedef struct SUserAuthVersion {
   char    user[TSDB_USER_LEN];
   int32_t version;
@@ -96,6 +111,8 @@ typedef struct SUserAuthVersion {
 typedef SDbCfgRsp SDbCfgInfo;
 typedef SUserIndexRsp SIndexInfo;
 
+typedef void (*catalogCallback)(SMetaData* pResult, void* param, int32_t code);
+
 int32_t catalogInit(SCatalogCfg *cfg);
 
 /**
@@ -131,7 +148,7 @@ int32_t catalogUpdateDBVgInfo(SCatalog* pCatalog, const char* dbName, uint64_t d
 
 int32_t catalogRemoveDB(SCatalog* pCatalog, const char* dbName, uint64_t dbId);
 
-int32_t catalogRemoveTableMeta(SCatalog* pCtg, const SName* pTableName);
+int32_t catalogRemoveTableMeta(SCatalog* pCtg, SName* pTableName);
 
 int32_t catalogRemoveStbMeta(SCatalog* pCtg, const char* dbFName, uint64_t dbId, const char* stbName, uint64_t suid);
 
@@ -241,9 +258,9 @@ int32_t catalogGetExpiredUsers(SCatalog* pCtg, SUserAuthVersion **users, uint32_
 
 int32_t catalogGetDBCfg(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, SDbCfgInfo* pDbCfg);
 
-int32_t catalogGetIndexInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* indexName, SIndexInfo* pInfo);
+int32_t catalogGetIndexMeta(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* indexName, SIndexInfo* pInfo);
 
-int32_t catalogGetUdfInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* funcName, SFuncInfo** pInfo);
+int32_t catalogGetUdfInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* funcName, SFuncInfo* pInfo);
 
 int32_t catalogChkAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, AUTH_TYPE type, bool *pass);
 

include/libs/function/functionMgt.h

@@ -146,7 +146,8 @@ bool fmIsBuiltinFunc(const char* pFunc);
 
 bool fmIsAggFunc(int32_t funcId);
 bool fmIsScalarFunc(int32_t funcId);
-bool fmIsNonstandardSQLFunc(int32_t funcId);
+bool fmIsVectorFunc(int32_t funcId);
+bool fmIsIndefiniteRowsFunc(int32_t funcId);
 bool fmIsStringFunc(int32_t funcId);
 bool fmIsDatetimeFunc(int32_t funcId);
 bool fmIsSelectFunc(int32_t funcId);

include/libs/index/index.h

@@ -16,9 +16,11 @@
 #ifndef _TD_INDEX_H_
 #define _TD_INDEX_H_
 
+#include "nodes.h"
 #include "os.h"
 #include "taoserror.h"
 #include "tarray.h"
+#include "tglobal.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -189,6 +191,12 @@ void        indexTermDestroy(SIndexTerm* p);
  */
 void indexInit();
 
+/* index filter */
+typedef enum { SFLT_NOT_INDEX, SFLT_COARSE_INDEX, SFLT_ACCURATE_INDEX } SIdxFltStatus;
+
+SIdxFltStatus idxGetFltStatus(SNode* pFilterNode);
+
+int32_t doFilterTag(const SNode* pFilterNode, SArray* result);
 /*
  * destory index env
  *

include/libs/nodes/nodes.h

@@ -61,8 +61,8 @@ extern "C" {
 
 #define DESTORY_LIST(list)    \
   do {                        \
-    nodesDestroyList(list); \
-    list = NULL;            \
+    nodesDestroyList((list)); \
+    (list) = NULL;            \
   } while (0)
 
 typedef enum ENodeType {
@@ -96,6 +96,7 @@ typedef enum ENodeType {
   QUERY_NODE_EXPLAIN_OPTIONS,
   QUERY_NODE_STREAM_OPTIONS,
   QUERY_NODE_TOPIC_OPTIONS,
+  QUERY_NODE_LEFT_VALUE,
 
   // Statement nodes are used in parser and planner module.
   QUERY_NODE_SET_OPERATOR,

include/libs/nodes/plannodes.h

@@ -54,6 +54,7 @@ typedef struct SScanLogicNode {
   int64_t            sliding;
   int8_t             intervalUnit;
   int8_t             slidingUnit;
+  SNode*             pTagCond;
 } SScanLogicNode;
 
 typedef struct SJoinLogicNode {
@@ -343,6 +344,7 @@ typedef struct SSubplan {
   SNodeList*     pParents;      // the data destination subplan, get data from current subplan
   SPhysiNode*    pNode;         // physical plan of current subplan
   SDataSinkNode* pDataSink;     // data of the subplan flow into the datasink
+  SNode*         pTagCond;
 } SSubplan;
 
 typedef enum EExplainMode { EXPLAIN_MODE_DISABLE = 1, EXPLAIN_MODE_STATIC, EXPLAIN_MODE_ANALYZE } EExplainMode;

include/libs/nodes/querynodes.h

@@ -81,6 +81,7 @@ typedef struct SValueNode {
   char*     literal;
   bool      isDuration;
   bool      translate;
+  bool      notReserved;
   int16_t   placeholderNo;
   union {
     bool     b;
@@ -93,6 +94,10 @@ typedef struct SValueNode {
   char    unit;
 } SValueNode;
 
+typedef struct SLeftValueNode {
+  ENodeType type;
+} SLeftValueNode;
+
 typedef struct SOperatorNode {
   SExprNode     node;  // QUERY_NODE_OPERATOR
   EOperatorType opType;
@@ -236,7 +241,7 @@ typedef struct SSelectStmt {
   bool        isTimeOrderQuery;
   bool        hasAggFuncs;
   bool        hasRepeatScanFuncs;
-  bool        hasNonstdSQLFunc;
+  bool        hasIndefiniteRowsFunc;
 } SSelectStmt;
 
 typedef enum ESetOperatorType { SET_OP_TYPE_UNION_ALL = 1, SET_OP_TYPE_UNION } ESetOperatorType;

include/libs/stream/tstream.h

@@ -114,17 +114,12 @@ static FORCE_INLINE void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit)
 int32_t streamDataBlockEncode(void** buf, const SStreamDataBlock* pOutput);
 void*   streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput);
 
-typedef struct {
-  void* inputHandle;
-  void* executor;
-} SStreamRunner;
-
 typedef struct {
   int8_t parallelizable;
   char*  qmsg;
   // followings are not applicable to encoder and decoder
-  int8_t         numOfRunners;
-  SStreamRunner* runners;
+  void* inputHandle;
+  void* executor;
 } STaskExec;
 
 typedef struct {
@@ -320,17 +315,15 @@ int32_t streamEnqueueDataSubmit(SStreamTask* pTask, SStreamDataSubmit* input);
 int32_t streamEnqueueDataBlk(SStreamTask* pTask, SStreamDataBlock* input);
 int32_t streamDequeueOutput(SStreamTask* pTask, void** output);
 
-int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, int32_t inputType, int32_t workId);
-
 int32_t streamTaskRun(SStreamTask* pTask);
 
 int32_t streamTaskHandleInput(SStreamTask* pTask, void* data);
 
 int32_t streamTaskProcessRunReq(SStreamTask* pTask, SMsgCb* pMsgCb);
-int32_t streamTaskProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchReq* pReq, SRpcMsg* pMsg);
-int32_t streamTaskProcessDispatchRsp(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchRsp* pRsp);
-int32_t streamTaskProcessRecoverReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamTaskRecoverReq* pReq, SRpcMsg* pMsg);
-int32_t streamTaskProcessRecoverRsp(SStreamTask* pTask, SStreamTaskRecoverRsp* pRsp);
+int32_t streamProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchReq* pReq, SRpcMsg* pMsg);
+int32_t streamProcessDispatchRsp(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchRsp* pRsp);
+int32_t streamProcessRecoverReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamTaskRecoverReq* pReq, SRpcMsg* pMsg);
+int32_t streamProcessRecoverRsp(SStreamTask* pTask, SStreamTaskRecoverRsp* pRsp);
 
 #ifdef __cplusplus
 }

include/libs/sync/sync.h

@@ -78,6 +78,8 @@ typedef struct SFsmCbMeta {
   int32_t    code;
   ESyncState state;
   uint64_t   seqNum;
+  SyncTerm   term;
+  SyncTerm   currentTerm;
 } SFsmCbMeta;
 
 typedef struct SSyncFSM {
@@ -85,6 +87,7 @@ typedef struct SSyncFSM {
   void (*FpCommitCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
   void (*FpPreCommitCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
   void (*FpRollBackCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
+  void (*FpRestoreFinish)(struct SSyncFSM* pFsm);
   int32_t (*FpGetSnapshot)(struct SSyncFSM* pFsm, SSnapshot* pSnapshot);
   int32_t (*FpRestoreSnapshot)(struct SSyncFSM* pFsm, const SSnapshot* snapshot);
 } SSyncFSM;
@@ -117,7 +120,6 @@ typedef struct SSyncLogStore {
 
 } SSyncLogStore;
 
-
 typedef struct SSyncInfo {
   SyncGroupId vgId;
   SSyncCfg    syncCfg;

include/libs/transport/trpc.h

@@ -28,7 +28,7 @@ extern "C" {
 #define TAOS_CONN_CLIENT 1
 #define IsReq(pMsg)      (pMsg->msgType & 1U)
 
-extern int tsRpcHeadSize;
+extern int32_t tsRpcHeadSize;
 
 typedef struct {
   uint32_t clientIp;
@@ -69,10 +69,10 @@ typedef struct SRpcInit {
   char     localFqdn[TSDB_FQDN_LEN];
   uint16_t localPort;     // local port
   char *   label;         // for debug purpose
-  int      numOfThreads;  // number of threads to handle connections
-  int      sessions;      // number of sessions allowed
+  int32_t  numOfThreads;  // number of threads to handle connections
+  int32_t  sessions;      // number of sessions allowed
   int8_t   connType;      // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
-  int      idleTime;      // milliseconds, 0 means idle timer is disabled
+  int32_t  idleTime;      // milliseconds, 0 means idle timer is disabled
 
   // the following is for client app ecurity only
   char *user;  // user name
@@ -108,9 +108,9 @@ int32_t rpcInit();
 void    rpcCleanup();
 void *  rpcOpen(const SRpcInit *pRpc);
 void    rpcClose(void *);
-void *  rpcMallocCont(int contLen);
+void *  rpcMallocCont(int32_t contLen);
 void    rpcFreeCont(void *pCont);
-void *  rpcReallocCont(void *ptr, int contLen);
+void *  rpcReallocCont(void *ptr, int32_t contLen);
 
 // Because taosd supports multi-process mode
 // These functions should not be used on the server side
@@ -123,7 +123,7 @@ void rpcReleaseHandle(void *handle, int8_t type);  // just release client conn t
 // These functions will not be called in the child process
 void    rpcSendRedirectRsp(void *pConn, const SEpSet *pEpSet);
 void    rpcSendRequestWithCtx(void *thandle, const SEpSet *pEpSet, SRpcMsg *pMsg, int64_t *rid, SRpcCtx *ctx);
-int  rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo);
+int32_t rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo);
 void    rpcSendRecv(void *shandle, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp);
 
 #ifdef __cplusplus

include/util/taoserror.h

@@ -253,6 +253,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_TRANS_INVALID_STAGE       TAOS_DEF_ERROR_CODE(0, 0x03D2)
 #define TSDB_CODE_MND_TRANS_CONFLICT            TAOS_DEF_ERROR_CODE(0, 0x03D3)
 #define TSDB_CODE_MND_TRANS_UNKNOW_ERROR        TAOS_DEF_ERROR_CODE(0, 0x03D4)
+#define TSDB_CODE_MND_TRANS_CLOG_IS_NULL        TAOS_DEF_ERROR_CODE(0, 0x03D5)
 
 // mnode-mq
 #define TSDB_CODE_MND_TOPIC_ALREADY_EXIST       TAOS_DEF_ERROR_CODE(0, 0x03E0)
@@ -638,6 +639,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_NOT_ALLOWED_FUNC          TAOS_DEF_ERROR_CODE(0, 0x264F)
 #define TSDB_CODE_PAR_NOT_ALLOWED_WIN_QUERY     TAOS_DEF_ERROR_CODE(0, 0x2650)
 #define TSDB_CODE_PAR_INVALID_DROP_COL          TAOS_DEF_ERROR_CODE(0, 0x2651)
+#define TSDB_CODE_PAR_INVALID_COL_JSON          TAOS_DEF_ERROR_CODE(0, 0x2652)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

include/util/tdef.h

@@ -132,6 +132,7 @@ typedef enum EOperatorType {
   OP_TYPE_MOD,
   // unary arithmetic operator
   OP_TYPE_MINUS,
+  OP_TYPE_ASSIGN,
 
   // bit operator
   OP_TYPE_BIT_AND,
@@ -233,6 +234,7 @@ typedef enum ELogicConditionType {
 #define TSDB_MAX_TAG_CONDITIONS 1024
 
 #define TSDB_MAX_JSON_TAG_LEN 16384
+#define TSDB_MAX_JSON_KEY_LEN 256
 
 #define TSDB_AUTH_LEN          16
 #define TSDB_PASSWORD_LEN      32

include/util/tencode.h

@@ -39,7 +39,7 @@ typedef struct {
 } SEncoder;
 
 typedef struct {
-  const uint8_t* data;
+  uint8_t*      data;
   uint32_t      size;
   uint32_t      pos;
   SCoderMem*    mList;
@@ -82,7 +82,7 @@ typedef struct {
   do {                               \
     SEncoder coder = {0};            \
     tEncoderInit(&coder, NULL, 0);   \
-    if ((E)(&coder, S) == 0) {       \
+    if ((E)(&coder, S) >= 0) {       \
       SIZE = coder.pos;              \
       RET = 0;                       \
     } else {                         \
@@ -120,7 +120,7 @@ static int32_t tEncodeCStrWithLen(SEncoder* pCoder, const char* val, uint32_t le
 static int32_t tEncodeCStr(SEncoder* pCoder, const char* val);
 
 /* ------------------------ DECODE ------------------------ */
-void           tDecoderInit(SDecoder* pCoder, const uint8_t* data, uint32_t size);
+void           tDecoderInit(SDecoder* pCoder, uint8_t* data, uint32_t size);
 void           tDecoderClear(SDecoder* SDecoder);
 int32_t        tStartDecode(SDecoder* pCoder);
 void           tEndDecode(SDecoder* pCoder);
@@ -141,9 +141,9 @@ static int32_t tDecodeU64v(SDecoder* pCoder, uint64_t* val);
 static int32_t tDecodeI64v(SDecoder* pCoder, int64_t* val);
 static int32_t tDecodeFloat(SDecoder* pCoder, float* val);
 static int32_t tDecodeDouble(SDecoder* pCoder, double* val);
-static int32_t tDecodeBinary(SDecoder* pCoder, const uint8_t** val, uint32_t* len);
-static int32_t tDecodeCStrAndLen(SDecoder* pCoder, const char** val, uint32_t* len);
-static int32_t tDecodeCStr(SDecoder* pCoder, const char** val);
+static int32_t tDecodeBinary(SDecoder* pCoder, uint8_t** val, uint32_t* len);
+static int32_t tDecodeCStrAndLen(SDecoder* pCoder, char** val, uint32_t* len);
+static int32_t tDecodeCStr(SDecoder* pCoder, char** val);
 static int32_t tDecodeCStrTo(SDecoder* pCoder, char* val);
 
 /* ------------------------ IMPL ------------------------ */
@@ -317,7 +317,7 @@ static FORCE_INLINE int32_t tDecodeI16v(SDecoder* pCoder, int16_t* val) {
   if (tDecodeU16v(pCoder, &tval) < 0) {
     return -1;
   }
-  *val = ZIGZAGD(int16_t, tval);
+  if (val) *val = ZIGZAGD(int16_t, tval);
   return 0;
 }
 
@@ -331,7 +331,7 @@ static FORCE_INLINE int32_t tDecodeI32v(SDecoder* pCoder, int32_t* val) {
   if (tDecodeU32v(pCoder, &tval) < 0) {
     return -1;
   }
-  *val = ZIGZAGD(int32_t, tval);
+  if (val) *val = ZIGZAGD(int32_t, tval);
   return 0;
 }
 
@@ -345,7 +345,7 @@ static FORCE_INLINE int32_t tDecodeI64v(SDecoder* pCoder, int64_t* val) {
   if (tDecodeU64v(pCoder, &tval) < 0) {
     return -1;
   }
-  *val = ZIGZAGD(int64_t, tval);
+  if (val) *val = ZIGZAGD(int64_t, tval);
   return 0;
 }
 
@@ -377,7 +377,7 @@ static FORCE_INLINE int32_t tDecodeDouble(SDecoder* pCoder, double* val) {
   return 0;
 }
 
-static FORCE_INLINE int32_t tDecodeBinary(SDecoder* pCoder, const uint8_t** val, uint32_t* len) {
+static FORCE_INLINE int32_t tDecodeBinary(SDecoder* pCoder, uint8_t** val, uint32_t* len) {
   if (tDecodeU32v(pCoder, len) < 0) return -1;
 
   if (TD_CODER_CHECK_CAPACITY_FAILED(pCoder, *len)) return -1;
@@ -389,19 +389,19 @@ static FORCE_INLINE int32_t tDecodeBinary(SDecoder* pCoder, const uint8_t** val,
   return 0;
 }
 
-static FORCE_INLINE int32_t tDecodeCStrAndLen(SDecoder* pCoder, const char** val, uint32_t* len) {
-  if (tDecodeBinary(pCoder, (const uint8_t**)val, len) < 0) return -1;
+static FORCE_INLINE int32_t tDecodeCStrAndLen(SDecoder* pCoder, char** val, uint32_t* len) {
+  if (tDecodeBinary(pCoder, (uint8_t**)val, len) < 0) return -1;
   (*len) -= 1;
   return 0;
 }
 
-static FORCE_INLINE int32_t tDecodeCStr(SDecoder* pCoder, const char** val) {
+static FORCE_INLINE int32_t tDecodeCStr(SDecoder* pCoder, char** val) {
   uint32_t len;
   return tDecodeCStrAndLen(pCoder, val, &len);
 }
 
 static int32_t tDecodeCStrTo(SDecoder* pCoder, char* val) {
-  const char* pStr;
+  char*    pStr;
   uint32_t len;
   if (tDecodeCStrAndLen(pCoder, &pStr, &len) < 0) return -1;
 

source/client/inc/clientInt.h

@@ -121,7 +121,7 @@ struct SAppInstInfo {
   SCorEpSet        mgmtEp;
   SInstanceSummary summary;
   SList*           pConnList;  // STscObj linked list
-  int64_t          clusterId;
+  uint64_t         clusterId;
   void*            pTransporter;
   SAppHbMgr*       pAppHbMgr;
 };
@@ -286,6 +286,8 @@ void initMsgHandleFp();
 TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
                             uint16_t port, int connType);
 
+SRequestObj* launchQuery(STscObj* pTscObj, const char* sql, int sqlLen);
+
 int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtCallback* pStmtCb);
 
 int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList);

source/client/src/clientImpl.c

@@ -866,8 +866,7 @@ static char* parseTagDatatoJson(void* p) {
     if (j == 0) {
       if (*val == TSDB_DATA_TYPE_NULL) {
         string = taosMemoryCalloc(1, 8);
-        sprintf(varDataVal(string), "%s", TSDB_DATA_NULL_STR_L);
-        varDataSetLen(string, strlen(varDataVal(string)));
+        sprintf(string, "%s", TSDB_DATA_NULL_STR_L);
         goto end;
       }
       continue;
@@ -1003,7 +1002,7 @@ static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int
               length = 0;
             }
             varDataSetLen(dst, length + CHAR_BYTES * 2);
-            *(char*)(varDataVal(dst), length + CHAR_BYTES) = '\"';
+            *(char*)POINTER_SHIFT(varDataVal(dst), length + CHAR_BYTES) = '\"';
           } else if (jsonInnerType == TSDB_DATA_TYPE_DOUBLE) {
             double jsonVd = *(double*)(jsonInnerData);
             sprintf(varDataVal(dst), "%.9lf", jsonVd);

source/client/src/clientMain.c

@@ -565,10 +565,32 @@ const char *taos_get_server_info(TAOS *taos) {
 
 void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param) {
   if (taos == NULL || sql == NULL) {
-    // todo directly call fp
+    fp(param, NULL, TSDB_CODE_INVALID_PARA);
+    return;
+  }
+
+  SRequestObj* pRequest = NULL;
+  int32_t      retryNum = 0;
+  int32_t      code = 0;
+
+  size_t sqlLen = strlen(sql);
+
+  while (retryNum++ < REQUEST_MAX_TRY_TIMES) {
+    pRequest = launchQuery(taos, sql, sqlLen);
+    if (pRequest == NULL || TSDB_CODE_SUCCESS == pRequest->code || !NEED_CLIENT_HANDLE_ERROR(pRequest->code)) {
+      break;
+    }
+
+    code = refreshMeta(taos, pRequest);
+    if (code) {
+      pRequest->code = code;
+      break;
+    }
+
+    destroyRequest(pRequest);
   }
 
-  taos_query_l(taos, sql, (int32_t)strlen(sql));
+  fp(param, pRequest, code);
 }
 
 void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {

source/client/src/clientMsgHandler.c

@@ -125,10 +125,10 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
     struct SCatalog* pCatalog = NULL;
 
     if (usedbRsp.vgVersion >= 0) {
-      int32_t code1 = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
+      uint64_t clusterId = pRequest->pTscObj->pAppInfo->clusterId;
+      int32_t code1 = catalogGetHandle(clusterId, &pCatalog);
       if (code1 != TSDB_CODE_SUCCESS) {
-        tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pRequest->pTscObj->pAppInfo->clusterId,
-                tstrerror(code1));
+        tscWarn("0x%" PRIx64 "catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pRequest->requestId, clusterId, tstrerror(code1));
       } else {
         catalogRemoveDB(pCatalog, usedbRsp.db, usedbRsp.uid);
       }
@@ -158,7 +158,7 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
     if (output.dbVgroup) taosHashCleanup(output.dbVgroup->vgHash);
     taosMemoryFreeClear(output.dbVgroup);
 
-    tscError("failed to build use db output since %s", terrstr());
+    tscError("0x%" PRIx64" failed to build use db output since %s", pRequest->requestId, terrstr());
   } else if (output.dbVgroup) {
     struct SCatalog* pCatalog = NULL;
 

source/common/src/tdatablock.c

@@ -1538,7 +1538,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
   int32_t     msgLen = sizeof(SSubmitReq);
   int32_t     numOfBlks = 0;
   SRowBuilder rb = {0};
-  tdSRowInit(&rb, 0);  // TODO: use the latest version
+  tdSRowInit(&rb, pTSchema->version);  // TODO: use the latest version
 
   for (int32_t i = 0; i < sz; ++i) {
     SSDataBlock* pDataBlock = taosArrayGet(pDataBlocks, i);

source/common/src/tmsg.c

@@ -28,7 +28,7 @@
 #undef TD_MSG_SEG_CODE_
 #include "tmsgdef.h"
 
-int32_t tInitSubmitMsgIter(const SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
+int32_t tInitSubmitMsgIter(SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
   if (pMsg == NULL) {
     terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
     return -1;
@@ -102,7 +102,7 @@ STSRow *tGetSubmitBlkNext(SSubmitBlkIter *pIter) {
   }
 }
 
-int32_t tPrintFixedSchemaSubmitReq(const SSubmitReq *pReq, STSchema *pTschema) {
+int32_t tPrintFixedSchemaSubmitReq(SSubmitReq *pReq, STSchema *pTschema) {
   SSubmitMsgIter msgIter = {0};
   if (tInitSubmitMsgIter(pReq, &msgIter) < 0) return -1;
   while (true) {
@@ -3817,7 +3817,7 @@ int tDecodeSVCreateStbReq(SDecoder *pCoder, SVCreateStbReq *pReq) {
 
 STSchema *tdGetSTSChemaFromSSChema(SSchema **pSchema, int32_t nCols) {
   STSchemaBuilder schemaBuilder = {0};
-  if (tdInitTSchemaBuilder(&schemaBuilder, 0) < 0) {
+  if (tdInitTSchemaBuilder(&schemaBuilder, 1) < 0) {
     return NULL;
   }
 

source/common/src/trow.c

@@ -924,7 +924,7 @@ void tdSRowPrint(STSRow *row, STSchema *pSchema, const char *tag) {
   STSRowIter iter = {0};
   tdSTSRowIterInit(&iter, pSchema);
   tdSTSRowIterReset(&iter, row);
-  printf("%s >>>", tag);
+  printf("%s >>>type:%d,sver:%d ", tag, (int32_t)TD_ROW_TYPE(row), (int32_t)TD_ROW_SVER(row));
   for (int i = 0; i < pSchema->numOfCols; ++i) {
     STColumn *stCol = pSchema->columns + i;
     SCellVal  sVal = {255, NULL};

source/dnode/mgmt/mgmt_mnode/inc/mmInt.h

@@ -37,6 +37,9 @@ typedef struct SMnodeMgmt {
   SReplica       replicas[TSDB_MAX_REPLICA];
   int8_t         replica;
   int8_t         selfIndex;
+  bool           stopped;
+  int32_t        refCount;
+  TdThreadRwlock lock;
 } SMnodeMgmt;
 
 // mmFile.c
@@ -45,6 +48,8 @@ int32_t mmWriteFile(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pMsg, bool deployed);
 
 // mmInt.c
 int32_t mmAlter(SMnodeMgmt *pMgmt, SDAlterMnodeReq *pMsg);
+int32_t mmAcquire(SMnodeMgmt *pMgmt);
+void    mmRelease(SMnodeMgmt *pMgmt);
 
 // mmHandle.c
 SArray *mmGetMsgHandles();

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

@@ -237,6 +237,16 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_VNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT_VNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
 
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_TIMEOUT, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING_REPLY, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_CLIENT_REQUEST, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_CLIENT_REQUEST_REPLY, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_REQUEST_VOTE, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_REQUEST_VOTE_REPLY, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_APPEND_ENTRIES, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_APPEND_ENTRIES_REPLY, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
+
   code = 0;
 
 _OVER:

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

@@ -110,6 +110,7 @@ static void mmClose(SMnodeMgmt *pMgmt) {
   if (pMgmt->pMnode != NULL) {
     mmStopWorker(pMgmt);
     mndClose(pMgmt->pMnode);
+    taosThreadRwlockDestroy(&pMgmt->lock);
     pMgmt->pMnode = NULL;
   }
 
@@ -122,6 +123,11 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
     return -1;
   }
 
+  if (syncInit() != 0) {
+    dError("failed to init sync since %s", terrstr());
+    return -1;
+  }
+
   SMnodeMgmt *pMgmt = taosMemoryCalloc(1, sizeof(SMnodeMgmt));
   if (pMgmt == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -137,6 +143,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
   pMgmt->msgCb.queueFps[WRITE_QUEUE] = (PutToQueueFp)mmPutRpcMsgToWriteQueue;
   pMgmt->msgCb.queueFps[SYNC_QUEUE] = (PutToQueueFp)mmPutRpcMsgToSyncQueue;
   pMgmt->msgCb.mgmt = pMgmt;
+  taosThreadRwlockInit(&pMgmt->lock, NULL);
 
   bool deployed = false;
   if (mmReadFile(pMgmt, &deployed) != 0) {
@@ -206,3 +213,22 @@ SMgmtFunc mmGetMgmtFunc() {
 
   return mgmtFunc;
 }
+
+int32_t mmAcquire(SMnodeMgmt *pMgmt) {
+  int32_t code = 0;
+
+  taosThreadRwlockRdlock(&pMgmt->lock);
+  if (pMgmt->stopped) {
+    code = -1;
+  } else {
+    atomic_add_fetch_32(&pMgmt->refCount, 1);
+  }
+  taosThreadRwlockUnlock(&pMgmt->lock);
+  return code;
+}
+
+void mmRelease(SMnodeMgmt *pMgmt) {
+  taosThreadRwlockRdlock(&pMgmt->lock);
+  atomic_sub_fetch_32(&pMgmt->refCount, 1);
+  taosThreadRwlockUnlock(&pMgmt->lock);
+}
\ No newline at end of file

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

@@ -56,6 +56,12 @@ static void mmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   taosFreeQitem(pMsg);
 }
 
+static void mmProcessSyncQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
+  SMnodeMgmt *pMgmt = pInfo->ahandle;
+  pMsg->info.node = pMgmt->pMnode;
+  mndProcessSyncMsg(pMsg);
+}
+
 static int32_t mmPutNodeMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pMsg) {
   dTrace("msg:%p, put into worker %s, type:%s", pMsg, pWorker->name, TMSG_INFO(pMsg->msgType));
   taosWriteQitem(pWorker->queue, pMsg);
@@ -105,7 +111,10 @@ int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 }
 
 int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return mmPutRpcMsgToWorker(&pMgmt->syncWorker, pMsg);
+  if (mmAcquire(pMgmt) != 0) return -1;
+  int32_t code = mmPutRpcMsgToWorker(&pMgmt->syncWorker, pMsg);
+  mmRelease(pMgmt);
+  return code;
 }
 
 int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
@@ -149,7 +158,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
       .min = 1,
       .max = 1,
       .name = "mnode-sync",
-      .fp = (FItem)mmProcessQueue,
+      .fp = (FItem)mmProcessSyncQueue,
       .param = pMgmt,
   };
   if (tSingleWorkerInit(&pMgmt->syncWorker, &sCfg) != 0) {
@@ -174,6 +183,11 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
 }
 
 void mmStopWorker(SMnodeMgmt *pMgmt) {
+  taosThreadRwlockWrlock(&pMgmt->lock);
+  pMgmt->stopped = 1;
+  taosThreadRwlockUnlock(&pMgmt->lock);
+  while (pMgmt->refCount > 0) taosMsleep(10);
+
   tSingleWorkerCleanup(&pMgmt->monitorWorker);
   tSingleWorkerCleanup(&pMgmt->queryWorker);
   tSingleWorkerCleanup(&pMgmt->readWorker);

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

@@ -113,6 +113,8 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     SRpcMsg *pMsg = *(SRpcMsg **)taosArrayGet(pArray, i);
     SRpcMsg  rsp = {.info = pMsg->info};
 
+    vnodePreprocessReq(pVnode->pImpl, pMsg);
+
     int32_t ret = syncPropose(vnodeGetSyncHandle(pVnode->pImpl), pMsg, false);
     if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) {
       dTrace("msg:%p, is redirect since not leader, vgId:%d ", pMsg, pVnode->vgId);

source/dnode/mgmt/test/mnode/CMakeLists.txt

@@ -4,7 +4,7 @@ target_link_libraries(
     dmnodeTest sut
 )
 
-add_test(
-    NAME dmnodeTest
-    COMMAND dmnodeTest
-)
+#add_test(
+#    NAME dmnodeTest
+#    COMMAND dmnodeTest
+#)

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

@@ -19,6 +19,7 @@
 #include "mndDef.h"
 
 #include "sdb.h"
+#include "syncTools.h"
 #include "tcache.h"
 #include "tdatablock.h"
 #include "tglobal.h"
@@ -31,12 +32,14 @@
 extern "C" {
 #endif
 
+// clang-format off
 #define mFatal(...) { if (mDebugFlag & DEBUG_FATAL) { taosPrintLog("MND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}
 #define mError(...) { if (mDebugFlag & DEBUG_ERROR) { taosPrintLog("MND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}
 #define mWarn(...)  { if (mDebugFlag & DEBUG_WARN)  { taosPrintLog("MND WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}
 #define mInfo(...)  { if (mDebugFlag & DEBUG_INFO)  { taosPrintLog("MND ", DEBUG_INFO, 255, __VA_ARGS__); }}
 #define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", DEBUG_DEBUG, mDebugFlag, __VA_ARGS__); }}
 #define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", DEBUG_TRACE, mDebugFlag, __VA_ARGS__); }}
+// clang-format on
 
 #define SYSTABLE_SCH_TABLE_NAME_LEN ((TSDB_TABLE_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
 #define SYSTABLE_SCH_DB_NAME_LEN    ((TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
@@ -72,10 +75,11 @@ typedef struct {
 } STelemMgmt;
 
 typedef struct {
+  SWal      *pWal;
   int32_t    errCode;
+  bool       restored;
   sem_t      syncSem;
-  SWal      *pWal;
-  SSyncNode *pSyncNode;
+  int64_t    sync;
   ESyncState state;
 } SSyncMgmt;
 

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

@@ -26,6 +26,8 @@ int32_t mndInitSync(SMnode *pMnode);
 void    mndCleanupSync(SMnode *pMnode);
 bool    mndIsMaster(SMnode *pMnode);
 int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw);
+void    mndSyncStart(SMnode *pMnode);
+void    mndSyncStop(SMnode *pMnode);
 
 #ifdef __cplusplus
 }

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

@@ -35,7 +35,7 @@ int32_t mndDropTopicByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
 
 const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]);
 
-int32_t mndSetTopicRedoLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic);
+int32_t mndSetTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic);
 
 #ifdef __cplusplus
 }

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

@@ -419,7 +419,7 @@ static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
     SMqTopicObj topicObj = {0};
     memcpy(&topicObj, pTopic, sizeof(SMqTopicObj));
     topicObj.refConsumerCnt = pTopic->refConsumerCnt + 1;
-    if (mndSetTopicRedoLogs(pMnode, pTrans, &topicObj) != 0) goto SUBSCRIBE_OVER;
+    if (mndSetTopicCommitLogs(pMnode, pTrans, &topicObj) != 0) goto SUBSCRIBE_OVER;
 
     mndReleaseTopic(pMnode, pTopic);
   }

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

@@ -448,13 +448,13 @@ static int32_t mndCreateDnode(SMnode *pMnode, SRpcMsg *pReq, SCreateDnodeReq *pC
   }
   mDebug("trans:%d, used to create dnode:%s", pTrans->id, dnodeObj.ep);
 
-  SSdbRaw *pRedoRaw = mndDnodeActionEncode(&dnodeObj);
-  if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
-    mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+  SSdbRaw *pCommitRaw = mndDnodeActionEncode(&dnodeObj);
+  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
-  sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
@@ -524,13 +524,13 @@ static int32_t mndDropDnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode) {
   }
   mDebug("trans:%d, used to drop dnode:%d", pTrans->id, pDnode->id);
 
-  SSdbRaw *pRedoRaw = mndDnodeActionEncode(pDnode);
-  if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
-    mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+  SSdbRaw *pCommitRaw = mndDnodeActionEncode(pDnode);
+  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
-  sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPED);
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());

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

@@ -153,6 +153,7 @@ int32_t mndCreateOffsets(STrans *pTrans, const char *cgroup, const char *topicNa
       return -1;
     }
     sdbSetRawStatus(pOffsetRaw, SDB_STATUS_READY);
+    // commit log or redo log?
     if (mndTransAppendRedolog(pTrans, pOffsetRaw) < 0) {
       return -1;
     }
@@ -188,7 +189,7 @@ static int32_t mndProcessCommitOffsetReq(SRpcMsg *pMsg) {
     pOffsetObj->offset = pOffset->offset;
     SSdbRaw *pOffsetRaw = mndOffsetActionEncode(pOffsetObj);
     sdbSetRawStatus(pOffsetRaw, SDB_STATUS_READY);
-    mndTransAppendRedolog(pTrans, pOffsetRaw);
+    mndTransAppendCommitlog(pTrans, pOffsetRaw);
     if (create) {
       taosMemoryFree(pOffsetObj);
     } else {

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

@@ -743,9 +743,7 @@ static int32_t mndCreateStb(SMnode *pMnode, SRpcMsg *pReq, SMCreateStbReq *pCrea
 
   mDebug("trans:%d, used to create stb:%s", pTrans->id, pCreate->name);
 
-  if (mndBuildStbFromReq(pMnode, &stbObj, pCreate, pDb) != 0) {
-    goto _OVER;
-  }
+  if (mndBuildStbFromReq(pMnode, &stbObj, pCreate, pDb) != 0) goto _OVER;
 
   if (mndAddStbToTrans(pMnode, pTrans, pDb, &stbObj) < 0) goto _OVER;
 

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

@@ -279,13 +279,13 @@ int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast
   }
   mDebug("trans:%d, used to create stream:%s", pTrans->id, pStream->name);
 
-  SSdbRaw *pRedoRaw = mndStreamActionEncode(pStream);
-  if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
-    mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+  SSdbRaw *pCommitRaw = mndStreamActionEncode(pStream);
+  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
-  sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
 
   return 0;
 }

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

@@ -479,7 +479,7 @@ static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOu
       SMqTopicObj topicObj = {0};
       memcpy(&topicObj, pTopic, sizeof(SMqTopicObj));
       topicObj.refConsumerCnt = pTopic->refConsumerCnt - consumerNum;
-      if (mndSetTopicRedoLogs(pMnode, pTrans, &topicObj) != 0) goto REB_FAIL;
+      if (mndSetTopicCommitLogs(pMnode, pTrans, &topicObj) != 0) goto REB_FAIL;
     }
   }
 

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

@@ -17,178 +17,152 @@
 #include "mndSync.h"
 #include "mndTrans.h"
 
-static int32_t mndInitWal(SMnode *pMnode) {
-  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
-
-  char path[PATH_MAX] = {0};
-  snprintf(path, sizeof(path), "%s%swal", pMnode->path, TD_DIRSEP);
-  SWalCfg cfg = {
-      .vgId = 1,
-      .fsyncPeriod = 0,
-      .rollPeriod = -1,
-      .segSize = -1,
-      .retentionPeriod = -1,
-      .retentionSize = -1,
-      .level = TAOS_WAL_FSYNC,
-  };
-  pMgmt->pWal = walOpen(path, &cfg);
-  if (pMgmt->pWal == NULL) return -1;
-
-  return 0;
-}
-
-static void mndCloseWal(SMnode *pMnode) {
-  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
-  if (pMgmt->pWal != NULL) {
-    walClose(pMgmt->pWal);
-    pMgmt->pWal = NULL;
+int32_t mndSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) { 
+  int32_t code = tmsgPutToQueue(msgcb, SYNC_QUEUE, pMsg); 
+  if (code != 0) {
+    rpcFreeCont(pMsg->pCont);
   }
+  return code;
 }
 
-static int32_t mndRestoreWal(SMnode *pMnode) {
-  SWal   *pWal = pMnode->syncMgmt.pWal;
-  SSdb   *pSdb = pMnode->pSdb;
-  int64_t lastSdbVer = sdbUpdateVer(pSdb, 0);
-  int32_t code = -1;
-
-  SWalReadHandle *pHandle = walOpenReadHandle(pWal);
-  if (pHandle == NULL) return -1;
+int32_t mndSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) { return tmsgSendReq(pEpSet, pMsg); }
 
-  int64_t first = walGetFirstVer(pWal);
-  int64_t last = walGetLastVer(pWal);
-  mDebug("start to restore wal, sdbver:%" PRId64 ", first:%" PRId64 " last:%" PRId64, lastSdbVer, first, last);
-
-  first = TMAX(lastSdbVer + 1, first);
-  for (int64_t ver = first; ver >= 0 && ver <= last; ++ver) {
-    if (walReadWithHandle(pHandle, ver) < 0) {
-      mError("ver:%" PRId64 ", failed to read from wal since %s", ver, terrstr());
-      goto _OVER;
-    }
-
-    SWalHead *pHead = pHandle->pHead;
-    int64_t   sdbVer = sdbUpdateVer(pSdb, 0);
-    if (sdbVer + 1 != ver) {
-      terrno = TSDB_CODE_SDB_INVALID_WAl_VER;
-      mError("ver:%" PRId64 ", failed to write to sdb, since inconsistent with sdbver:%" PRId64, ver, sdbVer);
-      goto _OVER;
-    }
-
-    mTrace("ver:%" PRId64 ", will be restored, content:%p", ver, pHead->head.body);
-    if (sdbWriteWithoutFree(pSdb, (void *)pHead->head.body) < 0) {
-      mError("ver:%" PRId64 ", failed to write to sdb since %s", ver, terrstr());
-      goto _OVER;
-    }
+void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
+  SMnode    *pMnode = pFsm->data;
+  SSdb      *pSdb = pMnode->pSdb;
+  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
+  SSdbRaw   *pRaw = pMsg->pCont;
 
-    sdbUpdateVer(pSdb, 1);
-    mDebug("ver:%" PRId64 ", is restored", ver);
+  mTrace("ver:%" PRId64 ", apply raw:%p to sdb, role:%s", cbMeta.index, pRaw, syncStr(cbMeta.state));
+  sdbWriteWithoutFree(pSdb, pRaw);
+  sdbSetApplyIndex(pSdb, cbMeta.index);
+  sdbSetApplyTerm(pSdb, cbMeta.term);
+  if (cbMeta.state == TAOS_SYNC_STATE_LEADER) {
+    tsem_post(&pMgmt->syncSem);
   }
+}
 
-  int64_t sdbVer = sdbUpdateVer(pSdb, 0);
-  mDebug("restore wal finished, sdbver:%" PRId64, sdbVer);
+int32_t mndSyncGetSnapshot(struct SSyncFSM *pFsm, SSnapshot *pSnapshot) {
+  SMnode *pMnode = pFsm->data;
+  pSnapshot->lastApplyIndex = sdbGetApplyIndex(pMnode->pSdb);
+  pSnapshot->lastApplyTerm = sdbGetApplyTerm(pMnode->pSdb);
+  return 0;
+}
 
+void mndRestoreFinish(struct SSyncFSM *pFsm) {
+  SMnode *pMnode = pFsm->data;
   mndTransPullup(pMnode);
-  sdbVer = sdbUpdateVer(pSdb, 0);
-  mDebug("pullup trans finished, sdbver:%" PRId64, sdbVer);
-
-  if (sdbVer != lastSdbVer) {
-    mInfo("sdb restored from %" PRId64 " to %" PRId64 ", write file", lastSdbVer, sdbVer);
-    if (sdbWriteFile(pSdb) != 0) {
-      goto _OVER;
-    }
-
-    if (walCommit(pWal, sdbVer) != 0) {
-      goto _OVER;
-    }
-
-    if (walBeginSnapshot(pWal, sdbVer) < 0) {
-      goto _OVER;
-    }
-
-    if (walEndSnapshot(pWal) < 0) {
-      goto _OVER;
-    }
-  }
-
-  code = 0;
+  pMnode->syncMgmt.restored = true;
+}
 
-_OVER:
-  walCloseReadHandle(pHandle);
-  return code;
+SSyncFSM *mndSyncMakeFsm(SMnode *pMnode) {
+  SSyncFSM *pFsm = taosMemoryCalloc(1, sizeof(SSyncFSM));
+  pFsm->data = pMnode;
+  pFsm->FpCommitCb = mndSyncCommitMsg;
+  pFsm->FpPreCommitCb = NULL;
+  pFsm->FpRollBackCb = NULL;
+  pFsm->FpGetSnapshot = mndSyncGetSnapshot;
+  pFsm->FpRestoreFinish = mndRestoreFinish;
+  pFsm->FpRestoreSnapshot = NULL;
+  return pFsm;
 }
 
 int32_t mndInitSync(SMnode *pMnode) {
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
-  tsem_init(&pMgmt->syncSem, 0, 0);
 
-  if (mndInitWal(pMnode) < 0) {
+  char path[PATH_MAX + 20] = {0};
+  snprintf(path, sizeof(path), "%s%swal", pMnode->path, TD_DIRSEP);
+  SWalCfg cfg = {
+      .vgId = 1,
+      .fsyncPeriod = 0,
+      .rollPeriod = -1,
+      .segSize = -1,
+      .retentionPeriod = -1,
+      .retentionSize = -1,
+      .level = TAOS_WAL_FSYNC,
+  };
+
+  pMgmt->pWal = walOpen(path, &cfg);
+  if (pMgmt->pWal == NULL) {
     mError("failed to open wal since %s", terrstr());
     return -1;
   }
 
-  if (mndRestoreWal(pMnode) < 0) {
-    mError("failed to restore wal since %s", terrstr());
-    return -1;
+  SSyncInfo syncInfo = {.vgId = 1, .FpSendMsg = mndSyncSendMsg, .FpEqMsg = mndSyncEqMsg};
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s%ssync", pMnode->path, TD_DIRSEP);
+  syncInfo.pWal = pMgmt->pWal;
+  syncInfo.pFsm = mndSyncMakeFsm(pMnode);
+
+  SSyncCfg *pCfg = &syncInfo.syncCfg;
+  pCfg->replicaNum = pMnode->replica;
+  pCfg->myIndex = pMnode->selfIndex;
+  for (int32_t i = 0; i < pMnode->replica; ++i) {
+    SNodeInfo *pNode = &pCfg->nodeInfo[i];
+    tstrncpy(pNode->nodeFqdn, pMnode->replicas[i].fqdn, sizeof(pNode->nodeFqdn));
+    pNode->nodePort = pMnode->replicas[i].port;
   }
 
-  if (pMnode->selfId == 1) {
-    pMgmt->state = TAOS_SYNC_STATE_LEADER;
+  tsem_init(&pMgmt->syncSem, 0, 0);
+  pMgmt->sync = syncOpen(&syncInfo);
+  if (pMgmt->sync <= 0) {
+    mError("failed to open sync since %s", terrstr());
+    return -1;
   }
-  pMgmt->pSyncNode = NULL;
+
+  mDebug("mnode sync is opened, id:%" PRId64, pMgmt->sync);
   return 0;
 }
 
 void mndCleanupSync(SMnode *pMnode) {
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
-  tsem_destroy(&pMgmt->syncSem);
-  mndCloseWal(pMnode);
-}
+  syncStop(pMgmt->sync);
+  mDebug("sync:%" PRId64 " is stopped", pMgmt->sync);
 
-static int32_t mndSyncApplyCb(struct SSyncFSM *fsm, SyncIndex index, const SSyncBuffer *buf, void *pData) {
-  SMnode    *pMnode = pData;
-  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
-
-  pMgmt->errCode = 0;
-  tsem_post(&pMgmt->syncSem);
+  tsem_destroy(&pMgmt->syncSem);
+  if (pMgmt->pWal != NULL) {
+    walClose(pMgmt->pWal);
+  }
 
-  return 0;
+  memset(pMgmt, 0, sizeof(SSyncMgmt));
 }
 
 int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw) {
-  SWal *pWal = pMnode->syncMgmt.pWal;
-  SSdb *pSdb = pMnode->pSdb;
-
-  int64_t ver = sdbUpdateVer(pSdb, 1);
-  if (walWrite(pWal, ver, 1, pRaw, sdbGetRawTotalSize(pRaw)) < 0) {
-    sdbUpdateVer(pSdb, -1);
-    mError("ver:%" PRId64 ", failed to write raw:%p to wal since %s", ver, pRaw, terrstr());
-    return -1;
-  }
-
-  mTrace("ver:%" PRId64 ", write to wal, raw:%p", ver, pRaw);
-  walCommit(pWal, ver);
-  walFsync(pWal, true);
-
-#if 1
-  return 0;
-#else
-  if (pMnode->replica == 1) return 0;
-
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
   pMgmt->errCode = 0;
 
-  SSyncBuffer buf = {.data = pRaw, .len = sdbGetRawTotalSize(pRaw)};
+  SRpcMsg rsp = {.code = TDMT_MND_APPLY_MSG, .contLen = sdbGetRawTotalSize(pRaw)};
+  rsp.pCont = rpcMallocCont(rsp.contLen);
+  if (rsp.pCont == NULL) return -1;
+  memcpy(rsp.pCont, pRaw, rsp.contLen);
 
-  bool    isWeak = false;
-  int32_t code = syncPropose(pMgmt->pSyncNode, &buf, pMnode, isWeak);
+  const bool isWeak = false;
+  int32_t    code = syncPropose(pMgmt->sync, &rsp, isWeak);
+  if (code == 0) {
+    tsem_wait(&pMgmt->syncSem);
+  } else if (code == TAOS_SYNC_PROPOSE_NOT_LEADER) {
+    terrno = TSDB_CODE_APP_NOT_READY;
+  } else if (code == TAOS_SYNC_PROPOSE_OTHER_ERROR) {
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+  } else {
+    terrno = TSDB_CODE_APP_ERROR;
+  }
 
   if (code != 0) return code;
-
-  tsem_wait(&pMgmt->syncSem);
   return pMgmt->errCode;
-#endif
 }
 
+void mndSyncStart(SMnode *pMnode) {
+  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
+  syncSetMsgCb(pMgmt->sync, &pMnode->msgCb);
+  syncStart(pMgmt->sync);
+  mDebug("sync:%" PRId64 " is started", pMgmt->sync);
+}
+
+void mndSyncStop(SMnode *pMnode) {}
+
 bool mndIsMaster(SMnode *pMnode) {
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
-  return pMgmt->state == TAOS_SYNC_STATE_LEADER;
+  pMgmt->state = syncGetMyRole(pMgmt->sync);
+
+  return (pMgmt->state == TAOS_SYNC_STATE_LEADER) && (pMnode->syncMgmt.restored);
 }

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

@@ -386,14 +386,14 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
   }
   mDebug("trans:%d, used to create topic:%s", pTrans->id, pCreate->name);
 
-  SSdbRaw *pRedoRaw = mndTopicActionEncode(&topicObj);
-  if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
-    mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);
+  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
     taosMemoryFreeClear(topicObj.physicalPlan);
     mndTransDrop(pTrans);
     return -1;
   }
-  sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
@@ -473,13 +473,13 @@ CREATE_TOPIC_OVER:
 }
 
 static int32_t mndDropTopic(SMnode *pMnode, STrans *pTrans, SRpcMsg *pReq, SMqTopicObj *pTopic) {
-  SSdbRaw *pRedoRaw = mndTopicActionEncode(pTopic);
-  if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
-    mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+  SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic);
+  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
-  sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPED);
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
@@ -627,11 +627,11 @@ static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
   return numOfRows;
 }
 
-int32_t mndSetTopicRedoLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic) {
-  SSdbRaw *pRedoRaw = mndTopicActionEncode(pTopic);
-  if (pRedoRaw == NULL) return -1;
-  if (mndTransAppendCommitlog(pTrans, pRedoRaw) != 0) return -1;
-  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY) != 0) return -1;
+int32_t mndSetTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic) {
+  SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic);
+  if (pCommitRaw == NULL) return -1;
+  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
+  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;
 
   return 0;
 }

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

@@ -682,13 +682,6 @@ static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
   }
 
   mDebug("trans:%d, sync finished", pTrans->id);
-
-  code = sdbWrite(pMnode->pSdb, pRaw);
-  if (code != 0) {
-    mError("trans:%d, failed to write sdb since %s", pTrans->id, terrstr());
-    return -1;
-  }
-
   return 0;
 }
 
@@ -768,6 +761,12 @@ int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
     return -1;
   }
 
+  if (taosArrayGetSize(pTrans->commitLogs) <= 0) {
+    terrno = TSDB_CODE_MND_TRANS_CLOG_IS_NULL;
+    mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
+    return -1;
+  }
+
   mDebug("trans:%d, prepare transaction", pTrans->id);
   if (mndTransSync(pMnode, pTrans) != 0) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
@@ -1080,6 +1079,8 @@ static bool mndTransPerformRedoLogStage(SMnode *pMnode, STrans *pTrans) {
 }
 
 static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans) {
+  if (!mndIsMaster(pMnode)) return false;
+
   bool    continueExec = true;
   int32_t code = mndTransExecuteRedoActions(pMnode, pTrans);
 
@@ -1169,6 +1170,8 @@ static bool mndTransPerformUndoLogStage(SMnode *pMnode, STrans *pTrans) {
 }
 
 static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans) {
+  if (!mndIsMaster(pMnode)) return false;
+
   bool    continueExec = true;
   int32_t code = mndTransExecuteUndoActions(pMnode, pTrans);
 
@@ -1350,19 +1353,35 @@ _OVER:
   return code;
 }
 
+static int32_t mndCompareTransId(int32_t *pTransId1, int32_t *pTransId2) { return *pTransId1 >= *pTransId2 ? 1 : 0; }
+
 void mndTransPullup(SMnode *pMnode) {
-  STrans *pTrans = NULL;
-  void   *pIter = NULL;
+  SSdb   *pSdb = pMnode->pSdb;
+  SArray *pArray = taosArrayInit(sdbGetSize(pSdb, SDB_TRANS), sizeof(int32_t));
+  if (pArray == NULL) return;
 
+  void *pIter = NULL;
   while (1) {
+    STrans *pTrans = NULL;
     pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
     if (pIter == NULL) break;
+    taosArrayPush(pArray, &pTrans->id);
+    sdbRelease(pSdb, pTrans);
+  }
 
+  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);
+
+  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
+    int32_t *pTransId = taosArrayGet(pArray, i);
+    STrans  *pTrans = mndAcquireTrans(pMnode, *pTransId);
+    if (pTrans != NULL) {
       mndTransExecute(pMnode, pTrans);
-    sdbRelease(pMnode->pSdb, pTrans);
+    }
+    mndReleaseTrans(pMnode, pTrans);
   }
 
   sdbWriteFile(pMnode->pSdb);
+  taosArrayDestroy(pArray);
 }
 
 static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {

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

@@ -272,13 +272,13 @@ static int32_t mndCreateUser(SMnode *pMnode, char *acct, SCreateUserReq *pCreate
   }
   mDebug("trans:%d, used to create user:%s", pTrans->id, pCreate->user);
 
-  SSdbRaw *pRedoRaw = mndUserActionEncode(&userObj);
-  if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
-    mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+  SSdbRaw *pCommitRaw = mndUserActionEncode(&userObj);
+  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    mError("trans:%d, failed to commit redo log since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
-  sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
@@ -352,13 +352,13 @@ static int32_t mndAlterUser(SMnode *pMnode, SUserObj *pOld, SUserObj *pNew, SRpc
   }
   mDebug("trans:%d, used to alter user:%s", pTrans->id, pOld->user);
 
-  SSdbRaw *pRedoRaw = mndUserActionEncode(pNew);
-  if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
-    mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+  SSdbRaw *pCommitRaw = mndUserActionEncode(pNew);
+  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
-  sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
@@ -559,13 +559,13 @@ static int32_t mndDropUser(SMnode *pMnode, SRpcMsg *pReq, SUserObj *pUser) {
   }
   mDebug("trans:%d, used to drop user:%s", pTrans->id, pUser->user);
 
-  SSdbRaw *pRedoRaw = mndUserActionEncode(pUser);
-  if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
-    mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+  SSdbRaw *pCommitRaw = mndUserActionEncode(pUser);
+  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
-  sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPED);
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());

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

@@ -86,7 +86,6 @@ static void *mndThreadFp(void *param) {
     lastTime++;
     taosMsleep(100);
     if (pMnode->stopped) break;
-    if (!mndIsMaster(pMnode)) continue;
 
     if (lastTime % (tsTransPullupInterval * 10) == 0) {
       mndPullupTrans(pMnode);
@@ -336,9 +335,107 @@ int32_t mndAlter(SMnode *pMnode, const SMnodeOpt *pOption) {
   return 0;
 }
 
-int32_t mndStart(SMnode *pMnode) { return mndInitTimer(pMnode); }
+int32_t mndStart(SMnode *pMnode) {
+  mndSyncStart(pMnode);
+  return mndInitTimer(pMnode);
+}
+
+void mndStop(SMnode *pMnode) {
+  mndSyncStop(pMnode);
+  return mndCleanupTimer(pMnode);
+}
+
+int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
+  SMnode *pMnode = pMsg->info.node;
+  void   *ahandle = pMsg->info.ahandle;
+  int32_t ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
+
+  if (syncEnvIsStart()) {
+    SSyncNode *pSyncNode = syncNodeAcquire(pMnode->syncMgmt.sync);
+    assert(pSyncNode != NULL);
+
+    ESyncState state = syncGetMyRole(pMnode->syncMgmt.sync);
+    SyncTerm   currentTerm = syncGetMyTerm(pMnode->syncMgmt.sync);
+
+    SMsgHead *pHead = pMsg->pCont;
+
+    char  logBuf[512];
+    char *syncNodeStr = sync2SimpleStr(pMnode->syncMgmt.sync);
+    snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr);
+    syncRpcMsgLog2(logBuf, pMsg);
+    taosMemoryFree(syncNodeStr);
+
+    SRpcMsg *pRpcMsg = pMsg;
+
+    if (pRpcMsg->msgType == TDMT_VND_SYNC_TIMEOUT) {
+      SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
+
+      ret = syncNodeOnTimeoutCb(pSyncNode, pSyncMsg);
+      syncTimeoutDestroy(pSyncMsg);
+
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING) {
+      SyncPing *pSyncMsg = syncPingFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
+
+      ret = syncNodeOnPingCb(pSyncNode, pSyncMsg);
+      syncPingDestroy(pSyncMsg);
+
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING_REPLY) {
+      SyncPingReply *pSyncMsg = syncPingReplyFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
+
+      ret = syncNodeOnPingReplyCb(pSyncNode, pSyncMsg);
+      syncPingReplyDestroy(pSyncMsg);
 
-void mndStop(SMnode *pMnode) { return mndCleanupTimer(pMnode); }
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_CLIENT_REQUEST) {
+      SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
+
+      ret = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg);
+      syncClientRequestDestroy(pSyncMsg);
+
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE) {
+      SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
+
+      ret = syncNodeOnRequestVoteCb(pSyncNode, pSyncMsg);
+      syncRequestVoteDestroy(pSyncMsg);
+
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE_REPLY) {
+      SyncRequestVoteReply *pSyncMsg = syncRequestVoteReplyFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
+
+      ret = syncNodeOnRequestVoteReplyCb(pSyncNode, pSyncMsg);
+      syncRequestVoteReplyDestroy(pSyncMsg);
+
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES) {
+      SyncAppendEntries *pSyncMsg = syncAppendEntriesFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
+
+      ret = syncNodeOnAppendEntriesCb(pSyncNode, pSyncMsg);
+      syncAppendEntriesDestroy(pSyncMsg);
+
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES_REPLY) {
+      SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
+
+      ret = syncNodeOnAppendEntriesReplyCb(pSyncNode, pSyncMsg);
+      syncAppendEntriesReplyDestroy(pSyncMsg);
+
+    } else {
+      mError("==mndProcessSyncMsg== error msg type:%d", pRpcMsg->msgType);
+      ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
+    }
+
+    syncNodeRelease(pSyncNode);
+  } else {
+    mError("==mndProcessSyncMsg== error syncEnv stop");
+    ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
+  }
+
+  return ret;
+}
 
 int32_t mndProcessMsg(SRpcMsg *pMsg) {
   SMnode *pMnode = pMsg->info.node;
@@ -346,7 +443,8 @@ int32_t mndProcessMsg(SRpcMsg *pMsg) {
   mTrace("msg:%p, will be processed, type:%s app:%p", pMsg, TMSG_INFO(pMsg->msgType), ahandle);
 
   if (IsReq(pMsg)) {
-    if (!mndIsMaster(pMnode)) {
+    if (!mndIsMaster(pMnode) && pMsg->msgType != TDMT_MND_TRANS_TIMER && pMsg->msgType != TDMT_MND_MQ_TIMER &&
+        pMsg->msgType != TDMT_MND_TELEM_TIMER) {
       terrno = TSDB_CODE_APP_NOT_READY;
       mDebug("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
       return -1;

source/dnode/mnode/impl/test/sdb/sdbTest.cpp

@@ -493,9 +493,8 @@ TEST_F(MndTestSdb, 01_Write_Str) {
   ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 2);
   ASSERT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1);
   ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 2 );
-  ASSERT_EQ(sdbUpdateVer(pSdb, 0), -1);
-  ASSERT_EQ(sdbUpdateVer(pSdb, 1), 0);
-  ASSERT_EQ(sdbUpdateVer(pSdb, -1), -1);
+  sdbSetApplyIndex(pSdb, -1);
+  ASSERT_EQ(sdbGetApplyIndex(pSdb), -1);
   ASSERT_EQ(mnode.insertTimes, 2);
   ASSERT_EQ(mnode.deleteTimes, 0);
 
@@ -537,9 +536,6 @@ TEST_F(MndTestSdb, 01_Write_Str) {
 
     ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 3);
     ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 4);
-    ASSERT_EQ(sdbUpdateVer(pSdb, 0), -1);
-    ASSERT_EQ(sdbUpdateVer(pSdb, 1), 0);
-    ASSERT_EQ(sdbUpdateVer(pSdb, -1), -1);
     ASSERT_EQ(mnode.insertTimes, 3);
     ASSERT_EQ(mnode.deleteTimes, 0);
 
@@ -704,8 +700,9 @@ TEST_F(MndTestSdb, 01_Write_Str) {
   }
 
   // write version
-  ASSERT_EQ(sdbUpdateVer(pSdb, 1), 0);
-  ASSERT_EQ(sdbUpdateVer(pSdb, 1), 1);
+  sdbSetApplyIndex(pSdb, 0);
+  sdbSetApplyIndex(pSdb, 1);
+  ASSERT_EQ(sdbGetApplyIndex(pSdb), 1);
   ASSERT_EQ(sdbWriteFile(pSdb), 0);
   ASSERT_EQ(sdbWriteFile(pSdb), 0);
 
@@ -775,7 +772,7 @@ TEST_F(MndTestSdb, 01_Read_Str) {
   ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 2);
   ASSERT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1);
   ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 5);
-  ASSERT_EQ(sdbUpdateVer(pSdb, 0), 1);
+  ASSERT_EQ(sdbGetApplyIndex(pSdb), 1);
   ASSERT_EQ(mnode.insertTimes, 4);
   ASSERT_EQ(mnode.deleteTimes, 0);
 

source/dnode/mnode/impl/test/trans/CMakeLists.txt

@@ -31,7 +31,7 @@ target_include_directories(
     PUBLIC "${TD_SOURCE_DIR}/include/dnode/mnode"
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/../../inc"
 )
-add_test(
-    NAME transTest2
-    COMMAND transTest2
-)
+#add_test(
+#    NAME transTest2
+#    COMMAND transTest2
+#)

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

@@ -23,6 +23,11 @@ int32_t sendReq(const SEpSet *pEpSet, SRpcMsg *pMsg) {
   return -1;
 }
 
+int32_t putToQueue(void *pMgmt, SRpcMsg *pMsg) {
+  terrno = TSDB_CODE_INVALID_PTR;
+  return -1;
+}
+
 class MndTestTrans2 : public ::testing::Test {
  protected:
   static void InitLog() {
@@ -55,6 +60,9 @@ class MndTestTrans2 : public ::testing::Test {
     msgCb.reportStartupFp = reportStartup;
     msgCb.sendReqFp = sendReq;
     msgCb.sendRspFp = sendRsp;
+    msgCb.queueFps[SYNC_QUEUE] = putToQueue;
+    msgCb.queueFps[WRITE_QUEUE] = putToQueue;
+     msgCb.queueFps[READ_QUEUE] = putToQueue;
     msgCb.mgmt = (SMgmtWrapper *)(&msgCb);  // hack
     tmsgSetDefault(&msgCb);
 
@@ -77,6 +85,7 @@ class MndTestTrans2 : public ::testing::Test {
   static void SetUpTestSuite() {
     InitLog();
     walInit();
+    syncInit();
     InitMnode();
   }
 

source/dnode/mnode/sdb/src/sdb.c

@@ -31,11 +31,9 @@ SSdb *sdbInit(SSdbOpt *pOption) {
   char path[PATH_MAX + 100] = {0};
   snprintf(path, sizeof(path), "%s%sdata", pOption->path, TD_DIRSEP);
   pSdb->currDir = strdup(path);
-  snprintf(path, sizeof(path), "%s%ssync", pOption->path, TD_DIRSEP);
-  pSdb->syncDir = strdup(path);
   snprintf(path, sizeof(path), "%s%stmp", pOption->path, TD_DIRSEP);
   pSdb->tmpDir = strdup(path);
-  if (pSdb->currDir == NULL || pSdb->currDir == NULL || pSdb->currDir == NULL) {
+  if (pSdb->currDir == NULL || pSdb->tmpDir == NULL) {
     sdbCleanup(pSdb);
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     mError("failed to init sdb since %s", terrstr());
@@ -55,6 +53,7 @@ SSdb *sdbInit(SSdbOpt *pOption) {
   }
 
   pSdb->curVer = -1;
+  pSdb->curTerm = -1;
   pSdb->lastCommitVer = -1;
   pSdb->pMnode = pOption->pMnode;
   mDebug("sdb init successfully");
@@ -149,12 +148,6 @@ static int32_t sdbCreateDir(SSdb *pSdb) {
     return -1;
   }
 
-  if (taosMkDir(pSdb->syncDir) != 0) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    mError("failed to create dir:%s since %s", pSdb->syncDir, terrstr());
-    return -1;
-  }
-
   if (taosMkDir(pSdb->tmpDir) != 0) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     mError("failed to create dir:%s since %s", pSdb->tmpDir, terrstr());
@@ -164,4 +157,10 @@ static int32_t sdbCreateDir(SSdb *pSdb) {
   return 0;
 }
 
-int64_t sdbUpdateVer(SSdb *pSdb, int32_t val) { return atomic_add_fetch_64(&pSdb->curVer, val); }
+void sdbSetApplyIndex(SSdb *pSdb, int64_t index) { pSdb->curVer = index; }
\ No newline at end of file
+
+int64_t sdbGetApplyIndex(SSdb *pSdb) { return pSdb->curVer; }
+
+void sdbSetApplyTerm(SSdb *pSdb, int64_t term) { pSdb->curTerm = term; }
+
+int64_t sdbGetApplyTerm(SSdb *pSdb) { return pSdb->curTerm; }

source/dnode/mnode/sdb/src/sdbFile.c

@@ -65,6 +65,16 @@ static int32_t sdbReadFileHead(SSdb *pSdb, TdFilePtr pFile) {
     return -1;
   }
 
+  ret = taosReadFile(pFile, &pSdb->curTerm, sizeof(int64_t));
+  if (ret < 0) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    return -1;
+  }
+  if (ret != sizeof(int64_t)) {
+    terrno = TSDB_CODE_FILE_CORRUPTED;
+    return -1;
+  }
+
   for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) {
     int64_t maxId = 0;
     ret = taosReadFile(pFile, &maxId, sizeof(int64_t));
@@ -123,6 +133,11 @@ static int32_t sdbWriteFileHead(SSdb *pSdb, TdFilePtr pFile) {
     return -1;
   }
 
+  if (taosWriteFile(pFile, &pSdb->curTerm, sizeof(int64_t)) != sizeof(int64_t)) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    return -1;
+  }
+
   for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) {
     int64_t maxId = 0;
     if (i < SDB_MAX) {
@@ -182,6 +197,7 @@ int32_t sdbReadFile(SSdb *pSdb) {
   if (sdbReadFileHead(pSdb, pFile) != 0) {
     mError("failed to read file:%s head since %s", file, terrstr());
     pSdb->curVer = -1;
+    pSdb->curTerm = -1;
     taosMemoryFree(pRaw);
     taosCloseFile(&pFile);
     return -1;
@@ -256,8 +272,8 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
   char curfile[PATH_MAX] = {0};
   snprintf(curfile, sizeof(curfile), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP);
 
-  mDebug("start to write file:%s, current ver:%" PRId64 ", commit ver:%" PRId64, curfile, pSdb->curVer,
-         pSdb->lastCommitVer);
+  mDebug("start to write file:%s, current ver:%" PRId64 " term:%" PRId64 ", commit ver:%" PRId64, curfile, pSdb->curVer,
+         pSdb->curTerm, pSdb->lastCommitVer);
 
   TdFilePtr pFile = taosOpenFile(tmpfile, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
   if (pFile == NULL) {
@@ -350,7 +366,7 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
     mError("failed to write file:%s since %s", curfile, tstrerror(code));
   } else {
     pSdb->lastCommitVer = pSdb->curVer;
-    mDebug("write file:%s successfully, ver:%" PRId64, curfile, pSdb->lastCommitVer);
+    mDebug("write file:%s successfully, ver:%" PRId64 " term:%" PRId64, curfile, pSdb->lastCommitVer, pSdb->curTerm);
   }
 
   terrno = code;

source/dnode/snode/src/snode.c

@@ -57,9 +57,7 @@ void sndMetaDelete(SStreamMeta *pMeta) {
 }
 
 int32_t sndMetaDeployTask(SStreamMeta *pMeta, SStreamTask *pTask) {
-  for (int i = 0; i < pTask->exec.numOfRunners; i++) {
-    pTask->exec.runners[i].executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, NULL);
-  }
+  pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, NULL);
   return taosHashPut(pMeta->pHash, &pTask->taskId, sizeof(int32_t), pTask, sizeof(void *));
 }
 

source/dnode/vnode/CMakeLists.txt

@@ -48,7 +48,6 @@ target_sources(
     # tq
     "src/tq/tq.c"
     "src/tq/tqCommit.c"
-    "src/tq/tqMetaStore.c"
     "src/tq/tqOffset.c"
     "src/tq/tqPush.c"
     "src/tq/tqRead.c"

source/dnode/vnode/inc/vnode.h

@@ -51,7 +51,7 @@ int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs);
 void    vnodeDestroy(const char *path, STfs *pTfs);
 SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb);
 void    vnodeClose(SVnode *pVnode);
-int32_t vnodePreprocessWriteReqs(SVnode *pVnode, SArray *pMsgs, int64_t *version);
+int32_t vnodePreprocessReq(SVnode *pVnode, SRpcMsg *pMsg);
 int32_t vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg *pRsp);
 int32_t vnodeProcessCMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
 int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
@@ -126,7 +126,7 @@ STqReadHandle *tqInitSubmitMsgScanner(SMeta *pMeta);
 void    tqReadHandleSetColIdList(STqReadHandle *pReadHandle, SArray *pColIdList);
 int32_t tqReadHandleSetTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
 int32_t tqReadHandleAddTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
-int32_t tqReadHandleRemoveTbUidList(STqReadHandle* pHandle, const SArray* tbUidList);
+int32_t tqReadHandleRemoveTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
 
 int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
 bool    tqNextDataBlock(STqReadHandle *pHandle);
@@ -177,7 +177,7 @@ struct SMetaEntry {
   int64_t  version;
   int8_t   type;
   tb_uid_t uid;
-  const char *name;
+  char    *name;
   union {
     struct {
       SSchemaWrapper schema;
@@ -187,7 +187,7 @@ struct SMetaEntry {
       int64_t  ctime;
       int32_t  ttlDays;
       tb_uid_t suid;
-      const uint8_t *pTags;
+      uint8_t *pTags;
     } ctbEntry;
     struct {
       int64_t        ctime;

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

@@ -20,9 +20,9 @@
 
 #include "executor.h"
 #include "os.h"
-#include "tcache.h"
 #include "thash.h"
 #include "tmsg.h"
+#include "tqueue.h"
 #include "trpc.h"
 #include "ttimer.h"
 #include "wal.h"
@@ -41,45 +41,6 @@ extern "C" {
 #define tqTrace(...) do { if (tqDebugFlag & DEBUG_TRACE) { taosPrintLog("TQ  ", DEBUG_TRACE, tqDebugFlag, __VA_ARGS__); }} while(0)
 // clang-format on
 
-#define TQ_BUFFER_SIZE 4
-
-#define TQ_BUCKET_MASK 0xFF
-#define TQ_BUCKET_SIZE 256
-
-#define TQ_PAGE_SIZE 4096
-// key + offset + size
-#define TQ_IDX_SIZE 24
-// 4096 / 24
-#define TQ_MAX_IDX_ONE_PAGE 170
-// 24 * 170
-#define TQ_IDX_PAGE_BODY_SIZE 4080
-// 4096 - 4080
-#define TQ_IDX_PAGE_HEAD_SIZE 16
-
-#define TQ_ACTION_CONST      0
-#define TQ_ACTION_INUSE      1
-#define TQ_ACTION_INUSE_CONT 2
-#define TQ_ACTION_INTXN      3
-
-#define TQ_SVER 0
-
-// TODO: inplace mode is not implemented
-#define TQ_UPDATE_INPLACE 0
-#define TQ_UPDATE_APPEND  1
-
-#define TQ_DUP_INTXN_REWRITE 0
-#define TQ_DUP_INTXN_REJECT  2
-
-static inline bool tqUpdateAppend(int32_t tqConfigFlag) { return tqConfigFlag & TQ_UPDATE_APPEND; }
-
-static inline bool tqDupIntxnReject(int32_t tqConfigFlag) { return tqConfigFlag & TQ_DUP_INTXN_REJECT; }
-
-static const int8_t TQ_CONST_DELETE = TQ_ACTION_CONST;
-
-#define TQ_DELETE_TOKEN (void*)&TQ_CONST_DELETE
-
-typedef enum { TQ_ITEM_READY, TQ_ITEM_PROCESS, TQ_ITEM_EMPTY } STqItemStatus;
-
 typedef struct STqOffsetCfg   STqOffsetCfg;
 typedef struct STqOffsetStore STqOffsetStore;
 
@@ -98,53 +59,6 @@ struct STqReadHandle {
   STSchema*         pSchema;
 };
 
-typedef struct {
-  int16_t ver;
-  int16_t action;
-  int32_t checksum;
-  int64_t ssize;
-  char    content[];
-} STqSerializedHead;
-
-typedef int32_t (*FTqSerialize)(const void* pObj, STqSerializedHead** ppHead);
-typedef int32_t (*FTqDeserialize)(void* self, const STqSerializedHead* pHead, void** ppObj);
-typedef void (*FTqDelete)(void*);
-
-typedef struct {
-  int64_t key;
-  int64_t offset;
-  int64_t serializedSize;
-  void*   valueInUse;
-  void*   valueInTxn;
-} STqMetaHandle;
-
-typedef struct STqMetaList {
-  STqMetaHandle       handle;
-  struct STqMetaList* next;
-  // struct STqMetaList* inTxnPrev;
-  // struct STqMetaList* inTxnNext;
-  struct STqMetaList* unpersistPrev;
-  struct STqMetaList* unpersistNext;
-} STqMetaList;
-
-typedef struct {
-  STQ*         pTq;
-  STqMetaList* bucket[TQ_BUCKET_SIZE];
-  // a table head
-  STqMetaList* unpersistHead;
-  // topics that are not connectted
-  STqMetaList* unconnectTopic;
-
-  TdFilePtr pFile;
-  TdFilePtr pIdxFile;
-
-  char*          dirPath;
-  int32_t        tqConfigFlag;
-  FTqSerialize   pSerializer;
-  FTqDeserialize pDeserializer;
-  FTqDelete      pDeleter;
-} STqMetaStore;
-
 typedef struct {
   int64_t  consumerId;
   int32_t  epoch;
@@ -172,15 +86,17 @@ typedef struct {
   qTaskInfo_t    task[5];
 } STqExec;
 
+int32_t tEncodeSTqExec(SEncoder* pEncoder, const STqExec* pExec);
+int32_t tDecodeSTqExec(SDecoder* pDecoder, STqExec* pExec);
+
 struct STQ {
   char*     path;
-  // STqMetaStore* tqMeta;
   SHashObj* pushMgr;  // consumerId -> STqExec*
   SHashObj* execs;    // subKey -> STqExec
   SHashObj* pStreamTasks;
   SVnode*   pVnode;
   SWal*     pWal;
-  // TDB*      pTdb;
+  TDB*      pTdb;
 };
 
 typedef struct {
@@ -188,89 +104,12 @@ typedef struct {
   tmr_h  timer;
 } STqMgmt;
 
-static STqMgmt tqMgmt;
-
-typedef struct {
-  int8_t         status;
-  int64_t        offset;
-  qTaskInfo_t    task;
-  STqReadHandle* pReadHandle;
-} STqTaskItem;
-
-// new version
-typedef struct {
-  int64_t     firstOffset;
-  int64_t     lastOffset;
-  STqTaskItem output[TQ_BUFFER_SIZE];
-} STqBuffer;
-
-typedef struct {
-  char            topicName[TSDB_TOPIC_FNAME_LEN];
-  char*           sql;
-  char*           logicalPlan;
-  char*           physicalPlan;
-  char*           qmsg;
-  STqBuffer       buffer;
-  SWalReadHandle* pReadhandle;
-} STqTopic;
-
-typedef struct {
-  int64_t consumerId;
-  int32_t epoch;
-  char    cgroup[TSDB_TOPIC_FNAME_LEN];
-  SArray* topics;  // SArray<STqTopic>
-} STqConsumer;
-
-typedef struct {
-  int8_t      type;
-  int8_t      nodeType;
-  int8_t      reserved[6];
-  int64_t     streamId;
-  qTaskInfo_t task;
-  // TODO sync function
-} STqStreamPusher;
-
-typedef struct {
-  int8_t inited;
-  tmr_h  timer;
-} STqPushMgmt;
-
-static STqPushMgmt tqPushMgmt;
+static STqMgmt tqMgmt = {0};
 
 // init once
 int  tqInit();
 void tqCleanUp();
 
-// open in each vnode
-// required by vnode
-
-int32_t tqSerializeConsumer(const STqConsumer*, STqSerializedHead**);
-int32_t tqDeserializeConsumer(STQ*, const STqSerializedHead*, STqConsumer**);
-
-static int FORCE_INLINE tqQueryExecuting(int32_t status) { return status; }
-
-// tqMetaStore.h
-STqMetaStore* tqStoreOpen(STQ* pTq, const char* path, FTqSerialize pSerializer, FTqDeserialize pDeserializer,
-                          FTqDelete pDeleter, int32_t tqConfigFlag);
-int32_t       tqStoreClose(STqMetaStore*);
-// int32_t       tqStoreDelete(TqMetaStore*);
-// int32_t       tqStoreCommitAll(TqMetaStore*);
-int32_t tqStorePersist(STqMetaStore*);
-// clean deleted idx and data from persistent file
-int32_t tqStoreCompact(STqMetaStore*);
-
-void* tqHandleGet(STqMetaStore*, int64_t key);
-// make it unpersist
-void*   tqHandleTouchGet(STqMetaStore*, int64_t key);
-int32_t tqHandleMovePut(STqMetaStore*, int64_t key, void* value);
-int32_t tqHandleCopyPut(STqMetaStore*, int64_t key, void* value, size_t vsize);
-// delete committed kv pair
-// notice that a delete action still needs to be committed
-int32_t tqHandleDel(STqMetaStore*, int64_t key);
-int32_t tqHandlePurge(STqMetaStore*, int64_t key);
-int32_t tqHandleCommit(STqMetaStore*, int64_t key);
-int32_t tqHandleAbort(STqMetaStore*, int64_t key);
-
 // tqOffset
 STqOffsetStore* STqOffsetOpen(STqOffsetCfg*);
 void            STqOffsetClose(STqOffsetStore*);

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

@@ -104,7 +104,7 @@ int          tsdbOpen(SVnode* pVnode, STsdb** ppTsdb, const char* dir, STsdbKeep
 int          tsdbClose(STsdb** pTsdb);
 int          tsdbBegin(STsdb* pTsdb);
 int          tsdbCommit(STsdb* pTsdb);
-int          tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, const SSubmitReq* pMsg);
+int          tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq* pMsg);
 int          tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
 int          tsdbInsertTableData(STsdb* pTsdb, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkRsp* pRsp);
 tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
@@ -123,11 +123,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen);
 int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen);
 int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId);
 int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen);
-#if 0
-int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId);
-int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen, int32_t workerId);
-#endif
-int32_t tqProcessStreamTriggerNew(STQ* pTq, SSubmitReq* data);
+int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* data);
 int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg);

source/dnode/vnode/src/meta/metaQuery.c

@@ -278,12 +278,13 @@ STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
   pSW = metaGetTableSchema(pMeta, quid, sver, 0);
   if (!pSW) return NULL;
 
-  tdInitTSchemaBuilder(&sb, 0);
+  tdInitTSchemaBuilder(&sb, sver);
   for (int i = 0; i < pSW->nCols; i++) {
     pSchema = pSW->pSchema + i;
     tdAddColToSchema(&sb, pSchema->type, pSchema->flags, pSchema->colId, pSchema->bytes);
   }
   pTSchema = tdGetSchemaFromBuilder(&sb);
+
   tdDestroyTSchemaBuilder(&sb);
 
   taosMemoryFree(pSW->pSchema);

source/dnode/vnode/src/meta/metaTable.c

@@ -30,9 +30,9 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   int         vLen = 0;
   const void *pKey = NULL;
   const void *pVal = NULL;
-  void *      pBuf = NULL;
+  void       *pBuf = NULL;
   int32_t     szBuf = 0;
-  void *      p = NULL;
+  void       *p = NULL;
   SMetaReader mr = {0};
 
   // validate req
@@ -71,9 +71,9 @@ _err:
 }
 
 int metaDropSTable(SMeta *pMeta, int64_t verison, SVDropStbReq *pReq) {
-  TBC *       pNameIdxc = NULL;
-  TBC *       pUidIdxc = NULL;
-  TBC *       pCtbIdxc = NULL;
+  TBC        *pNameIdxc = NULL;
+  TBC        *pUidIdxc = NULL;
+  TBC        *pCtbIdxc = NULL;
   SCtbIdxKey *pCtbIdxKey;
   const void *pKey = NULL;
   int         nKey;
@@ -134,8 +134,8 @@ _err:
 int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   SMetaEntry  oStbEntry = {0};
   SMetaEntry  nStbEntry = {0};
-  TBC *       pUidIdxc = NULL;
-  TBC *       pTbDbc = NULL;
+  TBC        *pUidIdxc = NULL;
+  TBC        *pTbDbc = NULL;
   const void *pData;
   int         nData;
   int64_t     oversion;
@@ -165,7 +165,9 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   ret = tdbTbcGet(pTbDbc, NULL, NULL, &pData, &nData);
   ASSERT(ret == 0);
 
-  tDecoderInit(&dc, pData, nData);
+  oStbEntry.pBuf = taosMemoryMalloc(nData);
+  memcpy(oStbEntry.pBuf, pData, nData);
+  tDecoderInit(&dc, oStbEntry.pBuf, nData);
   metaDecodeEntry(&dc, &oStbEntry);
 
   nStbEntry.version = version;
@@ -193,6 +195,7 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   // update uid index
   tdbTbcUpsert(pUidIdxc, &pReq->suid, sizeof(tb_uid_t), &version, sizeof(version), 0);
 
+  if (oStbEntry.pBuf) taosMemoryFree(oStbEntry.pBuf);
   metaULock(pMeta);
   tDecoderClear(&dc);
   tdbTbcClose(pTbDbc);
@@ -220,9 +223,6 @@ int metaCreateTable(SMeta *pMeta, int64_t version, SVCreateTbReq *pReq) {
     terrno = TSDB_CODE_TDB_TABLE_ALREADY_EXIST;
     metaReaderClear(&mr);
     return -1;
-  } else {
-    pReq->uid = tGenIdPI64();
-    pReq->ctime = taosGetTimestampMs();
   }
   metaReaderClear(&mr);
 
@@ -256,9 +256,9 @@ _err:
 }
 
 int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUids) {
-  TBC *       pTbDbc = NULL;
-  TBC *       pUidIdxc = NULL;
-  TBC *       pNameIdxc = NULL;
+  TBC        *pTbDbc = NULL;
+  TBC        *pUidIdxc = NULL;
+  TBC        *pNameIdxc = NULL;
   const void *pData;
   int         nData;
   tb_uid_t    uid;
@@ -377,14 +377,14 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUi
 }
 
 static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) {
-  void *          pVal = NULL;
+  void           *pVal = NULL;
   int             nVal = 0;
-  const void *    pData = NULL;
+  const void     *pData = NULL;
   int             nData = 0;
   int             ret = 0;
   tb_uid_t        uid;
   int64_t         oversion;
-  SSchema *       pColumn = NULL;
+  SSchema        *pColumn = NULL;
   SMetaEntry      entry = {0};
   SSchemaWrapper *pSchema;
   int             c;
@@ -420,7 +420,9 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
 
   // get table entry
   SDecoder dc = {0};
-  tDecoderInit(&dc, pData, nData);
+  entry.pBuf = taosMemoryMalloc(nData);
+  memcpy(entry.pBuf, pData, nData);
+  tDecoderInit(&dc, entry.pBuf, nData);
   ret = metaDecodeEntry(&dc, &entry);
   ASSERT(ret == 0);
 
@@ -530,7 +532,7 @@ _err:
 static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) {
   SMetaEntry  ctbEntry = {0};
   SMetaEntry  stbEntry = {0};
-  void *      pVal = NULL;
+  void       *pVal = NULL;
   int         nVal = 0;
   int         ret;
   int         c;
@@ -561,7 +563,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
   oversion = *(int64_t *)pData;
 
   // search table.db
-  TBC *    pTbDbc = NULL;
+  TBC     *pTbDbc = NULL;
   SDecoder dc1 = {0};
   SDecoder dc2 = {0};
 
@@ -585,7 +587,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
   metaDecodeEntry(&dc2, &stbEntry);
 
   SSchemaWrapper *pTagSchema = &stbEntry.stbEntry.schemaTag;
-  SSchema *       pColumn = NULL;
+  SSchema        *pColumn = NULL;
   int32_t         iCol = 0;
   for (;;) {
     pColumn = NULL;
@@ -605,8 +607,15 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
   if (iCol == 0) {
     // TODO : need to update tag index
   }
-
   ctbEntry.version = version;
+  if(pTagSchema->nCols == 1 && pTagSchema->pSchema[0].type == TSDB_DATA_TYPE_JSON){
+    ctbEntry.ctbEntry.pTags = taosMemoryMalloc(pAlterTbReq->nTagVal);
+    if(ctbEntry.ctbEntry.pTags == NULL){
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      goto _err;
+    }
+    memcpy((void*)ctbEntry.ctbEntry.pTags, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal);
+  }else{
     SKVRowBuilder kvrb = {0};
     const SKVRow  pOldTag = (const SKVRow)ctbEntry.ctbEntry.pTags;
     SKVRow        pNewTag = NULL;
@@ -630,6 +639,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
 
     ctbEntry.ctbEntry.pTags = tdGetKVRowFromBuilder(&kvrb);
     tdDestroyKVRowBuilder(&kvrb);
+  }
 
   // save to table.db
   metaSaveToTbDb(pMeta, &ctbEntry);
@@ -639,6 +649,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
 
   tDecoderClear(&dc1);
   tDecoderClear(&dc2);
+  if (ctbEntry.ctbEntry.pTags) taosMemoryFree((void*)ctbEntry.ctbEntry.pTags);
   if (ctbEntry.pBuf) taosMemoryFree(ctbEntry.pBuf);
   if (stbEntry.pBuf) tdbFree(stbEntry.pBuf);
   tdbTbcClose(pTbDbc);
@@ -681,8 +692,8 @@ int metaAlterTable(SMeta *pMeta, int64_t version, SVAlterTbReq *pReq) {
 
 static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
   STbDbKey tbDbKey;
-  void *   pKey = NULL;
-  void *   pVal = NULL;
+  void    *pKey = NULL;
+  void    *pVal = NULL;
   int      kLen = 0;
   int      vLen = 0;
   SEncoder coder = {0};
@@ -797,14 +808,14 @@ static void metaDestroyTagIdxKey(STagIdxKey *pTagIdxKey) {
 }
 
 static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
-  void *         pData = NULL;
+  void          *pData = NULL;
   int            nData = 0;
   STbDbKey       tbDbKey = {0};
   SMetaEntry     stbEntry = {0};
-  STagIdxKey *   pTagIdxKey = NULL;
+  STagIdxKey    *pTagIdxKey = NULL;
   int32_t        nTagIdxKey;
   const SSchema *pTagColumn;       // = &stbEntry.stbEntry.schema.pSchema[0];
-  const void *   pTagData = NULL;  //
+  const void    *pTagData = NULL;  //
   SDecoder       dc = {0};
 
   // get super table
@@ -846,7 +857,7 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
 
 static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
   SEncoder              coder = {0};
-  void *                pVal = NULL;
+  void                 *pVal = NULL;
   int                   vLen = 0;
   int                   rcode = 0;
   SSkmDbKey             skmDbKey = {0};

source/dnode/vnode/src/tsdb/tsdbMemTable2.c

@@ -217,7 +217,7 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
     if (tDecodeIsEnd(&dc)) break;
 
     // decode row
-    if (tDecodeBinary(&dc, (const uint8_t **)&tRow.pRow, &tRow.szRow) < 0) {
+    if (tDecodeBinary(&dc, (uint8_t **)&tRow.pRow, &tRow.szRow) < 0) {
       terrno = TSDB_CODE_INVALID_MSG;
       return -1;
     }
@@ -273,7 +273,7 @@ static FORCE_INLINE int32_t tsdbEncodeRow(SEncoder *pEncoder, const STsdbRow *pR
 
 static FORCE_INLINE int32_t tsdbDecodeRow(SDecoder *pDecoder, STsdbRow *pRow) {
   if (tDecodeI64(pDecoder, &pRow->version) < 0) return -1;
-  if (tDecodeBinary(pDecoder, (const uint8_t **)&pRow->pRow, &pRow->szRow) < 0) return -1;
+  if (tDecodeBinary(pDecoder, (uint8_t **)&pRow->pRow, &pRow->szRow) < 0) return -1;
   return 0;
 }
 

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -1638,9 +1638,7 @@ static int32_t mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capa
   int32_t numOfColsOfRow1 = 0;
 
   if (pSchema1 == NULL) {
-    // pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row1));
-    // TODO: use the real schemaVersion
-    pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, 1);
+    pSchema1 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row1));
   }
 
 #ifdef TD_DEBUG_PRINT_ROW
@@ -1657,9 +1655,7 @@ static int32_t mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capa
   if (row2) {
     isRow2DataRow = TD_IS_TP_ROW(row2);
     if (pSchema2 == NULL) {
-      // pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row2));
-      // TODO: use the real schemaVersion
-      pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, 1);
+      pSchema2 = metaGetTbTSchema(REPO_META(pTsdbReadHandle->pTsdb), uid, TD_ROW_SVER(row2));
     }
     if (isRow2DataRow) {
       numOfColsOfRow2 = schemaNCols(pSchema2);

source/dnode/vnode/src/tsdb/tsdbWrite.c

@@ -85,7 +85,7 @@ static FORCE_INLINE int tsdbCheckRowRange(STsdb *pTsdb, tb_uid_t uid, STSRow *ro
   return 0;
 }
 
-int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, const SSubmitReq *pMsg) {
+int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg) {
   ASSERT(pMsg != NULL);
   // STsdbMeta *    pMeta = pTsdb->tsdbMeta;
   SSubmitMsgIter msgIter = {0};
@@ -150,7 +150,6 @@ int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, const SSubmitReq *pMsg) {
         return -1;
       }
     }
-
   }
 
   if (terrno != TSDB_CODE_SUCCESS) return -1;

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

@@ -24,26 +24,66 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
 static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
 static int vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp);
 
-int vnodePreprocessWriteReqs(SVnode *pVnode, SArray *pMsgs, int64_t *version) {
-#if 0
-  SRpcMsg *pMsg;
-  SRpcMsg  *pRpc;
-
-  *version = pVnode->state.processed;
-  for (int i = 0; i < taosArrayGetSize(pMsgs); i++) {
-    pMsg = *(SRpcMsg **)taosArrayGet(pMsgs, i);
-    pRpc = pMsg;
-
-    // set request version
-    if (walWrite(pVnode->pWal, pVnode->state.processed++, pRpc->msgType, pRpc->pCont, pRpc->contLen) < 0) {
-      vError("vnode:%d  write wal error since %s", TD_VID(pVnode), terrstr());
-      return -1;
+int32_t vnodePreprocessReq(SVnode *pVnode, SRpcMsg *pMsg) {
+  SDecoder dc = {0};
+
+  switch (pMsg->msgType) {
+    case TDMT_VND_CREATE_TABLE: {
+      int64_t ctime = taosGetTimestampMs();
+      int32_t nReqs;
+
+      tDecoderInit(&dc, (uint8_t *)pMsg->pCont + sizeof(SMsgHead), pMsg->contLen - sizeof(SMsgHead));
+      tStartDecode(&dc);
+
+      tDecodeI32v(&dc, &nReqs);
+      for (int32_t iReq = 0; iReq < nReqs; iReq++) {
+        tb_uid_t uid = tGenIdPI64();
+        tStartDecode(&dc);
+
+        tDecodeI32v(&dc, NULL);
+        *(int64_t *)(dc.data + dc.pos) = uid;
+        *(int64_t *)(dc.data + dc.pos + 8) = ctime;
+
+        tEndDecode(&dc);
+      }
+
+      tEndDecode(&dc);
+      tDecoderClear(&dc);
+    } break;
+    case TDMT_VND_SUBMIT: {
+      SSubmitMsgIter msgIter = {0};
+      SSubmitReq    *pSubmitReq = (SSubmitReq *)pMsg->pCont;
+      SSubmitBlk    *pBlock = NULL;
+      int64_t        ctime = taosGetTimestampMs();
+      tb_uid_t       uid;
+
+      tInitSubmitMsgIter(pSubmitReq, &msgIter);
+
+      for (;;) {
+        tGetSubmitMsgNext(&msgIter, &pBlock);
+        if (pBlock == NULL) break;
+
+        if (msgIter.schemaLen > 0) {
+          uid = tGenIdPI64();
+
+          tDecoderInit(&dc, pBlock->data, msgIter.schemaLen);
+          tStartDecode(&dc);
+
+          tDecodeI32v(&dc, NULL);
+          *(int64_t *)(dc.data + dc.pos) = uid;
+          *(int64_t *)(dc.data + dc.pos + 8) = ctime;
+          pBlock->uid = htobe64(uid);
+
+          tEndDecode(&dc);
+          tDecoderClear(&dc);
         }
       }
 
-  walFsync(pVnode->pWal, false);
+    } break;
+    default:
+      break;
+  }
 
-#endif
   return 0;
 }
 
@@ -106,13 +146,6 @@ int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg
                               pMsg->contLen - sizeof(SMsgHead)) < 0) {
       }
     } break;
-#if 0
-    case TDMT_VND_TASK_WRITE_EXEC: {
-      if (tqProcessTaskExec(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), pMsg->contLen - sizeof(SMsgHead),
-                            0) < 0) {
-      }
-    } break;
-#endif
     case TDMT_VND_ALTER_VNODE:
       break;
     default:
@@ -195,17 +228,6 @@ int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
     case TDMT_VND_TASK_RECOVER_RSP:
       return tqProcessTaskRecoverRsp(pVnode->pTq, pMsg);
 
-#if 0
-    case TDMT_VND_TASK_PIPE_EXEC:
-    case TDMT_VND_TASK_MERGE_EXEC:
-      return tqProcessTaskExec(pVnode->pTq, msgstr, msgLen, 0);
-    case TDMT_VND_STREAM_TRIGGER:{
-      // refactor, avoid double free
-      int code = tqProcessStreamTrigger(pVnode->pTq, pMsg->pCont, pMsg->contLen, 0);
-      pMsg->pCont = NULL;
-      return code;
-    }
-#endif
     case TDMT_VND_QUERY_HEARTBEAT:
       return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg);
     default:
@@ -675,7 +697,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
     goto _exit;
   }
 
-  for (int i = 0;;) {
+  for (;;) {
     tGetSubmitMsgNext(&msgIter, &pBlock);
     if (pBlock == NULL) break;
 

source/libs/command/src/command.c

@@ -21,6 +21,7 @@ static int32_t getSchemaBytes(const SSchema* pSchema) {
     case TSDB_DATA_TYPE_BINARY:
       return (pSchema->bytes - VARSTR_HEADER_SIZE);
     case TSDB_DATA_TYPE_NCHAR:
+    case TSDB_DATA_TYPE_JSON:
       return (pSchema->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
     default:
       return pSchema->bytes;

source/libs/executor/src/timewindowoperator.c

@@ -1181,6 +1181,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
 
   initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win);
   pInfo->invertible = allInvertible(pInfo->binfo.pCtx, numOfCols);
+  pInfo->invertible = false; // Todo(liuyao): Dependent TSDB API
 
   //  pInfo->pTableQueryInfo = initTableQueryInfo(pTableGroupInfo);
   if (code != TSDB_CODE_SUCCESS /* || pInfo->pTableQueryInfo == NULL*/) {