Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/data_format

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")

contrib/CMakeLists.txt

@@ -100,8 +100,10 @@ endif(${BUILD_WITH_NURAFT})
 
 # addr2line
 if(${BUILD_ADDR2LINE})
-    cat("${TD_SUPPORT_DIR}/libdwarf_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
-    cat("${TD_SUPPORT_DIR}/addr2line_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+    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,45 +337,47 @@ endif(${BUILD_WITH_SQLITE})
 
 # addr2line
 if(${BUILD_ADDR2LINE})
-    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   ) 
-    check_include_file( "stddef.h"        HAVE_STDDEF_H   ) 
-    check_include_file( "stdlib.h"        HAVE_STDLIB_H   ) 
-    check_include_file( "string.h"        HAVE_STRING_H   ) 
-    check_include_file( "memory.h"        HAVE_MEMORY_H   ) 
-    check_include_file( "strings.h"       HAVE_STRINGS_H  ) 
-    check_include_file( "stdint.h"        HAVE_STDINT_H   )
-    check_include_file( "unistd.h"        HAVE_UNISTD_H   )
-    check_include_file( "sgidefs.h"       HAVE_SGIDEFS_H  )
-    check_include_file( "stdafx.h"        HAVE_STDAFX_H   )
-    check_include_file( "elf.h"           HAVE_ELF_H      ) 
-    check_include_file( "libelf.h"        HAVE_LIBELF_H   ) 
-    check_include_file( "libelf/libelf.h" HAVE_LIBELF_LIBELF_H) 
-    check_include_file( "alloca.h"        HAVE_ALLOCA_H   )
-    check_include_file( "elfaccess.h"     HAVE_ELFACCESS_H)
-    check_include_file( "sys/elf_386.h"   HAVE_SYS_ELF_386_H  )
-    check_include_file( "sys/elf_amd64.h" HAVE_SYS_ELF_AMD64_H)
-    check_include_file( "sys/elf_sparc.h" HAVE_SYS_ELF_SPARC_H)
-    check_include_file( "sys/ia64/elf.h"  HAVE_SYS_IA64_ELF_H )
-    set(VERSION 0.3.1)
-    set(PACKAGE_VERSION "\"${VERSION}\"") 
-    configure_file(libdwarf/cmake/config.h.cmake config.h)
-    file(GLOB_RECURSE LIBDWARF_SOURCES "libdwarf/src/lib/libdwarf/*.c")
-    add_library(libdwarf STATIC ${LIBDWARF_SOURCES})
-    set_target_properties(libdwarf PROPERTIES OUTPUT_NAME "libdwarf")
-    if(HAVE_LIBELF_H OR HAVE_LIBELF_LIBELF_H)
-        target_link_libraries(libdwarf PUBLIC libelf)
-    endif()
-    target_include_directories(libdwarf SYSTEM PUBLIC "libdwarf/src/lib/libdwarf" ${CMAKE_CURRENT_BINARY_DIR})
-    file(READ "addr2line/addr2line.c" ADDR2LINE_CONTENT)
-    string(REPLACE "static int" "int" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")
-    string(REPLACE "static void" "void" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")
-    string(REPLACE "main(" "main_addr2line(" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")
-    file(WRITE "addr2line/addr2line.c" "${ADDR2LINE_CONTENT}")
-    add_library(addr2line STATIC "addr2line/addr2line.c")
-    target_link_libraries(addr2line PUBLIC libdwarf dl z)
-    target_include_directories(addr2line PUBLIC "libdwarf/src/lib/libdwarf" )
+    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   ) 
+        check_include_file( "stddef.h"        HAVE_STDDEF_H   ) 
+        check_include_file( "stdlib.h"        HAVE_STDLIB_H   ) 
+        check_include_file( "string.h"        HAVE_STRING_H   ) 
+        check_include_file( "memory.h"        HAVE_MEMORY_H   ) 
+        check_include_file( "strings.h"       HAVE_STRINGS_H  ) 
+        check_include_file( "stdint.h"        HAVE_STDINT_H   )
+        check_include_file( "unistd.h"        HAVE_UNISTD_H   )
+        check_include_file( "sgidefs.h"       HAVE_SGIDEFS_H  )
+        check_include_file( "stdafx.h"        HAVE_STDAFX_H   )
+        check_include_file( "elf.h"           HAVE_ELF_H      ) 
+        check_include_file( "libelf.h"        HAVE_LIBELF_H   ) 
+        check_include_file( "libelf/libelf.h" HAVE_LIBELF_LIBELF_H) 
+        check_include_file( "alloca.h"        HAVE_ALLOCA_H   )
+        check_include_file( "elfaccess.h"     HAVE_ELFACCESS_H)
+        check_include_file( "sys/elf_386.h"   HAVE_SYS_ELF_386_H  )
+        check_include_file( "sys/elf_amd64.h" HAVE_SYS_ELF_AMD64_H)
+        check_include_file( "sys/elf_sparc.h" HAVE_SYS_ELF_SPARC_H)
+        check_include_file( "sys/ia64/elf.h"  HAVE_SYS_IA64_ELF_H )
+        set(VERSION 0.3.1)
+        set(PACKAGE_VERSION "\"${VERSION}\"") 
+        configure_file(libdwarf/cmake/config.h.cmake config.h)
+        file(GLOB_RECURSE LIBDWARF_SOURCES "libdwarf/src/lib/libdwarf/*.c")
+        add_library(libdwarf STATIC ${LIBDWARF_SOURCES})
+        set_target_properties(libdwarf PROPERTIES OUTPUT_NAME "libdwarf")
+        if(HAVE_LIBELF_H OR HAVE_LIBELF_LIBELF_H)
+            target_link_libraries(libdwarf PUBLIC libelf)
+        endif()
+        target_include_directories(libdwarf SYSTEM PUBLIC "libdwarf/src/lib/libdwarf" ${CMAKE_CURRENT_BINARY_DIR})
+        file(READ "addr2line/addr2line.c" ADDR2LINE_CONTENT)
+        string(REPLACE "static int" "int" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")
+        string(REPLACE "static void" "void" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")
+        string(REPLACE "main(" "main_addr2line(" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")
+        file(WRITE "addr2line/addr2line.c" "${ADDR2LINE_CONTENT}")
+        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/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.
 

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,8 +1710,8 @@ int tDecodeSVCreateStbReq(SDecoder* pCoder, SVCreateStbReq* pReq);
 
 // TDMT_VND_DROP_STB ==============
 typedef struct SVDropStbReq {
-  const char* name;
-  tb_uid_t    suid;
+  char*    name;
+  tb_uid_t suid;
 } SVDropStbReq;
 
 int32_t tEncodeSVDropStbReq(SEncoder* pCoder, const SVDropStbReq* pReq);
@@ -1720,16 +1720,16 @@ int32_t tDecodeSVDropStbReq(SDecoder* pCoder, SVDropStbReq* pReq);
 // TDMT_VND_CREATE_TABLE ==============
 #define TD_CREATE_IF_NOT_EXISTS 0x1
 typedef struct SVCreateTbReq {
-  int32_t     flags;
-  tb_uid_t    uid;
-  int64_t     ctime;
-  const char* name;
-  int32_t     ttl;
-  int8_t      type;
+  int32_t  flags;
+  tb_uid_t uid;
+  int64_t  ctime;
+  char*    name;
+  int32_t  ttl;
+  int8_t   type;
   union {
     struct {
-      tb_uid_t       suid;
-      const uint8_t* pTag;
+      tb_uid_t suid;
+      uint8_t* pTag;
     } ctb;
     struct {
       SSchemaWrapper schema;
@@ -1777,8 +1777,8 @@ int32_t tDeserializeSVCreateTbBatchRsp(void* buf, int32_t bufLen, SVCreateTbBatc
 
 // TDMT_VND_DROP_TABLE =================
 typedef struct {
-  const char* name;
-  int8_t      igNotExists;
+  char*  name;
+  int8_t igNotExists;
 } SVDropTbReq;
 
 typedef struct {
@@ -1809,9 +1809,9 @@ int32_t tDecodeSVDropTbBatchRsp(SDecoder* pCoder, SVDropTbBatchRsp* pRsp);
 
 // TDMT_VND_ALTER_TABLE =====================
 typedef struct {
-  const char* tbName;
-  int8_t      action;
-  const char* colName;
+  char*  tbName;
+  int8_t action;
+  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;
-  int8_t         isNull;
-  uint32_t       nTagVal;
-  const uint8_t* pTagVal;
+  char*    tagName;
+  int8_t   isNull;
+  uint32_t nTagVal;
+  uint8_t* pTagVal;
   // TSDB_ALTER_TABLE_UPDATE_OPTIONS
-  int8_t      updateTTL;
-  int32_t     newTTL;
-  int8_t      updateComment;
-  const char* newComment;
+  int8_t  updateTTL;
+  int32_t newTTL;
+  int8_t  updateComment;
+  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;
@@ -2255,20 +2255,20 @@ int32_t tSerializeSMDropSmaReq(void* buf, int32_t bufLen, SMDropSmaReq* pReq);
 int32_t tDeserializeSMDropSmaReq(void* buf, int32_t bufLen, SMDropSmaReq* pReq);
 
 typedef struct {
-  int8_t      version;       // for compatibility(default 0)
-  int8_t      intervalUnit;  // MACRO: TIME_UNIT_XXX
-  int8_t      slidingUnit;   // MACRO: TIME_UNIT_XXX
-  int8_t      timezoneInt;   // sma data expired if timezone changes.
-  char        indexName[TSDB_INDEX_NAME_LEN];
-  int32_t     exprLen;
-  int32_t     tagsFilterLen;
-  int64_t     indexUid;
-  tb_uid_t    tableUid;  // super/child/common table uid
-  int64_t     interval;
-  int64_t     offset;  // use unit by precision of DB
-  int64_t     sliding;
-  const char* expr;  // sma expression
-  const char* tagsFilter;
+  int8_t   version;       // for compatibility(default 0)
+  int8_t   intervalUnit;  // MACRO: TIME_UNIT_XXX
+  int8_t   slidingUnit;   // MACRO: TIME_UNIT_XXX
+  int8_t   timezoneInt;   // sma data expired if timezone changes.
+  char     indexName[TSDB_INDEX_NAME_LEN];
+  int32_t  exprLen;
+  int32_t  tagsFilterLen;
+  int64_t  indexUid;
+  tb_uid_t tableUid;  // super/child/common table uid
+  int64_t  interval;
+  int64_t  offset;  // use unit by precision of DB
+  int64_t  sliding;
+  char*    expr;  // sma expression
+  char*    tagsFilter;
 } STSma;  // Time-range-wise SMA
 
 typedef STSma SVCreateTSmaReq;
@@ -2596,12 +2596,12 @@ static FORCE_INLINE void tDeleteSMqAskEpRsp(SMqAskEpRsp* pRsp) {
 
 #define TD_AUTO_CREATE_TABLE 0x1
 typedef struct {
-  int64_t        suid;
-  int64_t        uid;
-  int32_t        sver;
-  uint32_t       nData;
-  const uint8_t* pData;
-  SVCreateTbReq  cTbReq;
+  int64_t       suid;
+  int64_t       uid;
+  int32_t       sver;
+  uint32_t      nData;
+  uint8_t*      pData;
+  SVCreateTbReq cTbReq;
 } SVSubmitBlk;
 
 typedef struct {

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/util/tencode.h

@@ -39,11 +39,11 @@ typedef struct {
 } SEncoder;
 
 typedef struct {
-  const uint8_t* data;
-  uint32_t       size;
-  uint32_t       pos;
-  SCoderMem*     mList;
-  SDecoderNode*  dStack;
+  uint8_t*      data;
+  uint32_t      size;
+  uint32_t      pos;
+  SCoderMem*    mList;
+  SDecoderNode* dStack;
 } SDecoder;
 
 #define tPut(TYPE, BUF, VAL) ((TYPE*)(BUF))[0] = (VAL)
@@ -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,20 +389,20 @@ 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;
-  uint32_t    len;
+  char*    pStr;
+  uint32_t len;
   if (tDecodeCStrAndLen(pCoder, &pStr, &len) < 0) return -1;
 
   memcpy(val, pStr, len + 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/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/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) {

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/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);
@@ -174,20 +174,20 @@ typedef struct {
 } STableKeyInfo;
 
 struct SMetaEntry {
-  int64_t     version;
-  int8_t      type;
-  tb_uid_t    uid;
-  const char *name;
+  int64_t  version;
+  int8_t   type;
+  tb_uid_t uid;
+  char    *name;
   union {
     struct {
       SSchemaWrapper schema;
       SSchemaWrapper schemaTag;
     } stbEntry;
     struct {
-      int64_t        ctime;
-      int32_t        ttlDays;
-      tb_uid_t       suid;
-      const uint8_t *pTags;
+      int64_t  ctime;
+      int32_t  ttlDays;
+      tb_uid_t suid;
+      uint8_t *pTags;
     } ctbEntry;
     struct {
       int64_t        ctime;

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,

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;
@@ -681,8 +683,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 +799,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 +848,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

@@ -38,7 +38,7 @@ struct SMemTable {
 
 struct SMemSkipListNode {
   int8_t            level;
-  SMemSkipListNode *forwards[1]; // Windows does not allow 0 
+  SMemSkipListNode *forwards[1];  // Windows does not allow 0
 };
 
 struct SMemSkipList {
@@ -46,7 +46,7 @@ struct SMemSkipList {
   int8_t           maxLevel;
   int8_t           level;
   int32_t          size;
-  SMemSkipListNode pHead[1]; // Windows does not allow 0 
+  SMemSkipListNode pHead[1];  // Windows does not allow 0
 };
 
 struct SMemData {
@@ -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/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,62 @@ 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};
 
-  walFsync(pVnode->pWal, false);
+  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();
+
+      tInitSubmitMsgIter(pSubmitReq, &msgIter);
+
+      for (;;) {
+        tGetSubmitMsgNext(&msgIter, &pBlock);
+        if (pBlock == NULL) break;
+
+        if (msgIter.schemaLen > 0) {
+          tDecoderInit(&dc, pBlock->data, msgIter.schemaLen);
+          tStartDecode(&dc);
+
+          tDecodeI32v(&dc, NULL);
+          *(int64_t *)(dc.data + dc.pos) = tGenIdPI64();
+          *(int64_t *)(dc.data + dc.pos + 8) = ctime;
+
+          tEndDecode(&dc);
+          tDecoderClear(&dc);
+        }
+      }
+
+    } break;
+    default:
+      break;
+  }
 
-#endif
   return 0;
 }
 
@@ -675,7 +711,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/catalog/inc/catalogInt.h

@@ -58,6 +58,17 @@ enum {
   CTG_ACT_MAX
 };
 
+typedef enum {
+  CTG_TASK_GET_QNODE = 0,
+  CTG_TASK_GET_DB_VGROUP,
+  CTG_TASK_GET_DB_CFG,
+  CTG_TASK_GET_TB_META,
+  CTG_TASK_GET_TB_HASH,
+  CTG_TASK_GET_INDEX,
+  CTG_TASK_GET_UDF,
+  CTG_TASK_GET_USER,
+} CTG_TASK_TYPE;
+
 typedef struct SCtgDebug {
   bool     lockEnable;
   bool     cacheEnable;
@@ -66,6 +77,43 @@ typedef struct SCtgDebug {
   uint32_t showCachePeriodSec;
 } SCtgDebug;
 
+typedef struct SCtgTbCacheInfo {
+  bool     inCache;
+  uint64_t dbId;
+  uint64_t suid;
+  int32_t  tbType;
+} SCtgTbCacheInfo;
+
+typedef struct SCtgTbMetaCtx {
+  SCtgTbCacheInfo tbInfo;
+  SName* pName;
+  int32_t flag;
+} SCtgTbMetaCtx;
+
+typedef struct SCtgDbVgCtx {
+  char dbFName[TSDB_DB_FNAME_LEN];
+} SCtgDbVgCtx;
+
+typedef struct SCtgDbCfgCtx {
+  char dbFName[TSDB_DB_FNAME_LEN];
+} SCtgDbCfgCtx;
+
+typedef struct SCtgTbHashCtx {
+  char dbFName[TSDB_DB_FNAME_LEN];
+  SName* pName;
+} SCtgTbHashCtx;
+
+typedef struct SCtgIndexCtx {
+  char indexFName[TSDB_INDEX_FNAME_LEN];
+} SCtgIndexCtx;
+
+typedef struct SCtgUdfCtx {
+  char udfName[TSDB_FUNC_NAME_LEN];
+} SCtgUdfCtx;
+
+typedef struct SCtgUserCtx {
+  SUserAuthInfo user;
+} SCtgUserCtx;
 
 typedef struct SCtgTbMetaCache {
   SRWLatch  stbLock;
@@ -113,6 +161,55 @@ typedef struct SCatalog {
   SCtgRentMgmt     stbRent;
 } SCatalog;
 
+typedef struct SCtgJob {
+  int64_t          refId;
+  SArray*          pTasks;
+  int32_t          taskDone;
+  SMetaData        jobRes;
+  int32_t          rspCode;
+
+  uint64_t         queryId;
+  SCatalog*        pCtg; 
+  void*            pTrans; 
+  const SEpSet*    pMgmtEps;
+  void*            userParam;
+  catalogCallback  userFp;
+  int32_t          tbMetaNum;
+  int32_t          tbHashNum;
+  int32_t          dbVgNum;
+  int32_t          udfNum;
+  int32_t          qnodeNum;
+  int32_t          dbCfgNum;
+  int32_t          indexNum;
+  int32_t          userNum;
+} SCtgJob;
+
+typedef struct SCtgMsgCtx {
+  int32_t reqType;
+  void* lastOut;
+  void* out;
+  char* target;
+} SCtgMsgCtx;
+
+typedef struct SCtgTask {
+  CTG_TASK_TYPE type;
+  int32_t  taskId;
+  SCtgJob *pJob;
+  void* taskCtx;
+  SCtgMsgCtx msgCtx;
+  void* res;
+} SCtgTask;
+
+typedef int32_t (*ctgLanchTaskFp)(SCtgTask*);
+typedef int32_t (*ctgHandleTaskMsgRspFp)(SCtgTask*, int32_t, const SDataBuf *, int32_t);
+typedef int32_t (*ctgDumpTaskResFp)(SCtgTask*);
+
+typedef struct SCtgAsyncFps {
+  ctgLanchTaskFp launchFp;
+  ctgHandleTaskMsgRspFp handleRspFp;
+  ctgDumpTaskResFp dumpResFp;
+} SCtgAsyncFps;
+
 typedef struct SCtgApiStat {
 
 #ifdef WINDOWS
@@ -214,6 +311,7 @@ typedef struct SCtgQueue {
 
 typedef struct SCatalogMgmt {
   bool                  exit;
+  int32_t               jobPool;
   SRWLatch              lock;
   SCtgQueue             queue;
   TdThread              updateThread;  
@@ -327,10 +425,80 @@ typedef struct SCtgAction {
 #define CTG_API_LEAVE(c) do { int32_t __code = c; CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock); CTG_API_DEBUG("CTG API leave %s", __FUNCTION__); CTG_RET(__code); } while (0)
 #define CTG_API_ENTER() do { CTG_API_DEBUG("CTG API enter %s", __FUNCTION__); CTG_LOCK(CTG_READ, &gCtgMgmt.lock); if (atomic_load_8((int8_t*)&gCtgMgmt.exit)) { CTG_API_LEAVE(TSDB_CODE_CTG_OUT_OF_SERVICE); }  } while (0)
 
-
-extern void ctgdShowTableMeta(SCatalog* pCtg, const char *tbName, STableMeta* p);
-extern void ctgdShowClusterCache(SCatalog* pCtg);
-extern int32_t ctgdShowCacheInfo(void);
+#define CTG_PARAMS SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps
+#define CTG_PARAMS_LIST() pCtg, pTrans, pMgmtEps
+
+void ctgdShowTableMeta(SCatalog* pCtg, const char *tbName, STableMeta* p);
+void ctgdShowClusterCache(SCatalog* pCtg);
+int32_t ctgdShowCacheInfo(void);
+
+int32_t ctgRemoveTbMetaFromCache(SCatalog* pCtg, SName* pTableName, bool syncReq);
+int32_t ctgGetTbMetaFromCache(CTG_PARAMS, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta);
+
+int32_t ctgActUpdateVg(SCtgMetaAction *action);
+int32_t ctgActUpdateTb(SCtgMetaAction *action);
+int32_t ctgActRemoveDB(SCtgMetaAction *action);
+int32_t ctgActRemoveStb(SCtgMetaAction *action);
+int32_t ctgActRemoveTb(SCtgMetaAction *action);
+int32_t ctgActUpdateUser(SCtgMetaAction *action);
+int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache);
+void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache);
+void ctgReleaseVgInfo(SCtgDBCache *dbCache);
+int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache);
+int32_t ctgTbMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist);
+int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta);
+int32_t ctgReadTbSverFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *sver, int32_t *tbType, uint64_t *suid, char *stbName);
+int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass);
+int32_t ctgPutRmDBToQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId);
+int32_t ctgPutRmStbToQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncReq);
+int32_t ctgPutRmTbToQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncReq);
+int32_t ctgPutUpdateVgToQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, SDBVgInfo* dbInfo, bool syncReq);
+int32_t ctgPutUpdateTbToQueue(SCatalog* pCtg, STableMetaOutput *output, bool syncReq);
+int32_t ctgPutUpdateUserToQueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq);
+int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type);
+int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size);
+int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size);
+int32_t ctgUpdateTbMetaToCache(SCatalog* pCtg, STableMetaOutput* pOut, bool syncReq);
+int32_t ctgStartUpdateThread();
+int32_t ctgRelaunchGetTbMetaTask(SCtgTask *pTask);
+
+
+
+int32_t ctgProcessRspMsg(void* out, int32_t reqType, char* msg, int32_t msgSize, int32_t rspCode, char* target);
+int32_t ctgGetDBVgInfoFromMnode(CTG_PARAMS, SBuildUseDBInput *input, SUseDbOutput *out, SCtgTask* pTask);
+int32_t ctgGetQnodeListFromMnode(CTG_PARAMS, SArray *out, SCtgTask* pTask);
+int32_t ctgGetDBCfgFromMnode(CTG_PARAMS, const char *dbFName, SDbCfgInfo *out, SCtgTask* pTask);
+int32_t ctgGetIndexInfoFromMnode(CTG_PARAMS, const char *indexName, SIndexInfo *out, SCtgTask* pTask);
+int32_t ctgGetUdfInfoFromMnode(CTG_PARAMS, const char *funcName, SFuncInfo *out, SCtgTask* pTask);
+int32_t ctgGetUserDbAuthFromMnode(CTG_PARAMS, const char *user, SGetUserAuthRsp *out, SCtgTask* pTask);
+int32_t ctgGetTbMetaFromMnodeImpl(CTG_PARAMS, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTask* pTask);
+int32_t ctgGetTbMetaFromMnode(CTG_PARAMS, const SName* pTableName, STableMetaOutput* out, SCtgTask* pTask);
+int32_t ctgGetTbMetaFromVnode(CTG_PARAMS, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* out, SCtgTask* pTask);
+
+int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq* pReq, catalogCallback fp, void* param);
+int32_t ctgLaunchJob(SCtgJob *pJob);
+int32_t ctgMakeAsyncRes(SCtgJob *pJob);
+
+int32_t ctgCloneVgInfo(SDBVgInfo *src, SDBVgInfo **dst);
+int32_t ctgCloneMetaOutput(STableMetaOutput *output, STableMetaOutput **pOutput);
+int32_t ctgGenerateVgList(SCatalog *pCtg, SHashObj *vgHash, SArray** pList);
+void ctgFreeJob(void* job);
+void ctgFreeHandle(SCatalog* pCtg);
+void ctgFreeVgInfo(SDBVgInfo *vgInfo);
+int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName *pTableName, SVgroupInfo *pVgroup);
+void ctgResetTbMetaTask(SCtgTask* pTask);
+void ctgFreeDbCache(SCtgDBCache *dbCache);
+int32_t ctgStbVersionSortCompare(const void* key1, const void* key2);
+int32_t ctgDbVgVersionSortCompare(const void* key1, const void* key2);
+int32_t ctgStbVersionSearchCompare(const void* key1, const void* key2);
+int32_t ctgDbVgVersionSearchCompare(const void* key1, const void* key2);
+void ctgFreeSTableMetaOutput(STableMetaOutput* pOutput);
+int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* target);
+
+
+extern SCatalogMgmt gCtgMgmt;
+extern SCtgDebug gCTGDebug;
+extern SCtgAsyncFps gCtgAsyncFps[];
 
 #ifdef __cplusplus
 }

source/libs/catalog/inc/ctgRemote.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_CATALOG_REMOTE_H_
+#define _TD_CATALOG_REMOTE_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct SCtgTaskCallbackParam {
+  uint64_t                queryId;
+  int64_t                 refId;
+  uint64_t                taskId;
+  int32_t                 reqType;
+} SCtgTaskCallbackParam;
+
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_CATALOG_REMOTE_H_*/

source/libs/executor/src/executorimpl.c

@@ -13,7 +13,6 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include <executorimpl.h>
 #include "filter.h"
 #include "function.h"
 #include "functionMgt.h"

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*/) {

source/libs/planner/test/planTestUtil.cpp

@@ -14,6 +14,7 @@
  */
 
 #include "planTestUtil.h"
+#include <getopt.h>
 
 #include <algorithm>
 #include <array>

taos-tools @ 788929bd

-Subproject commit 0aad27d725f4ee6b18daf1db0c07d933aed16eea
+Subproject commit 788929bdc475d264d8306ceff30f7df006fd18d8