Merge pull request #19142 from taosdata/refact/submit_req

fix: merge from Refact/submit req

Jenkinsfile2

@@ -173,7 +173,7 @@ def pre_test_build_mac() {
     '''
     sh '''
         cd ${WK}/debug
-        cmake .. -DBUILD_TEST=true -DBUILD_HTTPS=false
+        cmake .. -DBUILD_TEST=true -DBUILD_HTTPS=false  -DCMAKE_BUILD_TYPE=Release
         make -j10
         ctest -j10 || exit 7
     '''

cmake/cmake.define

@@ -124,7 +124,9 @@ ELSE ()
     ENDIF ()
 
     INCLUDE(CheckCCompilerFlag)
-    IF (("${CMAKE_C_COMPILER_ID}" MATCHES "Clang") OR ("${CMAKE_C_COMPILER_ID}" MATCHES "AppleClang"))
+    IF (TD_ARM_64 OR TD_ARM_32)
+        SET(COMPILER_SUPPORT_SSE42 false)
+    ELSEIF (("${CMAKE_C_COMPILER_ID}" MATCHES "Clang") OR ("${CMAKE_C_COMPILER_ID}" MATCHES "AppleClang"))
         SET(COMPILER_SUPPORT_SSE42 true)
         MESSAGE(STATUS "Always enable sse4.2 for Clang/AppleClang")
     ELSE()

cmake/taosadapter_CMakeLists.txt.in

@@ -2,7 +2,7 @@
 # taosadapter
 ExternalProject_Add(taosadapter
         GIT_REPOSITORY https://github.com/taosdata/taosadapter.git
-        GIT_TAG 566540d
+        GIT_TAG f0c1753
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosadapter"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

cmake/taostools_CMakeLists.txt.in

@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
         GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG b20c9d1
+        GIT_TAG 11b60a4
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

docs/en/12-taos-sql/10-function.md

@@ -876,7 +876,8 @@ INTERP(expr)
 - The number of rows in the result set of `INTERP` is determined by the parameter `EVERY`. Starting from timestamp1, one interpolation is performed for every time interval specified `EVERY` parameter. The parameter `EVERY` must be an integer, with no quotes, with a time unit of: b(nanosecond), u(microsecond), a(millisecond)), s(second), m(minute), h(hour), d(day), or w(week). For example, `EVERY(500a)` will interpolate every 500 milliseconds.
 - Interpolation is performed based on `FILL` parameter.
 - `INTERP` can only be used to interpolate in single timeline. So it must be used with `partition by tbname` when it's used on a STable.
-- Pseudo column `_irowts` can be used along with `INTERP` to return the timestamps associated with interpolation points(support after version 3.0.1.4).
+- Pseudocolumn `_irowts` can be used along with `INTERP` to return the timestamps associated with interpolation points(support after version 3.0.1.4).
+- Pseudocolumn `_isfilled` can be used along with `INTERP` to indicate whether the results are original records or data points generated by interpolation algorithm(support after version 3.0.2.1).
 
 ### LAST
 

docs/en/12-taos-sql/14-stream.md

@@ -108,7 +108,7 @@ SHOW STREAMS;
 
 When you create a stream, you can use the TRIGGER parameter to specify triggering conditions for it.
 
-For non-windowed processing, triggering occurs in real time. For windowed processing, there are three methods of triggering:
+For non-windowed processing, triggering occurs in real time. For windowed processing, there are three methods of triggering，the default value is AT_ONCE:
 
 1. AT_ONCE: triggers on write
 

docs/en/12-taos-sql/24-show.md

@@ -178,6 +178,77 @@ SHOW TABLE DISTRIBUTED table_name;
 
 Shows how table data is distributed.
 
+Examples： show table distributed d0\G;   Display the block distribution of table `d0` in detailed format.
+
+*************************** 1.row ***************************
+_block_dist: Total_Blocks=[5] Total_Size=[93.65 Kb] Average_size=[18.73 Kb] Compression_Ratio=[23.98 %]
+
+Total_Blocks :  Table `d0` contains total 5 blocks
+Total_Size:  The total size of all the data blocks in table `d0` is 93.65 KB 
+Average_size:  The average size of each block is 18.73 KB
+Compression_Ratio: The data compression rate is 23.98%
+ 
+*************************** 2.row ***************************
+_block_dist: Total_Rows=[20000] Inmem_Rows=[0] MinRows=[3616] MaxRows=[4096] Average_Rows=[4000]
+
+Total_Rows: Table `d0` contains 20,000 rows
+Inmem_Rows： The rows still in memory, i.e. not committed in disk, is 0, i.e. none such rows
+MinRows：  The minimum number of rows in a block is 3,616 
+MaxRows： The maximum number of rows in a block is 4,096B
+Average_Rows： The average number of rows in a block is 4,000
+
+*************************** 3.row ***************************
+_block_dist: Total_Tables=[1] Total_Files=[2]
+
+Total_Tables:  The number of child tables, 1 in this example
+Total_Files：   The number of files storing the table's data, 2 in this example
+
+*************************** 4.row ***************************
+_block_dist: --------------------------------------------------------------------------------
+*************************** 5.row ***************************
+_block_dist: 0100 |
+*************************** 6.row ***************************
+_block_dist: 0299 |
+*************************** 7.row ***************************
+_block_dist: 0498 |
+*************************** 8.row ***************************
+_block_dist: 0697 |
+*************************** 9.row ***************************
+_block_dist: 0896 |
+*************************** 10.row ***************************
+_block_dist: 1095 |
+*************************** 11.row ***************************
+_block_dist: 1294 |
+*************************** 12.row ***************************
+_block_dist: 1493 |
+*************************** 13.row ***************************
+_block_dist: 1692 |
+*************************** 14.row ***************************
+_block_dist: 1891 |
+*************************** 15.row ***************************
+_block_dist: 2090 |
+*************************** 16.row ***************************
+_block_dist: 2289 |
+*************************** 17.row ***************************
+_block_dist: 2488 |
+*************************** 18.row ***************************
+_block_dist: 2687 |
+*************************** 19.row ***************************
+_block_dist: 2886 |
+*************************** 20.row ***************************
+_block_dist: 3085 |
+*************************** 21.row ***************************
+_block_dist: 3284 |
+*************************** 22.row ***************************
+_block_dist: 3483 |||||||||||||||||  1 (20.00%)
+*************************** 23.row ***************************
+_block_dist: 3682 |
+*************************** 24.row ***************************
+_block_dist: 3881 |||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||  4 (80.00%)
+Query OK, 24 row(s) in set (0.002444s)
+
+  The above show the block distribution percentage according to the number of rows in each block. In the above example, `_block_dist: 3483 |||||||||||||||||  1 (20.00%)` means there is one block whose rows is between 3,483 and 3,681. `_block_dist: 3881 |||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||  4 (80.00%)` means there are 4 blocks whose rows is between 3,881 and 4,096. The number of blocks whose rows fall in other range is zero.
+
 ## SHOW TAGS
 
 ```sql

docs/en/28-releases/01-tdengine.md

@@ -10,6 +10,10 @@ For TDengine 2.x installation packages by version, please visit [here](https://w
 
 import Release from "/components/ReleaseV3";
 
+## 3.0.2.1
+
+<Release type="tdengine" version="3.0.2.1" />
+
 ## 3.0.2.0
 
 <Release type="tdengine" version="3.0.2.0" />

docs/en/28-releases/02-tools.md

@@ -10,6 +10,10 @@ For other historical version installers, please visit [here](https://www.taosdat
 
 import Release from "/components/ReleaseV3";
 
+## 2.3.3
+
+<Release type="tools" version="2.3.3" />
+
 ## 2.3.2
 
 <Release type="tools" version="2.3.2" />

docs/examples/go/sub/main.go

@@ -24,7 +24,7 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
-	_, err = db.Exec("create topic if not exists example_tmq_topic with meta as DATABASE example_tmq")
+	_, err = db.Exec("create topic if not exists example_tmq_topic as DATABASE example_tmq")
 	if err != nil {
 		panic(err)
 	}
@@ -84,20 +84,6 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
-	for {
-		result, err := consumer.Poll(time.Second)
-		if err != nil {
-			panic(err)
-		}
-		if result.Type != common.TMQ_RES_TABLE_META {
-			panic("want message type 2 got " + strconv.Itoa(int(result.Type)))
-		}
-		data, _ := json.Marshal(result.Meta)
-		fmt.Println(string(data))
-		consumer.Commit(context.Background(), result.Message)
-		consumer.FreeMessage(result.Message)
-		break
-	}
 	_, err = db.Exec("insert into example_tmq.t1 values(now,1)")
 	if err != nil {
 		panic(err)

docs/zh/08-connector/05-schemaless-api.mdx

@@ -19,6 +19,7 @@ TDengine 提供了兼容 InfluxDB (v1) 和 OpenTSDB 行协议的 Schemaless API
 - `precision` TDengine 使用的时间精度
 - `u` TDengine 用户名
 - `p` TDengine 密码
+- `ttl` 自动创建的子表生命周期，以子表的第一条数据的 TTL 参数为准，不可更新。更多信息请参考[创建表文档](taos-sql/table/#创建表)的 TTL 参数
 
 注意： 目前不支持 InfluxDB 的 token 验证方式，仅支持 Basic 验证和查询参数验证。
 

docs/zh/12-taos-sql/04-stable.md

@@ -139,10 +139,10 @@ alter_table_option: {
 
 - ADD COLUMN：添加列。
 - DROP COLUMN：删除列。
-- MODIFY COLUMN：修改列定义，如果数据列的类型是可变长类型，那么可以使用此指令修改其宽度，只能改大，不能改小。
+- MODIFY COLUMN：修改列的宽度，数据列的类型必须是 nchar 和 binary，使用此指令可以修改其宽度，只能改大，不能改小。
 - ADD TAG：给超级表添加一个标签。
 - DROP TAG：删除超级表的一个标签。从超级表删除某个标签后，该超级表下的所有子表也会自动删除该标签。
-- MODIFY TAG：修改超级表的一个标签的定义。如果标签的类型是可变长类型，那么可以使用此指令修改其宽度，只能改大，不能改小。
+- MODIFY TAG：修改超级表的一个标签的列宽度。标签的类型只能是 nchar 和 binary，使用此指令可以修改其宽度，只能改大，不能改小。
 - RENAME TAG：修改超级表的一个标签的名称。从超级表修改某个标签名后，该超级表下的所有子表也会自动更新该标签名。
 
 ### 增加列

docs/zh/12-taos-sql/10-function.md

@@ -880,6 +880,7 @@ INTERP(expr)
 - INTERP 根据 FILL 字段来决定在每个符合输出条件的时刻如何进行插值。
 - INTERP 只能在一个时间序列内进行插值，因此当作用于超级表时必须跟 partition by tbname 一起使用。
 - INTERP 可以与伪列 _irowts 一起使用，返回插值点所对应的时间戳(3.0.1.4版本以后支持)。
+- INTERP 可以与伪列 _isfilled 一起使用，显示返回结果是否为原始记录或插值算法产生的数据(3.0.2.1版本以后支持)。
 
 ### LAST
 

docs/zh/12-taos-sql/14-stream.md

@@ -114,7 +114,7 @@ SELECT * from information_schema.`ins_streams`;
 
 在创建流时，可以通过 TRIGGER 指令指定流式计算的触发模式。
 
-对于非窗口计算，流式计算的触发是实时的；对于窗口计算，目前提供 3 种触发模式：
+对于非窗口计算，流式计算的触发是实时的；对于窗口计算，目前提供 3 种触发模式，默认为 AT_ONCE：
 
 1. AT_ONCE：写入立即触发
 

docs/zh/12-taos-sql/24-show.md

@@ -179,6 +179,75 @@ SHOW TABLE DISTRIBUTED table_name;
 
 显示表的数据分布信息。
 
+示例说明：
+
+语句： show table distributed d0\G;   竖行显示表 d0 的 BLOCK 分布情况
+
+*************************** 1.row ***************************
+
+_block_dist: Total_Blocks=[5] Total_Size=[93.65 Kb] Average_size=[18.73 Kb] Compression_Ratio=[23.98 %]
+
+Total_Blocks :  表d0 占用的 block 个数为 5 个
+
+Total_Size.    :  表 d0 所有 block 在文件中占用的大小为 93.65 KB 
+
+Average_size:  平均每个 block 在文件中占用的空间大小为 18.73 KB
+
+Compression_Ratio: 数据压缩率为 23.98%
+
+ 
+*************************** 2.row ***************************
+
+_block_dist: Total_Rows=[20000] Inmem_Rows=[0] MinRows=[3616] MaxRows=[4096] Average_Rows=[4000]
+
+Total_Rows: 统计表 d0 的所有行数 为20000 行
+
+Inmem_Rows： 表示仍然还存放在内存中的行数，即没有落盘的行数，为 0行，表示没有
+
+MinRows：  BLOCK 中最小的行数，为 3616 行
+
+MaxRows： BLOCK 中最大的行数，为 4096行
+
+Average_Rows： BLOCK 中的平均行数，为4000 行
+
+
+*************************** 3.row ***************************
+
+_block_dist: Total_Tables=[1] Total_Files=[2]
+
+Total_Tables:  表示子表的个数，这里为1
+
+Total_Files：   表数据保存在几个文件中，这里保存在 2 个文件中
+
+
+*************************** 5.row ***************************
+
+_block_dist: 0100 |
+
+*************************** 6.row ***************************
+
+_block_dist: 0299 |
+
+......
+
+*************************** 22.row ***************************
+
+_block_dist: 3483 |||||||||||||||||  1 (20.00%)
+
+*************************** 23.row ***************************
+
+_block_dist: 3682 |
+
+*************************** 24.row ***************************
+
+_block_dist: 3881 |||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||  4 (80.00%)
+
+Query OK, 24 row(s) in set (0.002444s)
+
+
+  上面是块中包含数据行数的块儿分布情况图，这里的 0100 0299 0498 … 表示的是每个块中包含的数据行数，上面的意思就是这个表的 5 个块，分布在 3483 ~3681 行的块有 1 个，占整个块的 20%，分布在 3881 ~ 4096（最大行数）的块数为 4 个，占整个块的 80%， 其它区域内分布块数为 0。
+
+
 ## SHOW TAGS
 
 ```sql

docs/zh/28-releases/01-tdengine.md

@@ -10,6 +10,10 @@ TDengine 2.x 各版本安装包请访问[这里](https://www.taosdata.com/all-do
 
 import Release from "/components/ReleaseV3";
 
+## 3.0.2.1
+
+<Release type="tdengine" version="3.0.2.1" />
+
 ## 3.0.2.0
 
 <Release type="tdengine" version="3.0.2.0" />

docs/zh/28-releases/02-tools.md

@@ -10,6 +10,10 @@ taosTools 各版本安装包下载链接如下：
 
 import Release from "/components/ReleaseV3";
 
+## 2.3.3
+
+<Release type="tools" version="2.3.3" />
+
 ## 2.3.2
 
 <Release type="tools" version="2.3.2" />

examples/c/asyncdemo.c

@@ -35,7 +35,7 @@ int64_t st, et;
 typedef struct {
   int       id;
   TAOS     *taos;
-  char      name[16];
+  char      name[32];
   time_t    timeStamp;
   int       value;
   int       rowsInserted;

examples/c/tmq.c

@@ -20,7 +20,7 @@
 #include <time.h>
 #include "taos.h"
 
-static int  running = 1;
+static int running = 1;
 
 static int32_t msg_process(TAOS_RES* msg) {
   char    buf[1024];
@@ -40,8 +40,8 @@ static int32_t msg_process(TAOS_RES* msg) {
 
     TAOS_FIELD* fields = taos_fetch_fields(msg);
     int32_t     numOfFields = taos_field_count(msg);
-    //int32_t*    length = taos_fetch_lengths(msg);
-    int32_t     precision = taos_result_precision(msg);
+    // int32_t*    length = taos_fetch_lengths(msg);
+    int32_t precision = taos_result_precision(msg);
     rows++;
     taos_print_row(buf, row, fields, numOfFields);
     printf("precision: %d, row content: %s\n", precision, buf);
@@ -62,14 +62,12 @@ static int32_t init_env() {
   pRes = taos_query(pConn, "drop topic topicname");
   if (taos_errno(pRes) != 0) {
     printf("error in drop tmqdb, reason:%s\n", taos_errstr(pRes));
-    return -1;
   }
   taos_free_result(pRes);
 
   pRes = taos_query(pConn, "drop database if exists tmqdb");
   if (taos_errno(pRes) != 0) {
     printf("error in drop tmqdb, reason:%s\n", taos_errstr(pRes));
-    return -1;
   }
   taos_free_result(pRes);
 
@@ -77,7 +75,7 @@ static int32_t init_env() {
   pRes = taos_query(pConn, "create database tmqdb precision 'ns'");
   if (taos_errno(pRes) != 0) {
     printf("error in create tmqdb, reason:%s\n", taos_errstr(pRes));
-    return -1;
+    goto END;
   }
   taos_free_result(pRes);
 
@@ -87,7 +85,7 @@ static int32_t init_env() {
       pConn, "create table tmqdb.stb (ts timestamp, c1 int, c2 float, c3 varchar(16)) tags(t1 int, t3 varchar(16))");
   if (taos_errno(pRes) != 0) {
     printf("failed to create super table stb, reason:%s\n", taos_errstr(pRes));
-    return -1;
+    goto END;
   }
   taos_free_result(pRes);
 
@@ -96,28 +94,28 @@ static int32_t init_env() {
   pRes = taos_query(pConn, "create table tmqdb.ctb0 using tmqdb.stb tags(0, 'subtable0')");
   if (taos_errno(pRes) != 0) {
     printf("failed to create super table ctb0, reason:%s\n", taos_errstr(pRes));
-    return -1;
+    goto END;
   }
   taos_free_result(pRes);
 
   pRes = taos_query(pConn, "create table tmqdb.ctb1 using tmqdb.stb tags(1, 'subtable1')");
   if (taos_errno(pRes) != 0) {
     printf("failed to create super table ctb1, reason:%s\n", taos_errstr(pRes));
-    return -1;
+    goto END;
   }
   taos_free_result(pRes);
 
   pRes = taos_query(pConn, "create table tmqdb.ctb2 using tmqdb.stb tags(2, 'subtable2')");
   if (taos_errno(pRes) != 0) {
     printf("failed to create super table ctb2, reason:%s\n", taos_errstr(pRes));
-    return -1;
+    goto END;
   }
   taos_free_result(pRes);
 
   pRes = taos_query(pConn, "create table tmqdb.ctb3 using tmqdb.stb tags(3, 'subtable3')");
   if (taos_errno(pRes) != 0) {
     printf("failed to create super table ctb3, reason:%s\n", taos_errstr(pRes));
-    return -1;
+    goto END;
   }
   taos_free_result(pRes);
 
@@ -126,33 +124,37 @@ static int32_t init_env() {
   pRes = taos_query(pConn, "insert into tmqdb.ctb0 values(now, 0, 0, 'a0')(now+1s, 0, 0, 'a00')");
   if (taos_errno(pRes) != 0) {
     printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes));
-    return -1;
+    goto END;
   }
   taos_free_result(pRes);
 
   pRes = taos_query(pConn, "insert into tmqdb.ctb1 values(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11')");
   if (taos_errno(pRes) != 0) {
     printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes));
-    return -1;
+    goto END;
   }
   taos_free_result(pRes);
 
   pRes = taos_query(pConn, "insert into tmqdb.ctb2 values(now, 2, 2, 'a1')(now+1s, 22, 22, 'a22')");
   if (taos_errno(pRes) != 0) {
     printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes));
-    return -1;
+    goto END;
   }
   taos_free_result(pRes);
 
   pRes = taos_query(pConn, "insert into tmqdb.ctb3 values(now, 3, 3, 'a1')(now+1s, 33, 33, 'a33')");
   if (taos_errno(pRes) != 0) {
     printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes));
-    return -1;
+    goto END;
   }
   taos_free_result(pRes);
-
   taos_close(pConn);
   return 0;
+
+END:
+  taos_free_result(pRes);
+  taos_close(pConn);
+  return -1;
 }
 
 int32_t create_topic() {

include/client/taos.h

@@ -59,6 +59,7 @@ typedef enum {
   TSDB_OPTION_TIMEZONE,
   TSDB_OPTION_CONFIGDIR,
   TSDB_OPTION_SHELL_ACTIVITY_TIMER,
+  TSDB_OPTION_USE_ADAPTER,
   TSDB_MAX_OPTIONS
 } TSDB_OPTION;
 
@@ -218,7 +219,7 @@ DLL_EXPORT const void *taos_get_raw_block(TAOS_RES *res);
 DLL_EXPORT int taos_get_db_route_info(TAOS *taos, const char *db, TAOS_DB_ROUTE_INFO *dbInfo);
 DLL_EXPORT int taos_get_table_vgId(TAOS *taos, const char *db, const char *table, int *vgId);
 
-DLL_EXPORT int       taos_load_table_info(TAOS *taos, const char *tableNameList);
+DLL_EXPORT int taos_load_table_info(TAOS *taos, const char *tableNameList);
 
 /*  --------------------------schemaless INTERFACE------------------------------- */
 
@@ -229,13 +230,14 @@ DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw(TAOS *taos, char *lines, int len
                                                 int precision);
 DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows,
                                                            int protocol, int precision, int64_t reqid);
-DLL_EXPORT TAOS_RES *taos_schemaless_insert_ttl(TAOS *taos, char *lines[], int numLines, int protocol, int precision, int32_t ttl);
+DLL_EXPORT TAOS_RES *taos_schemaless_insert_ttl(TAOS *taos, char *lines[], int numLines, int protocol, int precision,
+                                                int32_t ttl);
 DLL_EXPORT TAOS_RES *taos_schemaless_insert_ttl_with_reqid(TAOS *taos, char *lines[], int numLines, int protocol,
-                                                       int precision, int32_t ttl, int64_t reqid);
+                                                           int precision, int32_t ttl, int64_t reqid);
 DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw_ttl(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
-                                                int precision, int32_t ttl);
+                                                    int precision, int32_t ttl);
 DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw_ttl_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows,
-                                                           int protocol, int precision, int32_t ttl, int64_t reqid);
+                                                               int protocol, int precision, int32_t ttl, int64_t reqid);
 
 /* --------------------------TMQ INTERFACE------------------------------- */
 
@@ -308,7 +310,8 @@ DLL_EXPORT tmq_res_t   tmq_get_res_type(TAOS_RES *res);
 DLL_EXPORT int32_t     tmq_get_raw(TAOS_RES *res, tmq_raw_data *raw);
 DLL_EXPORT int32_t     tmq_write_raw(TAOS *taos, tmq_raw_data raw);
 DLL_EXPORT int         taos_write_raw_block(TAOS *taos, int numOfRows, char *pData, const char *tbname);
-DLL_EXPORT int         taos_write_raw_block_with_fields(TAOS* taos, int rows, char* pData, const char* tbname, TAOS_FIELD *fields, int numFields);
+DLL_EXPORT int         taos_write_raw_block_with_fields(TAOS *taos, int rows, char *pData, const char *tbname,
+                                                        TAOS_FIELD *fields, int numFields);
 DLL_EXPORT void        tmq_free_raw(tmq_raw_data raw);
 // Returning null means error. Returned result need to be freed by tmq_free_json_meta
 DLL_EXPORT char *tmq_get_json_meta(TAOS_RES *res);

include/common/tcommon.h

@@ -195,7 +195,7 @@ typedef struct SDataBlockInfo {
   uint32_t    capacity;
   SBlockID    id;
   int16_t     hasVarCol;
-  int16_t     dataLoad;   // denote if the data is loaded or not
+  int16_t     dataLoad;  // denote if the data is loaded or not
 
   // TODO: optimize and remove following
   int64_t     version;    // used for stream, and need serialization
@@ -204,8 +204,9 @@ typedef struct SDataBlockInfo {
   STimeWindow calWin;     // used for stream, do not serialize
   TSKEY       watermark;  // used for stream
 
-  char        parTbName[TSDB_TABLE_NAME_LEN];  // used for stream partition
-  STag*       pTag;                            // used for stream partition
+  char    parTbName[TSDB_TABLE_NAME_LEN];  // used for stream partition
+  int32_t tagLen;
+  void*   pTag;  // used for stream partition
 } SDataBlockInfo;
 
 typedef struct SSDataBlock {
@@ -239,22 +240,22 @@ typedef struct SVarColAttr {
 // pBlockAgg->numOfNull == info.rows, all data are null
 // pBlockAgg->numOfNull == 0, no data are null.
 typedef struct SColumnInfoData {
-  char*         pData;       // the corresponding block data in memory
+  char* pData;  // the corresponding block data in memory
   union {
     char*       nullbitmap;  // bitmap, one bit for each item in the list
     SVarColAttr varmeta;
   };
-  SColumnInfo   info;        // column info
-  bool          hasNull;     // if current column data has null value.
+  SColumnInfo info;     // column info
+  bool        hasNull;  // if current column data has null value.
 } SColumnInfoData;
 
 typedef struct SQueryTableDataCond {
   uint64_t     suid;
-  int32_t      order;    // desc|asc order to iterate the data block
+  int32_t      order;  // desc|asc order to iterate the data block
   int32_t      numOfCols;
   SColumnInfo* colList;
-  int32_t*     pSlotList; // the column output destation slot, and it may be null
-  int32_t      type;     // data block load type:
+  int32_t*     pSlotList;  // the column output destation slot, and it may be null
+  int32_t      type;       // data block load type:
   STimeWindow  twindows;
   int64_t      startVersion;
   int64_t      endVersion;

include/common/tdatablock.h

@@ -265,7 +265,7 @@ void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag);
 // for debug
 char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** dumpBuf);
 
-int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pDataBlocks, STSchema* pTSchema, int32_t vgId,
+int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pDataBlocks, const STSchema* pTSchema, int64_t uid, int32_t vgId,
                                     tb_uid_t suid);
 
 char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId);

include/common/tdataformat.h

@@ -59,9 +59,23 @@ const static uint8_t BIT2_MAP[4] = {0b11111100, 0b11110011, 0b11001111, 0b001111
 #define BIT1_SIZE(n)      (DIV_8((n)-1) + 1)
 #define BIT2_SIZE(n)      (DIV_4((n)-1) + 1)
 #define SET_BIT1(p, i, v) ((p)[DIV_8(i)] = (p)[DIV_8(i)] & BIT1_MAP[MOD_8(i)] | ((v) << MOD_8(i)))
+#define SET_BIT1_EX(p, i, v) \
+  do {                       \
+    if (MOD_8(i) == 0) {     \
+      (p)[DIV_8(i)] = 0;     \
+    }                        \
+    SET_BIT1(p, i, v);       \
+  } while (0)
 #define GET_BIT1(p, i)    (((p)[DIV_8(i)] >> MOD_8(i)) & ONE)
 #define SET_BIT2(p, i, v) ((p)[DIV_4(i)] = (p)[DIV_4(i)] & BIT2_MAP[MOD_4(i)] | ((v) << MOD_4_TIME_2(i)))
-#define GET_BIT2(p, i)    (((p)[DIV_4(i)] >> MOD_4_TIME_2(i)) & THREE)
+#define SET_BIT2_EX(p, i, v) \
+  do {                       \
+    if (MOD_4(i) == 0) {     \
+      (p)[DIV_4(i)] = 0;     \
+    }                        \
+    SET_BIT2(p, i, v);       \
+  } while (0)
+#define GET_BIT2(p, i) (((p)[DIV_4(i)] >> MOD_4_TIME_2(i)) & THREE)
 
 // SBuffer ================================
 struct SBuffer {
@@ -90,7 +104,7 @@ int32_t tBufferReserve(SBuffer *pBuffer, int64_t nData, void **ppData);
 #define COL_VAL_IS_VALUE(CV) ((CV)->flag == CV_FLAG_VALUE)
 
 // SRow ================================
-int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SRow **ppRow);
+int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow);
 void    tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
 void    tRowDestroy(SRow *pRow);
 void    tRowSort(SArray *aRowP);
@@ -132,6 +146,9 @@ extern void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max,
 int32_t tColDataAddValueByBind(SColData *pColData, TAOS_MULTI_BIND *pBind);
 void    tColDataSortMerge(SArray *colDataArr);
 
+//for raw block
+int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes,
+                                    int32_t nRows, char* lengthOrbitmap, char *data);
 // for encode/decode
 int32_t tPutColData(uint8_t *pBuf, SColData *pColData);
 int32_t tGetColData(uint8_t *pBuf, SColData *pColData);

include/common/tglobal.h

@@ -101,6 +101,7 @@ extern int32_t tsRedirectPeriod;
 extern int32_t tsRedirectFactor;
 extern int32_t tsRedirectMaxPeriod;
 extern int32_t tsMaxRetryWaitTime;
+extern bool    tsUseAdapter;
 
 // client
 extern int32_t tsMinSlidingTime;

include/common/ttypes.h

@@ -266,6 +266,7 @@ typedef struct {
 #define IS_FLOAT_TYPE(_t)            ((_t) == TSDB_DATA_TYPE_FLOAT || (_t) == TSDB_DATA_TYPE_DOUBLE)
 #define IS_INTEGER_TYPE(_t)          ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t)))
 #define IS_TIMESTAMP_TYPE(_t)        ((_t) == TSDB_DATA_TYPE_TIMESTAMP)
+#define IS_BOOLEAN_TYPE(_t)          ((_t) == TSDB_DATA_TYPE_BOOL)
 
 #define IS_NUMERIC_TYPE(_t) ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t)) || (IS_FLOAT_TYPE(_t)))
 #define IS_MATHABLE_TYPE(_t) \

include/libs/function/functionMgt.h

@@ -120,6 +120,7 @@ typedef enum EFunctionType {
   FUNCTION_TYPE_WEND,
   FUNCTION_TYPE_WDURATION,
   FUNCTION_TYPE_IROWTS,
+  FUNCTION_TYPE_ISFILLED,
   FUNCTION_TYPE_TAGS,
 
   // internal function

include/libs/nodes/cmdnodes.h

@@ -171,10 +171,10 @@ typedef struct SCreateSubTableClause {
   STableOptions* pOptions;
 } SCreateSubTableClause;
 
-typedef struct SCreateMultiTableStmt {
+typedef struct SCreateMultiTablesStmt {
   ENodeType  type;
   SNodeList* pSubTables;
-} SCreateMultiTableStmt;
+} SCreateMultiTablesStmt;
 
 typedef struct SDropTableClause {
   ENodeType type;
@@ -209,14 +209,14 @@ typedef struct SAlterTableStmt {
 
 typedef struct SCreateUserStmt {
   ENodeType type;
-  char      useName[TSDB_USER_LEN];
+  char      userName[TSDB_USER_LEN];
   char      password[TSDB_USET_PASSWORD_LEN];
   int8_t    sysinfo;
 } SCreateUserStmt;
 
 typedef struct SAlterUserStmt {
   ENodeType type;
-  char      useName[TSDB_USER_LEN];
+  char      userName[TSDB_USER_LEN];
   int8_t    alterType;
   char      password[TSDB_USET_PASSWORD_LEN];
   int8_t    enable;
@@ -225,7 +225,7 @@ typedef struct SAlterUserStmt {
 
 typedef struct SDropUserStmt {
   ENodeType type;
-  char      useName[TSDB_USER_LEN];
+  char      userName[TSDB_USER_LEN];
 } SDropUserStmt;
 
 typedef struct SCreateDnodeStmt {

include/libs/nodes/nodes.h

@@ -117,7 +117,7 @@ typedef enum ENodeType {
   // Statement nodes are used in parser and planner module.
   QUERY_NODE_SET_OPERATOR = 100,
   QUERY_NODE_SELECT_STMT,
-  QUERY_NODE_VNODE_MODIF_STMT,
+  QUERY_NODE_VNODE_MODIFY_STMT,
   QUERY_NODE_CREATE_DATABASE_STMT,
   QUERY_NODE_DROP_DATABASE_STMT,
   QUERY_NODE_ALTER_DATABASE_STMT,
@@ -125,7 +125,7 @@ typedef enum ENodeType {
   QUERY_NODE_TRIM_DATABASE_STMT,
   QUERY_NODE_CREATE_TABLE_STMT,
   QUERY_NODE_CREATE_SUBTABLE_CLAUSE,
-  QUERY_NODE_CREATE_MULTI_TABLE_STMT,
+  QUERY_NODE_CREATE_MULTI_TABLES_STMT,
   QUERY_NODE_DROP_TABLE_CLAUSE,
   QUERY_NODE_DROP_TABLE_STMT,
   QUERY_NODE_DROP_SUPER_TABLE_STMT,

include/libs/nodes/querynodes.h

@@ -364,7 +364,7 @@ typedef struct SVgDataBlocks {
 typedef void (*FFreeTableBlockHash)(SHashObj*);
 typedef void (*FFreeVgourpBlockArray)(SArray*);
 
-typedef struct SVnodeModifOpStmt {
+typedef struct SVnodeModifyOpStmt {
   ENodeType             nodeType;
   ENodeType             sqlNodeType;
   SArray*               pDataBlocks;  // data block for each vgroup, SArray<SVgDataBlocks*>.
@@ -388,7 +388,7 @@ typedef struct SVnodeModifOpStmt {
   FFreeVgourpBlockArray freeArrayFunc;
   bool                  usingTableProcessing;
   bool                  fileProcessing;
-} SVnodeModifOpStmt;
+} SVnodeModifyOpStmt;
 
 typedef struct SExplainOptions {
   ENodeType type;

include/libs/parser/parser.h

@@ -115,6 +115,8 @@ int32_t smlBindData(SQuery* handle, bool dataFormat, SArray* tags, SArray* colsS
                     char* tableName, const char* sTableName, int32_t sTableNameLen, int32_t ttl, char* msgBuf, int16_t msgBufLen);
 int32_t smlBuildOutput(SQuery* handle, SHashObj* pVgHash);
 
+int     rawBlockBindData(SQuery *query, STableMeta* pTableMeta, void* data, SVCreateTbReq* pCreateTb, TAOS_FIELD *fields, int numFields);
+
 int32_t rewriteToVnodeModifyOpStmt(SQuery* pQuery, SArray* pBufArray);
 SArray* serializeVgroupsCreateTableBatch(SHashObj* pVgroupHashmap);
 SArray* serializeVgroupsDropTableBatch(SHashObj* pVgroupHashmap);

include/libs/stream/streamState.h

@@ -35,6 +35,7 @@ typedef struct STdbState {
   TTB*         pFillStateDb;  // todo refactor
   TTB*         pSessionStateDb;
   TTB*         pParNameDb;
+  TTB*         pParTagDb;
   TXN*         txn;
 } STdbState;
 
@@ -108,6 +109,9 @@ int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur);
 int32_t streamStatePutParName(SStreamState* pState, int64_t groupId, const char* tbname);
 int32_t streamStateGetParName(SStreamState* pState, int64_t groupId, void** pVal);
 
+int32_t streamStatePutParTag(SStreamState* pState, int64_t groupId, const void* tag, int32_t tagLen);
+int32_t streamStateGetParTag(SStreamState* pState, int64_t groupId, void** tagVal, int32_t* tagLen);
+
 #if 0
 char* streamStateSessionDump(SStreamState* pState);
 #endif

include/util/taoserror.h

@@ -701,7 +701,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_RSMA_INVALID_ENV               TAOS_DEF_ERROR_CODE(0, 0x3150)
 #define TSDB_CODE_RSMA_INVALID_STAT              TAOS_DEF_ERROR_CODE(0, 0x3151)
 #define TSDB_CODE_RSMA_QTASKINFO_CREATE          TAOS_DEF_ERROR_CODE(0, 0x3152)
-// #define TSDB_CODE_RSMA_FILE_CORRUPTED         TAOS_DEF_ERROR_CODE(0, 0x3153)
+#define TSDB_CODE_RSMA_FS_COMMIT                 TAOS_DEF_ERROR_CODE(0, 0x3153)
 #define TSDB_CODE_RSMA_REMOVE_EXISTS             TAOS_DEF_ERROR_CODE(0, 0x3154)
 #define TSDB_CODE_RSMA_FETCH_MSG_MSSED_UP        TAOS_DEF_ERROR_CODE(0, 0x3155)
 #define TSDB_CODE_RSMA_EMPTY_INFO                TAOS_DEF_ERROR_CODE(0, 0x3156)
@@ -709,6 +709,9 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_RSMA_REGEX_MATCH               TAOS_DEF_ERROR_CODE(0, 0x3158)
 #define TSDB_CODE_RSMA_STREAM_STATE_OPEN         TAOS_DEF_ERROR_CODE(0, 0x3159)
 #define TSDB_CODE_RSMA_STREAM_STATE_COMMIT       TAOS_DEF_ERROR_CODE(0, 0x3160)
+#define TSDB_CODE_RSMA_FS_REF                    TAOS_DEF_ERROR_CODE(0, 0x3161)
+#define TSDB_CODE_RSMA_FS_SYNC                   TAOS_DEF_ERROR_CODE(0, 0x3162)
+#define TSDB_CODE_RSMA_FS_UPDATE                 TAOS_DEF_ERROR_CODE(0, 0x3163)
 
 //index
 #define TSDB_CODE_INDEX_REBUILDING               TAOS_DEF_ERROR_CODE(0, 0x3200)

packaging/release.sh

@@ -3,7 +3,7 @@
 # Generate the deb package for ubuntu, or rpm package for centos, or tar.gz package for other linux os
 
 set -e
-#set -x
+# set -x
 
 # release.sh  -v [cluster | edge]
 #             -c [aarch32 | aarch64 | x64 | x86 | mips64 | loongarch64...]
@@ -96,6 +96,8 @@ while getopts "hv:V:c:o:l:s:d:a:n:m:H:" arg; do
   esac
 done
 
+osType=$(uname)
+
 echo "verMode=${verMode} verType=${verType} cpuType=${cpuType} osType=${osType} pagMode=${pagMode} soMode=${soMode} dbName=${dbName} allocator=${allocator} verNumber=${verNumber} verNumberComp=${verNumberComp} httpdBuild=${httpdBuild}"
 
 curr_dir=$(pwd)
@@ -233,7 +235,12 @@ else
   exit 1
 fi
 
-CORES=$(grep -c ^processor /proc/cpuinfo)
+ostype=`uname`
+if [ "${ostype}" == "Darwin" ]; then
+    CORES=$(sysctl -n hw.ncpu)
+else
+    CORES=$(grep -c ^processor /proc/cpuinfo)
+fi
 
 if [[ "$allocator" == "jemalloc" ]]; then
   # jemalloc need compile first, so disable parallel build
@@ -306,7 +313,7 @@ if [ "$osType" != "Darwin" ]; then
   ${csudo}./makeclient.sh ${compile_dir} ${verNumber} "${build_time}" ${cpuType} ${osType} ${verMode} ${verType} ${pagMode} ${dbName}
 
 else
-  # only make client for Darwin
   cd ${script_dir}/tools
+  ./makepkg.sh ${compile_dir} ${verNumber} "${build_time}" ${cpuType} ${osType} ${verMode} ${verType} ${pagMode} ${verNumberComp} ${dbName}
   ./makeclient.sh ${compile_dir} ${verNumber} "${build_time}" ${cpuType} ${osType} ${verMode} ${verType} ${pagMode} ${dbName}
 fi

packaging/tools/makeclient.sh

@@ -2,7 +2,7 @@
 #
 # Generate tar.gz package for linux client in all os system
 set -e
-#set -x
+# set -x
 
 curr_dir=$(pwd)
 compile_dir=$1
@@ -249,9 +249,9 @@ if [ "$osType" != "Darwin" ]; then
   tar -zcv -f "$(basename ${pkg_name}).tar.gz" $(basename ${install_dir}) --remove-files || :
 else
   tar -zcv -f "$(basename ${pkg_name}).tar.gz" $(basename ${install_dir}) || :
-  mv "$(basename ${pkg_name}).tar.gz" ..
-  rm -rf ./*
-  mv ../"$(basename ${pkg_name}).tar.gz" .
+#  mv "$(basename ${pkg_name}).tar.gz" ..
+  rm -rf ${install_dir} ||:
+#  mv ../"$(basename ${pkg_name}).tar.gz" .
 fi
 
 cd ${curr_dir}

packaging/tools/makepkg.sh

@@ -3,7 +3,7 @@
 # Generate tar.gz package for all os system
 
 set -e
-#set -x
+# set -x
 
 curr_dir=$(pwd)
 compile_dir=$1
@@ -74,14 +74,16 @@ else
   tdinsight_caches=""
   cd ${build_dir}/bin/ && \
     chmod +x TDinsight.sh
-  tdinsight_caches=$(./TDinsight.sh --download-only | xargs -i printf "${build_dir}/bin/{} ")
+  ./TDinsight.sh --download-only ||:
+#  tdinsight_caches=$(./TDinsight.sh --download-only | xargs -I printf "${build_dir}/bin/{} ")
   cd $orig_pwd
   echo "TDinsight caches: $tdinsight_caches"
 
   taostools_bin_files=" ${build_dir}/bin/taosdump \
       ${build_dir}/bin/taosBenchmark \
       ${build_dir}/bin/TDinsight.sh \
-      $tdinsight_caches"
+      ${build_dir}/bin/tdengine-datasource.zip \
+      ${build_dir}/bin/tdengine-datasource.zip.md5sum"
   [ -f ${build_dir}/bin/taosx ] && taosx_bin="${build_dir}/bin/taosx"
 
   bin_files="${build_dir}/bin/${serverName} \
@@ -96,8 +98,13 @@ else
       ${script_dir}/taosd-dump-cfg.gdb"
 fi
 
-lib_files="${build_dir}/lib/libtaos.so.${version}"
-wslib_files="${build_dir}/lib/libtaosws.so"
+if [ "$osType" == "Darwin" ]; then
+    lib_files="${build_dir}/lib/libtaos.${version}.dylib"
+    wslib_files="${build_dir}/lib/libtaosws.dylib"
+else
+    lib_files="${build_dir}/lib/libtaos.so.${version}"
+    wslib_files="${build_dir}/lib/libtaosws.so"
+fi
 header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h ${code_dir}/include/libs/function/taosudf.h"
 
 wsheader_files="${build_dir}/include/taosws.h"
@@ -226,7 +233,12 @@ if [ "$verMode" == "cloud" ]; then
 fi
 
 cd ${install_dir}
-tar -zcv -f ${tarName} * --remove-files || :
+if [ "$osType" != "Darwin" ]; then
+    tar -zcv -f ${tarName} * --remove-files || :
+else
+    tar -zcv -f ${tarName} * || :
+fi
+
 exitcode=$?
 if [ "$exitcode" != "0" ]; then
   echo "tar ${tarName} error !!!"
@@ -288,7 +300,7 @@ if [[ $dbName == "taos" ]]; then
   if [ "$verMode" == "cluster" ] || [ "$verMode" == "cloud" ]; then
     if [ -d "${web_dir}/admin" ] ; then
       mkdir -p ${install_dir}/share/
-      cp ${web_dir}/admin ${install_dir}/share/ -r
+      cp -Rfap ${web_dir}/admin ${install_dir}/share/
       cp ${web_dir}/png/taos.png ${install_dir}/share/admin/images/taos.png
     else
       echo "directory not found for enterprise release: ${web_dir}/admin"
@@ -362,7 +374,15 @@ if [ "$pagMode" == "lite" ]; then
   pkg_name=${pkg_name}-Lite
 fi
 
-tar -zcv -f "$(basename ${pkg_name}).tar.gz" "$(basename ${install_dir})" --remove-files || :
+
+if [ "$osType" != "Darwin" ]; then
+    tar -zcv -f "$(basename ${pkg_name}).tar.gz" "$(basename ${install_dir})" --remove-files || :
+else
+    tar -zcv -f "$(basename ${pkg_name}).tar.gz" "$(basename ${install_dir})" || :
+    rm -rf ${install_dir} ||:
+    ([ -d build-taoskeeper ] && rm -rf build-taoskeeper ) ||:
+fi
+
 exitcode=$?
 if [ "$exitcode" != "0" ]; then
   echo "tar ${pkg_name}.tar.gz error !!!"
@@ -371,7 +391,12 @@ fi
 
 if [ -n "${taostools_bin_files}" ]; then
     wget https://github.com/taosdata/grafanaplugin/releases/latest/download/TDinsight.sh -O ${taostools_install_dir}/bin/TDinsight.sh && echo "TDinsight.sh downloaded!"|| echo "failed to download TDinsight.sh"
-    tar -zcv -f "$(basename ${taostools_pkg_name}).tar.gz" "$(basename ${taostools_install_dir})" --remove-files || :
+    if [ "$osType" != "Darwin" ]; then
+        tar -zcv -f "$(basename ${taostools_pkg_name}).tar.gz" "$(basename ${taostools_install_dir})" --remove-files || :
+    else
+        tar -zcv -f "$(basename ${taostools_pkg_name}).tar.gz" "$(basename ${taostools_install_dir})" || :
+        rm -rf ${taostools_install_dir} ||:
+    fi
     exitcode=$?
     if [ "$exitcode" != "0" ]; then
         echo "tar ${taostools_pkg_name}.tar.gz error !!!"

source/client/jni/com_taosdata_jdbc_TSDBJNIConnector.h

@@ -260,6 +260,14 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_insertLinesImp(JN
 JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertImp(JNIEnv *, jobject, jobjectArray,
                                                                                     jlong, jint, jint);
 
+/**
+ * Class:     com_taosdata_jdbc_TSDBJNIConnector
+ * Method:    getTableVgID
+ * Signature: (Ljava/lang/String;Ljava/lang/String)Lcom/taosdata/jdbc/VGroupIDResp
+ */
+JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_getTableVgID(JNIEnv *, jobject, jlong, jstring,
+                                                                               jstring, jobject);
+
 #ifdef __cplusplus
 }
 #endif

source/client/src/clientEnv.c

@@ -77,19 +77,18 @@ static void deregisterRequest(SRequestObj *pRequest) {
            pRequest->self, pTscObj->id, pRequest->requestId, duration / 1000.0, num, currentInst);
 
   tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
-              "us, exec:%" PRId64 "us, stmtType:%d",
+          "us, exec:%" PRId64 "us, stmtType:%d",
           duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
-          pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
-                                                               pRequest->metric.ctgEnd, pRequest->metric.execEnd - pRequest->metric.semanticEnd,
-          pRequest->stmtType);
-
-  if (QUERY_NODE_VNODE_MODIF_STMT == pRequest->stmtType) {
-//        tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
-//                "us, exec:%" PRId64 "us",
-//                duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
-//                pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
-//                pRequest->metric.ctgEnd, pRequest->metric.execEnd - pRequest->metric.semanticEnd);
-//    atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
+          pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd - pRequest->metric.ctgEnd,
+          pRequest->metric.execEnd - pRequest->metric.semanticEnd, pRequest->stmtType);
+
+  if (QUERY_NODE_VNODE_MODIFY_STMT == pRequest->stmtType) {
+    //        tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
+    //                "us, exec:%" PRId64 "us",
+    //                duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
+    //                pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
+    //                pRequest->metric.ctgEnd, pRequest->metric.execEnd - pRequest->metric.semanticEnd);
+    //    atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
   } else if (QUERY_NODE_SELECT_STMT == pRequest->stmtType) {
     //    tscPerf("select duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
     //            "us, planner:%" PRId64 "us, exec:%" PRId64 "us, reqId:0x%" PRIx64,
@@ -475,6 +474,9 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
     case TSDB_OPTION_TIMEZONE:
       pItem = cfgGetItem(pCfg, "timezone");
       break;
+    case TSDB_OPTION_USE_ADAPTER:
+      pItem = cfgGetItem(pCfg, "useAdapter");
+      break;
     default:
       break;
   }

source/client/src/clientImpl.c

@@ -734,7 +734,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
 }
 
 int32_t handleSubmitExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {
-  SArray*     pArray = NULL;
+  SArray*      pArray = NULL;
   SSubmitRsp2* pRsp = (SSubmitRsp2*)res;
   if (NULL == pRsp->aCreateTbRsp) {
     return TSDB_CODE_SUCCESS;
@@ -849,7 +849,7 @@ int32_t handleQueryExecRsp(SRequestObj* pRequest) {
 }
 
 static bool incompletaFileParsing(SNode* pStmt) {
-  return QUERY_NODE_VNODE_MODIF_STMT != nodeType(pStmt) ? false : ((SVnodeModifOpStmt*)pStmt)->fileProcessing;
+  return QUERY_NODE_VNODE_MODIFY_STMT != nodeType(pStmt) ? false : ((SVnodeModifyOpStmt*)pStmt)->fileProcessing;
 }
 
 // todo refacto the error code  mgmt
@@ -928,7 +928,7 @@ SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQue
   if (pQuery->pRoot && !pRequest->inRetry) {
     STscObj*            pTscObj = pRequest->pTscObj;
     SAppClusterSummary* pActivity = &pTscObj->pAppInfo->summary;
-    if (QUERY_NODE_VNODE_MODIF_STMT == pQuery->pRoot->type) {
+    if (QUERY_NODE_VNODE_MODIFY_STMT == pQuery->pRoot->type) {
       atomic_add_fetch_64((int64_t*)&pActivity->numOfInsertsReq, 1);
     } else if (QUERY_NODE_SELECT_STMT == pQuery->pRoot->type) {
       atomic_add_fetch_64((int64_t*)&pActivity->numOfQueryReq, 1);
@@ -1033,7 +1033,7 @@ static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaDat
   }
   if (TSDB_CODE_SUCCESS == code && !pRequest->validateOnly) {
     SArray* pNodeList = NULL;
-    if (QUERY_NODE_VNODE_MODIF_STMT != nodeType(pQuery->pRoot)) {
+    if (QUERY_NODE_VNODE_MODIFY_STMT != nodeType(pQuery->pRoot)) {
       buildAsyncExecNodeList(pRequest, &pNodeList, pMnodeList, pResultMeta);
     }
 

source/client/src/clientJniConnector.c

@@ -488,7 +488,8 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_fetchRowImp(JNIEn
                numOfFields);
       return JNI_FETCH_END;
     } else {
-      jniDebug("jobj:%p, conn:%p, interrupted query. fetch row error code: %d, msg:%s", jobj, tscon, code, taos_errstr(result));
+      jniDebug("jobj:%p, conn:%p, interrupted query. fetch row error code: %d, msg:%s", jobj, tscon, code,
+               taos_errstr(result));
       return JNI_RESULT_SET_NULL;
     }
   }
@@ -583,7 +584,8 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_fetchBlockImp(JNI
       jniDebug("jobj:%p, conn:%p, resultset:%p, no data to retrieve", jobj, tscon, (void *)res);
       return JNI_FETCH_END;
     } else {
-      jniError("jobj:%p, conn:%p, query interrupted. fetch block error code:%d, msg:%s", jobj, tscon, error_code, taos_errstr(tres));
+      jniError("jobj:%p, conn:%p, query interrupted. fetch block error code:%d, msg:%s", jobj, tscon, error_code,
+               taos_errstr(tres));
       return JNI_RESULT_SET_NULL;
     }
   }
@@ -1028,3 +1030,62 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsert
   }
   return (jlong)tres;
 }
+
+// TABLE_VG_ID_FID_CACHE cache resp object for getTableVgID
+typedef struct TABLE_VG_ID_FIELD_CACHE {
+  int      cached;
+  jclass   clazz;
+  jfieldID codeField;
+  jfieldID vgIDField;
+} TABLE_VG_ID_FIELD_CACHE;
+
+TABLE_VG_ID_FIELD_CACHE tableVgIdFieldCache;
+
+void cacheTableVgIDField(JNIEnv *env, jobject jobj) {
+  if (tableVgIdFieldCache.cached) {
+    return;
+  }
+
+  tableVgIdFieldCache.clazz = (*env)->GetObjectClass(env, jobj);
+  tableVgIdFieldCache.codeField = (*env)->GetFieldID(env, tableVgIdFieldCache.clazz, "code", "I");
+  tableVgIdFieldCache.vgIDField = (*env)->GetFieldID(env, tableVgIdFieldCache.clazz, "vgID", "I");
+  tableVgIdFieldCache.cached = 1;
+}
+
+JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_getTableVgID(JNIEnv *env, jobject jobj, jlong conn,
+                                                                               jstring jdb, jstring jtable,
+                                                                               jobject resp) {
+  if (!tableVgIdFieldCache.cached) {
+    cacheTableVgIDField(env, resp);
+  }
+
+  TAOS *taos = (TAOS *)conn;
+  if (taos == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    (*env)->SetIntField(env, resp, tableVgIdFieldCache.codeField, JNI_CONNECTION_NULL);
+    return resp;
+  }
+
+  const char *db = NULL;
+  const char *table = NULL;
+  int         vgID = 0;
+
+  if (jdb != NULL) {
+    db = (*env)->GetStringUTFChars(env, jdb, NULL);
+  }
+  if (jtable != NULL) {
+    table = (*env)->GetStringUTFChars(env, jtable, NULL);
+  }
+
+  int code = taos_get_table_vgId(taos, db, table, &vgID);
+  if (db != NULL) {
+    (*env)->ReleaseStringUTFChars(env, jdb, db);
+  }
+  if (table != NULL) {
+    (*env)->ReleaseStringUTFChars(env, jtable, table);
+  }
+
+  (*env)->SetIntField(env, resp, tableVgIdFieldCache.codeField, code);
+  (*env)->SetIntField(env, resp, tableVgIdFieldCache.vgIDField, vgID);
+  return resp;
+}

source/client/src/clientTmq.c

@@ -438,6 +438,7 @@ int32_t tmqCommitCb(void* param, SDataBuf* pBuf, int32_t code) {
 
   taosMemoryFree(pParam->pOffset);
   taosMemoryFree(pBuf->pData);
+  taosMemoryFree(pBuf->pEpSet);
 
   /*tscDebug("receive offset commit cb of %s on vgId:%d, offset is %" PRId64, pParam->pOffset->subKey, pParam->->vgId,
    * pOffset->version);*/
@@ -529,7 +530,6 @@ static int32_t tmqSendCommitReq(tmq_t* tmq, SMqClientVg* pVg, SMqClientTopic* pT
 int32_t tmqCommitMsgImpl(tmq_t* tmq, const TAOS_RES* msg, int8_t async, tmq_commit_cb* userCb, void* userParam) {
   char*   topic;
   int32_t vgId;
-  ASSERT(msg != NULL);
   if (TD_RES_TMQ(msg)) {
     SMqRspObj* pRspObj = (SMqRspObj*)msg;
     topic = pRspObj->topic;
@@ -724,7 +724,10 @@ void tmqAssignDelayedReportTask(void* param, void* tmrId) {
 }
 
 int32_t tmqHbCb(void* param, SDataBuf* pMsg, int32_t code) {
-  if (pMsg && pMsg->pData) taosMemoryFree(pMsg->pData);
+  if (pMsg) {
+    taosMemoryFree(pMsg->pData);
+    taosMemoryFree(pMsg->pEpSet);
+  }
   return 0;
 }
 
@@ -805,8 +808,6 @@ int32_t tmqHandleAllDelayedTask(tmq_t* tmq) {
 
       taosTmrReset(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, pRefId, tmqMgmt.timer, &tmq->commitTimer);
     } else if (*pTaskType == TMQ_DELAYED_TASK__REPORT) {
-    } else {
-      ASSERT(0);
     }
     taosFreeQitem(pTaskType);
   }
@@ -869,6 +870,8 @@ void tmqClearUnhandleMsg(tmq_t* tmq) {
 int32_t tmqSubscribeCb(void* param, SDataBuf* pMsg, int32_t code) {
   SMqSubscribeCbParam* pParam = (SMqSubscribeCbParam*)param;
   pParam->rspErr = code;
+
+  taosMemoryFree(pMsg->pEpSet);
   tsem_post(&pParam->rspSem);
   return 0;
 }
@@ -947,10 +950,6 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
   const char* user = conf->user == NULL ? TSDB_DEFAULT_USER : conf->user;
   const char* pass = conf->pass == NULL ? TSDB_DEFAULT_PASS : conf->pass;
 
-  ASSERT(user);
-  ASSERT(pass);
-  ASSERT(conf->groupId[0]);
-
   pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic));
   pTmq->mqueue = taosOpenQueue();
   pTmq->qall = taosAllocateQall();
@@ -1166,6 +1165,8 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
   if (code != 0) {
     tscWarn("msg discard from vgId:%d, epoch %d, since %s", vgId, epoch, terrstr());
     if (pMsg->pData) taosMemoryFree(pMsg->pData);
+    if (pMsg->pEpSet) taosMemoryFree(pMsg->pEpSet);
+
     if (code == TSDB_CODE_TMQ_CONSUMER_MISMATCH) {
       atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__RECOVER);
       goto CREATE_MSG_FAIL;
@@ -1239,8 +1240,6 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
     tDecodeSTaosxRsp(&decoder, &pRspWrapper->taosxRsp);
     tDecoderClear(&decoder);
     memcpy(&pRspWrapper->taosxRsp, pMsg->pData, sizeof(SMqRspHead));
-  } else {
-    ASSERT(0);
   }
 
   taosMemoryFree(pMsg->pData);
@@ -1365,6 +1364,7 @@ int32_t tmqAskEpCb(void* param, SDataBuf* pMsg, int32_t code) {
       taosMemoryFree(pParam);
     }
     taosMemoryFree(pMsg->pData);
+    taosMemoryFree(pMsg->pEpSet);
     terrno = TSDB_CODE_TMQ_CONSUMER_CLOSED;
     return -1;
   }
@@ -1416,6 +1416,8 @@ END:
   } else {
     taosMemoryFree(pParam);
   }
+
+  taosMemoryFree(pMsg->pEpSet);
   taosMemoryFree(pMsg->pData);
   return code;
 }

source/common/src/tdatablock.c

@@ -360,7 +360,7 @@ size_t blockDataGetNumOfRows(const SSDataBlock* pBlock) { return pBlock->info.ro
 
 int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock, int32_t tsColumnIndex) {
   if (pDataBlock->info.rows > 0) {
-//    ASSERT(pDataBlock->info.dataLoad == 1);
+    //    ASSERT(pDataBlock->info.dataLoad == 1);
   }
 
   if (pDataBlock == NULL || pDataBlock->info.rows <= 0 || pDataBlock->info.dataLoad == 0) {
@@ -1169,7 +1169,8 @@ void blockDataEmpty(SSDataBlock* pDataBlock) {
 
 // todo temporarily disable it
 
-static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo* pBlockInfo, uint32_t numOfRows, bool clearPayload) {
+static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo* pBlockInfo, uint32_t numOfRows,
+                                bool clearPayload) {
   ASSERT(numOfRows > 0);
 
   if (numOfRows <= pBlockInfo->capacity) {
@@ -1228,7 +1229,7 @@ static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo*
   return TSDB_CODE_SUCCESS;
 }
 
-void  colInfoDataCleanup(SColumnInfoData* pColumn, uint32_t numOfRows) {
+void colInfoDataCleanup(SColumnInfoData* pColumn, uint32_t numOfRows) {
   pColumn->hasNull = false;
 
   if (IS_VAR_DATA_TYPE(pColumn->info.type)) {
@@ -1956,7 +1957,8 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
                   "===stream===%s|block type %d|child id %d|group id:%" PRIu64 "|uid:%" PRId64
                   "|rows:%d|version:%" PRIu64 "|cal start:%" PRIu64 "|cal end:%" PRIu64 "\n",
                   flag, (int32_t)pDataBlock->info.type, pDataBlock->info.childId, pDataBlock->info.id.groupId,
-                  pDataBlock->info.id.uid, pDataBlock->info.rows, pDataBlock->info.version, pDataBlock->info.calWin.skey, pDataBlock->info.calWin.ekey);
+                  pDataBlock->info.id.uid, pDataBlock->info.rows, pDataBlock->info.version,
+                  pDataBlock->info.calWin.skey, pDataBlock->info.calWin.ekey);
   if (len >= size - 1) return dumpBuf;
 
   for (int32_t j = 0; j < rows; j++) {
@@ -2226,22 +2228,24 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
 }
 #endif
 
-int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDataBlock, STSchema* pTSchema,
-                                    int32_t vgId, tb_uid_t suid) {
-  SSubmitReq2* pReq = NULL;
+int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDataBlock, const STSchema* pTSchema,
+                                    int64_t uid, int32_t vgId, tb_uid_t suid) {
+  SSubmitReq2* pReq = *ppReq;
   SArray*      pVals = NULL;
   int32_t      numOfBlks = 0;
   int32_t      sz = 1;
 
   terrno = TSDB_CODE_SUCCESS;
 
-  if (!(pReq = taosMemoryMalloc(sizeof(SSubmitReq2)))) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    goto _end;
-  }
+  if (NULL == pReq) {
+    if (!(pReq = taosMemoryMalloc(sizeof(SSubmitReq2)))) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      goto _end;
+    }
 
-  if (!(pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {
-    goto _end;
+    if (!(pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {
+      goto _end;
+    }
   }
 
   for (int32_t i = 0; i < sz; ++i) {
@@ -2261,7 +2265,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDat
       goto _end;
     }
     tbData.suid = suid;
-    tbData.uid = pDataBlock->info.id.groupId;
+    tbData.uid = uid;
     tbData.sver = pTSchema->version;
 
     if (!pVals && !(pVals = taosArrayInit(colNum, sizeof(SColVal)))) {
@@ -2277,7 +2281,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDat
       int32_t offset = 0;
       for (int32_t k = 0; k < colNum; ++k) {  // iterate by column
         SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
-        STColumn*        pCol = &pTSchema->columns[k];
+        const STColumn*  pCol = &pTSchema->columns[k];
         void*            var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
 
         switch (pColInfoData->info.type) {
@@ -2285,13 +2289,14 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDat
             ASSERT(pColInfoData->info.type == pCol->type);
             if (!isStartKey) {
               isStartKey = true;
-              SColVal cv = COL_VAL_VALUE(PRIMARYKEY_TIMESTAMP_COL_ID, pCol->type, (SValue){.val = *(TSKEY*)var});
+              ASSERT(PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId);
+              SColVal cv = COL_VAL_VALUE(pCol->colId, pCol->type, (SValue){.val = *(TSKEY*)var});
               taosArrayPush(pVals, &cv);
             } else if (colDataIsNull_s(pColInfoData, j)) {
-              SColVal cv = COL_VAL_NULL(PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type);
+              SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
               taosArrayPush(pVals, &cv);
             } else {
-              SColVal cv = COL_VAL_VALUE(PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, (SValue){.val = *(int64_t*)var});
+              SColVal cv = COL_VAL_VALUE(pCol->colId, pCol->type, (SValue){.val = *(int64_t*)var});
               taosArrayPush(pVals, &cv);
             }
             break;
@@ -2299,12 +2304,12 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDat
           case TSDB_DATA_TYPE_VARCHAR: {  // TSDB_DATA_TYPE_BINARY
             ASSERT(pColInfoData->info.type == pCol->type);
             if (colDataIsNull_s(pColInfoData, j)) {
-              SColVal cv = COL_VAL_NULL(PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type);
+              SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
               taosArrayPush(pVals, &cv);
             } else {
               void*   data = colDataGetVarData(pColInfoData, j);
               SValue  sv = (SValue){.nData = varDataLen(data), .pData = varDataVal(data)};  // address copy, no value
-              SColVal cv = COL_VAL_VALUE(PRIMARYKEY_TIMESTAMP_COL_ID, pCol->type, sv);
+              SColVal cv = COL_VAL_VALUE(pCol->colId, pCol->type, sv);
               taosArrayPush(pVals, &cv);
             }
             break;
@@ -2320,7 +2325,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDat
           default:
             if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
               if (colDataIsNull_s(pColInfoData, j)) {
-                SColVal cv = COL_VAL_NULL(PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type);  // should use pCol->type
+                SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);  // should use pCol->type
                 taosArrayPush(pVals, &cv);
               } else {
                 SValue sv;
@@ -2351,7 +2356,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDat
                   }
                   memcpy(&sv.val, tv, tDataTypes[pCol->type].bytes);
                 }
-                SColVal cv = COL_VAL_VALUE(PRIMARYKEY_TIMESTAMP_COL_ID + k, pColInfoData->info.type, sv);
+                SColVal cv = COL_VAL_VALUE(pCol->colId, pColInfoData->info.type, sv);
                 taosArrayPush(pVals, &cv);
               }
             } else {
@@ -2385,7 +2390,7 @@ _end:
 
 char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
   ASSERT(stbFullName[0] != 0);
-  SArray* tags = taosArrayInit(0, sizeof(void*));
+  SArray* tags = taosArrayInit(0, sizeof(SSmlKv));
   if (tags == NULL) {
     return NULL;
   }
@@ -2396,8 +2401,10 @@ char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
     return NULL;
   }
 
-  SSmlKv pTag = {.key = "group_id", .keyLen = sizeof("group_id") - 1,
-                 .type = TSDB_DATA_TYPE_UBIGINT, .u = groupId,
+  SSmlKv pTag = {.key = "group_id",
+                 .keyLen = sizeof("group_id") - 1,
+                 .type = TSDB_DATA_TYPE_UBIGINT,
+                 .u = groupId,
                  .length = sizeof(uint64_t)};
   taosArrayPush(tags, &pTag);
 

source/common/src/tdataformat.c

@@ -97,7 +97,7 @@ typedef struct {
     }                                             \
   } while (0)
 
-int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SRow **ppRow) {
+int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow) {
   int32_t code = 0;
 
   ASSERT(TARRAY_SIZE(aColVal) > 0);
@@ -105,18 +105,18 @@ int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SRow **ppRow) {
   ASSERT(((SColVal *)aColVal->pData)[0].type == TSDB_DATA_TYPE_TIMESTAMP);
 
   // scan ---------------
-  SRow         *pRow = NULL;
-  SColVal      *colVals = (SColVal *)TARRAY_DATA(aColVal);
-  uint8_t       flag = 0;
-  int32_t       iColVal = 1;
-  const int32_t nColVal = TARRAY_SIZE(aColVal);
-  SColVal      *pColVal = (iColVal < nColVal) ? &colVals[iColVal] : NULL;
-  int32_t       iTColumn = 1;
-  STColumn     *pTColumn = pTSchema->columns + iTColumn;
-  int32_t       ntp = 0;
-  int32_t       nkv = 0;
-  int32_t       maxIdx = 0;
-  int32_t       nIdx = 0;
+  SRow           *pRow = NULL;
+  SColVal        *colVals = (SColVal *)TARRAY_DATA(aColVal);
+  uint8_t         flag = 0;
+  int32_t         iColVal = 1;
+  const int32_t   nColVal = TARRAY_SIZE(aColVal);
+  SColVal        *pColVal = (iColVal < nColVal) ? &colVals[iColVal] : NULL;
+  int32_t         iTColumn = 1;
+  const STColumn *pTColumn = pTSchema->columns + iTColumn;
+  int32_t         ntp = 0;
+  int32_t         nkv = 0;
+  int32_t         maxIdx = 0;
+  int32_t         nIdx = 0;
   while (pTColumn) {
     if (pColVal) {
       if (pColVal->cid == pTColumn->colId) {
@@ -309,6 +309,14 @@ int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SRow **ppRow) {
         break;
     }
 
+    if (pb) {
+      if (flag == (HAS_VALUE | HAS_NULL | HAS_NONE)) {
+        memset(pb, 0, BIT2_SIZE(pTSchema->numOfCols - 1));
+      } else {
+        memset(pb, 0, BIT1_SIZE(pTSchema->numOfCols - 1));
+      }
+    }
+
     // build impl
     while (pTColumn) {
       if (pColVal) {
@@ -1612,7 +1620,7 @@ static FORCE_INLINE int32_t tColDataAppendValue10(SColData *pColData, uint8_t *p
   if (code) return code;
 
   memset(pColData->pBitMap, 0, nBit);
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 1);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 1);
 
   pColData->flag |= HAS_VALUE;
 
@@ -1644,7 +1652,7 @@ static FORCE_INLINE int32_t tColDataAppendValue12(SColData *pColData, uint8_t *p
   if (code) return code;
 
   memset(pColData->pBitMap, 0, nBit);
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 1);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 1);
 
   pColData->flag |= HAS_NULL;
   pColData->nVal++;
@@ -1659,7 +1667,7 @@ static FORCE_INLINE int32_t tColDataAppendValue20(SColData *pColData, uint8_t *p
   if (code) return code;
 
   memset(pColData->pBitMap, 0, nBit);
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 1);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 1);
 
   pColData->flag |= HAS_VALUE;
 
@@ -1687,7 +1695,7 @@ static FORCE_INLINE int32_t tColDataAppendValue21(SColData *pColData, uint8_t *p
   if (code) return code;
 
   memset(pColData->pBitMap, 255, nBit);
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 0);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 0);
 
   pColData->flag |= HAS_NONE;
   pColData->nVal++;
@@ -1708,9 +1716,9 @@ static FORCE_INLINE int32_t tColDataAppendValue30(SColData *pColData, uint8_t *p
   if (code) return code;
 
   for (int32_t iVal = 0; iVal < pColData->nVal; iVal++) {
-    SET_BIT2(pBitMap, iVal, GET_BIT1(pColData->pBitMap, iVal));
+    SET_BIT2_EX(pBitMap, iVal, GET_BIT1(pColData->pBitMap, iVal));
   }
-  SET_BIT2(pBitMap, pColData->nVal, 2);
+  SET_BIT2_EX(pBitMap, pColData->nVal, 2);
 
   tFree(pColData->pBitMap);
   pColData->pBitMap = pBitMap;
@@ -1737,7 +1745,7 @@ static FORCE_INLINE int32_t tColDataAppendValue31(SColData *pColData, uint8_t *p
   code = tRealloc(&pColData->pBitMap, BIT1_SIZE(pColData->nVal + 1));
   if (code) return code;
 
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 0);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 0);
   pColData->nVal++;
 
   return code;
@@ -1748,7 +1756,7 @@ static FORCE_INLINE int32_t tColDataAppendValue32(SColData *pColData, uint8_t *p
   code = tRealloc(&pColData->pBitMap, BIT1_SIZE(pColData->nVal + 1));
   if (code) return code;
 
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 1);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 1);
   pColData->nVal++;
 
   return code;
@@ -1764,7 +1772,7 @@ static FORCE_INLINE int32_t tColDataAppendValue41(SColData *pColData, uint8_t *p
   if (code) return code;
 
   memset(pColData->pBitMap, 255, nBit);
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 0);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 0);
 
   return tColDataPutValue(pColData, NULL, 0);
 }
@@ -1778,7 +1786,7 @@ static FORCE_INLINE int32_t tColDataAppendValue42(SColData *pColData, uint8_t *p
   if (code) return code;
 
   memset(pColData->pBitMap, 255, nBit);
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 0);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 0);
 
   return tColDataPutValue(pColData, NULL, 0);
 }
@@ -1788,7 +1796,7 @@ static FORCE_INLINE int32_t tColDataAppendValue50(SColData *pColData, uint8_t *p
   code = tRealloc(&pColData->pBitMap, BIT1_SIZE(pColData->nVal + 1));
   if (code) return code;
 
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 1);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 1);
 
   return tColDataPutValue(pColData, pData, nData);
 }
@@ -1798,7 +1806,7 @@ static FORCE_INLINE int32_t tColDataAppendValue51(SColData *pColData, uint8_t *p
   code = tRealloc(&pColData->pBitMap, BIT1_SIZE(pColData->nVal + 1));
   if (code) return code;
 
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 0);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 0);
 
   return tColDataPutValue(pColData, NULL, 0);
 }
@@ -1812,9 +1820,9 @@ static FORCE_INLINE int32_t tColDataAppendValue52(SColData *pColData, uint8_t *p
   if (code) return code;
 
   for (int32_t iVal = 0; iVal < pColData->nVal; iVal++) {
-    SET_BIT2(pBitMap, iVal, GET_BIT1(pColData->pBitMap, iVal) ? 2 : 0);
+    SET_BIT2_EX(pBitMap, iVal, GET_BIT1(pColData->pBitMap, iVal) ? 2 : 0);
   }
-  SET_BIT2(pBitMap, pColData->nVal, 1);
+  SET_BIT2_EX(pBitMap, pColData->nVal, 1);
 
   tFree(pColData->pBitMap);
   pColData->pBitMap = pBitMap;
@@ -1826,7 +1834,7 @@ static FORCE_INLINE int32_t tColDataAppendValue60(SColData *pColData, uint8_t *p
 
   code = tRealloc(&pColData->pBitMap, BIT1_SIZE(pColData->nVal + 1));
   if (code) return code;
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 1);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 1);
 
   return tColDataPutValue(pColData, pData, nData);
 }
@@ -1840,9 +1848,9 @@ static FORCE_INLINE int32_t tColDataAppendValue61(SColData *pColData, uint8_t *p
   if (code) return code;
 
   for (int32_t iVal = 0; iVal < pColData->nVal; iVal++) {
-    SET_BIT2(pBitMap, iVal, GET_BIT1(pColData->pBitMap, iVal) ? 2 : 1);
+    SET_BIT2_EX(pBitMap, iVal, GET_BIT1(pColData->pBitMap, iVal) ? 2 : 1);
   }
-  SET_BIT2(pBitMap, pColData->nVal, 0);
+  SET_BIT2_EX(pBitMap, pColData->nVal, 0);
 
   tFree(pColData->pBitMap);
   pColData->pBitMap = pBitMap;
@@ -1854,7 +1862,7 @@ static FORCE_INLINE int32_t tColDataAppendValue62(SColData *pColData, uint8_t *p
 
   code = tRealloc(&pColData->pBitMap, BIT1_SIZE(pColData->nVal + 1));
   if (code) return code;
-  SET_BIT1(pColData->pBitMap, pColData->nVal, 0);
+  SET_BIT1_EX(pColData->pBitMap, pColData->nVal, 0);
 
   return tColDataPutValue(pColData, NULL, 0);
 }
@@ -1863,7 +1871,7 @@ static FORCE_INLINE int32_t tColDataAppendValue70(SColData *pColData, uint8_t *p
 
   code = tRealloc(&pColData->pBitMap, BIT2_SIZE(pColData->nVal + 1));
   if (code) return code;
-  SET_BIT2(pColData->pBitMap, pColData->nVal, 2);
+  SET_BIT2_EX(pColData->pBitMap, pColData->nVal, 2);
 
   return tColDataPutValue(pColData, pData, nData);
 }
@@ -1872,7 +1880,7 @@ static FORCE_INLINE int32_t tColDataAppendValue71(SColData *pColData, uint8_t *p
 
   code = tRealloc(&pColData->pBitMap, BIT2_SIZE(pColData->nVal + 1));
   if (code) return code;
-  SET_BIT2(pColData->pBitMap, pColData->nVal, 0);
+  SET_BIT2_EX(pColData->pBitMap, pColData->nVal, 0);
 
   return tColDataPutValue(pColData, NULL, 0);
 }
@@ -1881,7 +1889,7 @@ static FORCE_INLINE int32_t tColDataAppendValue72(SColData *pColData, uint8_t *p
 
   code = tRealloc(&pColData->pBitMap, BIT2_SIZE(pColData->nVal + 1));
   if (code) return code;
-  SET_BIT2(pColData->pBitMap, pColData->nVal, 1);
+  SET_BIT2_EX(pColData->pBitMap, pColData->nVal, 1);
 
   return tColDataPutValue(pColData, NULL, 0);
 }
@@ -2083,6 +2091,61 @@ _exit:
   return code;
 }
 
+int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes,
+                                    int32_t nRows, char* lengthOrbitmap, char *data) {
+  int32_t code = 0;
+
+  if (IS_VAR_DATA_TYPE(type)) {  // var-length data type
+    for (int32_t i = 0; i < nRows; ++i) {
+      int32_t offset = *((int32_t*)lengthOrbitmap + i);
+      if (offset == -1) {
+        code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NULL](pColData, NULL, 0);
+        if (code) goto _exit;
+      } else {
+        code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_VALUE](
+            pColData, (uint8_t *)varDataVal(data + offset), varDataLen(data + offset));
+      }
+    }
+  } else {  // fixed-length data type
+    bool allValue = true;
+    bool allNull  = true;
+    for (int32_t i = 0; i < nRows; ++i) {
+      if(!colDataIsNull_f(lengthOrbitmap, i)){
+        allNull = false;
+      }else{
+        allValue = false;
+      }
+    }
+
+    if (allValue) {
+      // optimize (todo)
+      for (int32_t i = 0; i < nRows; ++i) {
+        code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_VALUE](
+            pColData, (uint8_t *)data + bytes * i, bytes);
+      }
+    } else if (allNull) {
+      // optimize (todo)
+      for (int32_t i = 0; i < nRows; ++i) {
+        code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NULL](pColData, NULL, 0);
+        if (code) goto _exit;
+      }
+    } else {
+      for (int32_t i = 0; i < nRows; ++i) {
+        if (colDataIsNull_f(lengthOrbitmap, i)) {
+          code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NULL](pColData, NULL, 0);
+          if (code) goto _exit;
+        } else {
+          code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_VALUE](
+              pColData, (uint8_t *)data + bytes * i, bytes);
+        }
+      }
+    }
+  }
+
+  _exit:
+  return code;
+}
+
 int32_t tColDataAddValueByBind(SColData *pColData, TAOS_MULTI_BIND *pBind) {
   int32_t code = 0;
 

source/common/src/tglobal.c

@@ -96,6 +96,7 @@ int32_t tsRedirectPeriod = 10;
 int32_t tsRedirectFactor = 2;
 int32_t tsRedirectMaxPeriod = 1000;
 int32_t tsMaxRetryWaitTime = 10000;
+bool    tsUseAdapter = false;
 
 /*
  * denote if the server needs to compress response message at the application layer to client, including query rsp,
@@ -201,9 +202,7 @@ int32_t taosSetTfsCfg(SConfig *pCfg) {
 int32_t taosSetTfsCfg(SConfig *pCfg);
 #endif
 
-struct SConfig *taosGetCfg() {
-  return tsCfg;
-}
+struct SConfig *taosGetCfg() { return tsCfg; }
 
 static int32_t taosLoadCfg(SConfig *pCfg, const char **envCmd, const char *inputCfgDir, const char *envFile,
                            char *apolloUrl) {
@@ -314,6 +313,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "smlBatchSize", tsSmlBatchSize, 1, INT32_MAX, true) != 0) return -1;
   if (cfgAddInt32(pCfg, "maxMemUsedByInsert", tsMaxMemUsedByInsert, 1, INT32_MAX, true) != 0) return -1;
   if (cfgAddInt32(pCfg, "maxRetryWaitTime", tsMaxRetryWaitTime, 0, 86400000, 0) != 0) return -1;
+  if (cfgAddBool(pCfg, "useAdapter", tsUseAdapter, true) != 0) return -1;
 
   tsNumOfTaskQueueThreads = tsNumOfCores / 2;
   tsNumOfTaskQueueThreads = TMAX(tsNumOfTaskQueueThreads, 4);
@@ -668,6 +668,7 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
   tsQueryNodeChunkSize = cfgGetItem(pCfg, "queryNodeChunkSize")->i32;
   tsQueryUseNodeAllocator = cfgGetItem(pCfg, "queryUseNodeAllocator")->bval;
   tsKeepColumnName = cfgGetItem(pCfg, "keepColumnName")->bval;
+  tsUseAdapter = cfgGetItem(pCfg, "useAdapter")->bval;
 
   tsMaxRetryWaitTime = cfgGetItem(pCfg, "maxRetryWaitTime")->i32;
   return 0;

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

@@ -174,6 +174,7 @@ typedef struct {
   void*       param;
   char        opername[TSDB_TRANS_OPER_LEN];
   SArray*     pRpcArray;
+  SRWLatch    lockRpcArray;
 } STrans;
 
 typedef struct {

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

@@ -133,7 +133,10 @@ static int32_t mndProcessConsumerRecoverMsg(SRpcMsg *pMsg) {
   SMnode                *pMnode = pMsg->info.node;
   SMqConsumerRecoverMsg *pRecoverMsg = pMsg->pCont;
   SMqConsumerObj        *pConsumer = mndAcquireConsumer(pMnode, pRecoverMsg->consumerId);
-  ASSERT(pConsumer);
+  if (pConsumer == NULL) {
+    mError("cannot find consumer %" PRId64 " when processing consumer recover msg", pRecoverMsg->consumerId);
+    return -1;
+  }
 
   mInfo("receive consumer recover msg, consumer id %" PRId64 ", status %s", pRecoverMsg->consumerId,
         mndConsumerStatusName(pConsumer->status));
@@ -381,8 +384,6 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
     return -1;
   }
 
-  ASSERT(strcmp(req.cgroup, pConsumer->cgroup) == 0);
-
   atomic_store_32(&pConsumer->hbStatus, 0);
 
   // 1. check consumer status
@@ -428,9 +429,8 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
     for (int32_t i = 0; i < numOfTopics; i++) {
       char            *topic = taosArrayGetP(pConsumer->currentTopics, i);
       SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, pConsumer->cgroup, topic);
-
       // txn guarantees pSub is created
-      ASSERT(pSub);
+
       taosRLockLatch(&pSub->lock);
 
       SMqSubTopicEp topicEp = {0};
@@ -438,7 +438,6 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
 
       // 2.1 fetch topic schema
       SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic);
-      ASSERT(pTopic);
       taosRLockLatch(&pTopic->lock);
       tstrncpy(topicEp.db, pTopic->db, TSDB_DB_FNAME_LEN);
       topicEp.schema.nCols = pTopic->schema.nCols;
@@ -779,8 +778,8 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
   taosWLockLatch(&pOldConsumer->lock);
 
   if (pNewConsumer->updateType == CONSUMER_UPDATE__MODIFY) {
-    ASSERT(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);
-    ASSERT(taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0);
+    /*A(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);*/
+    /*A(taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0);*/
 
     if (taosArrayGetSize(pNewConsumer->rebNewTopics) == 0 && taosArrayGetSize(pNewConsumer->rebRemovedTopics) == 0) {
       pOldConsumer->status = MQ_CONSUMER_STATUS__READY;
@@ -802,8 +801,8 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
       pOldConsumer->status = MQ_CONSUMER_STATUS__MODIFY;
     }
   } else if (pNewConsumer->updateType == CONSUMER_UPDATE__LOST) {
-    ASSERT(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);
-    ASSERT(taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0);
+    /*A(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);*/
+    /*A(taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0);*/
 
     int32_t sz = taosArrayGetSize(pOldConsumer->currentTopics);
     /*pOldConsumer->rebRemovedTopics = taosArrayInit(sz, sizeof(void *));*/
@@ -816,8 +815,8 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
 
     pOldConsumer->status = MQ_CONSUMER_STATUS__LOST;
   } else if (pNewConsumer->updateType == CONSUMER_UPDATE__RECOVER) {
-    ASSERT(taosArrayGetSize(pOldConsumer->currentTopics) == 0);
-    ASSERT(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);
+    /*A(taosArrayGetSize(pOldConsumer->currentTopics) == 0);*/
+    /*A(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);*/
 
     int32_t sz = taosArrayGetSize(pOldConsumer->assignedTopics);
     for (int32_t i = 0; i < sz; i++) {
@@ -834,15 +833,15 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
     pOldConsumer->rebalanceTime = pNewConsumer->upTime;
 
   } else if (pNewConsumer->updateType == CONSUMER_UPDATE__ADD) {
-    ASSERT(taosArrayGetSize(pNewConsumer->rebNewTopics) == 1);
-    ASSERT(taosArrayGetSize(pNewConsumer->rebRemovedTopics) == 0);
+    /*A(taosArrayGetSize(pNewConsumer->rebNewTopics) == 1);*/
+    /*A(taosArrayGetSize(pNewConsumer->rebRemovedTopics) == 0);*/
 
     char *addedTopic = strdup(taosArrayGetP(pNewConsumer->rebNewTopics, 0));
     // not exist in current topic
-#if 1
+#if 0
     for (int32_t i = 0; i < taosArrayGetSize(pOldConsumer->currentTopics); i++) {
       char *topic = taosArrayGetP(pOldConsumer->currentTopics, i);
-      ASSERT(strcmp(topic, addedTopic) != 0);
+      A(strcmp(topic, addedTopic) != 0);
     }
 #endif
 
@@ -883,15 +882,15 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
 
     atomic_add_fetch_32(&pOldConsumer->epoch, 1);
   } else if (pNewConsumer->updateType == CONSUMER_UPDATE__REMOVE) {
-    ASSERT(taosArrayGetSize(pNewConsumer->rebNewTopics) == 0);
-    ASSERT(taosArrayGetSize(pNewConsumer->rebRemovedTopics) == 1);
+    /*A(taosArrayGetSize(pNewConsumer->rebNewTopics) == 0);*/
+    /*A(taosArrayGetSize(pNewConsumer->rebRemovedTopics) == 1);*/
     char *removedTopic = taosArrayGetP(pNewConsumer->rebRemovedTopics, 0);
 
     // not exist in new topic
-#if 1
+#if 0
     for (int32_t i = 0; i < taosArrayGetSize(pOldConsumer->rebNewTopics); i++) {
       char *topic = taosArrayGetP(pOldConsumer->rebNewTopics, i);
-      ASSERT(strcmp(topic, removedTopic) != 0);
+      A(strcmp(topic, removedTopic) != 0);
     }
 #endif
 
@@ -917,7 +916,7 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
       }
     }
     // must find the topic
-    ASSERT(i < sz);
+    /*A(i < sz);*/
 
     // set status
     if (taosArrayGetSize(pOldConsumer->rebNewTopics) == 0 && taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0) {

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

@@ -115,13 +115,11 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, SStreamObj* pStream, SStream
 
   if (pStream->fixedSinkVgId == 0) {
     SDbObj* pDb = mndAcquireDb(pMnode, pStream->targetDb);
-    ASSERT(pDb);
     if (pDb->cfg.numOfVgroups > 1) {
       isShuffle = true;
       pTask->outputType = TASK_OUTPUT__SHUFFLE_DISPATCH;
       pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH;
       if (mndExtractDbInfo(pMnode, pDb, &pTask->shuffleDispatcher.dbInfo, NULL) < 0) {
-        ASSERT(0);
         return -1;
       }
     }
@@ -140,9 +138,7 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, SStreamObj* pStream, SStream
       for (int32_t j = 0; j < sinkLvSize; j++) {
         SStreamTask* pLastLevelTask = taosArrayGetP(sinkLv, j);
         if (pLastLevelTask->nodeId == pVgInfo->vgId) {
-          ASSERT(pVgInfo->vgId > 0);
           pVgInfo->taskId = pLastLevelTask->taskId;
-          ASSERT(pVgInfo->taskId != 0);
           break;
         }
       }
@@ -152,7 +148,6 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, SStreamObj* pStream, SStream
     pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH;
     SArray* pArray = taosArrayGetP(pStream->tasks, 0);
     // one sink only
-    ASSERT(taosArrayGetSize(pArray) == 1);
     SStreamTask* lastLevelTask = taosArrayGetP(pArray, 0);
     pTask->fixedEpDispatcher.taskId = lastLevelTask->taskId;
     pTask->fixedEpDispatcher.nodeId = lastLevelTask->nodeId;
@@ -170,7 +165,6 @@ int32_t mndAssignTaskToVg(SMnode* pMnode, SStreamTask* pTask, SSubplan* plan, co
   plan->execNode.epSet = pTask->epSet;
 
   if (qSubPlanToString(plan, &pTask->exec.qmsg, &msgLen) < 0) {
-    ASSERT(0);
     terrno = TSDB_CODE_QRY_INVALID_INPUT;
     return -1;
   }
@@ -195,7 +189,6 @@ int32_t mndAssignTaskToSnode(SMnode* pMnode, SStreamTask* pTask, SSubplan* plan,
   plan->execNode.epSet = pTask->epSet;
 
   if (qSubPlanToString(plan, &pTask->exec.qmsg, &msgLen) < 0) {
-    ASSERT(0);
     terrno = TSDB_CODE_QRY_INVALID_INPUT;
     return -1;
   }
@@ -222,8 +215,6 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SStreamObj* pStream) {
   void*   pIter = NULL;
   SArray* tasks = taosArrayGetP(pStream->tasks, 0);
 
-  ASSERT(taosArrayGetSize(pStream->tasks) == 1);
-
   while (1) {
     SVgObj* pVgroup = NULL;
     pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
@@ -257,7 +248,10 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SStreamObj* pStream) {
       pTask->tbSink.stbUid = pStream->targetStbUid;
       memcpy(pTask->tbSink.stbFullName, pStream->targetSTbName, TSDB_TABLE_FNAME_LEN);
       pTask->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema);
-      ASSERT(pTask->tbSink.pSchemaWrapper);
+      if (pTask->tbSink.pSchemaWrapper == NULL) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        return -1;
+      }
     }
     sdbRelease(pSdb, pVgroup);
   }
@@ -265,7 +259,6 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SStreamObj* pStream) {
 }
 
 int32_t mndAddFixedSinkTaskToStream(SMnode* pMnode, SStreamObj* pStream) {
-  ASSERT(pStream->fixedSinkVgId != 0);
   SArray*      tasks = taosArrayGetP(pStream->tasks, 0);
   SStreamTask* pTask = tNewSStreamTask(pStream->uid);
   if (pTask == NULL) {
@@ -275,8 +268,6 @@ int32_t mndAddFixedSinkTaskToStream(SMnode* pMnode, SStreamObj* pStream) {
   pTask->fillHistory = pStream->fillHistory;
   mndAddTaskToTaskSet(tasks, pTask);
 
-  ASSERT(pStream->fixedSinkVg.vgId == pStream->fixedSinkVgId);
-
   pTask->nodeId = pStream->fixedSinkVgId;
 #if 0
   SVgObj* pVgroup = mndAcquireVgroup(pMnode, pStream->fixedSinkVgId);
@@ -311,13 +302,16 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
     return -1;
   }
   int32_t planTotLevel = LIST_LENGTH(pPlan->pSubplans);
-  ASSERT(planTotLevel <= 2);
+
   pStream->tasks = taosArrayInit(planTotLevel, sizeof(void*));
 
   bool    hasExtraSink = false;
   bool    externalTargetDB = strcmp(pStream->sourceDb, pStream->targetDb) != 0;
   SDbObj* pDbObj = mndAcquireDb(pMnode, pStream->targetDb);
-  ASSERT(pDbObj != NULL);
+  if (pDbObj == NULL) {
+    terrno = TSDB_CODE_QRY_INVALID_INPUT;
+    return -1;
+  }
 
   bool multiTarget = pDbObj->cfg.numOfVgroups > 1;
   sdbRelease(pSdb, pDbObj);
@@ -351,7 +345,10 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
 
       SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
       SSubplan*      plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
-      ASSERT(plan->subplanType == SUBPLAN_TYPE_MERGE);
+      if (plan->subplanType != SUBPLAN_TYPE_MERGE) {
+        terrno = TSDB_CODE_QRY_INVALID_INPUT;
+        return -1;
+      }
 
       pInnerTask = tNewSStreamTask(pStream->uid);
       if (pInnerTask == NULL) {
@@ -409,7 +406,10 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
 
     SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 1);
     SSubplan*      plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
-    ASSERT(plan->subplanType == SUBPLAN_TYPE_SCAN);
+    if (plan->subplanType != SUBPLAN_TYPE_SCAN) {
+      terrno = TSDB_CODE_QRY_INVALID_INPUT;
+      return -1;
+    }
 
     void* pIter = NULL;
     while (1) {
@@ -471,9 +471,15 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
     taosArrayPush(pStream->tasks, &taskOneLevel);
 
     SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
-    ASSERT(LIST_LENGTH(inner->pNodeList) == 1);
+    if (LIST_LENGTH(inner->pNodeList) != 1) {
+      terrno = TSDB_CODE_QRY_INVALID_INPUT;
+      return -1;
+    }
     SSubplan* plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
-    ASSERT(plan->subplanType == SUBPLAN_TYPE_SCAN);
+    if (plan->subplanType != SUBPLAN_TYPE_SCAN) {
+      terrno = TSDB_CODE_QRY_INVALID_INPUT;
+      return -1;
+    }
 
     void* pIter = NULL;
     while (1) {
@@ -550,9 +556,6 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
   }
 
-  ASSERT(pSub->unassignedVgs);
-  ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
-
   void* pIter = NULL;
   while (1) {
     pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
@@ -590,10 +593,6 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     sdbRelease(pSdb, pVgroup);
   }
 
-  ASSERT(pSub->unassignedVgs->size > 0);
-
-  ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
-
   qDestroyQueryPlan(pPlan);
 
   return 0;

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

@@ -326,13 +326,11 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj,
 
   // deserialize ast
   if (nodesStringToNode(pObj->ast, &pAst) < 0) {
-    /*ASSERT(0);*/
     goto FAIL;
   }
 
   // extract output schema from ast
   if (qExtractResultSchema(pAst, (int32_t *)&pObj->outputSchema.nCols, &pObj->outputSchema.pSchema) != 0) {
-    /*ASSERT(0);*/
     goto FAIL;
   }
 
@@ -347,13 +345,11 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj,
 
   // using ast and param to build physical plan
   if (qCreateQueryPlan(&cxt, &pPlan, NULL) < 0) {
-    /*ASSERT(0);*/
     goto FAIL;
   }
 
   // save physcial plan
   if (nodesNodeToString((SNode *)pPlan, false, &pObj->physicalPlan, NULL) != 0) {
-    /*ASSERT(0);*/
     goto FAIL;
   }
 
@@ -361,7 +357,7 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj,
   if (pCreate->numOfTags) {
     pObj->tagSchema.pSchema = taosMemoryCalloc(pCreate->numOfTags, sizeof(SSchema));
   }
-  ASSERT(pCreate->numOfTags == taosArrayGetSize(pCreate->pTags));
+  /*A(pCreate->numOfTags == taosArrayGetSize(pCreate->pTags));*/
   for (int32_t i = 0; i < pCreate->numOfTags; i++) {
     SField *pField = taosArrayGet(pCreate->pTags, i);
     pObj->tagSchema.pSchema[i].colId = pObj->outputSchema.nCols + i + 1;
@@ -378,9 +374,6 @@ FAIL:
 }
 
 int32_t mndPersistTaskDeployReq(STrans *pTrans, const SStreamTask *pTask) {
-  if (pTask->taskLevel == TASK_LEVEL__AGG) {
-    ASSERT(taosArrayGetSize(pTask->childEpInfo) != 0);
-  }
   SEncoder encoder;
   tEncoderInit(&encoder, NULL, 0);
   tEncodeSStreamTask(&encoder, pTask);
@@ -545,8 +538,6 @@ _OVER:
 }
 
 static int32_t mndPersistTaskDropReq(STrans *pTrans, SStreamTask *pTask) {
-  ASSERT(pTask->nodeId != 0);
-
   // vnode
   /*if (pTask->nodeId > 0) {*/
   SVDropStreamTaskReq *pReq = taosMemoryCalloc(1, sizeof(SVDropStreamTaskReq));
@@ -800,10 +791,9 @@ static int32_t mndProcessStreamDoCheckpoint(SRpcMsg *pReq) {
       int32_t sz = taosArrayGetSize(pLevel);
       for (int32_t j = 0; j < sz; j++) {
         SStreamTask *pTask = taosArrayGetP(pLevel, j);
-        ASSERT(pTask->nodeId > 0);
+        /*A(pTask->nodeId > 0);*/
         SVgObj *pVgObj = mndAcquireVgroup(pMnode, pTask->nodeId);
         if (pVgObj == NULL) {
-          ASSERT(0);
           taosRUnLockLatch(&pStream->lock);
           mndReleaseStream(pMnode, pStream);
           mndTransDrop(pTrans);
@@ -863,7 +853,6 @@ static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) {
 
   SMDropStreamReq dropReq = {0};
   if (tDeserializeSMDropStreamReq(pReq->pCont, pReq->contLen, &dropReq) < 0) {
-    ASSERT(0);
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }

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

@@ -628,6 +628,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict,
   pTrans->undoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
   pTrans->commitActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
   pTrans->pRpcArray = taosArrayInit(1, sizeof(SRpcHandleInfo));
+  taosInitRWLatch(&pTrans->lockRpcArray);
 
   if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
       pTrans->pRpcArray == NULL) {
@@ -737,12 +738,14 @@ int32_t mndSetRpcInfoForDbTrans(SMnode *pMnode, SRpcMsg *pMsg, EOperType oper, c
     if (pTrans->oper == oper) {
       if (strcasecmp(dbname, pTrans->dbname) == 0) {
         mInfo("trans:%d, db:%s oper:%d matched with input", pTrans->id, dbname, oper);
+        taosWLockLatch(&pTrans->lockRpcArray);
         if (pTrans->pRpcArray == NULL) {
-          pTrans->pRpcArray = taosArrayInit(1, sizeof(SRpcHandleInfo));
+          pTrans->pRpcArray = taosArrayInit(4, sizeof(SRpcHandleInfo));
         }
         if (pTrans->pRpcArray != NULL && taosArrayPush(pTrans->pRpcArray, &pMsg->info) != NULL) {
           code = 0;
         }
+        taosWUnLockLatch(&pTrans->lockRpcArray);
 
         sdbRelease(pMnode->pSdb, pTrans);
         break;
@@ -944,8 +947,12 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
           pTrans->failedTimes, code);
   }
 
+  taosWLockLatch(&pTrans->lockRpcArray);
   int32_t size = taosArrayGetSize(pTrans->pRpcArray);
-  if (size <= 0) return;
+  if (size <= 0) {
+    taosWUnLockLatch(&pTrans->lockRpcArray);
+    return;
+  }
 
   for (int32_t i = 0; i < size; ++i) {
     SRpcHandleInfo *pInfo = taosArrayGet(pTrans->pRpcArray, i);
@@ -997,6 +1004,7 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
     }
   }
   taosArrayClear(pTrans->pRpcArray);
+  taosWUnLockLatch(&pTrans->lockRpcArray);
 }
 
 int32_t mndTransProcessRsp(SRpcMsg *pRsp) {

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

@@ -838,9 +838,9 @@ static int32_t mndRetrievePrivileges(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock
 
     if (pUser->superUser) {
       cols = 0;
-      SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-      char             userName[TSDB_USER_LEN + VARSTR_HEADER_SIZE] = {0};
+      char userName[TSDB_USER_LEN + VARSTR_HEADER_SIZE] = {0};
       STR_WITH_MAXSIZE_TO_VARSTR(userName, pUser->user, pShow->pMeta->pSchemas[cols].bytes);
+      SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
       colDataAppend(pColInfo, numOfRows, (const char *)userName, false);
 
       char privilege[20] = {0};
@@ -859,9 +859,9 @@ static int32_t mndRetrievePrivileges(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock
     char *db = taosHashIterate(pUser->readDbs, NULL);
     while (db != NULL) {
       cols = 0;
-      SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-      char             userName[TSDB_USER_LEN + VARSTR_HEADER_SIZE] = {0};
+      char userName[TSDB_USER_LEN + VARSTR_HEADER_SIZE] = {0};
       STR_WITH_MAXSIZE_TO_VARSTR(userName, pUser->user, pShow->pMeta->pSchemas[cols].bytes);
+      SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
       colDataAppend(pColInfo, numOfRows, (const char *)userName, false);
 
       char privilege[20] = {0};

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

@@ -44,7 +44,6 @@ typedef struct SRSmaInfoItem SRSmaInfoItem;
 typedef struct SRSmaFS       SRSmaFS;
 typedef struct SQTaskFile    SQTaskFile;
 typedef struct SQTaskFReader SQTaskFReader;
-typedef struct SQTaskFWriter SQTaskFWriter;
 
 struct SSmaEnv {
   SRWLatch  lock;
@@ -85,22 +84,20 @@ struct STSmaStat {
 
 struct SQTaskFile {
   volatile int32_t nRef;
-  int32_t          padding;
+  int8_t           level;
+  int64_t          suid;
   int64_t          version;
   int64_t          size;
+  int64_t          mtime;
 };
 
 struct SQTaskFReader {
   SSma     *pSma;
+  int8_t    level;
+  int64_t   suid;
   int64_t   version;
   TdFilePtr pReadH;
 };
-struct SQTaskFWriter {
-  SSma     *pSma;
-  int64_t   version;
-  TdFilePtr pWriteH;
-  char     *fname;
-};
 
 struct SRSmaFS {
   SArray *aQTaskInf;  // array of SQTaskFile
@@ -214,85 +211,40 @@ static FORCE_INLINE void tdUnRefSmaStat(SSma *pSma, SSmaStat *pStat) {
 
 // rsma
 void   *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree);
-int32_t tdRSmaFSOpen(SSma *pSma, int64_t version);
+int32_t tdRSmaFSOpen(SSma *pSma, int64_t version, int8_t rollback);
 void    tdRSmaFSClose(SRSmaFS *fs);
-int32_t tdRSmaFSRef(SSma *pSma, SRSmaStat *pStat, int64_t version);
-void    tdRSmaFSUnRef(SSma *pSma, SRSmaStat *pStat, int64_t version);
-int64_t tdRSmaFSMaxVer(SSma *pSma, SRSmaStat *pStat);
-int32_t tdRSmaFSUpsertQTaskFile(SRSmaFS *pFS, SQTaskFile *qTaskFile);
-int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer);
+int32_t tdRSmaFSPrepareCommit(SSma *pSma, SRSmaFS *pFSNew);
+int32_t tdRSmaFSCommit(SSma *pSma);
+int32_t tdRSmaFSFinishCommit(SSma *pSma);
+int32_t tdRSmaFSCopy(SSma *pSma, SRSmaFS *pFS);
+int32_t tdRSmaFSTakeSnapshot(SSma *pSma, SRSmaFS *pFS);
+int32_t tdRSmaFSRef(SSma *pSma, SRSmaFS *pFS);
+void    tdRSmaFSUnRef(SSma *pSma, SRSmaFS *pFS);
+int32_t tdRSmaFSUpsertQTaskFile(SSma *pSma, SRSmaFS *pFS, SQTaskFile *qTaskFile, int32_t nSize);
+int32_t tdRSmaFSRollback(SSma *pSma);
+int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer, int8_t rollback);
 int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName);
 int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type);
 int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
-int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer);
-void    tdRSmaQTaskInfoGetFileName(int32_t vgId, int64_t version, char *outputName);
-void    tdRSmaQTaskInfoGetFullName(int32_t vgId, int64_t version, const char *path, char *outputName);
+int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer, int8_t rollback);
+void    tdRSmaQTaskInfoGetFileName(int32_t vgId, int64_t suid, int8_t level, int64_t version, char *outputName);
+void    tdRSmaQTaskInfoGetFullName(int32_t vgId, int64_t suid, int8_t level, int64_t version, const char *path,
+                                   char *outputName);
 void    tdRSmaQTaskInfoGetFullPath(int32_t vgId, int8_t level, const char *path, char *outputName);
 void    tdRSmaQTaskInfoGetFullPathEx(int32_t vgId, tb_uid_t suid, int8_t level, const char *path, char *outputName);
 
 static FORCE_INLINE void tdRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) {
   int32_t ref = T_REF_INC(pRSmaInfo);
-  smaDebug("vgId:%d, ref rsma info:%p, val:%d", SMA_VID(pSma), pRSmaInfo, ref);
+  smaTrace("vgId:%d, ref rsma info:%p, val:%d", SMA_VID(pSma), pRSmaInfo, ref);
 }
 static FORCE_INLINE void tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) {
   int32_t ref = T_REF_DEC(pRSmaInfo);
-  smaDebug("vgId:%d, unref rsma info:%p, val:%d", SMA_VID(pSma), pRSmaInfo, ref);
+  smaTrace("vgId:%d, unref rsma info:%p, val:%d", SMA_VID(pSma), pRSmaInfo, ref);
 }
 
-// smaFileUtil ================
-
-#define TD_FILE_HEAD_SIZE 512
-
-typedef struct STFInfo STFInfo;
-typedef struct STFile  STFile;
-
-struct STFInfo {
-  // common fields
-  uint32_t magic;
-  uint32_t ftype;
-  uint32_t fver;
-  int64_t  fsize;
-};
-
-enum {
-  TD_FTYPE_RSMA_QTASKINFO = 0,
-};
-
-#if 0
-struct STFile {
-  uint8_t   state;
-  STFInfo   info;
-  char     *fname;
-  TdFilePtr pFile;
-};
-
-#define TD_TFILE_PFILE(tf)        ((tf)->pFile)
-#define TD_TFILE_OPENED(tf)       (TD_TFILE_PFILE(tf) != NULL)
-#define TD_TFILE_FULL_NAME(tf)    ((tf)->fname)
-#define TD_TFILE_OPENED(tf)       (TD_TFILE_PFILE(tf) != NULL)
-#define TD_TFILE_CLOSED(tf)       (!TD_TFILE_OPENED(tf))
-#define TD_TFILE_SET_CLOSED(f)    (TD_TFILE_PFILE(f) = NULL)
-#define TD_TFILE_SET_STATE(tf, s) ((tf)->state = (s))
-
-int32_t tdInitTFile(STFile *pTFile, const char *dname, const char *fname);
-int32_t tdCreateTFile(STFile *pTFile, bool updateHeader, int8_t fType);
-int32_t tdOpenTFile(STFile *pTFile, int flags);
-int64_t tdReadTFile(STFile *pTFile, void *buf, int64_t nbyte);
-int64_t tdSeekTFile(STFile *pTFile, int64_t offset, int whence);
-int64_t tdWriteTFile(STFile *pTFile, void *buf, int64_t nbyte);
-int64_t tdAppendTFile(STFile *pTFile, void *buf, int64_t nbyte, int64_t *offset);
-int64_t tdGetTFileSize(STFile *pTFile, int64_t *size);
-int32_t tdRemoveTFile(STFile *pTFile);
-int32_t tdLoadTFileHeader(STFile *pTFile, STFInfo *pInfo);
-int32_t tdUpdateTFileHeader(STFile *pTFile);
-void    tdUpdateTFileMagic(STFile *pTFile, void *pCksm);
-void    tdCloseTFile(STFile *pTFile);
-void    tdDestroyTFile(STFile *pTFile);
-#endif
-
-void tdGetVndFileName(int32_t vgId, const char *pdname, const char *dname, const char *fname, int64_t version,
-                      char *outputName);
-void tdGetVndDirName(int32_t vgId, const char *pdname, const char *dname, bool endWithSep, char *outputName);
+void tdRSmaGetFileName(int32_t vgId, const char *pdname, const char *dname, const char *fname, int64_t suid,
+                       int8_t level, int64_t version, char *outputName);
+void tdRSmaGetDirName(int32_t vgId, const char *pdname, const char *dname, bool endWithSep, char *outputName);
 
 #ifdef __cplusplus
 }

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

@@ -210,9 +210,6 @@ void    smaCleanUp();
 int32_t smaOpen(SVnode* pVnode, int8_t rollback);
 int32_t smaClose(SSma* pSma);
 int32_t smaBegin(SSma* pSma);
-int32_t smaSyncPreCommit(SSma* pSma);
-int32_t smaSyncCommit(SSma* pSma);
-int32_t smaSyncPostCommit(SSma* pSma);
 int32_t smaPrepareAsyncCommit(SSma* pSma);
 int32_t smaCommit(SSma* pSma, SCommitInfo* pInfo);
 int32_t smaFinishCommit(SSma* pSma);
@@ -227,7 +224,6 @@ int32_t tdProcessRSmaSubmit(SSma* pSma, int64_t version, void* pReq, void* pMsg,
 int32_t tdProcessRSmaDrop(SSma* pSma, SVDropStbReq* pReq);
 int32_t tdFetchTbUidList(SSma* pSma, STbUidStore** ppStore, tb_uid_t suid, tb_uid_t uid);
 int32_t tdUpdateTbUidList(SSma* pSma, STbUidStore* pUidStore, bool isAdd);
-void    tdUidStoreDestory(STbUidStore* pStore);
 void*   tdUidStoreFree(STbUidStore* pStore);
 
 // SMetaSnapReader ========================================
@@ -274,6 +270,7 @@ int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData);
 // SRSmaSnapWriter ========================================
 int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWriter** ppWriter);
 int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData);
+int32_t rsmaSnapWriterPrepareClose(SRSmaSnapWriter* pWriter);
 int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback);
 
 typedef struct {
@@ -417,6 +414,7 @@ enum {
 
 struct SSnapDataHdr {
   int8_t  type;
+  int8_t  flag;
   int64_t index;
   int64_t size;
   uint8_t data[];

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

@@ -161,7 +161,10 @@ int32_t metaSnapWriterClose(SMetaSnapWriter** ppWriter, int8_t rollback) {
   SMetaSnapWriter* pWriter = *ppWriter;
 
   if (rollback) {
+    metaInfo("vgId:%d, meta snapshot writer close and rollback start ", TD_VID(pWriter->pMeta->pVnode));
     code = metaAbort(pWriter->pMeta);
+    metaInfo("vgId:%d, meta snapshot writer close and rollback finished, code:0x%x", TD_VID(pWriter->pMeta->pVnode),
+             code);
     if (code) goto _err;
   } else {
     code = metaCommit(pWriter->pMeta, pWriter->pMeta->txn);

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

@@ -710,6 +710,9 @@ int metaUpdateCtimeIdx(SMeta *pMeta, const SMetaEntry *pME) {
   if (metaBuildCtimeIdxKey(&ctimeKey, pME) < 0) {
     return 0;
   }
+  metaDebug("vgId:%d, start to save ctime:%" PRId64 " uid:%" PRId64 " ct:%" PRId64, TD_VID(pMeta->pVnode), pME->version,
+            pME->uid, ctimeKey.ctime);
+
   return tdbTbInsert(pMeta->pCtimeIdx, &ctimeKey, sizeof(ctimeKey), NULL, 0, pMeta->txn);
 }
 

source/dnode/vnode/src/sma/smaCommit.c

@@ -17,42 +17,11 @@
 
 extern SSmaMgmt smaMgmt;
 
-#if 0
-static int32_t tdProcessRSmaSyncPreCommitImpl(SSma *pSma);
-static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma);
-static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma);
-#endif
 static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma);
 static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo);
 static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma);
 static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pRSmaStat);
 
-#if 0
-/**
- * @brief Only applicable to Rollup SMA
- *
- * @param pSma
- * @return int32_t
- */
-int32_t smaSyncPreCommit(SSma *pSma) { return tdProcessRSmaSyncPreCommitImpl(pSma); }
-
-/**
- * @brief Only applicable to Rollup SMA
- *
- * @param pSma
- * @return int32_t
- */
-int32_t smaSyncCommit(SSma *pSma) { return tdProcessRSmaSyncCommitImpl(pSma); }
-
-/**
- * @brief Only applicable to Rollup SMA
- *
- * @param pSma
- * @return int32_t
- */
-int32_t smaSyncPostCommit(SSma *pSma) { return tdProcessRSmaSyncPostCommitImpl(pSma); }
-#endif
-
 /**
  * @brief async commit, only applicable to Rollup SMA
  *
@@ -128,84 +97,24 @@ _exit:
 
 int32_t smaFinishCommit(SSma *pSma) {
   int32_t code = 0;
+  int32_t lino = 0;
   SVnode *pVnode = pSma->pVnode;
 
+  code = tdRSmaFSFinishCommit(pSma);
+  TSDB_CHECK_CODE(code, lino, _exit);
+
   if (VND_RSMA1(pVnode) && (code = tsdbFinishCommit(VND_RSMA1(pVnode))) < 0) {
-    smaError("vgId:%d, failed to finish commit tsdb rsma1 since %s", TD_VID(pVnode), tstrerror(code));
-    goto _exit;
+    TSDB_CHECK_CODE(code, lino, _exit);
   }
   if (VND_RSMA2(pVnode) && (code = tsdbFinishCommit(VND_RSMA2(pVnode))) < 0) {
-    smaError("vgId:%d, failed to finish commit tsdb rsma2 since %s", TD_VID(pVnode), tstrerror(code));
-    goto _exit;
+    TSDB_CHECK_CODE(code, lino, _exit);
   }
 _exit:
-  terrno = code;
-  return code;
-}
-
-#if 0
-/**
- * @brief pre-commit for rollup sma(sync commit).
- *  1) set trigger stat of rsma timer TASK_TRIGGER_STAT_PAUSED.
- *  2) wait for all triggered fetch tasks to finish
- *  3) perform persist task for qTaskInfo
- *
- * @param pSma
- * @return int32_t
- */
-static int32_t tdProcessRSmaSyncPreCommitImpl(SSma *pSma) {
-  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
-  if (!pSmaEnv) {
-    return TSDB_CODE_SUCCESS;
-  }
-
-  SSmaStat  *pStat = SMA_ENV_STAT(pSmaEnv);
-  SRSmaStat *pRSmaStat = SMA_STAT_RSMA(pStat);
-
-  // step 1: set rsma stat paused
-  atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_PAUSED);
-
-  // step 2: wait for all triggered fetch tasks to finish
-  int32_t nLoops = 0;
-  while (1) {
-    if (T_REF_VAL_GET(pStat) == 0) {
-      smaDebug("vgId:%d, rsma fetch tasks are all finished", SMA_VID(pSma));
-      break;
-    } else {
-      smaDebug("vgId:%d, rsma fetch tasks are not all finished yet", SMA_VID(pSma));
-    }
-    ++nLoops;
-    if (nLoops > 1000) {
-      sched_yield();
-      nLoops = 0;
-    }
-  }
-
-  // step 3: perform persist task for qTaskInfo
-  pRSmaStat->commitAppliedVer = pSma->pVnode->state.applied;
-  tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat));
-
-  smaDebug("vgId:%d, rsma pre commit success", SMA_VID(pSma));
-
-  return TSDB_CODE_SUCCESS;
-}
-
-/**
- * @brief commit for rollup sma
- *
- * @param pSma
- * @return int32_t
- */
-static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma) {
-#if 0
-  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
-  if (!pSmaEnv) {
-    return TSDB_CODE_SUCCESS;
+  if (code) {
+    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
   }
-#endif
-  return TSDB_CODE_SUCCESS;
+  return code;
 }
-#endif
 
 // SQTaskFile ======================================================
 
@@ -218,6 +127,7 @@ static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma) {
  * @return int32_t
  */
 static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pStat) {
+#if 0
   SVnode  *pVnode = pSma->pVnode;
   SRSmaFS *pFS = RSMA_FS(pStat);
   int64_t  committed = pStat->commitAppliedVer;
@@ -264,31 +174,10 @@ static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pStat) {
   }
 
   taosWUnLockLatch(RSMA_FS_LOCK(pStat));
+#endif
   return TSDB_CODE_SUCCESS;
 }
 
-#if 0
-/**
- * @brief post-commit for rollup sma
- *  1) clean up the outdated qtaskinfo files
- *
- * @param pSma
- * @return int32_t
- */
-static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma) {
-  SVnode *pVnode = pSma->pVnode;
-  if (!VND_IS_RSMA(pVnode)) {
-    return TSDB_CODE_SUCCESS;
-  }
-
-  SRSmaStat *pRSmaStat = SMA_RSMA_STAT(pSma);
-
-  tdUpdateQTaskInfoFiles(pSma, pRSmaStat);
-
-  return TSDB_CODE_SUCCESS;
-}
-#endif
-
 /**
  * @brief Rsma async commit implementation(only do some necessary light weighted task)
  *  1) set rsma stat TASK_TRIGGER_STAT_PAUSED
@@ -298,9 +187,11 @@ static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma) {
  * @return int32_t
  */
 static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
+  int32_t  code = 0;
+
   SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
   if (!pEnv) {
-    return TSDB_CODE_SUCCESS;
+    return code;
   }
 
   SSmaStat  *pStat = SMA_ENV_STAT(pEnv);
@@ -317,7 +208,7 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
     }
   }
   pRSmaStat->commitAppliedVer = pSma->pVnode->state.applied;
-  ASSERT(pRSmaStat->commitAppliedVer > 0);
+  // ASSERT(pRSmaStat->commitAppliedVer > 0);
 
   // step 2: wait for all triggered fetch tasks to finish
   nLoops = 0;
@@ -351,8 +242,8 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
     }
   }
   smaInfo("vgId:%d, rsma commit, all items are consumed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());
-  if (tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat)) < 0) {
-    return TSDB_CODE_FAILED;
+  if ((code = tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat))) != 0) {
+    return code;
   }
   smaInfo("vgId:%d, rsma commit, operator state committed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());
 
@@ -383,7 +274,7 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
   if ((pTsdb = VND_RSMA1(pSma->pVnode))) tsdbPrepareCommit(pTsdb);
   if ((pTsdb = VND_RSMA2(pSma->pVnode))) tsdbPrepareCommit(pTsdb);
 
-  return TSDB_CODE_SUCCESS;
+  return code;
 }
 
 /**
@@ -394,26 +285,22 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
  */
 static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo) {
   int32_t code = 0;
+  int32_t lino = 0;
   SVnode *pVnode = pSma->pVnode;
-#if 0
-  SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pSmaEnv);
 
-  // perform persist task for qTaskInfo operator
-  if (tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat)) < 0) {
-    return TSDB_CODE_FAILED;
-  }
-#endif
+  code = tdRSmaFSCommit(pSma);
+  TSDB_CHECK_CODE(code, lino, _exit);
+
+  code = tsdbCommit(VND_RSMA1(pVnode), pInfo);
+  TSDB_CHECK_CODE(code, lino, _exit);
+
+  code = tsdbCommit(VND_RSMA2(pVnode), pInfo);
+  TSDB_CHECK_CODE(code, lino, _exit);
 
-  if ((code = tsdbCommit(VND_RSMA1(pVnode), pInfo)) < 0) {
-    smaError("vgId:%d, failed to commit tsdb rsma1 since %s", TD_VID(pVnode), tstrerror(code));
-    goto _exit;
-  }
-  if ((code = tsdbCommit(VND_RSMA2(pVnode), pInfo)) < 0) {
-    smaError("vgId:%d, failed to commit tsdb rsma2 since %s", TD_VID(pVnode), tstrerror(code));
-    goto _exit;
-  }
 _exit:
-  terrno = code;
+  if (code) {
+    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
+  }
   return code;
 }
 

source/dnode/vnode/src/sma/smaEnv.c

@@ -243,10 +243,7 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pS
         return TSDB_CODE_FAILED;
       }
 
-      if (!(RSMA_FS(pRSmaStat)->aQTaskInf = taosArrayInit(1, sizeof(SQTaskFile)))) {
-        terrno = TSDB_CODE_OUT_OF_MEMORY;
-        return TSDB_CODE_FAILED;
-      }
+      taosInitRWLatch(RSMA_FS_LOCK(pRSmaStat));
     } else if (smaType == TSDB_SMA_TYPE_TIME_RANGE) {
       // TODO
     } else {

source/dnode/vnode/src/sma/smaOpen.c

@@ -121,8 +121,6 @@ int smaSetKeepCfg(SVnode *pVnode, STsdbKeepCfg *pKeepCfg, STsdbCfg *pCfg, int ty
 int32_t smaOpen(SVnode *pVnode, int8_t rollback) {
   STsdbCfg *pCfg = &pVnode->config.tsdbCfg;
 
-  ASSERT(!pVnode->pSma);
-
   SSma *pSma = taosMemoryCalloc(1, sizeof(SSma));
   if (!pSma) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -137,7 +135,7 @@ int32_t smaOpen(SVnode *pVnode, int8_t rollback) {
 
   if (VND_IS_RSMA(pVnode)) {
     STsdbKeepCfg keepCfg = {0};
-    for (int i = 0; i < TSDB_RETENTION_MAX; ++i) {
+    for (int32_t i = 0; i < TSDB_RETENTION_MAX; ++i) {
       if (i == TSDB_RETENTION_L0) {
         SMA_OPEN_RSMA_IMPL(pVnode, 0);
       } else if (i == TSDB_RETENTION_L1) {
@@ -145,12 +143,14 @@ int32_t smaOpen(SVnode *pVnode, int8_t rollback) {
       } else if (i == TSDB_RETENTION_L2) {
         SMA_OPEN_RSMA_IMPL(pVnode, 2);
       } else {
-        ASSERT(0);
+        terrno = TSDB_CODE_APP_ERROR;
+        smaError("vgId:%d, sma open failed since %s, level:%d", TD_VID(pVnode), terrstr(), i);
+        goto _err;
       }
     }
 
     // restore the rsma
-    if (tdRSmaRestore(pSma, RSMA_RESTORE_REBOOT, pVnode->state.committed) < 0) {
+    if (tdRSmaRestore(pSma, RSMA_RESTORE_REBOOT, pVnode->state.committed, rollback) < 0) {
       goto _err;
     }
   }
@@ -181,8 +181,11 @@ int32_t smaClose(SSma *pSma) {
  * @param committedVer
  * @return int32_t
  */
-int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer) {
-  ASSERT(VND_IS_RSMA(pSma->pVnode));
+int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer, int8_t rollback) {
+  if (!VND_IS_RSMA(pSma->pVnode)) {
+    terrno = TSDB_CODE_RSMA_INVALID_ENV;
+    return TSDB_CODE_FAILED;
+  }
 
-  return tdRSmaProcessRestoreImpl(pSma, type, committedVer);
+  return tdRSmaProcessRestoreImpl(pSma, type, committedVer, rollback);
 }

source/dnode/vnode/src/sma/smaTimeRange.c

@@ -111,35 +111,48 @@ _err:
  * @return int32_t
  */
 static int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
-  SSmaCfg *pCfg = (SSmaCfg *)pMsg;
+  int32_t        code = 0;
+  int32_t        lino = 0;
+  SSmaCfg       *pCfg = (SSmaCfg *)pMsg;
+  SName          stbFullName = {0};
+  SVCreateStbReq pReq = {0};
 
   if (TD_VID(pSma->pVnode) == pCfg->dstVgId) {
     // create tsma meta in dstVgId
     if (metaCreateTSma(SMA_META(pSma), version, pCfg) < 0) {
-      return -1;
+      code = terrno;
+      TSDB_CHECK_CODE(code, lino, _exit);
     }
 
     // create stable to save tsma result in dstVgId
-    SName stbFullName = {0};
     tNameFromString(&stbFullName, pCfg->dstTbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
-    SVCreateStbReq pReq = {0};
     pReq.name = (char *)tNameGetTableName(&stbFullName);
     pReq.suid = pCfg->dstTbUid;
     pReq.schemaRow = pCfg->schemaRow;
     pReq.schemaTag = pCfg->schemaTag;
 
     if (metaCreateSTable(SMA_META(pSma), version, &pReq) < 0) {
-      return -1;
+      code = terrno;
+      TSDB_CHECK_CODE(code, lino, _exit);
     }
+  } else {
+    code = terrno = TSDB_CODE_TSMA_INVALID_STAT;
+    TSDB_CHECK_CODE(code, lino, _exit);
+  }
 
+_exit:
+  if (code) {
+    smaError("vgId:%d, failed at line %d to create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64
+             " dstTb:%s dstVg:%d",
+             SMA_VID(pSma), lino, pCfg->indexName, pCfg->indexUid, pCfg->tableUid, pCfg->dstTbUid, pReq.name,
+             pCfg->dstVgId);
+  } else {
     smaDebug("vgId:%d, success to create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64
              " dstTb:%s dstVg:%d",
              SMA_VID(pSma), pCfg->indexName, pCfg->indexUid, pCfg->tableUid, pCfg->dstTbUid, pReq.name, pCfg->dstVgId);
-  } else {
-    ASSERT(0);
   }
 
-  return 0;
+  return code;
 }
 
 /**
@@ -174,7 +187,7 @@ static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char
   STSmaStat *pTsmaStat = NULL;
 
   if (!pEnv || !(pStat = SMA_ENV_STAT(pEnv))) {
-    terrno = TSDB_CODE_TSMA_INVALID_STAT;
+    terrno = TSDB_CODE_TSMA_INVALID_ENV;
     return TSDB_CODE_FAILED;
   }
 
@@ -213,12 +226,16 @@ static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char
              indexUid, tstrerror(terrno));
     goto _err;
   }
+
   // TODO deleteReq
   taosArrayDestroy(deleteReq.deleteReqs);
-  
-
 #if 0
-   ASSERT(!strncasecmp("td.tsma.rst.tb", pTsmaStat->pTSma->dstTbName, 14));
+  if (!strncasecmp("td.tsma.rst.tb", pTsmaStat->pTSma->dstTbName, 14)) {
+    terrno = TSDB_CODE_APP_ERROR;
+    smaError("vgId:%d, tsma insert for smaIndex %" PRIi64 " failed since %s, %s", SMA_VID(pSma), indexUid,
+             pTsmaStat->pTSma->indexUid, tstrerror(terrno), pTsmaStat->pTSma->dstTbName);
+    goto _err;
+  }
 #endif
 
   SRpcMsg submitReqMsg = {

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

@@ -29,7 +29,6 @@ int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t
 
   int32_t actualLen = blockEncode(pBlock, pRetrieve->data, numOfCols);
   actualLen += sizeof(SRetrieveTableRsp);
-  ASSERT(actualLen <= dataStrLen);
   taosArrayPush(pRsp->blockDataLen, &actualLen);
   taosArrayPush(pRsp->blockData, &buf);
   return 0;
@@ -62,7 +61,6 @@ static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp, i
 
 int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset) {
   const STqExecHandle* pExec = &pHandle->execHandle;
-  ASSERT(pExec->subType == TOPIC_SUB_TYPE__COLUMN);
 
   qTaskInfo_t task = pExec->task;
 
@@ -87,7 +85,8 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
     uint64_t     ts = 0;
     tqDebug("vgId:%d, tmq task start to execute", pTq->pVnode->config.vgId);
     if (qExecTask(task, &pDataBlock, &ts) < 0) {
-      ASSERT(0);
+      tqError("vgId:%d, task exec error since %s", pTq->pVnode->config.vgId, terrstr());
+      return -1;
     }
     tqDebug("vgId:%d, tmq task executed, get %p", pTq->pVnode->config.vgId, pDataBlock);
 
@@ -105,10 +104,14 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
   }
 
   if (qStreamExtractOffset(task, &pRsp->rspOffset) < 0) {
-    ASSERT(0);
     return -1;
   }
-  ASSERT(pRsp->rspOffset.type != 0);
+
+  if (pRsp->rspOffset.type == 0) {
+    tqError("expected rsp offset: type %d %" PRId64 " %" PRId64 " %" PRId64, pRsp->rspOffset.type, pRsp->rspOffset.ts,
+            pRsp->rspOffset.uid, pRsp->rspOffset.version);
+    return -1;
+  }
 
   ASSERT(pRsp->withTbName == false);
   ASSERT(pRsp->withSchema == false);
@@ -141,7 +144,8 @@ int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMeta
     uint64_t     ts = 0;
     tqDebug("tmqsnap task start to execute");
     if (qExecTask(task, &pDataBlock, &ts) < 0) {
-      ASSERT(0);
+      tqError("vgId:%d, task exec error since %s", pTq->pVnode->config.vgId, terrstr());
+      return -1;
     }
     tqDebug("tmqsnap task execute end, get %p", pDataBlock);
 
@@ -207,17 +211,20 @@ int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMeta
     break;
   }
 
-  if (qStreamExtractOffset(task, &pRsp->rspOffset) < 0) {
-    ASSERT(0);
+  qStreamExtractOffset(task, &pRsp->rspOffset);
+
+  if (pRsp->rspOffset.type == 0) {
+    tqError("expected rsp offset: type %d %" PRId64 " %" PRId64 " %" PRId64, pRsp->rspOffset.type, pRsp->rspOffset.ts,
+            pRsp->rspOffset.uid, pRsp->rspOffset.version);
+    return -1;
   }
 
-  ASSERT(pRsp->rspOffset.type != 0);
   return 0;
 }
 
 int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxRsp* pRsp) {
   STqExecHandle* pExec = &pHandle->execHandle;
-  ASSERT(pExec->subType != TOPIC_SUB_TYPE__COLUMN);
+  /*A(pExec->subType != TOPIC_SUB_TYPE__COLUMN);*/
 
   SArray* pBlocks = taosArrayInit(0, sizeof(SSDataBlock));
   SArray* pSchemas = taosArrayInit(0, sizeof(void*));

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

@@ -100,8 +100,6 @@ int32_t tqSnapRead(STqSnapReader* pReader, uint8_t** ppData) {
     }
   }
 
-  ASSERT(pVal && vLen);
-
   *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen);
   if (*ppData == NULL) {
     code = TSDB_CODE_OUT_OF_MEMORY;