Merge remote-tracking branch 'origin/3.0' into fix/mnode

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

cmake/cmake.options

@@ -146,5 +146,6 @@ option(
 option(
    BUILD_WITH_INVERTEDINDEX
    "If use invertedIndex" 
-   ON 
+   OFF 
 )
+

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/07-develop/01-connect/index.md

@@ -33,7 +33,7 @@ TDengine 提供了丰富的应用程序开发接口，为了便于用户快速
 关键不同点在于：
 
 1. 使用 REST 连接，用户无需安装客户端驱动程序 taosc，具有跨平台易用的优势，但性能要下降 30%左右。
-2. 使用原生连接可以体验 TDengine 的全部功能，如[参数绑定接口](/reference/connector/cpp#参数绑定-api)、[订阅](reference/connector/cpp#数据订阅接口)等等。
+2. 使用原生连接可以体验 TDengine 的全部功能，如[参数绑定接口](/reference/connector/cpp#参数绑定-api)、[订阅](/reference/connector/cpp#订阅和消费-api)等等。
 
 ## 安装客户端驱动 taosc
 

docs-cn/12-taos-sql/02-database.md

@@ -20,21 +20,21 @@ CREATE DATABASE [IF NOT EXISTS] db_name [KEEP keep] [DAYS days] [UPDATE 1];
 3. 数据库名最大长度为 33；
 4. 一条 SQL 语句的最大长度为 65480 个字符；
 5. 创建数据库时可用的参数有：
-   - cache: [Description](/reference/config/#cache)
-   - blocks: [Description](/reference/config/#blocks)
-   - days: [Description](/reference/config/#days)
-   - keep: [Description](/reference/config/#keep)
-   - minRows: [Description](/reference/config/#minrows)
-   - maxRows: [Description](/reference/config/#maxrows)
-   - wal: [Description](/reference/config/#wallevel)
-   - fsync: [Description](/reference/config/#fsync)
-   - update: [Description](/reference/config/#update)
-   - cacheLast: [Description](/reference/config/#cachelast)
-   - replica: [Description](/reference/config/#replica)
-   - quorum: [Description](/reference/config/#quorum)
-   - maxVgroupsPerDb: [Description](/reference/config/#maxvgroupsperdb)
-   - comp: [Description](/reference/config/#comp)
-   - precision: [Description](/reference/config/#precision)
+   - cache: [详细说明](/reference/config/#cache)
+   - blocks: [详细说明](/reference/config/#blocks)
+   - days: [详细说明](/reference/config/#days)
+   - keep: [详细说明](/reference/config/#keep)
+   - minRows: [详细说明](/reference/config/#minrows)
+   - maxRows: [详细说明](/reference/config/#maxrows)
+   - wal: [详细说明](/reference/config/#wallevel)
+   - fsync: [详细说明](/reference/config/#fsync)
+   - update: [详细说明](/reference/config/#update)
+   - cacheLast: [详细说明](/reference/config/#cachelast)
+   - replica: [详细说明](/reference/config/#replica)
+   - quorum: [详细说明](/reference/config/#quorum)
+   - maxVgroupsPerDb: [详细说明](/reference/config/#maxvgroupsperdb)
+   - comp: [详细说明](/reference/config/#comp)
+   - precision: [详细说明](/reference/config/#precision)
 6. 请注意上面列出的所有参数都可以配置在配置文件 `taosd.cfg` 中作为创建数据库时使用的默认配置， `create database` 的参数中明确指定的会覆盖配置文件中的设置。
 
 :::

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/07-develop/01-connect/index.md

@@ -33,7 +33,7 @@ Either way, same or similar APIs are provided by connectors to access database o
 Key differences：
 
 1. With REST connection, it's not necessary to install TDengine client driver (taosc), it's more friendly for cross-platform with the cost of 30% performance downgrade. When taosc has an upgrade, application does not need to make changes. 
-2. With native connection, full compatibility of TDengine can be utilized, like [Parameter Binding](/reference/connector/cpp#Parameter Binding-api), [Subscription](reference/connector/cpp#Subscription), etc. But taosc has to be installed, some platforms may not be supported.
+2. With native connection, full compatibility of TDengine can be utilized, like [Parameter Binding](/reference/connector/cpp#parameter-binding-api), [Subscription](/reference/connector/cpp#subscription-and-consumption-api), etc. But taosc has to be installed, some platforms may not be supported.
 
 ## Install Client Driver taosc
 

docs-en/12-taos-sql/02-database.md

@@ -34,7 +34,7 @@ CREATE DATABASE [IF NOT EXISTS] db_name [KEEP keep] [DAYS days] [UPDATE 1];
    - quorum: [Description](/reference/config/#quorum)
    - maxVgroupsPerDb: [Description](/reference/config/#maxvgroupsperdb)
    - comp: [Description](/reference/config/#comp)
-   - precision: [Description](reference/config/#precision)
+   - precision: [Description](/reference/config/#precision)
 6. Please be noted that all of the parameters mentioned in this section can be configured in configuration file `taosd.cfg` at server side and used by default,  can be override if they are specified in `create database` statement.
    
 :::

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.
 

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

@@ -78,7 +78,7 @@ Manually install the following tools.
 - Install [Python](https://www.python.org/downloads/) 2.7 (`v3.x.x` is not supported) and execute `npm config set python python2.7`.
 - Go to the `cmd` command-line interface, `npm config set msvs_version 2017`
 
-Refer to Microsoft's Node.js User Manual [Microsoft's Node.js Guidelines for Windows](https://github.com/Microsoft/nodejs-guidelines/blob/master/windows- environment. md#compiling-native-addon-modules).
+Refer to Microsoft's Node.js User Manual [Microsoft's Node.js Guidelines for Windows](https://github.com/Microsoft/nodejs-guidelines/blob/master/windows-environment.md#compiling-native-addon-modules).
 
 If using ARM64 Node.js on Windows 10 ARM, you must add "Visual C++ compilers and libraries for ARM64" and "Visual C++ ATL for ARM64".
 

docs-en/25-application/03-immigrate.md

@@ -32,7 +32,7 @@ We will explain how to migrate OpenTSDB applications to TDengine quickly, secure
 The following figure (Figure 1) shows the system's overall architecture for a typical DevOps application scenario.
 
 **Figure 1. Typical architecture in a DevOps scenario**
-Figure 1. [IT-DevOps-Solutions-Immigrate-OpenTSDB-Arch](/img/IT-DevOps-Solutions-Immigrate-OpenTSDB-Arch.jpg "Figure 1. Typical architecture in a DevOps scenario")
+![IT-DevOps-Solutions-Immigrate-OpenTSDB-Arch](/img/IT-DevOps-Solutions-Immigrate-OpenTSDB-Arch.jpg "Figure 1. Typical architecture in a DevOps scenario")
 
 In this application scenario, there are Agent tools deployed in the application environment to collect machine metrics, network metrics, and application metrics. Data collectors to aggregate information collected by agents, systems for persistent data storage and management, and tools for monitoring data visualization (e.g., Grafana, etc.).
 
@@ -75,7 +75,7 @@ After writing the data to TDengine properly, you can adapt Grafana to visualize
 TDengine provides two sets of Dashboard templates by default, and users only need to import the templates from the Grafana directory into Grafana to activate their use.
 
 **Importing Grafana Templates** Figure 2.
-! [](/img/IT-DevOps-Solutions-Immigrate-OpenTSDB-Dashboard.jpg "Figure 2. Importing a Grafana Template")
+![](/img/IT-DevOps-Solutions-Immigrate-OpenTSDB-Dashboard.jpg "Figure 2. Importing a Grafana Template")
 
 After the above steps, you completed the migration to replace OpenTSDB with TDengine. You can see that the whole process is straightforward, there is no need to write any code, and only some configuration files need to be adjusted to meet the migration work.
 
@@ -88,7 +88,7 @@ In most DevOps scenarios, if you have a small OpenTSDB cluster (3 or fewer nodes
 Suppose your application is particularly complex, or the application domain is not a DevOps scenario. You can continue reading subsequent chapters for a more comprehensive and in-depth look at the advanced topics of migrating an OpenTSDB application to TDengine.
 
 **Figure 3. System architecture after migration**
-! [IT-DevOps-Solutions-Immigrate-TDengine-Arch](/img/IT-DevOps-Solutions-Immigrate-TDengine-Arch.jpg "Figure 3. System architecture after migration completion")
+![IT-DevOps-Solutions-Immigrate-TDengine-Arch](/img/IT-DevOps-Solutions-Immigrate-TDengine-Arch.jpg "Figure 3. System architecture after migration completion")
 
 ## Migration evaluation and strategy for other scenarios
 
@@ -96,7 +96,7 @@ Suppose your application is particularly complex, or the application domain is n
 
 This chapter describes the differences between OpenTSDB and TDengine at the system functionality level. After reading this chapter, you can fully evaluate whether you can migrate some complex OpenTSDB-based applications to TDengine, and what you should pay attention to after migration.
 
-TDengine currently only supports Grafana for visual kanban rendering, so if your application uses front-end kanban boards other than Grafana (e.g., [TSDash](https://github.com/facebook/tsdash), [Status Wolf](https://github) .com/box/StatusWolf), etc.). You cannot directly migrate those front-end kanbans to TDengine, and the front-end kanban will need to be ported to Grafana to work correctly.
+TDengine currently only supports Grafana for visual kanban rendering, so if your application uses front-end kanban boards other than Grafana (e.g., [TSDash](https://github.com/facebook/tsdash), [Status Wolf](https://github.com/box/StatusWolf), etc.). You cannot directly migrate those front-end kanbans to TDengine, and the front-end kanban will need to be ported to Grafana to work correctly.
 
 TDengine version 2.3.0.x only supports collectd and StatsD as data collection aggregation software but will provide more data collection aggregation software in the future. If you use other data aggregators on the collection side, your application needs to be ported to these two data aggregation systems to write data correctly.
 In addition to the two data aggregator software protocols mentioned above, TDengine also supports writing data directly via InfluxDB's line protocol and OpenTSDB's data writing protocol, JSON format. You can rewrite the logic on the data push side to write data using the line protocols supported by TDengine.

include/common/taosdef.h

@@ -86,6 +86,11 @@ typedef enum {
   TSDB_RETENTION_MAX = 3
 } ERetentionLevel;
 
+typedef enum {
+  TSDB_BITMODE_DEFAULT = 0,  // 2 bits
+  TSDB_BITMODE_ONE_BIT = 1,  // 1 bit
+} EBitmapMode;
+
 extern char *qtypeStr[];
 
 #define TSDB_PORT_HTTP 11

include/common/tdataformat.h

@@ -313,8 +313,9 @@ typedef struct {
   SDataCol *cols;
 } SDataCols;
 
-static FORCE_INLINE bool tdDataColsIsBitmapI(SDataCols *pCols) { return pCols->bitmapMode != 0; }
-static FORCE_INLINE void tdDataColsSetBitmapI(SDataCols *pCols) { pCols->bitmapMode = 1; }
+static FORCE_INLINE bool tdDataColsIsBitmapI(SDataCols *pCols) { return pCols->bitmapMode != TSDB_BITMODE_DEFAULT; }
+static FORCE_INLINE void tdDataColsSetBitmapI(SDataCols *pCols) { pCols->bitmapMode = TSDB_BITMODE_ONE_BIT; }
+static FORCE_INLINE bool tdIsBitmapModeI(int8_t bitmapMode) { return bitmapMode != TSDB_BITMODE_DEFAULT; }
 
 #define keyCol(pCols)              (&((pCols)->cols[0]))                    // Key column
 #define dataColsTKeyAt(pCols, idx) ((TKEY *)(keyCol(pCols)->pData))[(idx)]  // the idx row of column-wised data

include/common/tmsg.h

@@ -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/os/osDir.h

@@ -31,6 +31,12 @@
 extern "C" {
 #endif
 
+#ifdef WINDOWS
+#define TD_TMP_DIR_PATH "C:\\Windows\\Temp\\"
+#else
+#define TD_TMP_DIR_PATH "/tmp/"
+#endif
+
 typedef struct TdDir *TdDirPtr;
 typedef struct TdDirEntry *TdDirEntryPtr;
 

include/util/taoserror.h

@@ -419,6 +419,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_TQ_META_KEY_NOT_IN_TXN        TAOS_DEF_ERROR_CODE(0, 0x0A09)
 #define TSDB_CODE_TQ_META_KEY_DUP_IN_TXN        TAOS_DEF_ERROR_CODE(0, 0x0A0A)
 #define TSDB_CODE_TQ_GROUP_NOT_SET              TAOS_DEF_ERROR_CODE(0, 0x0A0B)
+#define TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND     TAOS_DEF_ERROR_CODE(0, 0x0A0B)
 
 // wal
 #define TSDB_CODE_WAL_APP_ERROR                 TAOS_DEF_ERROR_CODE(0, 0x1000)
@@ -636,6 +637,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_COMMENT_TOO_LONG          TAOS_DEF_ERROR_CODE(0, 0x264E)
 #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)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

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 ------------------------ */
@@ -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/clientHb.c

@@ -310,6 +310,8 @@ int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) {
         taosArrayDestroy(desc.subDesc);
         desc.subDesc = NULL;
       }
+    } else {
+      desc.subDesc = NULL;
     }
 
     releaseRequest(*rid);

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/tdataformat.c

@@ -855,7 +855,7 @@ SDataCols *tdNewDataCols(int maxCols, int maxRows) {
   pCols->maxCols = maxCols;
   pCols->numOfRows = 0;
   pCols->numOfCols = 0;
-  // pCols->bitmapMode = 0; // calloc already set 0
+  pCols->bitmapMode = TSDB_BITMODE_DEFAULT;
 
   if (maxCols > 0) {
     pCols->cols = (SDataCol *)taosMemoryCalloc(maxCols, sizeof(SDataCol));
@@ -899,7 +899,7 @@ int tdInitDataCols(SDataCols *pCols, STSchema *pSchema) {
 #endif
 
   pCols->numOfRows = 0;
-  pCols->bitmapMode = 0;
+  pCols->bitmapMode = TSDB_BITMODE_DEFAULT;
   pCols->numOfCols = schemaNCols(pSchema);
 
   for (i = 0; i < schemaNCols(pSchema); ++i) {

source/common/src/trow.c

@@ -341,18 +341,19 @@ int32_t tdSetBitmapValTypeN(void *pBitmap, int16_t nEle, TDRowValT valType, int8
 bool tdIsBitmapBlkNorm(const void *pBitmap, int32_t numOfBits, int8_t bitmapMode) {
   int32_t nBytes = (bitmapMode == 0 ? numOfBits / TD_VTYPE_PARTS : numOfBits / TD_VTYPE_PARTS_I);
   uint8_t vTypeByte = tdVTypeByte[bitmapMode][TD_VTYPE_NORM];
+  uint8_t *qBitmap = (uint8_t*)pBitmap;
   for (int i = 0; i < nBytes; ++i) {
-    if (*((uint8_t *)pBitmap) != vTypeByte) {
+    if (*qBitmap != vTypeByte) {
       return false;
     }
-    pBitmap = POINTER_SHIFT(pBitmap, i);
+    qBitmap = (uint8_t *)POINTER_SHIFT(pBitmap, i);
   }
 
   int32_t nLeft = numOfBits - nBytes * (bitmapMode == 0 ? TD_VTYPE_BITS : TD_VTYPE_BITS_I);
 
   for (int j = 0; j < nLeft; ++j) {
     uint8_t vType;
-    tdGetBitmapValType(pBitmap, j, &vType, bitmapMode);
+    tdGetBitmapValType(qBitmap, j, &vType, bitmapMode);
     if (vType != TD_VTYPE_NORM) {
       return false;
     }

source/dnode/mgmt/test/bnode/dbnode.cpp

@@ -14,7 +14,7 @@
 class DndTestBnode : public ::testing::Test {
  protected:
   static void SetUpTestSuite() {
-    test.Init("/tmp/dbnodeTest", 9112);
+    test.Init(TD_TMP_DIR_PATH "dbnodeTest", 9112);
     taosMsleep(1100);
   }
   static void TearDownTestSuite() { test.Cleanup(); }

source/dnode/mgmt/test/mnode/dmnode.cpp

@@ -13,7 +13,7 @@
 
 class DndTestMnode : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/dmnodeTest", 9114); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "dmnodeTest", 9114); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/mgmt/test/qnode/dqnode.cpp

@@ -13,7 +13,7 @@
 
 class DndTestQnode : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/dqnodeTest", 9111); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "dqnodeTest", 9111); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/mgmt/test/snode/dsnode.cpp

@@ -13,7 +13,7 @@
 
 class DndTestSnode : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/dsnodeTest", 9113); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "dsnodeTest", 9113); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/mgmt/test/sut/src/sut.cpp

@@ -48,7 +48,7 @@ void Testbase::Init(const char* path, int16_t port) {
   strcpy(tsDataDir, path);
   taosRemoveDir(path);
   taosMkDir(path);
-  InitLog("/tmp/td");
+  InitLog(TD_TMP_DIR_PATH "td");
 
   server.Start();
   client.Init("root", "taosdata");

source/dnode/mgmt/test/vnode/vnode.cpp

@@ -13,7 +13,7 @@
 
 class DndTestVnode : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/dvnodeTest", 9115); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "dvnodeTest", 9115); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/mnode/impl/test/acct/acct.cpp

@@ -13,7 +13,7 @@
 
 class MndTestAcct : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/acctTest", 9012); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "acctTest", 9012); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/mnode/impl/test/bnode/mbnode.cpp

@@ -18,11 +18,11 @@ class MndTestBnode : public ::testing::Test {
 
  public:
   static void SetUpTestSuite() {
-    test.Init("/tmp/mnode_test_bnode1", 9018);
+    test.Init(TD_TMP_DIR_PATH "mnode_test_bnode1", 9018);
     const char* fqdn = "localhost";
     const char* firstEp = "localhost:9018";
 
-    server2.Start("/tmp/mnode_test_bnode2", 9019);
+    server2.Start(TD_TMP_DIR_PATH "mnode_test_bnode2", 9019);
     taosMsleep(300);
   }
 

source/dnode/mnode/impl/test/db/db.cpp

@@ -13,7 +13,7 @@
 
 class MndTestDb : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_db", 9030); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "mnode_test_db", 9030); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/mnode/impl/test/dnode/mdnode.cpp

@@ -18,14 +18,14 @@ class MndTestDnode : public ::testing::Test {
 
  public:
   static void SetUpTestSuite() {
-    test.Init("/tmp/dnode_test_dnode1", 9023);
+    test.Init(TD_TMP_DIR_PATH "dnode_test_dnode1", 9023);
     const char* fqdn = "localhost";
     const char* firstEp = "localhost:9023";
 
-    // server2.Start("/tmp/dnode_test_dnode2", fqdn, 9024, firstEp);
-    // server3.Start("/tmp/dnode_test_dnode3", fqdn, 9025, firstEp);
-    // server4.Start("/tmp/dnode_test_dnode4", fqdn, 9026, firstEp);
-    // server5.Start("/tmp/dnode_test_dnode5", fqdn, 9027, firstEp);
+    // server2.Start(TD_TMP_DIR_PATH "dnode_test_dnode2", fqdn, 9024, firstEp);
+    // server3.Start(TD_TMP_DIR_PATH "dnode_test_dnode3", fqdn, 9025, firstEp);
+    // server4.Start(TD_TMP_DIR_PATH "dnode_test_dnode4", fqdn, 9026, firstEp);
+    // server5.Start(TD_TMP_DIR_PATH "dnode_test_dnode5", fqdn, 9027, firstEp);
     taosMsleep(300);
   }
 

source/dnode/mnode/impl/test/func/func.cpp

@@ -13,7 +13,7 @@
 
 class MndTestFunc : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_func", 9038); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "mnode_test_func", 9038); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

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

@@ -18,11 +18,11 @@ class MndTestMnode : public ::testing::Test {
 
  public:
   static void SetUpTestSuite() {
-    test.Init("/tmp/mnode_test_mnode1", 9028);
+    test.Init(TD_TMP_DIR_PATH "mnode_test_mnode1", 9028);
     const char* fqdn = "localhost";
     const char* firstEp = "localhost:9028";
 
-    // server2.Start("/tmp/mnode_test_mnode2", fqdn, 9029, firstEp);
+    // server2.Start(TD_TMP_DIR_PATH "mnode_test_mnode2", fqdn, 9029, firstEp);
     taosMsleep(300);
   }
 

source/dnode/mnode/impl/test/profile/profile.cpp

@@ -13,7 +13,7 @@
 
 class MndTestProfile : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_profile", 9031); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "mnode_test_profile", 9031); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/mnode/impl/test/qnode/qnode.cpp

@@ -18,11 +18,11 @@ class MndTestQnode : public ::testing::Test {
 
  public:
   static void SetUpTestSuite() {
-    test.Init("/tmp/mnode_test_qnode1", 9014);
+    test.Init(TD_TMP_DIR_PATH "mnode_test_qnode1", 9014);
     const char* fqdn = "localhost";
     const char* firstEp = "localhost:9014";
 
-    // server2.Start("/tmp/mnode_test_qnode2", fqdn, 9015, firstEp);
+    // server2.Start(TD_TMP_DIR_PATH "mnode_test_qnode2", fqdn, 9015, firstEp);
     taosMsleep(300);
   }
 

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

@@ -31,7 +31,7 @@ class MndTestSdb : public ::testing::Test {
     tsLogEmbedded = 1;
     tsAsyncLog = 0;
 
-    const char *path = "/tmp/td";
+    const char *path = TD_TMP_DIR_PATH "td";
     taosRemoveDir(path);
     taosMkDir(path);
     tstrncpy(tsLogDir, path, PATH_MAX);
@@ -385,7 +385,7 @@ TEST_F(MndTestSdb, 01_Write_Str) {
   mnode.v100 = 100;
   mnode.v200 = 200;
   opt.pMnode = &mnode;
-  opt.path = "/tmp/mnode_test_sdb";
+  opt.path = TD_TMP_DIR_PATH "mnode_test_sdb";
   taosRemoveDir(opt.path);
 
   SSdbTable strTable1;
@@ -727,7 +727,7 @@ TEST_F(MndTestSdb, 01_Read_Str) {
   mnode.v100 = 100;
   mnode.v200 = 200;
   opt.pMnode = &mnode;
-  opt.path = "/tmp/mnode_test_sdb";
+  opt.path = TD_TMP_DIR_PATH "mnode_test_sdb";
 
   SSdbTable strTable1;
   memset(&strTable1, 0, sizeof(SSdbTable));

source/dnode/mnode/impl/test/show/show.cpp

@@ -13,7 +13,7 @@
 
 class MndTestShow : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_show", 9021); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "mnode_test_show", 9021); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/mnode/impl/test/sma/sma.cpp

@@ -13,7 +13,7 @@
 
 class MndTestSma : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_sma", 9035); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "mnode_test_sma", 9035); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/mnode/impl/test/snode/snode.cpp

@@ -18,11 +18,11 @@ class MndTestSnode : public ::testing::Test {
 
  public:
   static void SetUpTestSuite() {
-    test.Init("/tmp/mnode_test_snode1", 9016);
+    test.Init(TD_TMP_DIR_PATH "mnode_test_snode1", 9016);
     const char* fqdn = "localhost";
     const char* firstEp = "localhost:9016";
 
-    // server2.Start("/tmp/mnode_test_snode2", fqdn, 9017, firstEp);
+    // server2.Start(TD_TMP_DIR_PATH "mnode_test_snode2", fqdn, 9017, firstEp);
     taosMsleep(300);
   }
 

source/dnode/mnode/impl/test/stb/stb.cpp

@@ -13,7 +13,7 @@
 
 class MndTestStb : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_stb", 9034); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "mnode_test_stb", 9034); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/mnode/impl/test/topic/topic.cpp

@@ -13,7 +13,7 @@
 
 class MndTestTopic : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_topic", 9039); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "mnode_test_topic", 9039); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

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

@@ -14,10 +14,10 @@
 class MndTestTrans1 : public ::testing::Test {
  protected:
   static void SetUpTestSuite() {
-    test.Init("/tmp/mnode_test_trans1", 9013);
+    test.Init(TD_TMP_DIR_PATH "mnode_test_trans1", 9013);
     const char* fqdn = "localhost";
     const char* firstEp = "localhost:9013";
-    // server2.Start("/tmp/mnode_test_trans2", fqdn, 9020, firstEp);
+    // server2.Start(TD_TMP_DIR_PATH "mnode_test_trans2", fqdn, 9020, firstEp);
   }
 
   static void TearDownTestSuite() {
@@ -26,7 +26,7 @@ class MndTestTrans1 : public ::testing::Test {
   }
 
   static void KillThenRestartServer() {
-    char      file[PATH_MAX] = "/tmp/mnode_test_trans1/mnode/data/sdb.data";
+    char      file[PATH_MAX] = TD_TMP_DIR_PATH "mnode_test_trans1/mnode/data/sdb.data";
     TdFilePtr pFile = taosOpenFile(file, TD_FILE_READ);
     int32_t   size = 3 * 1024 * 1024;
     void*     buffer = taosMemoryMalloc(size);

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

@@ -46,7 +46,7 @@ class MndTestTrans2 : public ::testing::Test {
     tsLogEmbedded = 1;
     tsAsyncLog = 0;
 
-    const char *logpath = "/tmp/td";
+    const char *logpath = TD_TMP_DIR_PATH "td";
     taosRemoveDir(logpath);
     taosMkDir(logpath);
     tstrncpy(tsLogDir, logpath, PATH_MAX);
@@ -76,7 +76,7 @@ class MndTestTrans2 : public ::testing::Test {
 
     tsTransPullupInterval = 1;
 
-    const char *mnodepath = "/tmp/mnode_test_trans";
+    const char *mnodepath = TD_TMP_DIR_PATH "mnode_test_trans";
     taosRemoveDir(mnodepath);
     pMnode = mndOpen(mnodepath, &opt);
     mndStart(pMnode);

source/dnode/mnode/impl/test/user/user.cpp

@@ -13,7 +13,7 @@
 
 class MndTestUser : public ::testing::Test {
  protected:
-  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_user", 9011); }
+  static void SetUpTestSuite() { test.Init(TD_TMP_DIR_PATH "mnode_test_user", 9011); }
   static void TearDownTestSuite() { test.Cleanup(); }
 
   static Testbase test;

source/dnode/vnode/CMakeLists.txt

@@ -76,9 +76,14 @@ target_link_libraries(
     #PUBLIC scalar
     PUBLIC transport
     PUBLIC stream
+    PUBLIC index
 )
 target_compile_definitions(vnode PUBLIC -DMETA_REFACT)
-
+if (${BUILD_WITH_INVERTEDINDEX})
+  add_definitions(-DUSE_INVERTED_INDEX) 
+endif(${BUILD_WITH_INVERTEDINDEX})
 if(${BUILD_TEST})
     add_subdirectory(test)
 endif(${BUILD_TEST})
+
+

source/dnode/vnode/inc/vnode.h

@@ -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/meta.h

@@ -17,6 +17,7 @@
 #define _TD_VNODE_META_H_
 
 #include "vnodeInt.h"
+#include "index.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -61,16 +62,20 @@ static FORCE_INLINE tb_uid_t metaGenerateUid(SMeta* pMeta) { return tGenIdPI64()
 struct SMeta {
   TdThreadRwlock lock;
 
-  char*     path;
-  SVnode*   pVnode;
-  TDB*      pEnv;
-  TXN       txn;
-  TTB*      pTbDb;
-  TTB*      pSkmDb;
-  TTB*      pUidIdx;
-  TTB*      pNameIdx;
-  TTB*      pCtbIdx;
-  TTB*      pTagIdx;
+  char*   path;
+  SVnode* pVnode;
+  TDB*    pEnv;
+  TXN     txn;
+  TTB*    pTbDb;
+  TTB*    pSkmDb;
+  TTB*    pUidIdx;
+  TTB*    pNameIdx;
+  TTB*    pCtbIdx;
+#ifdef USE_INVERTED_INDEX
+  void* pTagIvtIdx;
+#else
+  TTB* pTagIdx;
+#endif
   TTB*      pTtlIdx;
   TTB*      pSmaIdx;
   SMetaIdx* pIdx;

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

@@ -53,10 +53,10 @@ int metaOpenIdx(SMeta *pMeta) {
 #endif
 
 #ifdef USE_INVERTED_INDEX
-  SIndexOpts opts;
-  if (indexOpen(&opts, pMeta->path, &pMeta->pIdx->pIdx) != 0) {
-    return -1;
-  }
+  // SIndexOpts opts;
+  // if (indexOpen(&opts, pMeta->path, &pMeta->pIdx->pIdx) != 0) {
+  //  return -1;
+  //}
 
 #endif
   return 0;
@@ -71,36 +71,37 @@ void metaCloseIdx(SMeta *pMeta) { /* TODO */
 #endif
 
 #ifdef USE_INVERTED_INDEX
-  SIndexOpts opts;
-  if (indexClose(pMeta->pIdx->pIdx) != 0) {
-    return -1;
-  }
+  // SIndexOpts opts;
+  // if (indexClose(pMeta->pIdx->pIdx) != 0) {
+  //  return -1;
+  //}
+  // return 0;
 
 #endif
 }
 
 int metaSaveTableToIdx(SMeta *pMeta, const STbCfg *pTbCfg) {
 #ifdef USE_INVERTED_INDEX
-  if (pTbCfgs->type == META_CHILD_TABLE) {
-    char    buf[8] = {0};
-    int16_t colId = (kvRowColIdx(pTbCfg->ctbCfg.pTag))[0].colId;
-    sprintf(buf, "%d", colId);  // colname
-
-    char *pTagVal = (char *)tdGetKVRowValOfCol(pTbCfg->ctbCfg.pTag, (kvRowColIdx(pTbCfg->ctbCfg.pTag))[0].colId);
-
-    tb_uid_t         suid = pTbCfg->ctbCfg.suid;  // super id
-    tb_uid_t         tuid = 0;                    // child table uid
-    SIndexMultiTerm *terms = indexMultiTermCreate();
-    SIndexTerm      *term =
-        indexTermCreate(suid, ADD_VALUE, TSDB_DATA_TYPE_BINARY, buf, strlen(buf), pTagVal, strlen(pTagVal), tuid);
-    indexMultiTermAdd(terms, term);
-
-    int ret = indexPut(pMeta->pIdx->pIdx, terms);
-    indexMultiTermDestroy(terms);
-    return ret;
-  } else {
-    return DB_DONOTINDEX;
-  }
+  // if (pTbCfgs->type == META_CHILD_TABLE) {
+  //  char    buf[8] = {0};
+  //  int16_t colId = (kvRowColIdx(pTbCfg->ctbCfg.pTag))[0].colId;
+  //  sprintf(buf, "%d", colId);  // colname
+
+  //  char *pTagVal = (char *)tdGetKVRowValOfCol(pTbCfg->ctbCfg.pTag, (kvRowColIdx(pTbCfg->ctbCfg.pTag))[0].colId);
+
+  //  tb_uid_t         suid = pTbCfg->ctbCfg.suid;  // super id
+  //  tb_uid_t         tuid = 0;                    // child table uid
+  //  SIndexMultiTerm *terms = indexMultiTermCreate();
+  //  SIndexTerm      *term =
+  //      indexTermCreate(suid, ADD_VALUE, TSDB_DATA_TYPE_BINARY, buf, strlen(buf), pTagVal, strlen(pTagVal), tuid);
+  //  indexMultiTermAdd(terms, term);
+
+  //  int ret = indexPut(pMeta->pIdx->pIdx, terms);
+  //  indexMultiTermDestroy(terms);
+  //  return ret;
+  //} else {
+  //  return DB_DONOTINDEX;
+  //}
 #endif
   // TODO
   return 0;
@@ -112,4 +113,4 @@ int metaRemoveTableFromIdx(SMeta *pMeta, tb_uid_t uid) {
 #endif
   // TODO
   return 0;
-}
+}
\ No newline at end of file

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

@@ -93,11 +93,24 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta) {
   }
 
   // open pTagIdx
+#ifdef USE_INVERTED_INDEX
+  // TODO(yihaoDeng), refactor later
+  char indexFullPath[128] = {0};
+  sprintf(indexFullPath, "%s/%s", pMeta->path, "invert");
+  taosMkDir(indexFullPath);
+  ret = indexOpen(indexOptsCreate(), indexFullPath, (SIndex **)&pMeta->pTagIvtIdx);
+  if (ret < 0) {
+    metaError("vgId:%d failed to open meta tag index since %s", TD_VID(pVnode), tstrerror(terrno));
+    goto _err;
+  }
+
+#else
   ret = tdbTbOpen("tag.idx", -1, 0, tagIdxKeyCmpr, pMeta->pEnv, &pMeta->pTagIdx);
   if (ret < 0) {
     metaError("vgId:%d failed to open meta tag index since %s", TD_VID(pVnode), tstrerror(terrno));
     goto _err;
   }
+#endif
 
   // open pTtlIdx
   ret = tdbTbOpen("ttl.idx", sizeof(STtlIdxKey), 0, ttlIdxKeyCmpr, pMeta->pEnv, &pMeta->pTtlIdx);
@@ -128,7 +141,11 @@ _err:
   if (pMeta->pIdx) metaCloseIdx(pMeta);
   if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx);
   if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx);
+#ifdef USE_INVERTED_INDEX
+  if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx);
+#else
   if (pMeta->pTagIdx) tdbTbClose(pMeta->pTagIdx);
+#endif
   if (pMeta->pCtbIdx) tdbTbClose(pMeta->pCtbIdx);
   if (pMeta->pNameIdx) tdbTbClose(pMeta->pNameIdx);
   if (pMeta->pUidIdx) tdbTbClose(pMeta->pUidIdx);
@@ -145,7 +162,11 @@ int metaClose(SMeta *pMeta) {
     if (pMeta->pIdx) metaCloseIdx(pMeta);
     if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx);
     if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx);
+#ifdef USE_INVERTED_INDEX
+    if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx);
+#else
     if (pMeta->pTagIdx) tdbTbClose(pMeta->pTagIdx);
+#endif
     if (pMeta->pCtbIdx) tdbTbClose(pMeta->pCtbIdx);
     if (pMeta->pNameIdx) tdbTbClose(pMeta->pNameIdx);
     if (pMeta->pUidIdx) tdbTbClose(pMeta->pUidIdx);

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

@@ -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);
@@ -420,7 +423,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);
 
@@ -820,16 +825,27 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
   pTagData = tdGetKVRowValOfCol((const SKVRow)pCtbEntry->ctbEntry.pTags, pTagColumn->colId);
 
   // update tag index
+#ifdef USE_INVERTED_INDEX
+  tb_uid_t suid = pCtbEntry->ctbEntry.suid;
+  tb_uid_t tuid = pCtbEntry->uid;
+
+  SIndexMultiTerm *tmGroup = indexMultiTermCreate();
+
+  SIndexTerm *tm = indexTermCreate(suid, ADD_VALUE, pTagColumn->type, pTagColumn->name, sizeof(pTagColumn->name),
+                                   pTagData, pTagData == NULL ? 0 : strlen(pTagData));
+  indexMultiTermAdd(tmGroup, tm);
+  int ret = indexPut((SIndex *)pMeta->pTagIvtIdx, tmGroup, tuid);
+  indexMultiTermDestroy(tmGroup);
+#else
   if (metaCreateTagIdxKey(pCtbEntry->ctbEntry.suid, pTagColumn->colId, pTagData, pTagColumn->type, pCtbEntry->uid,
                           &pTagIdxKey, &nTagIdxKey) < 0) {
     return -1;
   }
   tdbTbInsert(pMeta->pTagIdx, pTagIdxKey, nTagIdxKey, NULL, 0, &pMeta->txn);
   metaDestroyTagIdxKey(pTagIdxKey);
-
+#endif
   tDecoderClear(&dc);
   tdbFree(pData);
-
   return 0;
 }
 

source/dnode/vnode/src/tq/tq.c

@@ -600,6 +600,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
           SSDataBlock block = {0};
           if (tqRetrieveDataBlock(&block.pDataBlock, pReader, &block.info.groupId, &block.info.uid, &block.info.rows,
                                   &block.info.numOfCols) < 0) {
+            if (terrno == TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND) continue;
             ASSERT(0);
           }
           int32_t            dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(&block);

source/dnode/vnode/src/tq/tqRead.c

@@ -91,12 +91,22 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
   if (pHandle->sver != sversion || pHandle->cachedSchemaUid != pHandle->msgIter.suid) {
     pHandle->pSchema = metaGetTbTSchema(pHandle->pVnodeMeta, pHandle->msgIter.uid, sversion);
     if (pHandle->pSchema == NULL) {
-      tqError("cannot found schema for table: %ld, version %d", pHandle->msgIter.suid, pHandle->sver);
+      tqWarn("cannot found tsschema for table: uid: %ld (suid: %ld), version %d, possibly dropped table",
+             pHandle->msgIter.uid, pHandle->msgIter.suid, pHandle->sver);
+      /*ASSERT(0);*/
+      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
       return -1;
     }
 
     // this interface use suid instead of uid
     pHandle->pSchemaWrapper = metaGetTableSchema(pHandle->pVnodeMeta, pHandle->msgIter.suid, sversion, true);
+    if (pHandle->pSchemaWrapper == NULL) {
+      tqWarn("cannot found schema wrapper for table: suid: %ld, version %d, possibly dropped table",
+             pHandle->msgIter.suid, pHandle->sver);
+      /*ASSERT(0);*/
+      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
+      return -1;
+    }
     pHandle->sver = sversion;
     pHandle->cachedSchemaUid = pHandle->msgIter.suid;
   }

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

@@ -1138,6 +1138,9 @@ int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDF
       memcpy(tptr, pDataCol->pData, flen);
       if (tBitmaps > 0) {
         bptr = POINTER_SHIFT(pBlockData, lsize + flen);
+        if (isSuper && !tdDataColsIsBitmapI(pDataCols)) {
+          tdMergeBitmap((uint8_t *)pDataCol->pBitmap, rowsToWrite, (uint8_t *)pDataCol->pBitmap);
+        }
         memcpy(bptr, pDataCol->pBitmap, tBitmaps);
         tBitmapsLen = tBitmaps;
         flen += tBitmapsLen;
@@ -1503,13 +1506,16 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt
       tSkipListIterNext(pCommitIter->pIter);
     } else {
       if (lastKey != key1) {
+        if (lastKey != TSKEY_INITIAL_VAL) {
+          ++pTarget->numOfRows;
+        }
         lastKey = key1;
-        ++pTarget->numOfRows;
       }
 
       // copy disk data
       for (int i = 0; i < pDataCols->numOfCols; ++i) {
         SCellVal sVal = {0};
+        // no duplicated TS keys in pDataCols from file
         if (tdGetColDataOfRow(&sVal, pDataCols->cols + i, *iter, pDataCols->bitmapMode) < 0) {
           TASSERT(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/tsdbRead.c

@@ -1732,6 +1732,7 @@ static int32_t mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capa
           if (*lastRowKey != TSKEY_INITIAL_VAL) {
             ++(*curRow);
           }
+          *lastRowKey = rowKey;
           ++nResult;
         } else if (update) {
           mergeOption = 2;
@@ -1739,8 +1740,6 @@ static int32_t mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capa
           mergeOption = 0;
           break;
         }
-
-        *lastRowKey = rowKey;
       }
     } else {
       // TODO: use STSRowIter
@@ -1753,6 +1752,7 @@ static int32_t mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capa
           if (*lastRowKey != TSKEY_INITIAL_VAL) {
             ++(*curRow);
           }
+          *lastRowKey = rowKey;
           ++nResult;
         } else if (update) {
           mergeOption = 2;
@@ -1760,7 +1760,6 @@ static int32_t mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capa
           mergeOption = 0;
           break;
         }
-        *lastRowKey = rowKey;
       } else {
         SKvRowIdx* pColIdx = tdKvRowColIdxAt(row, chosen_itr - 1);
         colId = pColIdx->colId;
@@ -1965,7 +1964,7 @@ static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInf
   SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
   assert(pCols->cols[0].type == TSDB_DATA_TYPE_TIMESTAMP && pCols->cols[0].colId == PRIMARYKEY_TIMESTAMP_COL_ID &&
          cur->pos >= 0 && cur->pos < pBlock->numOfRows);
-
+  // Even Multi-Version supported, the records with duplicated TSKEY would be merged inside of tsdbLoadData interface.
   TSKEY* tsArray = pCols->cols[0].pData;
   assert(pCols->numOfRows == pBlock->numOfRows && tsArray[0] == pBlock->keyFirst &&
          tsArray[pBlock->numOfRows - 1] == pBlock->keyLast);
@@ -1995,6 +1994,7 @@ static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInf
 
   int32_t pos = cur->pos;
   cur->win = TSWINDOW_INITIALIZER;
+  bool adjustPos = false;
 
   // no data in buffer, load data from file directly
   if (pCheckInfo->iiter == NULL && pCheckInfo->iter == NULL) {
@@ -2016,6 +2016,13 @@ static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInf
         break;
       }
 
+      if (adjustPos) {
+        if (key == lastKeyAppend) {
+          pos -= step;
+        }
+        adjustPos = false;
+      }
+
       if (((pos > endPos || tsArray[pos] > pTsdbReadHandle->window.ekey) && ascScan) ||
           ((pos < endPos || tsArray[pos] < pTsdbReadHandle->window.ekey) && !ascScan)) {
         break;
@@ -2107,7 +2114,9 @@ static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInf
           moveToNextRowInMem(pCheckInfo);
 
           pos += step;
+          adjustPos = true;
         } else {
+          // discard the memory record
           moveToNextRowInMem(pCheckInfo);
         }
       } else if ((key > tsArray[pos] && ascScan) || (key < tsArray[pos] && !ascScan)) {

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

@@ -20,11 +20,11 @@
 static void tsdbResetReadTable(SReadH *pReadh);
 static void tsdbResetReadFile(SReadH *pReadh);
 static int  tsdbLoadBlockOffset(SReadH *pReadh, SBlock *pBlock);
-static int  tsdbLoadBlockDataImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDataCols);
+static int  tsdbLoadBlockDataImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDataCols, int8_t bitmapMode);
 static int  tsdbCheckAndDecodeColumnData(SDataCol *pDataCol, void *content, int32_t len, int32_t bitmapLen, int8_t comp,
                                          int numOfRows, int numOfBitmaps, int maxPoints, char *buffer, int bufferSize);
 static int  tsdbLoadBlockDataColsImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDataCols, const int16_t *colIds,
-                                      int numOfColIds);
+                                      int numOfColIds, int8_t bitmapMode);
 static int  tsdbLoadColData(SReadH *pReadh, SDFile *pDFile, SBlock *pBlock, SBlockCol *pBlockCol, SDataCol *pDataCol);
 
 int tsdbInitReadH(SReadH *pReadh, STsdb *pRepo) {
@@ -266,10 +266,11 @@ int tsdbLoadBlockData(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlkInfo) {
     }
   }
 
-  if (tsdbLoadBlockDataImpl(pReadh, iBlock, pReadh->pDCols[0]) < 0) return -1;
+  
+  if (tsdbLoadBlockDataImpl(pReadh, iBlock, pReadh->pDCols[0], TSDB_BITMODE_ONE_BIT) < 0) return -1;
   for (int i = 1; i < pBlock->numOfSubBlocks; i++) {
     iBlock++;
-    if (tsdbLoadBlockDataImpl(pReadh, iBlock, pReadh->pDCols[1]) < 0) return -1;
+    if (tsdbLoadBlockDataImpl(pReadh, iBlock, pReadh->pDCols[1], TSDB_BITMODE_DEFAULT) < 0) return -1;
     // TODO: use the real maxVersion to replace the UINT64_MAX to support Multi-Version
     if (tdMergeDataCols(pReadh->pDCols[0], pReadh->pDCols[1], pReadh->pDCols[1]->numOfRows, NULL,
                         TD_SUPPORT_UPDATE(update), TD_VER_MAX) < 0)
@@ -309,10 +310,10 @@ int tsdbLoadBlockDataCols(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlkInfo,
     }
   }
 
-  if (tsdbLoadBlockDataColsImpl(pReadh, iBlock, pReadh->pDCols[0], colIds, numOfColsIds) < 0) return -1;
+  if (tsdbLoadBlockDataColsImpl(pReadh, iBlock, pReadh->pDCols[0], colIds, numOfColsIds, TSDB_BITMODE_ONE_BIT) < 0) return -1;
   for (int i = 1; i < pBlock->numOfSubBlocks; i++) {
     iBlock++;
-    if (tsdbLoadBlockDataColsImpl(pReadh, iBlock, pReadh->pDCols[1], colIds, numOfColsIds) < 0) return -1;
+    if (tsdbLoadBlockDataColsImpl(pReadh, iBlock, pReadh->pDCols[1], colIds, numOfColsIds, TSDB_BITMODE_DEFAULT) < 0) return -1;
     // TODO: use the real maxVersion to replace the UINT64_MAX to support Multi-Version
     if (tdMergeDataCols(pReadh->pDCols[0], pReadh->pDCols[1], pReadh->pDCols[1]->numOfRows, NULL,
                         TD_SUPPORT_UPDATE(update), TD_VER_MAX) < 0)
@@ -543,14 +544,14 @@ static void tsdbResetReadFile(SReadH *pReadh) {
   tsdbCloseDFileSet(TSDB_READ_FSET(pReadh));
 }
 
-static int tsdbLoadBlockDataImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDataCols) {
+static int tsdbLoadBlockDataImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDataCols, int8_t bitmapMode) {
   ASSERT(pBlock->numOfSubBlocks == 0 || pBlock->numOfSubBlocks == 1);
 
   SDFile *pDFile = (pBlock->last) ? TSDB_READ_LAST_FILE(pReadh) : TSDB_READ_DATA_FILE(pReadh);
 
   tdResetDataCols(pDataCols);
 
-  if (tsdbIsSupBlock(pBlock)) {
+  if (tdIsBitmapModeI(bitmapMode)) {
     tdDataColsSetBitmapI(pDataCols);
   }
 
@@ -730,7 +731,7 @@ static int tsdbCheckAndDecodeColumnData(SDataCol *pDataCol, void *content, int32
 }
 
 static int tsdbLoadBlockDataColsImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDataCols, const int16_t *colIds,
-                                     int numOfColIds) {
+                                     int numOfColIds, int8_t bitmapMode) {
   ASSERT(pBlock->numOfSubBlocks == 0 || pBlock->numOfSubBlocks == 1);
   ASSERT(colIds[0] == PRIMARYKEY_TIMESTAMP_COL_ID);
 
@@ -739,7 +740,7 @@ static int tsdbLoadBlockDataColsImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *
 
   tdResetDataCols(pDataCols);
 
-  if (tsdbIsSupBlock(pBlock)) {
+  if (tdIsBitmapModeI(bitmapMode)) {
     tdDataColsSetBitmapI(pDataCols);
   }
 

source/dnode/vnode/test/tqMetaTest.cpp

@@ -41,7 +41,7 @@ class TqMetaUpdateAppendTest : public ::testing::Test {
   void TearDown() override { tqStoreClose(pMeta); }
 
   STqMetaStore* pMeta;
-  const char*   pathName = "/tmp/tq_test";
+  const char*   pathName = TD_TMP_DIR_PATH "tq_test";
 };
 
 TEST_F(TqMetaUpdateAppendTest, copyPutTest) {

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/catalog/src/ctgRemote.c

+/*
+ * 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/>.
+ */
+
+#include "trpc.h"
+#include "query.h"
+#include "tname.h"
+#include "catalogInt.h"
+#include "systable.h"
+#include "ctgRemote.h"
+#include "tref.h"
+
+int32_t ctgProcessRspMsg(void* out, int32_t reqType, char* msg, int32_t msgSize, int32_t rspCode, char* target) {
+  int32_t code = 0;
+  
+  switch (reqType) {
+    case TDMT_MND_QNODE_LIST: {
+      if (TSDB_CODE_SUCCESS != rspCode) {
+        qError("error rsp for qnode list, error:%s", tstrerror(rspCode));
+        CTG_ERR_RET(rspCode);
+      }
+      
+      code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
+      if (code) {
+        qError("Process qnode list rsp failed, error:%s", tstrerror(rspCode));
+        CTG_ERR_RET(code);
+      }
+      
+      qDebug("Got qnode list from mnode, listNum:%d", (int32_t)taosArrayGetSize(out));
+      break;
+    }
+    case TDMT_MND_USE_DB: {
+      if (TSDB_CODE_SUCCESS != rspCode) {
+        qError("error rsp for use db, error:%s, dbFName:%s", tstrerror(rspCode), target);
+        CTG_ERR_RET(rspCode);
+      }
+      
+      code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
+      if (code) {
+        qError("Process use db rsp failed, error:%s, dbFName:%s", tstrerror(code), target);
+        CTG_ERR_RET(code);
+      }
+      
+      qDebug("Got db vgInfo from mnode, dbFName:%s", target);
+      break;
+    }
+    case TDMT_MND_GET_DB_CFG: {
+      if (TSDB_CODE_SUCCESS != rspCode) {
+        qError("error rsp for get db cfg, error:%s, db:%s", tstrerror(rspCode), target);
+        CTG_ERR_RET(rspCode);
+      }
+      
+      code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
+      if (code) {
+        qError("Process get db cfg rsp failed, error:%s, db:%s", tstrerror(code), target);
+        CTG_ERR_RET(code);
+      }
+      
+      qDebug("Got db cfg from mnode, dbFName:%s", target);
+      break;
+    }
+    case TDMT_MND_GET_INDEX: {
+      if (TSDB_CODE_SUCCESS != rspCode) {
+        qError("error rsp for get index, error:%s, indexName:%s", tstrerror(rspCode), target);
+        CTG_ERR_RET(rspCode);
+      }
+      
+      code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
+      if (code) {
+        qError("Process get index rsp failed, error:%s, indexName:%s", tstrerror(code), target);
+        CTG_ERR_RET(code);
+      }
+      
+      qDebug("Got index from mnode, indexName:%s", target);
+      break;
+    }
+    case TDMT_MND_RETRIEVE_FUNC: {
+      if (TSDB_CODE_SUCCESS != rspCode) {
+        qError("error rsp for get udf, error:%s, funcName:%s", tstrerror(rspCode), target);
+        CTG_ERR_RET(rspCode);
+      }
+      
+      code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
+      if (code) {
+        qError("Process get udf rsp failed, error:%s, funcName:%s", tstrerror(code), target);
+        CTG_ERR_RET(code);
+      }
+      
+      qDebug("Got udf from mnode, funcName:%s", target);
+      break;
+    }
+    case TDMT_MND_GET_USER_AUTH: {
+      if (TSDB_CODE_SUCCESS != rspCode) {
+        qError("error rsp for get user auth, error:%s, user:%s", tstrerror(rspCode), target);
+        CTG_ERR_RET(rspCode);
+      }
+      
+      code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
+      if (code) {
+        qError("Process get user auth rsp failed, error:%s, user:%s", tstrerror(code), target);
+        CTG_ERR_RET(code);
+      }
+      
+      qDebug("Got user auth from mnode, user:%s", target);
+      break;
+    }
+    case TDMT_MND_TABLE_META: {
+      if (TSDB_CODE_SUCCESS != rspCode) {
+        if (CTG_TABLE_NOT_EXIST(rspCode)) {
+          SET_META_TYPE_NULL(((STableMetaOutput*)out)->metaType);
+          qDebug("stablemeta not exist in mnode, tbFName:%s", target);
+          return TSDB_CODE_SUCCESS;
+        }
+        
+        qError("error rsp for stablemeta from mnode, error:%s, tbFName:%s", tstrerror(rspCode), target);
+        CTG_ERR_RET(rspCode);
+      }
+      
+      code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
+      if (code) {
+        qError("Process mnode stablemeta rsp failed, error:%s, tbFName:%s", tstrerror(code), target);
+        CTG_ERR_RET(code);
+      }
+      
+      qDebug("Got table meta from mnode, tbFName:%s", target);
+      break;
+    }
+    case TDMT_VND_TABLE_META: {
+      if (TSDB_CODE_SUCCESS != rspCode) {
+        if (CTG_TABLE_NOT_EXIST(rspCode)) {
+          SET_META_TYPE_NULL(((STableMetaOutput*)out)->metaType);
+          qDebug("tablemeta not exist in vnode, tbFName:%s", target);
+          return TSDB_CODE_SUCCESS;
+        }
+      
+        qError("error rsp for table meta from vnode, code:%s, tbFName:%s", tstrerror(rspCode), target);
+        CTG_ERR_RET(rspCode);
+      }
+      
+      code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
+      if (code) {
+        qError("Process vnode tablemeta rsp failed, code:%s, tbFName:%s", tstrerror(code), target);
+        CTG_ERR_RET(code);
+      }
+      
+      qDebug("Got table meta from vnode, tbFName:%s", target);
+      break;
+    }
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
+int32_t ctgHandleMsgCallback(void *param, const SDataBuf *pMsg, int32_t rspCode) {
+  SCtgTaskCallbackParam* cbParam = (SCtgTaskCallbackParam*)param;
+  int32_t code = 0;
+  
+  CTG_API_ENTER();
+
+  SCtgJob* pJob = taosAcquireRef(gCtgMgmt.jobPool, cbParam->refId);
+  if (NULL == pJob) {
+    qDebug("job refId %" PRIx64 " already dropped", cbParam->refId);
+    goto _return;
+  }
+
+  SCtgTask *pTask = taosArrayGet(pJob->pTasks, cbParam->taskId);
+
+  qDebug("QID:%" PRIx64 " task %d start to handle rsp %s", pJob->queryId, pTask->taskId, TMSG_INFO(cbParam->reqType + 1));
+  
+  CTG_ERR_JRET((*gCtgAsyncFps[pTask->type].handleRspFp)(pTask, cbParam->reqType, pMsg, rspCode));
+ 
+_return:
+
+  if (pJob) {
+    taosReleaseRef(gCtgMgmt.jobPool, cbParam->refId);
+  }
+  
+  taosMemoryFree(param);
+
+  CTG_API_LEAVE(code);
+}
+
+
+int32_t ctgMakeMsgSendInfo(SCtgTask* pTask, int32_t msgType, SMsgSendInfo **pMsgSendInfo) {
+  int32_t       code = 0;
+  SMsgSendInfo *msgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
+  if (NULL == msgSendInfo) {
+    qError("calloc %d failed", (int32_t)sizeof(SMsgSendInfo));
+    CTG_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
+  }
+
+  SCtgTaskCallbackParam *param = taosMemoryCalloc(1, sizeof(SCtgTaskCallbackParam));
+  if (NULL == param) {
+    qError("calloc %d failed", (int32_t)sizeof(SCtgTaskCallbackParam));
+    CTG_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
+  }
+
+  param->reqType = msgType;
+  param->queryId = pTask->pJob->queryId;
+  param->refId = pTask->pJob->refId;
+  param->taskId = pTask->taskId;
+
+  msgSendInfo->param = param;
+  msgSendInfo->fp = ctgHandleMsgCallback;
+
+  *pMsgSendInfo = msgSendInfo;
+
+  return TSDB_CODE_SUCCESS;
+
+_return:
+
+  taosMemoryFree(param);
+  taosMemoryFree(msgSendInfo);
+
+  CTG_RET(code);
+}
+
+int32_t ctgAsyncSendMsg(CTG_PARAMS, SCtgTask* pTask, int32_t msgType, void *msg, uint32_t msgSize) {
+  int32_t code = 0;
+  SMsgSendInfo *pMsgSendInfo = NULL;
+  CTG_ERR_JRET(ctgMakeMsgSendInfo(pTask, msgType, &pMsgSendInfo));
+
+  pMsgSendInfo->msgInfo.pData = msg;
+  pMsgSendInfo->msgInfo.len = msgSize;
+  pMsgSendInfo->msgInfo.handle = NULL;
+  pMsgSendInfo->msgType = msgType;
+
+  int64_t transporterId = 0;
+  code = asyncSendMsgToServer(pTrans, (SEpSet*)pMgmtEps, &transporterId, pMsgSendInfo);
+  if (code) {
+    ctgError("asyncSendMsgToSever failed, error: %s", tstrerror(code));
+    CTG_ERR_JRET(code);
+  }
+
+  ctgDebug("req msg sent, reqId:%" PRIx64 ", msg type:%d, %s", pTask->pJob->queryId, msgType, TMSG_INFO(msgType));
+  return TSDB_CODE_SUCCESS;
+
+_return:
+
+  if (pMsgSendInfo) {
+    taosMemoryFreeClear(pMsgSendInfo->param);
+    taosMemoryFreeClear(pMsgSendInfo);
+  }
+
+  CTG_RET(code);
+}
+
+
+
+
+int32_t ctgGetQnodeListFromMnode(CTG_PARAMS, SArray *out, SCtgTask* pTask) {
+  char *msg = NULL;
+  int32_t msgLen = 0;
+  int32_t reqType = TDMT_MND_QNODE_LIST;
+
+  ctgDebug("try to get qnode list from mnode, mgmtEpInUse:%d", pMgmtEps->inUse);
+
+  int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](NULL, &msg, 0, &msgLen);
+  if (code) {
+    ctgError("Build qnode list msg failed, error:%s", tstrerror(code));
+    CTG_ERR_RET(code);
+  }
+
+  if (pTask) {
+    void* pOut = taosArrayInit(4, sizeof(struct SQueryNodeAddr));
+    if (NULL == pOut) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+    CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, NULL));
+    CTG_RET(ctgAsyncSendMsg(CTG_PARAMS_LIST(), pTask, reqType, msg, msgLen));
+  }
+  
+  SRpcMsg rpcMsg = {
+      .msgType = reqType,
+      .pCont   = msg,
+      .contLen = msgLen,
+  };
+
+  SRpcMsg rpcRsp = {0};
+  rpcSendRecv(pTrans, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
+
+  CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, NULL));
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
+int32_t ctgGetDBVgInfoFromMnode(CTG_PARAMS, SBuildUseDBInput *input, SUseDbOutput *out, SCtgTask* pTask) {
+  char *msg = NULL;
+  int32_t msgLen = 0;
+  int32_t reqType = TDMT_MND_USE_DB;
+
+  ctgDebug("try to get db vgInfo from mnode, dbFName:%s", input->db);
+
+  int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](input, &msg, 0, &msgLen);
+  if (code) {
+    ctgError("Build use db msg failed, code:%x, db:%s", code, input->db);
+    CTG_ERR_RET(code);
+  }
+
+  if (pTask) {
+    void* pOut = taosMemoryCalloc(1, sizeof(SUseDbOutput));
+    if (NULL == pOut) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+    CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, input->db));
+    
+    CTG_RET(ctgAsyncSendMsg(CTG_PARAMS_LIST(), pTask, reqType, msg, msgLen));
+  }
+  
+  SRpcMsg rpcMsg = {
+      .msgType = reqType,
+      .pCont   = msg,
+      .contLen = msgLen,
+  };
+
+  SRpcMsg rpcRsp = {0};
+  rpcSendRecv(pTrans, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
+
+  CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, input->db));
+  
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t ctgGetDBCfgFromMnode(CTG_PARAMS, const char *dbFName, SDbCfgInfo *out, SCtgTask* pTask) {
+  char *msg = NULL;
+  int32_t msgLen = 0;
+  int32_t reqType = TDMT_MND_GET_DB_CFG;
+
+  ctgDebug("try to get db cfg from mnode, dbFName:%s", dbFName);
+
+  int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)dbFName, &msg, 0, &msgLen);
+  if (code) {
+    ctgError("Build get db cfg msg failed, code:%x, db:%s", code, dbFName);
+    CTG_ERR_RET(code);
+  }
+
+  if (pTask) {
+    void* pOut = taosMemoryCalloc(1, sizeof(SDbCfgInfo));
+    if (NULL == pOut) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+    CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, (char*)dbFName));
+    
+    CTG_RET(ctgAsyncSendMsg(CTG_PARAMS_LIST(), pTask, reqType, msg, msgLen));
+  }
+  
+  SRpcMsg rpcMsg = {
+      .msgType = TDMT_MND_GET_DB_CFG,
+      .pCont   = msg,
+      .contLen = msgLen,
+  };
+
+  SRpcMsg rpcRsp = {0};
+  rpcSendRecv(pTrans, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
+
+  CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)dbFName));
+
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t ctgGetIndexInfoFromMnode(CTG_PARAMS, const char *indexName, SIndexInfo *out, SCtgTask* pTask) {
+  char *msg = NULL;
+  int32_t msgLen = 0;
+  int32_t reqType = TDMT_MND_GET_INDEX;
+
+  ctgDebug("try to get index from mnode, indexName:%s", indexName);
+
+  int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)indexName, &msg, 0, &msgLen);
+  if (code) {
+    ctgError("Build get index msg failed, code:%x, db:%s", code, indexName);
+    CTG_ERR_RET(code);
+  }
+
+  if (pTask) {
+    void* pOut = taosMemoryCalloc(1, sizeof(SIndexInfo));
+    if (NULL == pOut) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+    CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, (char*)indexName));
+    
+    CTG_RET(ctgAsyncSendMsg(CTG_PARAMS_LIST(), pTask, reqType, msg, msgLen));
+  }
+  
+  SRpcMsg rpcMsg = {
+      .msgType = reqType,
+      .pCont   = msg,
+      .contLen = msgLen,
+  };
+
+  SRpcMsg rpcRsp = {0};
+  rpcSendRecv(pTrans, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
+
+  CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)indexName));
+  
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t ctgGetUdfInfoFromMnode(CTG_PARAMS, const char *funcName, SFuncInfo *out, SCtgTask* pTask) {
+  char *msg = NULL;
+  int32_t msgLen = 0;
+  int32_t reqType = TDMT_MND_RETRIEVE_FUNC;
+
+  ctgDebug("try to get udf info from mnode, funcName:%s", funcName);
+
+  int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)funcName, &msg, 0, &msgLen);
+  if (code) {
+    ctgError("Build get udf msg failed, code:%x, db:%s", code, funcName);
+    CTG_ERR_RET(code);
+  }
+
+  if (pTask) {
+    void* pOut = taosMemoryCalloc(1, sizeof(SFuncInfo));
+    if (NULL == pOut) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+    CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, (char*)funcName));
+    
+    CTG_RET(ctgAsyncSendMsg(CTG_PARAMS_LIST(), pTask, reqType, msg, msgLen));
+  }
+  
+  SRpcMsg rpcMsg = {
+      .msgType = reqType,
+      .pCont   = msg,
+      .contLen = msgLen,
+  };
+
+  SRpcMsg rpcRsp = {0};
+  rpcSendRecv(pTrans, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
+
+  CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)funcName));
+
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t ctgGetUserDbAuthFromMnode(CTG_PARAMS, const char *user, SGetUserAuthRsp *out, SCtgTask* pTask) {
+  char *msg = NULL;
+  int32_t msgLen = 0;
+  int32_t reqType = TDMT_MND_GET_USER_AUTH;
+
+  ctgDebug("try to get user auth from mnode, user:%s", user);
+
+  int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)user, &msg, 0, &msgLen);
+  if (code) {
+    ctgError("Build get user auth msg failed, code:%x, db:%s", code, user);
+    CTG_ERR_RET(code);
+  }
+
+  if (pTask) {
+    void* pOut = taosMemoryCalloc(1, sizeof(SGetUserAuthRsp));
+    if (NULL == pOut) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+    CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, (char*)user));
+    
+    CTG_RET(ctgAsyncSendMsg(CTG_PARAMS_LIST(), pTask, reqType, msg, msgLen));
+  }
+  
+  SRpcMsg rpcMsg = {
+      .msgType = reqType,
+      .pCont   = msg,
+      .contLen = msgLen,
+  };
+
+  SRpcMsg rpcRsp = {0};
+  rpcSendRecv(pTrans, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
+
+  CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)user));
+  
+  return TSDB_CODE_SUCCESS;
+}
+
+
+int32_t ctgGetTbMetaFromMnodeImpl(CTG_PARAMS, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTask* pTask) {
+  SBuildTableMetaInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = tbName};
+  char *msg = NULL;
+  SEpSet *pVnodeEpSet = NULL;
+  int32_t msgLen = 0;
+  int32_t reqType = TDMT_MND_TABLE_META;
+  char tbFName[TSDB_TABLE_FNAME_LEN];
+  sprintf(tbFName, "%s.%s", dbFName, tbName);
+
+  ctgDebug("try to get table meta from mnode, tbFName:%s", tbFName);
+
+  int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](&bInput, &msg, 0, &msgLen);
+  if (code) {
+    ctgError("Build mnode stablemeta msg failed, code:%x", code);
+    CTG_ERR_RET(code);
+  }
+
+  if (pTask) {
+    void* pOut = taosMemoryCalloc(1, sizeof(STableMetaOutput));
+    if (NULL == pOut) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+    CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, tbFName));
+    
+    CTG_RET(ctgAsyncSendMsg(CTG_PARAMS_LIST(), pTask, reqType, msg, msgLen));
+  }
+
+  SRpcMsg rpcMsg = {
+      .msgType = reqType,
+      .pCont   = msg,
+      .contLen = msgLen,
+  };
+
+  SRpcMsg rpcRsp = {0};
+  rpcSendRecv(pTrans, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
+  
+  CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, tbFName));
+
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t ctgGetTbMetaFromMnode(CTG_PARAMS, const SName* pTableName, STableMetaOutput* out, SCtgTask* pTask) {
+  char dbFName[TSDB_DB_FNAME_LEN];
+  tNameGetFullDbName(pTableName, dbFName);
+
+  return ctgGetTbMetaFromMnodeImpl(CTG_PARAMS_LIST(), dbFName, (char *)pTableName->tname, out, pTask);
+}
+
+int32_t ctgGetTbMetaFromVnode(CTG_PARAMS, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* out, SCtgTask* pTask) {
+  char dbFName[TSDB_DB_FNAME_LEN];
+  tNameGetFullDbName(pTableName, dbFName);
+  int32_t reqType = TDMT_VND_TABLE_META;
+  char tbFName[TSDB_TABLE_FNAME_LEN];
+  sprintf(tbFName, "%s.%s", dbFName, pTableName->tname);
+
+  ctgDebug("try to get table meta from vnode, vgId:%d, tbFName:%s", vgroupInfo->vgId, tbFName);
+
+  SBuildTableMetaInput bInput = {.vgId = vgroupInfo->vgId, .dbFName = dbFName, .tbName = (char *)tNameGetTableName(pTableName)};
+  char *msg = NULL;
+  int32_t msgLen = 0;
+
+  int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](&bInput, &msg, 0, &msgLen);
+  if (code) {
+    ctgError("Build vnode tablemeta msg failed, code:%x, tbFName:%s", code, tbFName);
+    CTG_ERR_RET(code);
+  }
+
+  if (pTask) {
+    void* pOut = taosMemoryCalloc(1, sizeof(STableMetaOutput));
+    if (NULL == pOut) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+    CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, tbFName));
+    
+    CTG_RET(ctgAsyncSendMsg(CTG_PARAMS_LIST(), pTask, reqType, msg, msgLen));
+  }
+
+  SRpcMsg rpcMsg = {
+      .msgType = reqType,
+      .pCont   = msg,
+      .contLen = msgLen,
+  };
+
+  SRpcMsg rpcRsp = {0};
+  rpcSendRecv(pTrans, &vgroupInfo->epSet, &rpcMsg, &rpcRsp);
+
+  CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, tbFName));  
+
+  return TSDB_CODE_SUCCESS;
+}
+
+

source/libs/catalog/test/catalogTests.cpp

@@ -40,10 +40,8 @@
 
 namespace {
 
-extern "C" int32_t ctgGetTableMetaFromCache(struct SCatalog *pCatalog, const SName *pTableName, STableMeta **pTableMeta,
-                                            bool *inCache, int32_t flag, uint64_t *dbId);
 extern "C" int32_t ctgdGetClusterCacheNum(struct SCatalog* pCatalog, int32_t type);
-extern "C" int32_t ctgActUpdateTbl(SCtgMetaAction *action);
+extern "C" int32_t ctgActUpdateTb(SCtgMetaAction *action);
 extern "C" int32_t ctgdEnableDebug(char *option);
 extern "C" int32_t ctgdGetStatNum(char *option, void *res);
 
@@ -52,7 +50,7 @@ void ctgTestSetRspCTableMeta();
 void ctgTestSetRspSTableMeta();
 void ctgTestSetRspMultiSTableMeta();
 
-extern "C" SCatalogMgmt gCtgMgmt;
+//extern "C" SCatalogMgmt gCtgMgmt;
 
 enum {
   CTGT_RSP_VGINFO = 1,
@@ -859,8 +857,12 @@ void *ctgTestGetCtableMetaThread(void *param) {
   strcpy(cn.dbname, "db1");
   strcpy(cn.tname, ctgTestCTablename);
 
+  SCtgTbMetaCtx ctx = {0};
+  ctx.pName = &cn;
+  ctx.flag = CTG_FLAG_UNKNOWN_STB;
+
   while (!ctgTestStop) {
-    code = ctgGetTableMetaFromCache(pCtg, &cn, &tbMeta, &inCache, 0, NULL);
+    code = ctgReadTbMetaFromCache(pCtg, &ctx, &tbMeta);
     if (code || !inCache) {
       assert(0);
     }
@@ -899,7 +901,7 @@ void *ctgTestSetCtableMetaThread(void *param) {
     msg->output = output;
     action.data = msg;
 
-    code = ctgActUpdateTbl(&action);
+    code = ctgActUpdateTb(&action);
     if (code) {
       assert(0);
     }

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"
@@ -4059,7 +4058,7 @@ int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t n
     defaultBufsz = defaultPgsz * 4;
   }
 
-  int32_t code = createDiskbasedBuf(&pAggSup->pResultBuf, defaultPgsz, defaultBufsz, pKey, "/tmp/");
+  int32_t code = createDiskbasedBuf(&pAggSup->pResultBuf, defaultPgsz, defaultBufsz, pKey, TD_TMP_DIR_PATH);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
   }

source/libs/executor/src/groupoperator.c

@@ -614,7 +614,7 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SExprInfo*
     goto _error;
   }
 
-  int32_t code = createDiskbasedBuf(&pInfo->pBuf, 4096, 4096 * 256, pTaskInfo->id.str, "/tmp/");
+  int32_t code = createDiskbasedBuf(&pInfo->pBuf, 4096, 4096 * 256, pTaskInfo->id.str, TD_TMP_DIR_PATH);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }

source/libs/executor/src/indexoperator.c

@@ -19,38 +19,11 @@
 #include "nodes.h"
 #include "tdatablock.h"
 
-typedef struct SIFCtx {
-  int32_t   code;
-  SHashObj *pRes;    /* element is SScalarParam */
-  bool      noExec;  // true: just iterate condition tree, and add hint to executor plan
-  // SIdxFltStatus st;
-} SIFCtx;
-
-#define SIF_ERR_RET(c)                \
-  do {                                \
-    int32_t _code = c;                \
-    if (_code != TSDB_CODE_SUCCESS) { \
-      terrno = _code;                 \
-      return _code;                   \
-    }                                 \
-  } while (0)
-#define SIF_RET(c)                    \
-  do {                                \
-    int32_t _code = c;                \
-    if (_code != TSDB_CODE_SUCCESS) { \
-      terrno = _code;                 \
-    }                                 \
-    return _code;                     \
-  } while (0)
-#define SIF_ERR_JRET(c)              \
-  do {                               \
-    code = c;                        \
-    if (code != TSDB_CODE_SUCCESS) { \
-      terrno = code;                 \
-      goto _return;                  \
-    }                                \
-  } while (0)
-
+// clang-format off
+#define SIF_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
+#define SIF_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
+#define SIF_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
+// clang-format on
 typedef struct SIFParam {
   SHashObj *pFilter;
 
@@ -65,6 +38,13 @@ typedef struct SIFParam {
   char          colName[TSDB_COL_NAME_LEN];
 } SIFParam;
 
+typedef struct SIFCtx {
+  int32_t   code;
+  SHashObj *pRes;    /* element is SIFParam */
+  bool      noExec;  // true: just iterate condition tree, and add hint to executor plan
+  // SIdxFltStatus st;
+} SIFCtx;
+
 static int32_t sifGetFuncFromSql(EOperatorType src, EIndexQueryType *dst) {
   if (src == OP_TYPE_GREATER_THAN) {
     *dst = QUERY_GREATER_THAN;

source/libs/executor/src/scanoperator.c

@@ -990,7 +990,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataR
 
   size_t childKeyBufSize = sizeof(int64_t) + sizeof(int64_t) + sizeof(TSKEY);
   initCatchSupporter(&pInfo->childAggSup, 1024, childKeyBufSize,
-      "StreamFinalInterval", "/tmp/"); // TODO(liuyao) get row size from phy plan
+      "StreamFinalInterval", TD_TMP_DIR_PATH); // TODO(liuyao) get row size from phy plan
 
   pOperator->name       = "StreamBlockScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;

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/executor/src/tlinearhash.c

@@ -247,7 +247,7 @@ SLHashObj* tHashInit(int32_t inMemPages, int32_t pageSize, _hash_fn_t fn, int32_
     return NULL;
   }
 
-  int32_t code = createDiskbasedBuf(&pHashObj->pBuf, pageSize, inMemPages * pageSize, 0, "/tmp");
+  int32_t code = createDiskbasedBuf(&pHashObj->pBuf, pageSize, inMemPages * pageSize, 0, TD_TMP_DIR_PATH);
   if (code != 0) {
     terrno = code;
     return NULL;

source/libs/executor/src/tsort.c

@@ -155,7 +155,7 @@ static int32_t doAddToBuf(SSDataBlock* pDataBlock, SSortHandle* pHandle) {
   int32_t start = 0;
 
   if (pHandle->pBuf == NULL) {
-    int32_t code = createDiskbasedBuf(&pHandle->pBuf, pHandle->pageSize, pHandle->numOfPages * pHandle->pageSize, "doAddToBuf", "/tmp");
+    int32_t code = createDiskbasedBuf(&pHandle->pBuf, pHandle->pageSize, pHandle->numOfPages * pHandle->pageSize, "doAddToBuf", TD_TMP_DIR_PATH);
     dBufSetPrintInfo(pHandle->pBuf);
     if (code != TSDB_CODE_SUCCESS) {
       return code;
@@ -217,7 +217,7 @@ static int32_t sortComparInit(SMsortComparParam* cmpParam, SArray* pSources, int
   } else {
     // multi-pass internal merge sort is required
     if (pHandle->pBuf == NULL) {
-      code = createDiskbasedBuf(&pHandle->pBuf, pHandle->pageSize, pHandle->numOfPages * pHandle->pageSize, "sortComparInit", "/tmp");
+      code = createDiskbasedBuf(&pHandle->pBuf, pHandle->pageSize, pHandle->numOfPages * pHandle->pageSize, "sortComparInit", TD_TMP_DIR_PATH);
       dBufSetPrintInfo(pHandle->pBuf);
       if (code != TSDB_CODE_SUCCESS) {
         return code;

source/libs/executor/test/index_executor_tests.cpp

@@ -57,7 +57,7 @@ void sifInitLogFile() {
 
   tsAsyncLog = 0;
   qDebugFlag = 159;
-  strcpy(tsLogDir, "/tmp/sif");
+  strcpy(tsLogDir, TD_TMP_DIR_PATH "sif");
   taosRemoveDir(tsLogDir);
   taosMkDir(tsLogDir);
 

source/libs/function/src/functionMgt.c

@@ -66,22 +66,18 @@ static bool isSpecificClassifyFunc(int32_t funcId, uint64_t classification) {
 }
 
 static int32_t getUdfInfo(SFmGetFuncInfoParam* pParam, SFunctionNode* pFunc) {
-  SFuncInfo* pInfo = NULL;
-  int32_t    code = catalogGetUdfInfo(pParam->pCtg, pParam->pRpc, pParam->pMgmtEps, pFunc->functionName, &pInfo);
+  SFuncInfo funcInfo = {0};
+  int32_t    code = catalogGetUdfInfo(pParam->pCtg, pParam->pRpc, pParam->pMgmtEps, pFunc->functionName, &funcInfo);
   if (TSDB_CODE_SUCCESS != code) {
     return code;
   }
-  if (NULL == pInfo) {
-    snprintf(pParam->pErrBuf, pParam->errBufLen, "Invalid function name: %s", pFunc->functionName);
-    return TSDB_CODE_FUNC_INVALID_FUNTION;
-  }
+
   pFunc->funcType = FUNCTION_TYPE_UDF;
-  pFunc->funcId = TSDB_FUNC_TYPE_AGGREGATE == pInfo->funcType ? FUNC_AGGREGATE_UDF_ID : FUNC_SCALAR_UDF_ID;
-  pFunc->node.resType.type = pInfo->outputType;
-  pFunc->node.resType.bytes = pInfo->outputLen;
-  pFunc->udfBufSize = pInfo->bufSize;
-  tFreeSFuncInfo(pInfo);
-  taosMemoryFree(pInfo);
+  pFunc->funcId = TSDB_FUNC_TYPE_AGGREGATE == funcInfo.funcType ? FUNC_AGGREGATE_UDF_ID : FUNC_SCALAR_UDF_ID;
+  pFunc->node.resType.type = funcInfo.outputType;
+  pFunc->node.resType.bytes = funcInfo.outputLen;
+  pFunc->udfBufSize = funcInfo.bufSize;
+  tFreeSFuncInfo(&funcInfo);
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/function/src/tpercentile.c

@@ -255,7 +255,7 @@ tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval,
 
   resetSlotInfo(pBucket);
 
-  int32_t ret = createDiskbasedBuf(&pBucket->pBuffer, pBucket->bufPageSize, pBucket->bufPageSize * 512, "1", "/tmp");
+  int32_t ret = createDiskbasedBuf(&pBucket->pBuffer, pBucket->bufPageSize, pBucket->bufPageSize * 512, "1", TD_TMP_DIR_PATH);
   if (ret != 0) {
     tMemBucketDestroy(pBucket);
     return NULL;

source/libs/index/CMakeLists.txt

@@ -31,7 +31,7 @@ if (${BUILD_WITH_INVERTEDINDEX})
 endif(${BUILD_WITH_INVERTEDINDEX})
 
 
-if (${BUILD_TEST})
-  add_subdirectory(test)
-endif(${BUILD_TEST})
+#if (${BUILD_TEST})
+#  add_subdirectory(test)
+#endif(${BUILD_TEST})
 

source/libs/index/inc/indexCache.h

@@ -63,7 +63,10 @@ typedef struct CacheTerm {
 
 IndexCache* indexCacheCreate(SIndex* idx, uint64_t suid, const char* colName, int8_t type);
 
+void indexCacheForceToMerge(void* cache);
 void indexCacheDestroy(void* cache);
+void indexCacheBroadcast(void* cache);
+void indexCacheWait(void* cache);
 
 Iterate* indexCacheIteratorCreate(IndexCache* cache);
 void     indexCacheIteratorDestroy(Iterate* iiter);

source/libs/index/inc/indexInt.h

@@ -58,6 +58,8 @@ struct SIndex {
 
   SIndexStat    stat;
   TdThreadMutex mtx;
+  tsem_t        sem;
+  bool          quit;
 };
 
 struct SIndexOpts {
@@ -69,6 +71,7 @@ struct SIndexOpts {
   int32_t cacheSize;  // MB
   // add cache module later
 #endif
+  int32_t cacheOpt;  // MB
 };
 
 struct SIndexMultiTermQuery {
@@ -131,42 +134,14 @@ int32_t indexSerialCacheKey(ICacheKey* key, char* buf);
 // int32_t indexSerialKey(ICacheKey* key, char* buf);
 // int32_t indexSerialTermKey(SIndexTerm* itm, char* buf);
 
-#define indexFatal(...)                                            \
-  do {                                                             \
-    if (sDebugFlag & DEBUG_FATAL) {                                \
-      taosPrintLog("index FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); \
-    }                                                              \
-  } while (0)
-#define indexError(...)                                            \
-  do {                                                             \
-    if (sDebugFlag & DEBUG_ERROR) {                                \
-      taosPrintLog("index ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); \
-    }                                                              \
-  } while (0)
-#define indexWarn(...)                                           \
-  do {                                                           \
-    if (sDebugFlag & DEBUG_WARN) {                               \
-      taosPrintLog("index WARN ", DEBUG_WARN, 255, __VA_ARGS__); \
-    }                                                            \
-  } while (0)
-#define indexInfo(...)                                      \
-  do {                                                      \
-    if (sDebugFlag & DEBUG_INFO) {                          \
-      taosPrintLog("index ", DEBUG_INFO, 255, __VA_ARGS__); \
-    }                                                       \
-  } while (0)
-#define indexDebug(...)                                             \
-  do {                                                              \
-    if (sDebugFlag & DEBUG_DEBUG) {                                 \
-      taosPrintLog("index ", DEBUG_DEBUG, sDebugFlag, __VA_ARGS__); \
-    }                                                               \
-  } while (0)
-#define indexTrace(...)                                             \
-  do {                                                              \
-    if (sDebugFlag & DEBUG_TRACE) {                                 \
-      taosPrintLog("index ", DEBUG_TRACE, sDebugFlag, __VA_ARGS__); \
-    }                                                               \
-  } while (0)
+// clang-format off
+#define indexFatal(...) do { if (sDebugFlag & DEBUG_FATAL) {  taosPrintLog("INDEX FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }} while (0)
+#define indexError(...) do { if (sDebugFlag & DEBUG_ERROR) {  taosPrintLog("INDEX ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }} while (0)
+#define indexWarn(...)  do { if (sDebugFlag & DEBUG_WARN)  {  taosPrintLog("INDEX WARN ", DEBUG_WARN, 255, __VA_ARGS__); }} while (0)
+#define indexInfo(...)  do { if (sDebugFlag & DEBUG_INFO)  { taosPrintLog("INDEX ", DEBUG_INFO, 255, __VA_ARGS__); } } while (0)
+#define indexDebug(...) do { if (sDebugFlag & DEBUG_DEBUG) { taosPrintLog("INDEX ", DEBUG_DEBUG, sDebugFlag, __VA_ARGS__);} } while (0)
+#define indexTrace(...) do { if (sDebugFlag & DEBUG_TRACE) { taosPrintLog("INDEX ", DEBUG_TRACE, sDebugFlag, __VA_ARGS__);} } while (0)
+// clang-format on
 
 #define INDEX_TYPE_CONTAIN_EXTERN_TYPE(ty, exTy) (((ty >> 4) & (exTy)) != 0)
 

source/libs/index/src/index.c

@@ -90,6 +90,15 @@ static void indexMergeCacheAndTFile(SArray* result, IterateValue* icache, Iterat
 // static int32_t indexSerialTermKey(SIndexTerm* itm, char* buf);
 // int32_t        indexSerialKey(ICacheKey* key, char* buf);
 
+static void indexPost(void* idx) {
+  SIndex* pIdx = idx;
+  tsem_post(&pIdx->sem);
+}
+static void indexWait(void* idx) {
+  SIndex* pIdx = idx;
+  tsem_wait(&pIdx->sem);
+}
+
 int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
   taosThreadOnce(&isInit, indexInit);
   SIndex* sIdx = taosMemoryCalloc(1, sizeof(SIndex));
@@ -107,6 +116,8 @@ int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
   sIdx->cVersion = 1;
   sIdx->path = tstrdup(path);
   taosThreadMutexInit(&sIdx->mtx, NULL);
+  tsem_init(&sIdx->sem, 0, 0);
+  // taosThreadCondInit(&sIdx->finished, NULL);
 
   sIdx->refId = indexAddRef(sIdx);
   indexAcquireRef(sIdx->refId);
@@ -124,16 +135,8 @@ END:
 
 void indexDestroy(void* handle) {
   SIndex* sIdx = handle;
-  void*   iter = taosHashIterate(sIdx->colObj, NULL);
-  while (iter) {
-    IndexCache** pCache = iter;
-    if (*pCache) {
-      indexCacheUnRef(*pCache);
-    }
-    iter = taosHashIterate(sIdx->colObj, iter);
-  }
-  taosHashCleanup(sIdx->colObj);
   taosThreadMutexDestroy(&sIdx->mtx);
+  tsem_destroy(&sIdx->sem);
   indexTFileDestroy(sIdx->tindex);
   taosMemoryFree(sIdx->path);
   taosMemoryFree(sIdx);
@@ -141,6 +144,20 @@ void indexDestroy(void* handle) {
 }
 void indexClose(SIndex* sIdx) {
   indexReleaseRef(sIdx->refId);
+  bool ref = 0;
+  if (sIdx->colObj != NULL) {
+    void* iter = taosHashIterate(sIdx->colObj, NULL);
+    while (iter) {
+      IndexCache** pCache = iter;
+      indexCacheForceToMerge((void*)(*pCache));
+      indexWait((void*)(sIdx));
+      iter = taosHashIterate(sIdx->colObj, iter);
+      indexCacheUnRef(*pCache);
+    }
+    taosHashCleanup(sIdx->colObj);
+    sIdx->colObj = NULL;
+  }
+  // taosMsleep(1000 * 5);
   indexRemoveRef(sIdx->refId);
 }
 int64_t indexAddRef(void* p) {
@@ -451,6 +468,18 @@ int indexFlushCacheToTFile(SIndex* sIdx, void* cache) {
   }
   // handle flush
   Iterate* cacheIter = indexCacheIteratorCreate(pCache);
+  if (cacheIter == NULL) {
+    indexError("%p immtable is empty, ignore merge opera", pCache);
+    indexCacheDestroyImm(pCache);
+    tfileReaderUnRef(pReader);
+    if (sIdx->quit) {
+      indexPost(sIdx);
+      // indexCacheBroadcast(pCache);
+    }
+    indexReleaseRef(sIdx->refId);
+    return 0;
+  }
+
   Iterate* tfileIter = tfileIteratorCreate(pReader);
   if (tfileIter == NULL) {
     indexWarn("empty tfile reader iterator");
@@ -506,7 +535,11 @@ int indexFlushCacheToTFile(SIndex* sIdx, void* cache) {
   } else {
     indexInfo("success to merge , time cost: %" PRId64 "ms", cost / 1000);
   }
+  if (sIdx->quit) {
+    indexPost(sIdx);
+  }
   indexReleaseRef(sIdx->refId);
+
   return ret;
 }
 void iterateValueDestroy(IterateValue* value, bool destroy) {
@@ -563,10 +596,11 @@ int32_t indexSerialCacheKey(ICacheKey* key, char* buf) {
   bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(key->colType, TSDB_DATA_TYPE_JSON);
 
   char* p = buf;
-  SERIALIZE_MEM_TO_BUF(buf, key, suid);
+  char  tbuf[65] = {0};
+  indexInt2str((int64_t)key->suid, tbuf, 0);
+
+  SERIALIZE_STR_VAR_TO_BUF(buf, tbuf, strlen(tbuf));
   SERIALIZE_VAR_TO_BUF(buf, '_', char);
-  // SERIALIZE_MEM_TO_BUF(buf, key, colType);
-  // SERIALIZE_VAR_TO_BUF(buf, '_', char);
   if (hasJson) {
     SERIALIZE_STR_VAR_TO_BUF(buf, JSON_COLUMN, strlen(JSON_COLUMN));
   } else {

source/libs/index/test/fstTest.cc

@@ -15,7 +15,7 @@
 #include "tutil.h"
 void* callback(void* s) { return s; }
 
-static std::string fileName = "/tmp/tindex.tindex";
+static std::string fileName = TD_TMP_DIR_PATH "tindex.tindex";
 class FstWriter {
  public:
   FstWriter() {
@@ -48,7 +48,7 @@ class FstWriter {
 
 class FstReadMemory {
  public:
-  FstReadMemory(int32_t size, const std::string& fileName = "/tmp/tindex.tindex") {
+  FstReadMemory(int32_t size, const std::string& fileName = TD_TMP_DIR_PATH "tindex.tindex") {
     _wc = writerCtxCreate(TFile, fileName.c_str(), true, 64 * 1024);
     _w = fstCountingWriterCreate(_wc);
     _size = size;

source/libs/index/test/fstUT.cc

@@ -17,7 +17,7 @@
 #include "tskiplist.h"
 #include "tutil.h"
 
-static std::string dir = "/tmp/index";
+static std::string dir = TD_TMP_DIR_PATH "index";
 
 static char indexlog[PATH_MAX] = {0};
 static char tindex[PATH_MAX] = {0};