fix:fix length error in tag insert

CMakeLists.txt

@@ -35,7 +35,7 @@ endif(${BUILD_TEST})
 add_subdirectory(source)
 add_subdirectory(tools)
 add_subdirectory(tests)
-add_subdirectory(example)
+add_subdirectory(examples/c)
 
 # docs 
 add_subdirectory(docs)

cmake/cmake.define

@@ -46,7 +46,7 @@ ENDIF ()
 
 IF (TD_WINDOWS)
     MESSAGE("${Yellow} set compiler flag for Windows! ${ColourReset}")
-    SET(COMMON_FLAGS "/w /D_WIN32 /Zi")
+    SET(COMMON_FLAGS "/w /D_WIN32 /DWIN32 /Zi")
     SET(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} /MANIFEST:NO")
     # IF (MSVC AND (MSVC_VERSION GREATER_EQUAL 1900))
     #  SET(COMMON_FLAGS "${COMMON_FLAGS} /Wv:18")

docs-cn/12-taos-sql/07-function.md

 ---
-sidebar_label: SQL 函数
-title: SQL 函数
+sidebar_label: 函数
+title: 函数
 toc_max_heading_level: 4
 ---
 
@@ -20,7 +20,7 @@ toc_max_heading_level: 4
 
 **返回结果类型**：如果输入值为整数，输出值是 UBIGINT 类型。如果输入值是 FLOAT/DOUBLE 数据类型，输出值是 DOUBLE 数据类型。
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -38,7 +38,7 @@ toc_max_heading_level: 4
 
 **返回结果类型**：DOUBLE。如果输入值为 NULL，输出值也为 NULL
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -56,7 +56,7 @@ toc_max_heading_level: 4
 
 **返回结果类型**：DOUBLE。如果输入值为 NULL，输出值也为 NULL
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -75,7 +75,7 @@ toc_max_heading_level: 4
 
 **返回结果类型**：DOUBLE。如果输入值为 NULL，输出值也为 NULL
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -94,7 +94,7 @@ SELECT CEIL(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：与指定列的原始数据类型一致。例如，如果指定列的原始数据类型为 Float，那么返回的数据类型也为 Float；如果指定列的原始数据类型为 Double，那么返回的数据类型也为 Double。
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列，无论 tag 列的类型是什么类型。
+**适用数据类型**：数值类型。
 
 **适用于**: 普通表、超级表。
 
@@ -115,7 +115,7 @@ SELECT CEIL(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：DOUBLE。如果输入值为 NULL，输出值也为 NULL
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -142,7 +142,7 @@ SELECT FLOOR(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：DOUBLE。如果输入值为 NULL，输出值也为 NULL
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -161,7 +161,7 @@ SELECT FLOOR(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：DOUBLE。如果输入值为 NULL，输出值也为 NULL
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -190,7 +190,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：DOUBLE。如果输入值为 NULL，输出值也为 NULL
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -208,7 +208,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：DOUBLE。如果输入值为 NULL，输出值也为 NULL
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -226,7 +226,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：DOUBLE。如果输入值为 NULL，输出值也为 NULL
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在 tag 列
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -248,7 +248,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：INT。如果输入值为NULL，输出值为NULL。
 
-**适用数据类型**：输入参数是 BINARY 类型或者 NCHAR 类型的字符串或者列。不能应用在 TAG 列。
+**适用数据类型**：VARCHAR, NCHAR
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -262,9 +262,9 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **功能说明**：字符串连接函数。
 
-**返回结果类型**：如果所有参数均为BINARY类型，则结果类型为BINARY。如果参数包含NCHAR类型，则结果类型为NCHAR。如果输入值为NULL，输出值为NULL。
+**返回结果类型**：如果所有参数均为 VARCHAR 类型，则结果类型为 VARCHAR。如果参数包含NCHAR类型，则结果类型为NCHAR。如果输入值为NULL，输出值为NULL。
 
-**适用数据类型**：BINARY, NCHAR。不能应用在 TAG 列。 该函数最小参数个数为2个，最大参数个数为8个。
+**适用数据类型**：VARCHAR, NCHAR。 该函数最小参数个数为2个，最大参数个数为8个。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -279,9 +279,9 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **功能说明**：带分隔符的字符串连接函数。
 
-**返回结果类型**：如果所有参数均为BINARY类型，则结果类型为BINARY。如果参数包含NCHAR类型，则结果类型为NCHAR。如果输入值为NULL，输出值为NULL。如果separator值不为NULL，其他输入为NULL，输出为空串。
+**返回结果类型**：如果所有参数均为VARCHAR类型，则结果类型为VARCHAR。如果参数包含NCHAR类型，则结果类型为NCHAR。如果输入值为NULL，输出值为NULL。如果separator值不为NULL，其他输入为NULL，输出为空串。
 
-**适用数据类型**：BINARY, NCHAR。不能应用在 TAG 列。 该函数最小参数个数为3个，最大参数个数为9个。
+**适用数据类型**：VARCHAR, NCHAR。 该函数最小参数个数为3个，最大参数个数为9个。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -298,7 +298,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：INT。
 
-**适用数据类型**：输入参数是 BINARY 类型或者 NCHAR 类型的字符串或者列。不能应用在 TAG 列。
+**适用数据类型**：输入参数是 VARCHAR 类型或者 NCHAR 类型的字符串或者列。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -315,7 +315,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：同输入类型。如果输入值为NULL，输出值为NULL。
 
-**适用数据类型**：输入参数是 BINARY 类型或者 NCHAR 类型的字符串或者列。不能应用在 TAG 列。
+**适用数据类型**：输入参数是 VARCHAR 类型或者 NCHAR 类型的字符串或者列。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -332,7 +332,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：同输入类型。如果输入值为NULL，输出值为NULL。
 
-**适用数据类型**：输入参数是 BINARY 类型或者 NCHAR 类型的字符串或者列。不能应用在 TAG 列。
+**适用数据类型**：输入参数是 VARCHAR 类型或者 NCHAR 类型的字符串或者列。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -349,7 +349,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：同输入类型。如果输入值为NULL，输出值为NULL。
 
-**适用数据类型**：输入参数是 BINARY 类型或者 NCHAR 类型的字符串或者列。不能应用在 TAG 列。
+**适用数据类型**：输入参数是 VARCHAR 类型或者 NCHAR 类型的字符串或者列。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -366,7 +366,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：同输入类型。如果输入值为NULL，输出值为NULL。
 
-**适用数据类型**：输入参数是 BINARY 类型或者 NCHAR 类型的字符串或者列。不能应用在 TAG 列。输入参数pos可以为正数，也可以为负数。如果pos是正数，表示开始位置从字符串开头正数计算。如果pos为负数，表示开始位置从字符串结尾倒数计算。如果输入参数len被忽略，返回的子串包含从pos开始的整个字串。
+**适用数据类型**：输入参数是 VARCHAR 类型或者 NCHAR 类型的字符串或者列。输入参数pos可以为正数，也可以为负数。如果pos是正数，表示开始位置从字符串开头正数计算。如果pos为负数，表示开始位置从字符串结尾倒数计算。如果输入参数len被忽略，返回的子串包含从pos开始的整个字串。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -383,7 +383,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
 
 **返回结果类型**：同输入类型。如果输入值为NULL，输出值为NULL。
 
-**适用数据类型**：输入参数是 BINARY 类型或者 NCHAR 类型的字符串或者列。不能应用在 TAG 列。
+**适用数据类型**：输入参数是 VARCHAR 类型或者 NCHAR 类型的字符串或者列。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 
@@ -400,7 +400,7 @@ SELECT ROUND(field_name) FROM { tb_name | stb_name } [WHERE clause];
   SELECT CAST(expression AS type_name) FROM { tb_name | stb_name } [WHERE clause]
 ```
 
-**功能说明**：数据类型转换函数，输入参数 expression 支持普通列、常量、标量函数及它们之间的四则运算，不支持 tag 列，只适用于 select 子句中。
+**功能说明**：数据类型转换函数，输入参数 expression 支持普通列、常量、标量函数及它们之间的四则运算，只适用于 select 子句中。
 
 **返回结果类型**：CAST 中指定的类型（type_name），可以是 BIGINT、BIGINT UNSIGNED、BINARY、VARCHAR、NCHAR和TIMESTAMP。
 
@@ -423,7 +423,7 @@ SELECT TO_ISO8601(ts_val | ts_col) FROM { tb_name | stb_name } [WHERE clause];
 
 **功能说明**：将 UNIX 时间戳转换成为 ISO8601 标准的日期时间格式，并附加客户端时区信息。
 
-**返回结果数据类型**：BINARY 类型。
+**返回结果数据类型**：VARCHAR 类型。
 
 **适用数据类型**：UNIX 时间戳常量或是 TIMESTAMP 类型的列
 
@@ -462,7 +462,7 @@ SELECT TO_UNIXTIMESTAMP(datetime_string | ts_col) FROM { tb_name | stb_name } [W
 
 **返回结果数据类型**：长整型 INT64。
 
-**应用字段**：字符串常量或是 BINARY/NCHAR 类型的列。
+**应用字段**：字符串常量或是 VARCHAR/NCHAR 类型的列。
 
 **适用于**：表、超级表。
 
@@ -549,7 +549,7 @@ SELECT TIMEZONE() FROM { tb_name | stb_name } [WHERE clause];
 
 **功能说明**：返回客户端当前时区信息。
 
-**返回结果数据类型**：BINARY 类型。
+**返回结果数据类型**：VARCHAR 类型。
 
 **应用字段**：无
 
@@ -668,7 +668,7 @@ SELECT LEASTSQUARES(field_name, start_val, step_val) FROM tb_name [WHERE clause]
 SELECT MODE(field_name) FROM tb_name [WHERE clause];
 ```
 
-**功能说明**：返回出现频率最高的值，若存在多个频率相同的最高值，输出空。不能匹配标签、时间戳输出。
+**功能说明**：返回出现频率最高的值，若存在多个频率相同的最高值，输出空。
 
 **返回数据类型**：同应用的字段。
 
@@ -808,6 +808,25 @@ SELECT BOTTOM(field_name, K) FROM { tb_name | stb_name } [WHERE clause];
 - 系统同时返回该记录关联的时间戳列；
 - 限制：BOTTOM 函数不支持 FILL 子句。
 
+### FIRST
+
+```
+SELECT FIRST(field_name) FROM { tb_name | stb_name } [WHERE clause];
+```
+
+**功能说明**：统计表/超级表中某列的值最先写入的非 NULL 值。
+
+**返回数据类型**：同应用的字段。
+
+**适用数据类型**：所有字段。
+
+**适用于**：表和超级表。
+
+**使用说明**:
+
+- 如果要返回各个列的首个（时间戳最小）非 NULL 值，可以使用 FIRST(\*)；
+- 如果结果集中的某列全部为 NULL 值，则该列的返回结果也是 NULL；
+- 如果结果集中所有列全部为 NULL 值，则不返回结果。
 
 ### INTERP
 
@@ -919,25 +938,6 @@ SELECT PERCENTILE(field_name, P) FROM { tb_name } [WHERE clause];
 
 **使用说明**：*P*值取值范围 0≤*P*≤100，为 0 的时候等同于 MIN，为 100 的时候等同于 MAX。
 
-### FIRST
-
-```
-SELECT FIRST(field_name) FROM { tb_name | stb_name } [WHERE clause];
-```
-
-**功能说明**：统计表/超级表中某列的值最先写入的非 NULL 值。
-
-**返回数据类型**：同应用的字段。
-
-**适用数据类型**：所有字段。
-
-**适用于**：表和超级表。
-
-**使用说明**:
-
-- 如果要返回各个列的首个（时间戳最小）非 NULL 值，可以使用 FIRST(\*)；
-- 如果结果集中的某列全部为 NULL 值，则该列的返回结果也是 NULL；
-- 如果结果集中所有列全部为 NULL 值，则不返回结果。
 
 ### TAIL
 
@@ -1005,7 +1005,7 @@ SELECT UNIQUE(field_name) FROM {tb_name | stb_name} [WHERE clause];
 
 **返回结果类型**： 输入列如果是整数类型返回值为长整型 （int64_t），浮点数返回值为双精度浮点数（Double）。无符号整数类型返回值为无符号长整型（uint64_t）。 返回结果中同时带有每行记录对应的时间戳。
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上；在超级表查询中使用时，不能应用在标签之上。
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**： 适用于内层查询和外层查询。
 
@@ -1076,7 +1076,7 @@ SELECT IRATE(field_name) FROM tb_name WHERE clause;
 
   **返回结果类型**： 返回双精度浮点数类型。
 
-  **适用数据类型**： 不能应用在 timestamp、binary、nchar、bool 类型上；在超级表查询中使用时，不能应用在标签之上。
+  **适用数据类型**： 数值类型。
 
   **嵌套子查询支持**： 适用于内层查询和外层查询。
 
@@ -1124,7 +1124,7 @@ SELECT STATECOUNT(field_name, oper, val) FROM { tb_name | stb_name } [WHERE clau
 
 **返回结果类型**：整形。
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上。
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：不支持应用在子查询上。
 
@@ -1152,7 +1152,7 @@ SELECT stateDuration(field_name, oper, val, unit) FROM { tb_name | stb_name } [W
 
 **返回结果类型**：整形。
 
-**适用数据类型**：不能应用在 timestamp、binary、nchar、bool 类型字段上。
+**适用数据类型**：数值类型。
 
 **嵌套子查询支持**：不支持应用在子查询上。
 

docs-cn/12-taos-sql/13-operators.md

@@ -35,8 +35,8 @@ TDengine 支持 `UNION ALL` 和 `UNION` 操作符。UNION ALL 将查询返回的
 | --- | :---------------: | -------------------------------------------------------------------- | -------------------- |
 | 1   |         =         | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 相等                 |
 | 2   |      <\>, !=      | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型，且不可以为表的时间戳主键列 | 不相等               |
-| 3   |      \>, \<       | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 大于，小于           |
-| 4   |     \>=, \<=      | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 大于等于，小于等于   |
+| 3   |      \>, <       | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 大于，小于           |
+| 4   |     \>=, <=      | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型                             | 大于等于，小于等于   |
 | 5   |   IS [NOT] NULL   | 所有类型                                                             | 是否为空值           |
 | 6   | [NOT] BETWEEN AND | 除 BOOL、BLOB、MEDIUMBLOB 和 JSON 外的所有类型                       | 闭区间比较           |
 | 7   |        IN         | 除 BLOB、MEDIUMBLOB 和 JSON 外的所有类型，且不可以为表的时间戳主键列 | 与列表内的任意值相等 |

docs-en/12-taos-sql/13-operators.md

@@ -35,8 +35,8 @@ TDengine provides 2 set operators: `UNION ALL` and `UNION`. `UNION ALL` combines
 | --- | :---------------: | ------------------------------------------------------------------- | ----------------------------------------------- |
 | 1   |         =         | Except for BLOB, MEDIUMBLOB and JSON                                | Equal                                           |
 | 2   |      <\>, !=      | Except for BLOB, MEDIUMBLOB, JSON and primary key of timestamp type | Not equal                                       |
-| 3   |      \>, \<       | Except for BLOB, MEDIUMBLOB and JSON                                | Greater than, less than                         |
-| 4   |     \>=, \<=      | Except for BLOB, MEDIUMBLOB and JSON                                | Greater than or equal to, less than or equal to |
+| 3   |      \>, <       | Except for BLOB, MEDIUMBLOB and JSON                                | Greater than, less than                         |
+| 4   |     \>=, <=      | Except for BLOB, MEDIUMBLOB and JSON                                | Greater than or equal to, less than or equal to |
 | 5   |   IS [NOT] NULL   | Any types                                                           | Is NULL or NOT                                  |
 | 6   | [NOT] BETWEEN AND | Except for BLOB, MEDIUMBLOB and JSON                                | In a value range or not                         |
 | 7   |        IN         | Except for BLOB, MEDIUMBLOB, JSON and primary key of timestamp type | In a list of values or not                      |

example/CMakeLists.txt

-add_executable(tmq "")
-add_executable(tstream "")
-add_executable(demoapi "")
-
-target_sources(tmq
-    PRIVATE
-    "src/tmq.c"
-)
-
-target_sources(tstream
-    PRIVATE
-    "src/tstream.c"
-)
-
-target_sources(demoapi
-    PRIVATE
-    "src/demoapi.c"
-)
-
-target_link_libraries(tmq
-    taos_static
-)
-
-target_link_libraries(tstream
-    taos_static
-)
-
-target_link_libraries(demoapi
-    taos_static
-)
-
-target_include_directories(tmq
-    PUBLIC "${TD_SOURCE_DIR}/include/os"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
-
-target_include_directories(tstream
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
-
-target_include_directories(demoapi
-    PUBLIC "${TD_SOURCE_DIR}/include/client"
-    PUBLIC "${TD_SOURCE_DIR}/include/os"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
-
-SET_TARGET_PROPERTIES(tmq PROPERTIES OUTPUT_NAME tmq)
-SET_TARGET_PROPERTIES(tstream PROPERTIES OUTPUT_NAME tstream)
-SET_TARGET_PROPERTIES(demoapi PROPERTIES OUTPUT_NAME demoapi)

examples/c/CMakeLists.txt

@@ -3,20 +3,70 @@ PROJECT(TDengine)
 IF (TD_LINUX)
   INCLUDE_DIRECTORIES(. ${TD_SOURCE_DIR}/src/inc ${TD_SOURCE_DIR}/src/client/inc  ${TD_SOURCE_DIR}/inc)
   AUX_SOURCE_DIRECTORY(. SRC)
-  ADD_EXECUTABLE(demo apitest.c)
-  TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread )
-  ADD_EXECUTABLE(sml schemaless.c)
-  TARGET_LINK_LIBRARIES(sml taos_static trpc tutil pthread )
-  ADD_EXECUTABLE(subscribe subscribe.c)
-  TARGET_LINK_LIBRARIES(subscribe taos_static trpc tutil pthread )
-  ADD_EXECUTABLE(epoll epoll.c)
-  TARGET_LINK_LIBRARIES(epoll taos_static trpc tutil pthread lua)
+  #  ADD_EXECUTABLE(demo apitest.c)
+  #TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread )
+  #ADD_EXECUTABLE(sml schemaless.c)
+  #TARGET_LINK_LIBRARIES(sml taos_static trpc tutil pthread )
+  #ADD_EXECUTABLE(subscribe subscribe.c)
+  #TARGET_LINK_LIBRARIES(subscribe taos_static trpc tutil pthread )
+  #ADD_EXECUTABLE(epoll epoll.c)
+  #TARGET_LINK_LIBRARIES(epoll taos_static trpc tutil pthread lua)
+
+  add_executable(tmq "")
+  add_executable(tstream "")
+  add_executable(demoapi "")
+
+  target_sources(tmq
+      PRIVATE
+      "tmq.c"
+      )
+
+  target_sources(tstream
+      PRIVATE
+      "tstream.c"
+      )
+
+  target_sources(demoapi
+      PRIVATE
+      "demoapi.c"
+      )
+
+  target_link_libraries(tmq
+      taos_static
+      )
+
+  target_link_libraries(tstream
+      taos_static
+      )
+
+  target_link_libraries(demoapi
+      taos_static
+      )
+
+  target_include_directories(tmq
+      PUBLIC "${TD_SOURCE_DIR}/include/os"
+      PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+      )
+
+  target_include_directories(tstream
+      PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+      )
+
+  target_include_directories(demoapi
+      PUBLIC "${TD_SOURCE_DIR}/include/client"
+      PUBLIC "${TD_SOURCE_DIR}/include/os"
+      PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+      )
+
+  SET_TARGET_PROPERTIES(tmq PROPERTIES OUTPUT_NAME tmq)
+  SET_TARGET_PROPERTIES(tstream PROPERTIES OUTPUT_NAME tstream)
+  SET_TARGET_PROPERTIES(demoapi PROPERTIES OUTPUT_NAME demoapi)
 ENDIF ()
 IF (TD_DARWIN)
   INCLUDE_DIRECTORIES(. ${TD_SOURCE_DIR}/src/inc ${TD_SOURCE_DIR}/src/client/inc  ${TD_SOURCE_DIR}/inc)
   AUX_SOURCE_DIRECTORY(. SRC)
-  ADD_EXECUTABLE(demo demo.c)
-  TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread lua)
-  ADD_EXECUTABLE(epoll epoll.c)
-  TARGET_LINK_LIBRARIES(epoll taos_static trpc tutil pthread lua)
+  #ADD_EXECUTABLE(demo demo.c)
+  #TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread lua)
+  #ADD_EXECUTABLE(epoll epoll.c)
+  #TARGET_LINK_LIBRARIES(epoll taos_static trpc tutil pthread lua)
 ENDIF ()

example/src/tmq.c → examples/c/tmq.c

@@ -165,7 +165,6 @@ tmq_t* build_consumer() {
   tmq_conf_set(conf, "group.id", "tg2");
   tmq_conf_set(conf, "td.connect.user", "root");
   tmq_conf_set(conf, "td.connect.pass", "taosdata");
-  /*tmq_conf_set(conf, "td.connect.db", "abc1");*/
   tmq_conf_set(conf, "msg.with.table.name", "true");
   tmq_conf_set(conf, "enable.auto.commit", "false");
   tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
@@ -191,7 +190,6 @@ void basic_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
     return;
   }
   int32_t cnt = 0;
-  /*clock_t startTime = clock();*/
   while (running) {
     TAOS_RES* tmqmessage = tmq_consumer_poll(tmq, 0);
     if (tmqmessage) {
@@ -204,8 +202,6 @@ void basic_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
       /*break;*/
     }
   }
-  /*clock_t endTime = clock();*/
-  /*printf("log cnt: %d %f s\n", cnt, (double)(endTime - startTime) / CLOCKS_PER_SEC);*/
 
   err = tmq_consumer_close(tmq);
   if (err)

include/common/tdataformat.h

@@ -30,6 +30,7 @@ extern "C" {
 typedef struct SSchema       SSchema;
 typedef struct STColumn      STColumn;
 typedef struct STSchema      STSchema;
+typedef struct SValue        SValue;
 typedef struct SColVal       SColVal;
 typedef struct STSRow2       STSRow2;
 typedef struct STSRowBuilder STSRowBuilder;
@@ -40,24 +41,26 @@ typedef struct STag          STag;
 int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t nCols, STSchema **ppTSchema);
 void    tTSchemaDestroy(STSchema *pTSchema);
 
-// SColVal
-#define ColValNONE               ((SColVal){.type = COL_VAL_NONE, .nData = 0, .pData = NULL})
-#define ColValNULL               ((SColVal){.type = COL_VAL_NULL, .nData = 0, .pData = NULL})
-#define ColValDATA(nData, pData) ((SColVal){.type = COL_VAL_DATA, .nData = (nData), .pData = (pData)})
-
 // STSRow2
+#define COL_VAL_NONE(CID)     ((SColVal){.cid = (CID), .isNone = 1})
+#define COL_VAL_NULL(CID)     ((SColVal){.cid = (CID), .isNull = 1})
+#define COL_VAL_VALUE(CID, V) ((SColVal){.cid = (CID), .value = (V)})
+
+int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow);
+void    tTSRowFree(STSRow2 *pRow);
+void    tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
+int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray);
 int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow);
 int32_t tGetTSRow(uint8_t *p, STSRow2 *pRow);
-int32_t tTSRowDup(const STSRow2 *pRow, STSRow2 **ppRow);
-void    tTSRowFree(STSRow2 *pRow);
-int32_t tTSRowGet(const STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
 
 // STSRowBuilder
+#if 0
 int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, int32_t nCols, SSchema *pSchema);
 void    tTSRowBuilderClear(STSRowBuilder *pBuilder);
 void    tTSRowBuilderReset(STSRowBuilder *pBuilder);
 int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, uint8_t *pData, uint32_t nData);
 int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow);
+#endif
 
 // STag
 int32_t tTagNew(SArray *pArray, int32_t version, int8_t isJson, STag **ppTag);
@@ -90,7 +93,9 @@ struct STSchema {
 #define TSROW_HAS_NONE ((uint8_t)0x1)
 #define TSROW_HAS_NULL ((uint8_t)0x2U)
 #define TSROW_HAS_VAL  ((uint8_t)0x4U)
-#define TSROW_KV_ROW   ((uint8_t)0x10U)
+#define TSROW_KV_SMALL ((uint8_t)0x10U)
+#define TSROW_KV_MID   ((uint8_t)0x20U)
+#define TSROW_KV_BIG   ((uint8_t)0x40U)
 struct STSRow2 {
   TSKEY    ts;
   uint8_t  flags;
@@ -113,11 +118,31 @@ struct STSRowBuilder {
   STSRow2   row;
 };
 
-typedef enum { COL_VAL_NONE = 0, COL_VAL_NULL = 1, COL_VAL_DATA = 2 } EColValT;
+struct SValue {
+  union {
+    int8_t   i8;   // TSDB_DATA_TYPE_BOOL||TSDB_DATA_TYPE_TINYINT
+    uint8_t  u8;   // TSDB_DATA_TYPE_UTINYINT
+    int16_t  i16;  // TSDB_DATA_TYPE_SMALLINT
+    uint16_t u16;  // TSDB_DATA_TYPE_USMALLINT
+    int32_t  i32;  // TSDB_DATA_TYPE_INT
+    uint32_t u32;  // TSDB_DATA_TYPE_UINT
+    int64_t  i64;  // TSDB_DATA_TYPE_BIGINT
+    uint64_t u64;  // TSDB_DATA_TYPE_UBIGINT
+    TSKEY    ts;   // TSDB_DATA_TYPE_TIMESTAMP
+    float    f;    // TSDB_DATA_TYPE_FLOAT
+    double   d;    // TSDB_DATA_TYPE_DOUBLE
+    struct {
+      uint32_t nData;
+      uint8_t *pData;
+    };
+  };
+};
+
 struct SColVal {
-  EColValT type;
-  uint32_t nData;
-  uint8_t *pData;
+  int16_t cid;
+  int8_t  isNone;
+  int8_t  isNull;
+  SValue  value;
 };
 
 #pragma pack(push, 1)

include/common/tmsg.h

@@ -244,7 +244,7 @@ typedef struct {
   const void* pMsg;
 } SSubmitMsgIter;
 
-int32_t tInitSubmitMsgIter(SSubmitReq* pMsg, SSubmitMsgIter* pIter);
+int32_t tInitSubmitMsgIter(const SSubmitReq* pMsg, SSubmitMsgIter* pIter);
 int32_t tGetSubmitMsgNext(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
 int32_t tInitSubmitBlkIter(SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
 STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
@@ -945,7 +945,6 @@ typedef struct {
   int64_t timeInFetchQueue;
 } SQnodeLoad;
 
-
 typedef struct {
   int32_t     sver;      // software version
   int64_t     dnodeVer;  // dnode table version in sdb
@@ -1002,7 +1001,6 @@ typedef struct {
 
 typedef struct {
   int32_t  vgId;
-  int32_t  dnodeId;
   char     db[TSDB_DB_FNAME_LEN];
   int64_t  dbUid;
   int32_t  vgVersion;
@@ -1025,16 +1023,14 @@ typedef struct {
   int8_t   compression;
   int8_t   strict;
   int8_t   cacheLastRow;
+  int8_t   isTsma;
+  int8_t   standby;
   int8_t   replica;
   int8_t   selfIndex;
   SReplica replicas[TSDB_MAX_REPLICA];
   int32_t  numOfRetensions;
   SArray*  pRetensions;  // SRetention
-
-  // for tsma
-  int8_t isTsma;
-  void*  pTsma;
-
+  void*    pTsma;
 } SCreateVnodeReq;
 
 int32_t tSerializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
@@ -1072,8 +1068,8 @@ typedef struct {
   int8_t   walLevel;
   int8_t   strict;
   int8_t   cacheLastRow;
-  int8_t   replica;
   int8_t   selfIndex;
+  int8_t   replica;
   SReplica replicas[TSDB_MAX_REPLICA];
 } SAlterVnodeReq;
 
@@ -1977,7 +1973,7 @@ typedef struct {
   int8_t   killConnection;
   int8_t   align[3];
   SEpSet   epSet;
-  SArray  *pQnodeList;
+  SArray*  pQnodeList;
 } SQueryHbRspBasic;
 
 typedef struct {
@@ -2305,6 +2301,7 @@ typedef struct {
   int8_t   intervalUnit;  // MACRO: TIME_UNIT_XXX
   int8_t   slidingUnit;   // MACRO: TIME_UNIT_XXX
   int8_t   timezoneInt;   // sma data expired if timezone changes.
+  int32_t  dstVgId;
   char     indexName[TSDB_INDEX_NAME_LEN];
   int32_t  exprLen;
   int32_t  tagsFilterLen;
@@ -2662,6 +2659,23 @@ typedef struct {
 int32_t tEncodeSVSubmitReq(SEncoder* pCoder, const SVSubmitReq* pReq);
 int32_t tDecodeSVSubmitReq(SDecoder* pCoder, SVSubmitReq* pReq);
 
+// TDMT_VND_DELETE
+typedef struct {
+  TSKEY sKey;
+  TSKEY eKey;
+
+  // super table
+  char* stbName;
+
+  // child/normal
+  char* tbName;
+} SVDeleteReq;
+
+typedef struct {
+  int32_t code;
+  // TODO
+} SVDeleteRsp;
+
 #pragma pack(pop)
 
 #ifdef __cplusplus

include/common/tmsgdef.h

@@ -221,9 +221,11 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_APPLY_MSG, "vnode-sync-apply-msg", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_CONFIG_CHANGE, "vnode-sync-config-change", NULL, NULL)
   
-  TD_DEF_MSG_TYPE(TDMT_VND_SYNC_VNODE, "vnode-sync-vnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_ALTER_VNODE, "vnode-alter-vnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_COMPACT_VNODE, "vnode-compact-vnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_ALTER_CONFIG, "vnode-alter-config", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_ALTER_REPLICA, "vnode-alter-replica", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_COMPACT, "vnode-compact", NULL, NULL)
+
+  TD_DEF_MSG_TYPE(TDMT_VND_DELETE, "vnode-delete-data", SVDeleteReq, SVDeleteRsp)
 
   // Requests handled by QNODE
   TD_NEW_MSG_SEG(TDMT_QND_MSG)

include/libs/stream/tstream.h

@@ -61,11 +61,8 @@ enum {
 };
 
 typedef struct {
-  int8_t type;
-
-  int32_t sourceVg;
-  int64_t sourceVer;
-
+  int8_t      type;
+  int64_t     ver;
   int32_t*    dataRef;
   SSubmitReq* data;
 } SStreamDataSubmit;
@@ -111,6 +108,8 @@ static FORCE_INLINE void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit)
   }
 }
 
+SStreamDataSubmit* streamSubmitRefClone(SStreamDataSubmit* pSubmit);
+
 int32_t streamDataBlockEncode(void** buf, const SStreamDataBlock* pOutput);
 void*   streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput);
 
@@ -209,8 +208,6 @@ struct SStreamTask {
   int32_t nodeId;
   SEpSet  epSet;
 
-  // source preprocess
-
   // exec
   STaskExec exec;
 
@@ -318,8 +315,6 @@ int32_t streamDequeueOutput(SStreamTask* pTask, void** output);
 
 int32_t streamTaskRun(SStreamTask* pTask);
 
-int32_t streamTaskHandleInput(SStreamTask* pTask, void* data);
-
 int32_t streamTaskProcessRunReq(SStreamTask* pTask, SMsgCb* pMsgCb);
 int32_t streamProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchReq* pReq, SRpcMsg* pMsg);
 int32_t streamProcessDispatchRsp(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchRsp* pRsp);

include/os/osDir.h

@@ -33,8 +33,19 @@ extern "C" {
 
 #ifdef WINDOWS
 #define TD_TMP_DIR_PATH "C:\\Windows\\Temp\\"
+#define TD_CFG_DIR_PATH "C:\\TDengine\\cfg\\"
+#define TD_DATA_DIR_PATH "C:\\TDengine\\data\\"
+#define TD_LOG_DIR_PATH "C:\\TDengine\\log\\"
+#elif defined(_TD_DARWIN_64)
+#define TD_TMP_DIR_PATH "/tmp/taosd/"
+#define TD_CFG_DIR_PATH "/usr/local/etc/taos/"
+#define TD_DATA_DIR_PATH "/usr/local/var/lib/taos/"
+#define TD_LOG_DIR_PATH "/usr/local/var/log/taos/"
 #else
 #define TD_TMP_DIR_PATH "/tmp/"
+#define TD_CFG_DIR_PATH "/etc/taos/"
+#define TD_DATA_DIR_PATH "/var/lib/taos/"
+#define TD_LOG_DIR_PATH "/var/log/taos/"
 #endif
 
 typedef struct TdDir *TdDirPtr;

include/util/taoserror.h

@@ -70,6 +70,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_NEED_RETRY                    TAOS_DEF_ERROR_CODE(0, 0x0028)
 #define TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE       TAOS_DEF_ERROR_CODE(0, 0x0029)
 #define TSDB_CODE_INVALID_TIMESTAMP             TAOS_DEF_ERROR_CODE(0, 0x0030)
+#define TSDB_CODE_MSG_DECODE_ERROR              TAOS_DEF_ERROR_CODE(0, 0x0031)
 
 #define TSDB_CODE_REF_NO_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0040)
 #define TSDB_CODE_REF_FULL                      TAOS_DEF_ERROR_CODE(0, 0x0041)

include/util/tencode.h

@@ -461,64 +461,153 @@ static FORCE_INLINE void* tDecoderMalloc(SDecoder* pCoder, int32_t size) {
 }
 
 // ===========================================
-#define tPutV(p, v)                  \
-  int32_t n = 0;                     \
-  for (;;) {                         \
-    if (v <= 0x7f) {                 \
-      if (p) p[n] = v;               \
-      n++;                           \
-      break;                         \
-    }                                \
-    if (p) p[n] = (v & 0x7f) | 0x80; \
-    n++;                             \
-    v >>= 7;                         \
-  }                                  \
-  return n;
+#define tPutV(p, v)                    \
+  do {                                 \
+    int32_t n = 0;                     \
+    for (;;) {                         \
+      if (v <= 0x7f) {                 \
+        if (p) p[n] = v;               \
+        n++;                           \
+        break;                         \
+      }                                \
+      if (p) p[n] = (v & 0x7f) | 0x80; \
+      n++;                             \
+      v >>= 7;                         \
+    }                                  \
+    return n;                          \
+  } while (0)
 
-#define tGetV(p, v)                            \
-  int32_t n = 0;                               \
-  if (v) *v = 0;                               \
-  for (;;) {                                   \
-    if (p[n] <= 0x7f) {                        \
-      if (v) (*v) |= (p[n] << (7 * n));        \
-      n++;                                     \
-      break;                                   \
-    }                                          \
-    if (v) (*v) |= ((p[n] & 0x7f) << (7 * n)); \
-    n++;                                       \
-  }                                            \
-  return n;
+#define tGetV(p, v)                              \
+  do {                                           \
+    int32_t n = 0;                               \
+    if (v) *v = 0;                               \
+    for (;;) {                                   \
+      if (p[n] <= 0x7f) {                        \
+        if (v) (*v) |= (p[n] << (7 * n));        \
+        n++;                                     \
+        break;                                   \
+      }                                          \
+      if (v) (*v) |= ((p[n] & 0x7f) << (7 * n)); \
+      n++;                                       \
+    }                                            \
+    return n;                                    \
+  } while (0)
 
 // PUT
+static FORCE_INLINE int32_t tPutU8(uint8_t* p, uint8_t v) {
+  if (p) ((uint8_t*)p)[0] = v;
+  return sizeof(uint8_t);
+}
+
 static FORCE_INLINE int32_t tPutI8(uint8_t* p, int8_t v) {
   if (p) ((int8_t*)p)[0] = v;
   return sizeof(int8_t);
 }
 
+static FORCE_INLINE int32_t tPutU16(uint8_t* p, uint16_t v) {
+  if (p) ((uint16_t*)p)[0] = v;
+  return sizeof(uint16_t);
+}
+
+static FORCE_INLINE int32_t tPutI16(uint8_t* p, int16_t v) {
+  if (p) ((int16_t*)p)[0] = v;
+  return sizeof(int16_t);
+}
+
+static FORCE_INLINE int32_t tPutU32(uint8_t* p, uint32_t v) {
+  if (p) ((uint32_t*)p)[0] = v;
+  return sizeof(uint32_t);
+}
+
+static FORCE_INLINE int32_t tPutI32(uint8_t* p, int32_t v) {
+  if (p) ((int32_t*)p)[0] = v;
+  return sizeof(int32_t);
+}
+
+static FORCE_INLINE int32_t tPutU64(uint8_t* p, uint64_t v) {
+  if (p) ((uint64_t*)p)[0] = v;
+  return sizeof(uint64_t);
+}
+
 static FORCE_INLINE int32_t tPutI64(uint8_t* p, int64_t v) {
   if (p) ((int64_t*)p)[0] = v;
   return sizeof(int64_t);
 }
 
-static FORCE_INLINE int32_t tPutU16v(uint8_t* p, uint16_t v) { tPutV(p, v) }
+static FORCE_INLINE int32_t tPutFloat(uint8_t* p, float f) {
+  union {
+    uint32_t ui;
+    float    f;
+  } v;
+  v.f = f;
+
+  return tPutU32(p, v.ui);
+}
+
+static FORCE_INLINE int32_t tPutDouble(uint8_t* p, double d) {
+  union {
+    uint64_t ui;
+    double   d;
+  } v;
+  v.d = d;
+
+  return tPutU64(p, v.ui);
+}
+
+static FORCE_INLINE int32_t tPutU16v(uint8_t* p, uint16_t v) { tPutV(p, v); }
 
 static FORCE_INLINE int32_t tPutI16v(uint8_t* p, int16_t v) { return tPutU16v(p, ZIGZAGE(int16_t, v)); }
 
-static FORCE_INLINE int32_t tPutU32v(uint8_t* p, uint32_t v) { tPutV(p, v) }
+static FORCE_INLINE int32_t tPutU32v(uint8_t* p, uint32_t v) { tPutV(p, v); }
 
 static FORCE_INLINE int32_t tPutI32v(uint8_t* p, int32_t v) { return tPutU32v(p, ZIGZAGE(int32_t, v)); }
 
+static FORCE_INLINE int32_t tPutU64v(uint8_t* p, uint64_t v) { tPutV(p, v); }
+
+static FORCE_INLINE int32_t tPutI64v(uint8_t* p, int64_t v) { return tPutU64v(p, ZIGZAGE(int64_t, v)); }
+
+// GET
+static FORCE_INLINE int32_t tGetU8(uint8_t* p, uint8_t* v) {
+  if (v) *v = ((uint8_t*)p)[0];
+  return sizeof(uint8_t);
+}
+
 static FORCE_INLINE int32_t tGetI8(uint8_t* p, int8_t* v) {
   if (v) *v = ((int8_t*)p)[0];
   return sizeof(int8_t);
 }
 
+static FORCE_INLINE int32_t tGetU16(uint8_t* p, uint16_t* v) {
+  if (v) *v = ((uint16_t*)p)[0];
+  return sizeof(uint16_t);
+}
+
+static FORCE_INLINE int32_t tGetI16(uint8_t* p, int16_t* v) {
+  if (v) *v = ((int16_t*)p)[0];
+  return sizeof(int16_t);
+}
+
+static FORCE_INLINE int32_t tGetU32(uint8_t* p, uint32_t* v) {
+  if (v) *v = ((uint32_t*)p)[0];
+  return sizeof(uint32_t);
+}
+
+static FORCE_INLINE int32_t tGetI32(uint8_t* p, int32_t* v) {
+  if (v) *v = ((int32_t*)p)[0];
+  return sizeof(int32_t);
+}
+
+static FORCE_INLINE int32_t tGetU64(uint8_t* p, uint64_t* v) {
+  if (v) *v = ((uint64_t*)p)[0];
+  return sizeof(uint64_t);
+}
+
 static FORCE_INLINE int32_t tGetI64(uint8_t* p, int64_t* v) {
   if (v) *v = ((int64_t*)p)[0];
   return sizeof(int64_t);
 }
 
-static FORCE_INLINE int32_t tGetU16v(uint8_t* p, uint16_t* v) { tGetV(p, v) }
+static FORCE_INLINE int32_t tGetU16v(uint8_t* p, uint16_t* v) { tGetV(p, v); }
 
 static FORCE_INLINE int32_t tGetI16v(uint8_t* p, int16_t* v) {
   int32_t  n;
@@ -530,7 +619,7 @@ static FORCE_INLINE int32_t tGetI16v(uint8_t* p, int16_t* v) {
   return n;
 }
 
-static FORCE_INLINE int32_t tGetU32v(uint8_t* p, uint32_t* v) { tGetV(p, v) }
+static FORCE_INLINE int32_t tGetU32v(uint8_t* p, uint32_t* v) { tGetV(p, v); }
 
 static FORCE_INLINE int32_t tGetI32v(uint8_t* p, int32_t* v) {
   int32_t  n;
@@ -542,6 +631,46 @@ static FORCE_INLINE int32_t tGetI32v(uint8_t* p, int32_t* v) {
   return n;
 }
 
+static FORCE_INLINE int32_t tGetU64v(uint8_t* p, uint64_t* v) { tGetV(p, v); }
+
+static FORCE_INLINE int32_t tGetI64v(uint8_t* p, int64_t* v) {
+  int32_t  n;
+  uint64_t tv;
+
+  n = tGetU64v(p, &tv);
+  if (v) *v = ZIGZAGD(int64_t, tv);
+
+  return n;
+}
+
+static FORCE_INLINE int32_t tGetFloat(uint8_t* p, float* f) {
+  int32_t n = 0;
+
+  union {
+    uint32_t ui;
+    float    f;
+  } v;
+
+  n = tGetU32(p, &v.ui);
+
+  *f = v.f;
+  return n;
+}
+
+static FORCE_INLINE int32_t tGetDouble(uint8_t* p, double* d) {
+  int32_t n = 0;
+
+  union {
+    uint64_t ui;
+    double   d;
+  } v;
+
+  n = tGetU64(p, &v.ui);
+
+  *d = v.d;
+  return n;
+}
+
 // =====================
 static FORCE_INLINE int32_t tPutBinary(uint8_t* p, uint8_t* pData, uint32_t nData) {
   int n = 0;

source/common/src/tdataformat.c

@@ -21,15 +21,10 @@
 
 static int32_t tGetTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson);
 
-typedef struct SKVIdx {
-  int32_t cid;
-  int32_t offset;
-} SKVIdx;
-
 #pragma pack(push, 1)
 typedef struct {
   int16_t nCols;
-  SKVIdx  idx[];
+  uint8_t idx[];
 } STSKVRow;
 #pragma pack(pop)
 
@@ -43,171 +38,551 @@ typedef struct {
 
 static FORCE_INLINE int tSKVIdxCmprFn(const void *p1, const void *p2);
 
-// STSRow2
-int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow) {
+// SValue
+static FORCE_INLINE int32_t tPutValue(uint8_t *p, SValue *pValue, int8_t type) {
   int32_t n = 0;
 
-  n += tPutI64(p ? p + n : p, pRow->ts);
-  n += tPutI8(p ? p + n : p, pRow->flags);
-  n += tPutI32v(p ? p + n : p, pRow->sver);
+  if (IS_VAR_DATA_TYPE(type)) {
+    n += tPutBinary(p ? p + n : p, pValue->pData, pValue->nData);
+  } else {
+    switch (type) {
+      case TSDB_DATA_TYPE_BOOL:
+        n += tPutI8(p ? p + n : p, pValue->i8 ? 1 : 0);
+        break;
+      case TSDB_DATA_TYPE_TINYINT:
+        n += tPutI8(p ? p + n : p, pValue->i8);
+        break;
+      case TSDB_DATA_TYPE_SMALLINT:
+        n += tPutI16(p ? p + n : p, pValue->i16);
+        break;
+      case TSDB_DATA_TYPE_INT:
+        n += tPutI32(p ? p + n : p, pValue->i32);
+        break;
+      case TSDB_DATA_TYPE_BIGINT:
+        n += tPutI64(p ? p + n : p, pValue->i64);
+        break;
+      case TSDB_DATA_TYPE_FLOAT:
+        n += tPutFloat(p ? p + n : p, pValue->f);
+        break;
+      case TSDB_DATA_TYPE_DOUBLE:
+        n += tPutDouble(p ? p + n : p, pValue->d);
+        break;
+      case TSDB_DATA_TYPE_TIMESTAMP:
+        n += tPutI64(p ? p + n : p, pValue->ts);
+        break;
+      case TSDB_DATA_TYPE_UTINYINT:
+        n += tPutU8(p ? p + n : p, pValue->u8);
+        break;
+      case TSDB_DATA_TYPE_USMALLINT:
+        n += tPutU16(p ? p + n : p, pValue->u16);
+        break;
+      case TSDB_DATA_TYPE_UINT:
+        n += tPutU32(p ? p + n : p, pValue->u32);
+        break;
+      case TSDB_DATA_TYPE_UBIGINT:
+        n += tPutU64(p ? p + n : p, pValue->u64);
+        break;
+      default:
+        ASSERT(0);
+    }
+  }
 
-  ASSERT(pRow->flags & 0xf);
+  return n;
+}
 
-  switch (pRow->flags & 0xf) {
-    case TSROW_HAS_NONE:
-    case TSROW_HAS_NULL:
-      break;
-    default:
-      n += tPutBinary(p ? p + n : p, pRow->pData, pRow->nData);
-      break;
+static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type) {
+  int32_t n = 0;
+
+  if (IS_VAR_DATA_TYPE(type)) {
+    n += tGetBinary(p, &pValue->pData, pValue ? &pValue->nData : NULL);
+  } else {
+    switch (type) {
+      case TSDB_DATA_TYPE_BOOL:
+        n += tGetI8(p, &pValue->i8);
+        break;
+      case TSDB_DATA_TYPE_TINYINT:
+        n += tGetI8(p, &pValue->i8);
+        break;
+      case TSDB_DATA_TYPE_SMALLINT:
+        n += tGetI16(p, &pValue->i16);
+        break;
+      case TSDB_DATA_TYPE_INT:
+        n += tGetI32(p, &pValue->i32);
+        break;
+      case TSDB_DATA_TYPE_BIGINT:
+        n += tGetI64(p, &pValue->i64);
+        break;
+      case TSDB_DATA_TYPE_FLOAT:
+        n += tGetFloat(p, &pValue->f);
+        break;
+      case TSDB_DATA_TYPE_DOUBLE:
+        n += tGetDouble(p, &pValue->d);
+        break;
+      case TSDB_DATA_TYPE_TIMESTAMP:
+        n += tGetI64(p, &pValue->ts);
+        break;
+      case TSDB_DATA_TYPE_UTINYINT:
+        n += tGetU8(p, &pValue->u8);
+        break;
+      case TSDB_DATA_TYPE_USMALLINT:
+        n += tGetU16(p, &pValue->u16);
+        break;
+      case TSDB_DATA_TYPE_UINT:
+        n += tGetU32(p, &pValue->u32);
+        break;
+      case TSDB_DATA_TYPE_UBIGINT:
+        n += tGetU64(p, &pValue->u64);
+        break;
+      default:
+        ASSERT(0);
+    }
   }
 
   return n;
 }
 
-int32_t tGetTSRow(uint8_t *p, STSRow2 *pRow) {
-  int32_t n = 0;
-  uint8_t flags;
+// STSRow2 ========================================================================
+static void tTupleTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 *pRow) {
+  int32_t   nColVal = taosArrayGetSize(pArray);
+  STColumn *pTColumn;
+  SColVal  *pColVal;
+
+  ASSERT(nColVal > 0);
+
+  pRow->sver = pTSchema->version;
 
-  n += tGetI64(p + n, pRow ? &pRow->ts : NULL);
-  n += tGetI8(p + n, pRow ? &pRow->flags : &flags);
-  n += tGetI32v(p + n, pRow ? &pRow->sver : NULL);
+  // ts
+  pTColumn = &pTSchema->columns[0];
+  pColVal = (SColVal *)taosArrayGet(pArray, 0);
 
-  if (pRow) flags = pRow->flags;
+  ASSERT(pTColumn->colId == 0 && pColVal->cid == 0);
+  ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP);
+
+  pRow->ts = pColVal->value.ts;
+
+  // other fields
+  int32_t  iColVal = 1;
+  int32_t  bidx;
+  uint32_t nv = 0;
+  uint8_t *pb = NULL;
+  uint8_t *pf = NULL;
+  uint8_t *pv = NULL;
+  uint8_t  flags = 0;
+  for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) {
+    bidx = iColumn - 1;
+    pTColumn = &pTSchema->columns[iColumn];
+
+    if (iColVal < nColVal) {
+      pColVal = (SColVal *)taosArrayGet(pArray, iColVal);
+    } else {
+      pColVal = NULL;
+    }
+
+    if (pColVal) {
+      if (pColVal->cid == pTColumn->colId) {
+        iColVal++;
+        if (pColVal->isNone) {
+          goto _set_none;
+        } else if (pColVal->isNull) {
+          goto _set_null;
+        } else {
+          goto _set_value;
+        }
+      } else if (pColVal->cid > pTColumn->colId) {
+        goto _set_none;
+      } else {
+        ASSERT(0);
+      }
+    } else {
+      goto _set_none;
+    }
+
+  _set_none:
+    flags |= TSROW_HAS_NONE;
+    // SET_BIT2(pb, bidx, 0); (todo)
+    continue;
+
+  _set_null:
+    flags != TSROW_HAS_NULL;
+    // SET_BIT2(pb, bidx, 1); (todo)
+    continue;
+
+  _set_value:
+    flags != TSROW_HAS_VAL;
+    // SET_BIT2(pb, bidx, 2); (todo)
+    if (IS_VAR_DATA_TYPE(pTColumn->type)) {
+      // nv += tPutColVal(pv ? pv + nv : pv, pColVal, pTColumn->type, 1);
+    } else {
+      // tPutColVal(pf ? pf + pTColumn->offset : pf, pColVal, pTColumn->type, 1);
+    }
+    continue;
+  }
+
+  ASSERT(flags);
   switch (flags & 0xf) {
     case TSROW_HAS_NONE:
     case TSROW_HAS_NULL:
+      pRow->nData = 0;
+      break;
+    case TSROW_HAS_VAL:
+      pRow->nData = pTSchema->flen + nv;
+      break;
+    case TSROW_HAS_NULL | TSROW_HAS_NONE:
+      pRow->nData = BIT1_SIZE(pTSchema->numOfCols - 1);
+      break;
+    case TSROW_HAS_VAL | TSROW_HAS_NONE:
+    case TSROW_HAS_VAL | TSROW_HAS_NULL:
+      pRow->nData = BIT1_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + nv;
+      break;
+    case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
+      pRow->nData = BIT2_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + nv;
       break;
     default:
-      n += tGetBinary(p + n, pRow ? &pRow->pData : NULL, pRow ? &pRow->nData : NULL);
       break;
   }
+}
 
-  return n;
+static void tMapTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 *pRow) {
+  int32_t   nColVal = taosArrayGetSize(pArray);
+  STColumn *pTColumn;
+  SColVal  *pColVal;
+
+  ASSERT(nColVal > 0);
+
+  pRow->sver = pTSchema->version;
+
+  // ts
+  pTColumn = &pTSchema->columns[0];
+  pColVal = (SColVal *)taosArrayGet(pArray, 0);
+
+  ASSERT(pTColumn->colId == 0 && pColVal->cid == 0);
+  ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP);
+
+  pRow->ts = pColVal->value.ts;
+
+  // other fields
+  int32_t  iColVal = 1;
+  uint32_t nv = 0;
+  uint8_t *pv = NULL;
+  uint8_t *pidx = NULL;
+  uint8_t  flags = 0;
+  int16_t  nCol = 0;
+  for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) {
+    pTColumn = &pTSchema->columns[iColumn];
+
+    if (iColVal < nColVal) {
+      pColVal = (SColVal *)taosArrayGet(pArray, iColVal);
+    } else {
+      pColVal = NULL;
+    }
+
+    if (pColVal) {
+      if (pColVal->cid == pTColumn->colId) {
+        iColVal++;
+        if (pColVal->isNone) {
+          goto _set_none;
+        } else if (pColVal->isNull) {
+          goto _set_null;
+        } else {
+          goto _set_value;
+        }
+      } else if (pColVal->cid > pTColumn->colId) {
+        goto _set_none;
+      } else {
+        ASSERT(0);
+      }
+    } else {
+      goto _set_none;
+    }
+
+  _set_none:
+    flags |= TSROW_HAS_NONE;
+    continue;
+
+  _set_null:
+    flags != TSROW_HAS_NULL;
+    pidx[nCol++] = nv;
+    // nv += tPutColVal(pv ? pv + nv : pv, pColVal, pTColumn->type, 0);
+    continue;
+
+  _set_value:
+    flags != TSROW_HAS_VAL;
+    pidx[nCol++] = nv;
+    // nv += tPutColVal(pv ? pv + nv : pv, pColVal, pTColumn->type, 0);
+    continue;
+  }
+
+  if (nv <= UINT8_MAX) {
+    // small
+  } else if (nv <= UINT16_MAX) {
+    // mid
+  } else {
+    // large
+  }
+}
+
+// try-decide-build
+int32_t tTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow) {
+  int32_t code = 0;
+  STSRow2 rowT = {0};
+  STSRow2 rowM = {0};
+
+  // try
+  tTupleTSRowNew(pArray, pTSchema, &rowT);
+  tMapTSRowNew(pArray, pTSchema, &rowM);
+
+  // decide & build
+  if (rowT.nData <= rowM.nData) {
+    tTupleTSRowNew(pArray, pTSchema, &rowT);
+  } else {
+    tMapTSRowNew(pArray, pTSchema, &rowM);
+  }
+
+  return code;
 }
 
-int32_t tTSRowDup(const STSRow2 *pRow, STSRow2 **ppRow) {
-  (*ppRow) = taosMemoryMalloc(sizeof(*pRow) + pRow->nData);
+int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow) {
+  int32_t code = 0;
+
+  (*ppRow) = (STSRow2 *)taosMemoryMalloc(sizeof(**ppRow));
   if (*ppRow == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return -1;
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _exit;
   }
+  **ppRow = *pRow;
+  (*ppRow)->pData = NULL;
 
-  (*ppRow)->ts = pRow->ts;
-  (*ppRow)->flags = pRow->flags;
-  (*ppRow)->sver = pRow->sver;
-  (*ppRow)->nData = pRow->nData;
   if (pRow->nData) {
-    (*ppRow)->pData = (uint8_t *)(&(*ppRow)[1]);
+    (*ppRow)->pData = taosMemoryMalloc(pRow->nData);
+    if ((*ppRow)->pData == NULL) {
+      taosMemoryFree(*ppRow);
+      code = TSDB_CODE_OUT_OF_MEMORY;
+      goto _exit;
+    }
     memcpy((*ppRow)->pData, pRow->pData, pRow->nData);
-  } else {
-    (*ppRow)->pData = NULL;
   }
 
-  return 0;
+_exit:
+  return code;
 }
 
 void tTSRowFree(STSRow2 *pRow) {
-  if (pRow) taosMemoryFree(pRow);
+  if (pRow) {
+    if (pRow->pData) taosMemoryFree(pRow->pData);
+    taosMemoryFree(pRow);
+  }
 }
 
-int32_t tTSRowGet(const STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
-  uint32_t  n;
-  uint8_t  *p;
-  uint8_t   v;
-  int32_t   bidx = iCol - 1;
+void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
+  uint8_t   isTuple = (pRow->flags & 0xf0 == 0) ? 1 : 0;
   STColumn *pTColumn = &pTSchema->columns[iCol];
-  STSKVRow *pTSKVRow;
-  SKVIdx   *pKVIdx;
+  uint8_t   flags = pRow->flags & (uint8_t)0xf;
+  SValue    value;
 
-  ASSERT(iCol != 0);
-  ASSERT(pTColumn->colId != 0);
+  ASSERT(iCol < pTSchema->numOfCols);
+  ASSERT(flags);
+  ASSERT(pRow->sver == pTSchema->version);
 
-  ASSERT((pRow->flags & 0xf) != 0);
-  switch (pRow->flags & 0xf) {
-    case TSROW_HAS_NONE:
-      *pColVal = ColValNONE;
-      return 0;
-    case TSROW_HAS_NULL:
-      *pColVal = ColValNULL;
-      return 0;
+  if (iCol == 0) {
+    value.ts = pRow->ts;
+    goto _return_value;
   }
 
-  if (TSROW_IS_KV_ROW(pRow)) {
-    ASSERT((pRow->flags & 0xf) != TSROW_HAS_VAL);
+  if (flags == TSROW_HAS_NONE) {
+    goto _return_none;
+  } else if (flags == TSROW_HAS_NONE) {
+    goto _return_null;
+  }
 
-    pTSKVRow = (STSKVRow *)pRow->pData;
-    pKVIdx =
-        bsearch(&((SKVIdx){.cid = pTColumn->colId}), pTSKVRow->idx, pTSKVRow->nCols, sizeof(SKVIdx), tSKVIdxCmprFn);
-    if (pKVIdx == NULL) {
-      *pColVal = ColValNONE;
-    } else if (pKVIdx->offset < 0) {
-      *pColVal = ColValNULL;
-    } else {
-      p = pRow->pData + sizeof(STSKVRow) + sizeof(SKVIdx) * pTSKVRow->nCols + pKVIdx->offset;
-      pColVal->type = COL_VAL_DATA;
-      tGetBinary(p, &pColVal->pData, &pColVal->nData);
-    }
-  } else {
-    // get bitmap
-    p = pRow->pData;
-    switch (pRow->flags & 0xf) {
+  ASSERT(pRow->nData && pRow->pData);
+
+  if (isTuple) {
+    uint8_t *pb = pRow->pData;
+    uint8_t *pf = NULL;
+    uint8_t *pv = NULL;
+    uint8_t *p;
+    uint8_t  b;
+
+    // bit
+    switch (flags) {
+      case TSROW_HAS_VAL:
+        pf = pb;
+        break;
       case TSROW_HAS_NULL | TSROW_HAS_NONE:
-        v = GET_BIT1(p, bidx);
-        if (v == 0) {
-          *pColVal = ColValNONE;
+        b = GET_BIT1(pb, iCol - 1);
+        if (b == 0) {
+          goto _return_none;
         } else {
-          *pColVal = ColValNULL;
+          goto _return_null;
         }
-        return 0;
       case TSROW_HAS_VAL | TSROW_HAS_NONE:
-        v = GET_BIT1(p, bidx);
-        if (v == 1) {
-          p = p + BIT1_SIZE(pTSchema->numOfCols - 1);
-          break;
+        b = GET_BIT1(pb, iCol - 1);
+        if (b == 0) {
+          goto _return_none;
         } else {
-          *pColVal = ColValNONE;
-          return 0;
+          pf = pb + BIT1_SIZE(pTSchema->numOfCols - 1);
+          break;
         }
       case TSROW_HAS_VAL | TSROW_HAS_NULL:
-        v = GET_BIT1(p, bidx);
-        if (v == 1) {
-          p = p + BIT1_SIZE(pTSchema->numOfCols - 1);
-          break;
+        b = GET_BIT1(pb, iCol - 1);
+        if (b == 0) {
+          goto _return_null;
         } else {
-          *pColVal = ColValNULL;
-          return 0;
+          pf = pb + BIT1_SIZE(pTSchema->numOfCols - 1);
+          break;
         }
       case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
-        v = GET_BIT2(p, bidx);
-        if (v == 0) {
-          *pColVal = ColValNONE;
-          return 0;
-        } else if (v == 1) {
-          *pColVal = ColValNULL;
-          return 0;
-        } else if (v == 2) {
-          p = p + BIT2_SIZE(pTSchema->numOfCols - 1);
-          break;
+        b = GET_BIT2(pb, iCol - 1);
+        if (b == 0) {
+          goto _return_none;
+        } else if (b == 1) {
+          goto _return_null;
         } else {
-          ASSERT(0);
+          pf = pb + BIT2_SIZE(pTSchema->numOfCols - 1);
+          break;
         }
       default:
-        break;
+        ASSERT(0);
     }
 
-    // get real value
-    p = p + pTColumn->offset;
-    pColVal->type = COL_VAL_DATA;
+    ASSERT(pf);
+
+    p = pf + pTColumn->offset;
     if (IS_VAR_DATA_TYPE(pTColumn->type)) {
-      tGetBinary(p + pTSchema->flen + *(int32_t *)p, &pColVal->pData, &pColVal->nData);
+      pv = pf + pTSchema->flen;
+      p = pv + *(VarDataOffsetT *)p;
+    }
+    tGetValue(p, &value, pTColumn->type);
+    goto _return_value;
+  } else {
+    STSKVRow *pRowK = (STSKVRow *)pRow->pData;
+    int16_t   lidx = 0;
+    int16_t   ridx = pRowK->nCols - 1;
+    uint8_t  *p;
+    int16_t   midx;
+    uint32_t  n;
+    int16_t   cid;
+
+    ASSERT(pRowK->nCols > 0);
+
+    if (pRow->flags & TSROW_KV_SMALL) {
+      p = pRow->pData + sizeof(STSKVRow) + sizeof(uint8_t) * pRowK->nCols;
+    } else if (pRow->flags & TSROW_KV_MID) {
+      p = pRow->pData + sizeof(STSKVRow) + sizeof(uint16_t) * pRowK->nCols;
+    } else if (pRow->flags & TSROW_KV_BIG) {
+      p = pRow->pData + sizeof(STSKVRow) + sizeof(uint32_t) * pRowK->nCols;
     } else {
-      pColVal->pData = p;
-      pColVal->nData = pTColumn->bytes;
+      ASSERT(0);
+    }
+    while (lidx <= ridx) {
+      midx = (lidx + ridx) / 2;
+
+      if (pRow->flags & TSROW_KV_SMALL) {
+        n = ((uint8_t *)pRowK->idx)[midx];
+      } else if (pRow->flags & TSROW_KV_MID) {
+        n = ((uint16_t *)pRowK->idx)[midx];
+      } else {
+        n = ((uint32_t *)pRowK->idx)[midx];
+      }
+
+      n += tGetI16v(p + n, &cid);
+
+      if (TABS(cid) == pTColumn->colId) {
+        if (cid < 0) {
+          goto _return_null;
+        } else {
+          n += tGetValue(p + n, &value, pTColumn->type);
+          goto _return_value;
+        }
+
+        return;
+      } else if (TABS(cid) > pTColumn->colId) {
+        ridx = midx - 1;
+      } else {
+        lidx = midx + 1;
+      }
     }
+
+    // not found, return NONE
+    goto _return_none;
   }
 
-  return 0;
+_return_none:
+  *pColVal = COL_VAL_NONE(pTColumn->colId);
+  return;
+
+_return_null:
+  *pColVal = COL_VAL_NULL(pTColumn->colId);
+  return;
+
+_return_value:
+  *pColVal = COL_VAL_VALUE(pTColumn->colId, value);
+  return;
+}
+
+int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray) {
+  int32_t code = 0;
+  SColVal cv;
+
+  (*ppArray) = taosArrayInit(pTSchema->numOfCols, sizeof(SColVal));
+  if (*ppArray == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _exit;
+  }
+
+  for (int32_t iColumn = 0; iColumn < pTSchema->numOfCols; iColumn++) {
+    tTSRowGet(pRow, pTSchema, iColumn, &cv);
+    taosArrayPush(*ppArray, &cv);
+  }
+
+_exit:
+  return code;
+}
+
+int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow) {
+  int32_t n = 0;
+
+  n += tPutI64(p ? p + n : p, pRow->ts);
+  n += tPutI8(p ? p + n : p, pRow->flags);
+  n += tPutI32v(p ? p + n : p, pRow->sver);
+
+  ASSERT(pRow->flags & 0xf);
+
+  switch (pRow->flags & 0xf) {
+    case TSROW_HAS_NONE:
+    case TSROW_HAS_NULL:
+      ASSERT(pRow->nData == 0);
+      ASSERT(pRow->pData == NULL);
+      break;
+    default:
+      ASSERT(pRow->nData && pRow->pData);
+      n += tPutBinary(p ? p + n : p, pRow->pData, pRow->nData);
+      break;
+  }
+
+  return n;
+}
+
+int32_t tGetTSRow(uint8_t *p, STSRow2 *pRow) {
+  int32_t n = 0;
+
+  n += tGetI64(p + n, &pRow->ts);
+  n += tGetI8(p + n, &pRow->flags);
+  n += tGetI32v(p + n, &pRow->sver);
+
+  ASSERT(pRow->flags);
+  switch (pRow->flags & 0xf) {
+    case TSROW_HAS_NONE:
+    case TSROW_HAS_NULL:
+      pRow->nData = 0;
+      pRow->pData = NULL;
+      break;
+    default:
+      n += tGetBinary(p + n, &pRow->pData, &pRow->nData);
+      break;
+  }
+
+  return n;
 }
 
 // STSchema
@@ -251,6 +626,7 @@ void tTSchemaDestroy(STSchema *pTSchema) {
 }
 
 // STSRowBuilder
+#if 0
 int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, int32_t nCols, SSchema *pSchema) {
   if (tTSchemaCreate(sver, pSchema, nCols, &pBuilder->pTSchema) < 0) return -1;
 
@@ -508,6 +884,7 @@ int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow) {
 
   return 0;
 }
+#endif
 
 static int tTagValCmprFn(const void *p1, const void *p2) {
   if (((STagVal *)p1)->cid < ((STagVal *)p2)->cid) {
@@ -622,9 +999,9 @@ void debugPrintSTag(STag *pTag, const char *tag, int32_t ln) {
     }
     printf("%s:%d loop[%d-%d] offset=%d\n", __func__, __LINE__, (int32_t)pTag->nTag, (int32_t)n, (int32_t)offset);
     tGetTagVal(p + offset, &tagVal, isJson);
-    if(IS_VAR_DATA_TYPE(tagVal.type)){
+    if (IS_VAR_DATA_TYPE(tagVal.type)) {
       debugPrintTagVal(tagVal.type, tagVal.pData, tagVal.nData, __func__, __LINE__);
-    }else{
+    } else {
       debugPrintTagVal(tagVal.type, &tagVal.i64, tDataTypes[tagVal.type].bytes, __func__, __LINE__);
     }
   }
@@ -650,7 +1027,7 @@ static int32_t tPutTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson) {
   } else {
     p = p ? p + n : p;
     n += tDataTypes[pTagVal->type].bytes;
-    if(p) memcpy(p, &(pTagVal->i64), tDataTypes[pTagVal->type].bytes);
+    if (p) memcpy(p, &(pTagVal->i64), tDataTypes[pTagVal->type].bytes);
   }
 
   return n;
@@ -750,21 +1127,21 @@ void tTagFree(STag *pTag) {
   if (pTag) taosMemoryFree(pTag);
 }
 
-char *tTagValToData(const STagVal *value, bool isJson){
-  if(!value) return NULL;
-  char *data = NULL;
+char *tTagValToData(const STagVal *value, bool isJson) {
+  if (!value) return NULL;
+  char  *data = NULL;
   int8_t typeBytes = 0;
   if (isJson) {
     typeBytes = CHAR_BYTES;
   }
-  if(IS_VAR_DATA_TYPE(value->type)){
+  if (IS_VAR_DATA_TYPE(value->type)) {
     data = taosMemoryCalloc(1, typeBytes + VARSTR_HEADER_SIZE + value->nData);
-    if(data == NULL) return NULL;
-    if(isJson) *data = value->type;
+    if (data == NULL) return NULL;
+    if (isJson) *data = value->type;
     varDataLen(data + typeBytes) = value->nData;
     memcpy(varDataVal(data + typeBytes), value->pData, value->nData);
-  }else{
-    data = ((char*)&(value->i64)) - typeBytes;  // json with type
+  } else {
+    data = ((char *)&(value->i64)) - typeBytes;  // json with type
   }
 
   return data;

source/common/src/tmsg.c

@@ -28,7 +28,7 @@
 #undef TD_MSG_SEG_CODE_
 #include "tmsgdef.h"
 
-int32_t tInitSubmitMsgIter(SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
+int32_t tInitSubmitMsgIter(const SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
   if (pMsg == NULL) {
     terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
     return -1;
@@ -165,7 +165,6 @@ int32_t tDecodeSQueryNodeLoad(SDecoder *pDecoder, SQueryNodeLoad *pLoad) {
   return 0;
 }
 
-
 int32_t taosEncodeSEpSet(void **buf, const SEpSet *pEp) {
   int32_t tlen = 0;
   tlen += taosEncodeFixedI8(buf, pEp->inUse);
@@ -2934,7 +2933,6 @@ int32_t tSerializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *pR
 
   if (tStartEncode(&encoder) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->vgId) < 0) return -1;
-  if (tEncodeI32(&encoder, pReq->dnodeId) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->db) < 0) return -1;
   if (tEncodeI64(&encoder, pReq->dbUid) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->vgVersion) < 0) return -1;
@@ -2957,6 +2955,7 @@ int32_t tSerializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *pR
   if (tEncodeI8(&encoder, pReq->compression) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->strict) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->cacheLastRow) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->standby) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->replica) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->selfIndex) < 0) return -1;
   for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
@@ -2991,7 +2990,6 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
 
   if (tStartDecode(&decoder) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->vgId) < 0) return -1;
-  if (tDecodeI32(&decoder, &pReq->dnodeId) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->db) < 0) return -1;
   if (tDecodeI64(&decoder, &pReq->dbUid) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->vgVersion) < 0) return -1;
@@ -3014,6 +3012,7 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
   if (tDecodeI8(&decoder, &pReq->compression) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->strict) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->cacheLastRow) < 0) return -1;
+  if (tDecodeI8(&decoder, &pReq->standby) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->replica) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->selfIndex) < 0) return -1;
   for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
@@ -3053,7 +3052,7 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
 int32_t tFreeSCreateVnodeReq(SCreateVnodeReq *pReq) {
   taosArrayDestroy(pReq->pRetensions);
   pReq->pRetensions = NULL;
-  if(pReq->isTsma) {
+  if (pReq->isTsma) {
     taosMemoryFreeClear(pReq->pTsma);
   }
   return 0;
@@ -3134,8 +3133,8 @@ int32_t tSerializeSAlterVnodeReq(void *buf, int32_t bufLen, SAlterVnodeReq *pReq
   if (tEncodeI8(&encoder, pReq->walLevel) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->strict) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->cacheLastRow) < 0) return -1;
-  if (tEncodeI8(&encoder, pReq->replica) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->selfIndex) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->replica) < 0) return -1;
   for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
     SReplica *pReplica = &pReq->replicas[i];
     if (tEncodeSReplica(&encoder, pReplica) < 0) return -1;
@@ -3165,8 +3164,8 @@ int32_t tDeserializeSAlterVnodeReq(void *buf, int32_t bufLen, SAlterVnodeReq *pR
   if (tDecodeI8(&decoder, &pReq->walLevel) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->strict) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->cacheLastRow) < 0) return -1;
-  if (tDecodeI8(&decoder, &pReq->replica) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->selfIndex) < 0) return -1;
+  if (tDecodeI8(&decoder, &pReq->replica) < 0) return -1;
   for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
     SReplica *pReplica = &pReq->replicas[i];
     if (tDecodeSReplica(&decoder, pReplica) < 0) return -1;
@@ -3654,6 +3653,7 @@ int32_t tEncodeTSma(SEncoder *pCoder, const STSma *pSma) {
   if (tEncodeI8(pCoder, pSma->intervalUnit) < 0) return -1;
   if (tEncodeI8(pCoder, pSma->slidingUnit) < 0) return -1;
   if (tEncodeI8(pCoder, pSma->timezoneInt) < 0) return -1;
+  if (tEncodeI32(pCoder, pSma->dstVgId) < 0) return -1;
   if (tEncodeCStr(pCoder, pSma->indexName) < 0) return -1;
   if (tEncodeI32(pCoder, pSma->exprLen) < 0) return -1;
   if (tEncodeI32(pCoder, pSma->tagsFilterLen) < 0) return -1;
@@ -3676,6 +3676,7 @@ int32_t tDecodeTSma(SDecoder *pCoder, STSma *pSma) {
   if (tDecodeI8(pCoder, &pSma->version) < 0) return -1;
   if (tDecodeI8(pCoder, &pSma->intervalUnit) < 0) return -1;
   if (tDecodeI8(pCoder, &pSma->slidingUnit) < 0) return -1;
+  if (tDecodeI32(pCoder, &pSma->dstVgId) < 0) return -1;
   if (tDecodeI8(pCoder, &pSma->timezoneInt) < 0) return -1;
   if (tDecodeCStrTo(pCoder, pSma->indexName) < 0) return -1;
   if (tDecodeI32(pCoder, &pSma->exprLen) < 0) return -1;

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

@@ -219,9 +219,9 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TASK, mmPutNodeMsgToQueryQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_HEARTBEAT, mmPutNodeMsgToQueryQueue, 1) == NULL) goto _OVER;
 
-  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_VNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_VNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT_VNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
 
   if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_TIMEOUT, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING, mmPutNodeMsgToSyncQueue, 1) == NULL) goto _OVER;

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

@@ -150,20 +150,26 @@ static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
   pCfg->tsdbCfg.minRows = pCreate->minRows;
   pCfg->tsdbCfg.maxRows = pCreate->maxRows;
   for (size_t i = 0; i < taosArrayGetSize(pCreate->pRetensions); ++i) {
-    memcpy(&pCfg->tsdbCfg.retentions[i], taosArrayGet(pCreate->pRetensions, i), sizeof(SRetention));
+    SRetention *pRetention = &pCfg->tsdbCfg.retentions[i];
+    memcpy(pRetention, taosArrayGet(pCreate->pRetensions, i), sizeof(SRetention));
+    if (i == 0) {
+      if ((pRetention->freq > 0 && pRetention->keep > 0)) pCfg->isRsma = 1;
+    }
   }
+
   pCfg->walCfg.vgId = pCreate->vgId;
   pCfg->hashBegin = pCreate->hashBegin;
   pCfg->hashEnd = pCreate->hashEnd;
   pCfg->hashMethod = pCreate->hashMethod;
 
+  pCfg->standby = pCfg->standby;
   pCfg->syncCfg.myIndex = pCreate->selfIndex;
   pCfg->syncCfg.replicaNum = pCreate->replica;
   memset(&pCfg->syncCfg.nodeInfo, 0, sizeof(pCfg->syncCfg.nodeInfo));
   for (int i = 0; i < pCreate->replica; ++i) {
-    pCfg->syncCfg.nodeInfo[i].nodePort = pCreate->replicas[i].port;
-    snprintf(pCfg->syncCfg.nodeInfo[i].nodeFqdn, sizeof(pCfg->syncCfg.nodeInfo[i].nodeFqdn), "%s",
-             pCreate->replicas[i].fqdn);
+    SNodeInfo *pNode = &pCfg->syncCfg.nodeInfo[i];
+    pNode->nodePort = pCreate->replicas[i].port;
+    tstrncpy(pNode->nodeFqdn, pCreate->replicas[i].fqdn, sizeof(pNode->nodeFqdn));
   }
 }
 
@@ -176,6 +182,8 @@ static void vmGenerateWrapperCfg(SVnodeMgmt *pMgmt, SCreateVnodeReq *pCreate, SW
 
 int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SCreateVnodeReq createReq = {0};
+  SVnodeCfg       vnodeCfg = {0};
+  SWrapperCfg     wrapperCfg = {0};
   int32_t         code = -1;
   char            path[TSDB_FILENAME_LEN] = {0};
 
@@ -184,12 +192,9 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     return -1;
   }
 
-  dDebug("vgId:%d, create vnode req is received, tsma:%d", createReq.vgId, createReq.isTsma);
-
-  SVnodeCfg vnodeCfg = {0};
+  dDebug("vgId:%d, create vnode req is received, tsma:%d standby:%d", createReq.vgId, createReq.isTsma,
+         createReq.standby);
   vmGenerateVnodeCfg(&createReq, &vnodeCfg);
-
-  SWrapperCfg wrapperCfg = {0};
   vmGenerateWrapperCfg(pMgmt, &createReq, &wrapperCfg);
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, createReq.vgId);
@@ -218,9 +223,20 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   code = vmOpenVnode(pMgmt, &wrapperCfg, pImpl);
   if (code != 0) {
     dError("vgId:%d, failed to open vnode since %s", createReq.vgId, terrstr());
+    code = terrno;
     goto _OVER;
   }
 
+  if (createReq.isTsma) {
+    SMsgHead *smaMsg = createReq.pTsma;
+    uint32_t  contLen = (uint32_t)(htonl(smaMsg->contLen) - sizeof(SMsgHead));
+    if (vnodeProcessCreateTSma(pImpl, POINTER_SHIFT(smaMsg, sizeof(SMsgHead)), contLen) < 0) {
+      dError("vgId:%d, failed to create tsma since %s", createReq.vgId, terrstr());
+      code = terrno;
+      goto _OVER;
+    };
+  }
+
   code = vnodeStart(pImpl);
   if (code != 0) {
     dError("vgId:%d, failed to start sync since %s", createReq.vgId, terrstr());
@@ -228,7 +244,10 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   }
 
   code = vmWriteVnodeListToFile(pMgmt);
-  if (code != 0) goto _OVER;
+  if (code != 0) {
+    code = terrno;
+    goto _OVER;
+  }
 
 _OVER:
   if (code != 0) {
@@ -314,8 +333,9 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DISPATCH, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_RECOVER, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
 
-  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_VNODE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT_VNODE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
 

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

@@ -61,7 +61,7 @@ int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction);
 int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction);
 void    mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen);
 void    mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen);
-void    mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb);
+void    mndTransSetDbName(STrans *pTrans, const char *dbname);
 void    mndTransSetSerial(STrans *pTrans);
 
 int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);

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

@@ -36,7 +36,7 @@ SArray *mndBuildDnodesArray(SMnode *pMnode);
 int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray);
 int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray, SVnodeGid *del1, SVnodeGid *del2);
 
-void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
+void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen, bool standby);
 void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
 void *mndBuildAlterVnodeReq(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
 

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

@@ -261,7 +261,7 @@ void mndReleaseDb(SMnode *pMnode, SDbObj *pDb) {
   sdbRelease(pSdb, pDb);
 }
 
-static int32_t mndAddCreateVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid) {
+static int32_t mndAddCreateVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid, bool standby) {
   STransAction action = {0};
 
   SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
@@ -270,7 +270,7 @@ static int32_t mndAddCreateVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *p
   mndReleaseDnode(pMnode, pDnode);
 
   int32_t contLen = 0;
-  void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
+  void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen, standby);
   if (pReq == NULL) return -1;
 
   action.pCont = pReq;
@@ -286,7 +286,7 @@ static int32_t mndAddCreateVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *p
   return 0;
 }
 
-static int32_t mndAddAlterVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup) {
+static int32_t mndAddAlterVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, tmsg_t msgType) {
   STransAction action = {0};
   action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
 
@@ -296,7 +296,7 @@ static int32_t mndAddAlterVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pD
 
   action.pCont = pReq;
   action.contLen = contLen;
-  action.msgType = TDMT_VND_ALTER_VNODE;
+  action.msgType = msgType;
 
   if (mndTransAppendRedoAction(pTrans, &action) != 0) {
     taosMemoryFree(pReq);
@@ -388,7 +388,7 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
   }
 
   terrno = 0;
-  return TSDB_CODE_SUCCESS;
+  return terrno;
 }
 
 static void mndSetDefaultDbCfg(SDbCfg *pCfg) {
@@ -467,7 +467,7 @@ static int32_t mndSetCreateDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
 
     for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
       SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
-      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid) != 0) {
+      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, false) != 0) {
         return -1;
       }
     }
@@ -550,7 +550,7 @@ static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate,
 
   mDebug("trans:%d, used to create db:%s", pTrans->id, pCreate->db);
 
-  mndTransSetDbInfo(pTrans, &dbObj);
+  mndTransSetDbName(pTrans, dbObj.name);
   if (mndSetCreateDbRedoLogs(pMnode, pTrans, &dbObj, pVgroups) != 0) goto _OVER;
   if (mndSetCreateDbUndoLogs(pMnode, pTrans, &dbObj, pVgroups) != 0) goto _OVER;
   if (mndSetCreateDbCommitLogs(pMnode, pTrans, &dbObj, pVgroups) != 0) goto _OVER;
@@ -688,29 +688,37 @@ static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
 static int32_t mndSetAlterDbRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pOld, SDbObj *pNew) {
   SSdbRaw *pRedoRaw = mndDbActionEncode(pOld);
   if (pRedoRaw == NULL) return -1;
-  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
-  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY) != 0) return -1;
+  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
+    sdbFreeRaw(pRedoRaw);
+    return -1;
+  }
 
+  sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
   return 0;
 }
 
 static int32_t mndSetAlterDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pOld, SDbObj *pNew) {
   SSdbRaw *pCommitRaw = mndDbActionEncode(pNew);
   if (pCommitRaw == NULL) return -1;
-  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
-  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;
+  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+    sdbFreeRaw(pCommitRaw);
+    return -1;
+  }
 
+  sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
   return 0;
 }
 
 static int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SArray *pArray) {
   if (pVgroup->replica <= 0 || pVgroup->replica == pDb->cfg.replications) {
-    if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, pVgroup) != 0) {
+    if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, pVgroup, TDMT_VND_ALTER_CONFIG) != 0) {
       return -1;
     }
   } else {
     SVgObj newVgroup = {0};
     memcpy(&newVgroup, pVgroup, sizeof(SVgObj));
+    mndTransSetSerial(pTrans);
+
     if (newVgroup.replica < pDb->cfg.replications) {
       mInfo("db:%s, vgId:%d, will add 2 vnodes, vn:0 dnode:%d", pVgroup->dbName, pVgroup->vgId,
             pVgroup->vnodeGid[0].dnodeId);
@@ -720,9 +728,9 @@ static int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj
         return -1;
       }
       newVgroup.replica = pDb->cfg.replications;
-      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup) != 0) return -1;
-      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[1]) != 0) return -1;
-      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[2]) != 0) return -1;
+      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[1], true) != 0) return -1;
+      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[2], true) != 0) return -1;
+      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
     } else {
       mInfo("db:%s, vgId:%d, will remove 2 vnodes", pVgroup->dbName, pVgroup->vgId);
 
@@ -733,15 +741,18 @@ static int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj
         return -1;
       }
       newVgroup.replica = pDb->cfg.replications;
-      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup) != 0) return -1;
+      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
       if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del1, true) != 0) return -1;
       if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del2, true) != 0) return -1;
     }
 
     SSdbRaw *pVgRaw = mndVgroupActionEncode(&newVgroup);
     if (pVgRaw == NULL) return -1;
-    if (mndTransAppendCommitlog(pTrans, pVgRaw) != 0) return -1;
-    if (sdbSetRawStatus(pVgRaw, SDB_STATUS_READY) != 0) return -1;
+    if (mndTransAppendCommitlog(pTrans, pVgRaw) != 0) {
+      sdbFreeRaw(pVgRaw);
+      return -1;
+    }
+    sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
   }
 
   return 0;
@@ -774,18 +785,16 @@ static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
 }
 
 static int32_t mndAlterDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pOld, SDbObj *pNew) {
-  int32_t code = -1;
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq);
-  if (pTrans == NULL) goto _OVER;
-
+  if (pTrans == NULL) return -1;
   mDebug("trans:%d, used to alter db:%s", pTrans->id, pOld->name);
 
-  mndTransSetDbInfo(pTrans, pOld);
+  int32_t code = -1;
+  mndTransSetDbName(pTrans, pOld->name);
   if (mndSetAlterDbRedoLogs(pMnode, pTrans, pOld, pNew) != 0) goto _OVER;
   if (mndSetAlterDbCommitLogs(pMnode, pTrans, pOld, pNew) != 0) goto _OVER;
   if (mndSetAlterDbRedoActions(pMnode, pTrans, pOld, pNew) != 0) goto _OVER;
   if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
-
   code = 0;
 
 _OVER:
@@ -1040,7 +1049,7 @@ static int32_t mndDropDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
   if (pTrans == NULL) goto _OVER;
 
   mDebug("trans:%d, used to drop db:%s", pTrans->id, pDb->name);
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
 
   if (mndSetDropDbRedoLogs(pMnode, pTrans, pDb) != 0) goto _OVER;
   if (mndSetDropDbCommitLogs(pMnode, pTrans, pDb) != 0) goto _OVER;

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

@@ -426,7 +426,7 @@ static int32_t mndSetCreateSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans,
   pVgroup->pTsma = pSmaReq;
 
   int32_t contLen = 0;
-  void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
+  void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen, false);
   taosMemoryFreeClear(pSmaReq);
   if (pReq == NULL) return -1;
 
@@ -512,7 +512,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   if (pTrans == NULL) goto _OVER;
 
   mDebug("trans:%d, used to create sma:%s", pTrans->id, pCreate->name);
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
   mndTransSetSerial(pTrans);
 
   if (mndSetCreateSmaRedoLogs(pMnode, pTrans, &smaObj) != 0) goto _OVER;
@@ -757,7 +757,7 @@ static int32_t mndDropSma(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SSmaObj *p
   if (pTrans == NULL) goto _OVER;
 
   mDebug("trans:%d, used to drop sma:%s", pTrans->id, pSma->name);
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
 
   if (mndSetDropSmaRedoLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
   if (mndSetDropSmaVgroupRedoLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;

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

@@ -754,7 +754,7 @@ _OVER:
 }
 
 int32_t mndAddStbToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
   if (mndSetCreateStbRedoLogs(pMnode, pTrans, pDb, pStb) != 0) return -1;
   if (mndSetCreateStbUndoLogs(pMnode, pTrans, pDb, pStb) != 0) return -1;
   if (mndSetCreateStbCommitLogs(pMnode, pTrans, pDb, pStb) != 0) return -1;
@@ -1261,7 +1261,7 @@ static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *p
   if (pTrans == NULL) goto _OVER;
 
   mDebug("trans:%d, used to alter stb:%s", pTrans->id, pAlter->name);
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
 
   if (mndSetAlterStbRedoLogs(pMnode, pTrans, pDb, &stbObj) != 0) goto _OVER;
   if (mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, &stbObj) != 0) goto _OVER;
@@ -1407,7 +1407,7 @@ static int32_t mndDropStb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *p
   if (pTrans == NULL) goto _OVER;
 
   mDebug("trans:%d, used to drop stb:%s", pTrans->id, pStb->name);
-  mndTransSetDbInfo(pTrans, pDb);
+  mndTransSetDbName(pTrans, pDb->name);
 
   if (mndSetDropStbRedoLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
   if (mndSetDropStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER;

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

@@ -619,8 +619,8 @@ void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *
   pTrans->paramLen = paramLen;
 }
 
-void mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb) {
-  memcpy(pTrans->dbname, pDb->name, TSDB_DB_FNAME_LEN);
+void mndTransSetDbName(STrans *pTrans, const char *dbname) {
+  memcpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
 }
 
 void mndTransSetSerial(STrans *pTrans) { pTrans->exec = TRN_EXEC_SERIAL; }

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

@@ -51,9 +51,10 @@ int32_t mndInitVgroup(SMnode *pMnode) {
   };
 
   mndSetMsgHandle(pMnode, TDMT_DND_CREATE_VNODE_RSP, mndProcessCreateVnodeRsp);
-  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_VNODE_RSP, mndProcessAlterVnodeRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_REPLICA_RSP, mndProcessAlterVnodeRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_CONFIG_RSP, mndProcessAlterVnodeRsp);
   mndSetMsgHandle(pMnode, TDMT_DND_DROP_VNODE_RSP, mndProcessDropVnodeRsp);
-  mndSetMsgHandle(pMnode, TDMT_VND_COMPACT_VNODE_RSP, mndProcessCompactVnodeRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_COMPACT_RSP, mndProcessCompactVnodeRsp);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_VGROUP, mndRetrieveVgroups);
   mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_VGROUP, mndCancelGetNextVgroup);
@@ -188,10 +189,10 @@ void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup) {
   sdbRelease(pSdb, pVgroup);
 }
 
-void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen) {
+void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen,
+                             bool standby) {
   SCreateVnodeReq createReq = {0};
   createReq.vgId = pVgroup->vgId;
-  createReq.dnodeId = pDnode->id;
   memcpy(createReq.db, pDb->name, TSDB_DB_FNAME_LEN);
   createReq.dbUid = pDb->uid;
   createReq.vgVersion = pVgroup->version;
@@ -218,6 +219,7 @@ void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVg
   createReq.hashMethod = pDb->cfg.hashMethod;
   createReq.numOfRetensions = pDb->cfg.numOfRetensions;
   createReq.pRetensions = pDb->cfg.pRetensions;
+  createReq.standby = standby;
   createReq.isTsma = pVgroup->isTsma;
   createReq.pTsma = pVgroup->pTsma;
 
@@ -276,7 +278,6 @@ void *mndBuildAlterVnodeReq(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup, int32_
   alterReq.strict = pDb->cfg.strict;
   alterReq.cacheLastRow = pDb->cfg.cacheLastRow;
   alterReq.replica = pVgroup->replica;
-  alterReq.selfIndex = -1;
 
   for (int32_t v = 0; v < pVgroup->replica; ++v) {
     SReplica  *pReplica = &alterReq.replicas[v];
@@ -292,13 +293,6 @@ void *mndBuildAlterVnodeReq(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup, int32_
     mndReleaseDnode(pMnode, pVgidDnode);
   }
 
-#if 0
-  if (alterReq.selfIndex == -1) {
-    terrno = TSDB_CODE_MND_APP_ERROR;
-    return NULL;
-  }
-#endif
-
   int32_t contLen = tSerializeSAlterVnodeReq(NULL, 0, &alterReq);
   if (contLen < 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -510,7 +504,7 @@ _OVER:
   taosArrayDestroy(pArray);
   return code;
 }
-
+//--->
 int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
   taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
   for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
@@ -538,7 +532,7 @@ int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
 
     SVnodeGid *pVgid = &pVgroup->vnodeGid[maxPos];
     pVgid->dnodeId = pDnode->id;
-    pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
+    pVgid->role = TAOS_SYNC_STATE_ERROR;
     pDnode->numOfVnodes++;
 
     mInfo("db:%s, vgId:%d, vnode_index:%d dnode:%d is added", pVgroup->dbName, pVgroup->vgId, maxPos, pVgid->dnodeId);
@@ -549,16 +543,15 @@ int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
   terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
   return -1;
 }
-
+//--->
 int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray, SVnodeGid *del1, SVnodeGid *del2) {
-  int32_t removedNum = 0;
-
   taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
   for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
     SDnodeObj *pDnode = taosArrayGet(pArray, i);
     mDebug("dnode:%d, equivalent vnodes:%d", pDnode->id, pDnode->numOfVnodes);
   }
 
+  int32_t removedNum = 0;
   for (int32_t d = taosArrayGetSize(pArray) - 1; d >= 0; --d) {
     SDnodeObj *pDnode = taosArrayGet(pArray, d);
 
@@ -664,6 +657,7 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
   int32_t numOfRows = 0;
   SVgObj *pVgroup = NULL;
   int32_t cols = 0;
+  int64_t curMs = taosGetTimestampMs();
 
   SDbObj *pDb = NULL;
   if (strlen(pShow->db) > 0) {
@@ -703,12 +697,15 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
       if (i < pVgroup->replica) {
         colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->vnodeGid[i].dnodeId, false);
 
+        bool       online = false;
+        SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgroup->vnodeGid[i].dnodeId);
+        if (pDnode != NULL) {
+          online = mndIsDnodeOnline(pMnode, pDnode, curMs);
+          mndReleaseDnode(pMnode, pDnode);
+        }
+
         char        buf1[20] = {0};
-        SDnodeObj *pDnodeObj = mndAcquireDnode(pMnode, pVgroup->vnodeGid[i].dnodeId);
-        ASSERT(pDnodeObj != NULL);
-        bool isOffLine = !mndIsDnodeOnline(pMnode, pDnodeObj, taosGetTimestampMs());
-        const char *role = isOffLine ? "OFFLINE" : syncStr(pVgroup->vnodeGid[i].role);
-        
+        const char *role = online ? syncStr(pVgroup->vnodeGid[i].role) : "OFFLINE";
         STR_WITH_MAXSIZE_TO_VARSTR(buf1, role, pShow->pMeta->pSchemas[cols].bytes);
 
         pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

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

@@ -128,7 +128,7 @@ class MndTestTrans2 : public ::testing::Test {
     mndTransSetCb(pTrans, TRANS_START_FUNC_TEST, TRANS_STOP_FUNC_TEST, param, strlen(param) + 1);
 
     if (pDb != NULL) {
-      mndTransSetDbInfo(pTrans, pDb);
+      mndTransSetDbName(pTrans, pDb->name);
     }
 
     int32_t code = mndTransPrepare(pMnode, pTrans);
@@ -201,7 +201,7 @@ class MndTestTrans2 : public ::testing::Test {
     }
 
     if (pDb != NULL) {
-      mndTransSetDbInfo(pTrans, pDb);
+      mndTransSetDbName(pTrans, pDb->name);
     }
 
     int32_t code = mndTransPrepare(pMnode, pTrans);

source/dnode/vnode/CMakeLists.txt

@@ -35,7 +35,6 @@ target_sources(
     "src/sma/smaTimeRange.c"
 
     # tsdb
-    # "src/tsdb/tsdbTDBImpl.c"
     "src/tsdb/tsdbCommit.c"
     "src/tsdb/tsdbCommit2.c"
     "src/tsdb/tsdbFile.c"
@@ -45,17 +44,18 @@ target_sources(
     "src/tsdb/tsdbMemTable2.c"
     "src/tsdb/tsdbRead.c"
     "src/tsdb/tsdbReadImpl.c"
-    # "src/tsdb/tsdbSma.c"
     "src/tsdb/tsdbWrite.c"
     "src/tsdb/tsdbSnapshot.c"
 
     # tq
     "src/tq/tq.c"
     "src/tq/tqExec.c"
-    "src/tq/tqCommit.c"
+    "src/tq/tqMeta.c"
+    "src/tq/tqRead.c"
     "src/tq/tqOffset.c"
     "src/tq/tqPush.c"
-    "src/tq/tqRead.c"
+    "src/tq/tqSink.c"
+    "src/tq/tqCommit.c"
 )
 target_include_directories(
     vnode

source/dnode/vnode/inc/vnode.h

@@ -68,6 +68,7 @@ void    vnodeGetInfo(SVnode *pVnode, const char **dbname, int32_t *vgId);
 int32_t vnodeSnapshotReaderOpen(SVnode *pVnode, SVSnapshotReader **ppReader, int64_t sver, int64_t ever);
 int32_t vnodeSnapshotReaderClose(SVSnapshotReader *pReader);
 int32_t vnodeSnapshotRead(SVSnapshotReader *pReader, const void **ppData, uint32_t *nData);
+int32_t vnodeProcessCreateTSma(SVnode *pVnode, void *pCont, uint32_t contLen);
 
 // meta
 typedef struct SMeta       SMeta;  // todo: remove
@@ -172,7 +173,9 @@ struct SVnodeCfg {
   bool     isHeap;
   bool     isWeak;
   int8_t   isTsma;
+  int8_t   isRsma;
   int8_t   hashMethod;
+  int8_t   standby;
   STsdbCfg tsdbCfg;
   SWalCfg  walCfg;
   SSyncCfg syncCfg;

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

@@ -219,7 +219,7 @@ static int32_t  tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SDisk
 void *tdFreeRSmaInfo(SRSmaInfo *pInfo);
 
 int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg);
-int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version);
+int32_t tdUpdateExpiredWindowImpl(SSma *pSma, const SSubmitReq *pMsg, int64_t version);
 // TODO: This is the basic params, and should wrap the params to a queryHandle.
 int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult);
 
@@ -227,4 +227,4 @@ int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKEY query
 }
 #endif
 
-#endif /*_TD_VNODE_SMA_H_*/
+#endif /*_TD_VNODE_SMA_H_*/
\ No newline at end of file

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

@@ -66,33 +66,27 @@ struct STqReadHandle {
 // tqPush
 
 typedef struct {
-  int64_t        consumerId;
-  int32_t        epoch;
-  int32_t        skipLogNum;
-  int64_t        reqOffset;
-  SRpcHandleInfo info;
-  SRWLatch       lock;
-} STqPushHandle;
+  STaosQueue* queue;
+  STaosQall*  qall;
+  void*       qItem;
+} STqInputQ;
 
-#if 0
 typedef struct {
-  char    subKey[TSDB_SUBSCRIBE_KEY_LEN];
+  // msg info
   int64_t consumerId;
+  int64_t reqOffset;
+  int64_t processedVer;
   int32_t epoch;
-  int8_t  subType;
-  // int8_t        withTbName;
-  // int8_t        withSchema;
-  // int8_t        withTag;
-  char*         qmsg;
-  SHashObj*     pDropTbUid;
-  STqPushHandle pushHandle;
-  // SRWLatch        lock;
-  SWalReadHandle* pWalReader;
-  // task number should be the same with fetch thread
-  STqReadHandle* pExecReader[5];
-  qTaskInfo_t    task[5];
-} STqExec;
-#endif
+  int32_t skipLogNum;
+  // rpc info
+  int64_t        reqId;
+  SRpcHandleInfo rpcInfo;
+  // exec
+  int8_t    inputStatus;
+  int8_t    execStatus;
+  STqInputQ inputQ;
+  SRWLatch  lock;
+} STqPushHandle;
 
 // tqExec
 
@@ -154,20 +148,21 @@ typedef struct {
 
 static STqMgmt tqMgmt = {0};
 
-// init once
-int  tqInit();
-void tqCleanUp();
-
-// int32_t tEncodeSTqExec(SEncoder* pEncoder, const STqExec* pExec);
-// int32_t tDecodeSTqExec(SDecoder* pDecoder, STqExec* pExec);
-
-int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle);
-int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle);
-
+// tqRead
 int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalHead** pHeadWithCkSum);
 
+// tqExec
 int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkRsp* pRsp, int32_t workerId);
 
+// tqMeta
+int32_t tqMetaOpen(STQ* pTq);
+int32_t tqMetaClose(STQ* pTq);
+int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle);
+int32_t tqMetaDeleteHandle(STQ* pTq, const char* key);
+
+// tqSink
+void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
+
 // tqOffset
 STqOffsetStore* STqOffsetOpen(STqOffsetCfg*);
 void            STqOffsetClose(STqOffsetStore*);

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

@@ -32,14 +32,27 @@ extern "C" {
 #define tsdbTrace(...) do { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TSDB ", DEBUG_TRACE, tsdbDebugFlag, __VA_ARGS__); }} while(0)
 // clang-format on
 
+typedef struct TSDBROW TSDBROW;
+typedef struct TSDBKEY TSDBKEY;
+typedef struct SDelOp  SDelOp;
+
+static int tsdbKeyCmprFn(const void *p1, const void *p2);
+
+// tsdbMemTable2.c ==============================================================================================
+typedef struct SMemTable SMemTable;
+
+int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTable);
+void    tsdbMemTableDestroy2(SMemTable *pMemTable);
+
 // tsdbMemTable ================
+typedef struct STsdbRow      STsdbRow;
 typedef struct STbData       STbData;
 typedef struct STsdbMemTable STsdbMemTable;
 typedef struct SMergeInfo    SMergeInfo;
 typedef struct STable        STable;
 
 int  tsdbMemTableCreate(STsdb *pTsdb, STsdbMemTable **ppMemTable);
-void tsdbMemTableDestroy(STsdb *pTsdb, STsdbMemTable *pMemTable);
+void tsdbMemTableDestroy(STsdbMemTable *pMemTable);
 int  tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead,
                            SDataCols *pCols, TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo);
 
@@ -845,6 +858,42 @@ static FORCE_INLINE int tsdbUnLockFS(STsdbFS *pFs) {
   return 0;
 }
 
+struct TSDBROW {
+  int64_t version;
+  STSRow2 tsRow;
+};
+
+struct TSDBKEY {
+  int64_t version;
+  TSKEY   ts;
+};
+
+struct SDelOp {
+  int64_t version;
+  TSKEY   sKey;  // included
+  TSKEY   eKey;  // included
+  SDelOp *pNext;
+};
+
+static FORCE_INLINE int tsdbKeyCmprFn(const void *p1, const void *p2) {
+  TSDBKEY *pKey1 = (TSDBKEY *)p1;
+  TSDBKEY *pKey2 = (TSDBKEY *)p2;
+
+  if (pKey1->ts < pKey2->ts) {
+    return -1;
+  } else if (pKey1->ts > pKey2->ts) {
+    return 1;
+  }
+
+  if (pKey1->version < pKey2->version) {
+    return -1;
+  } else if (pKey1->version > pKey2->version) {
+    return 1;
+  }
+
+  return 0;
+}
+
 #endif
 
 #ifdef __cplusplus

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

@@ -81,9 +81,10 @@ int32_t vnodeSyncCommit(SVnode* pVnode);
 int32_t vnodeAsyncCommit(SVnode* pVnode);
 
 // vnodeSync.c
-int32_t   vnodeSyncOpen(SVnode* pVnode, char* path);
-void      vnodeSyncStart(SVnode* pVnode);
-void      vnodeSyncClose(SVnode* pVnode);
+int32_t vnodeSyncOpen(SVnode* pVnode, char* path);
+void    vnodeSyncStart(SVnode* pVnode);
+void    vnodeSyncClose(SVnode* pVnode);
+void    vnodeSyncAlter(SVnode* pVnode, SRpcMsg* pMsg);
 
 #ifdef __cplusplus
 }

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

@@ -15,7 +15,6 @@
 
 #include "sma.h"
 
-
 // TODO: Who is responsible for resource allocate and release?
 int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg) {
   int32_t code = TSDB_CODE_SUCCESS;
@@ -37,7 +36,7 @@ int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg) {
   return code;
 }
 
-int32_t tdUpdateExpireWindow(SSma* pSma, SSubmitReq* pMsg, int64_t version) {
+int32_t tdUpdateExpireWindow(SSma* pSma, const SSubmitReq* pMsg, int64_t version) {
   int32_t code = TSDB_CODE_SUCCESS;
   if ((code = tdUpdateExpiredWindowImpl(pSma, pMsg, version)) < 0) {
     smaWarn("vgId:%d update expired sma window failed since %s", SMA_VID(pSma), tstrerror(terrno));