other: merge 3.0

.gitmodules

@@ -4,21 +4,9 @@
 [submodule "src/connector/hivemq-tdengine-extension"]
 	path = src/connector/hivemq-tdengine-extension
 	url = git@github.com:taosdata/hivemq-tdengine-extension.git
-[submodule "deps/jemalloc"]
-	path = deps/jemalloc
-	url = https://github.com/jemalloc/jemalloc
 [submodule "deps/TSZ"]
 	path = deps/TSZ
 	url = https://github.com/taosdata/TSZ.git
 [submodule "examples/rust"]
 	path = examples/rust
 	url = https://github.com/songtianyi/tdengine-rust-bindings.git
-[submodule "tools/taos-tools"]
-	path = tools/taos-tools
-	url = https://github.com/taosdata/taos-tools
-[submodule "tools/taosadapter"]
-	path = tools/taosadapter
-	url = https://github.com/taosdata/taosadapter.git
-[submodule "tools/taosws-rs"]
-	path = tools/taosws-rs
-	url = https://github.com/taosdata/taosws-rs

Jenkinsfile2

@@ -113,6 +113,12 @@ def pre_test(){
             echo "unmatched reposiotry ${CHANGE_URL}"
         '''
     }
+    sh '''
+        cd ${WKC}
+        git rm --cached tools/taos-tools 2>/dev/null || :
+        git rm --cached tools/taosadapter 2>/dev/null || :
+        git rm --cached tools/taosws-rs 2>/dev/null || :
+    '''
     sh '''
         cd ${WKC}
         git submodule update --init --recursive
@@ -258,6 +264,13 @@ def pre_test_win(){
         git branch
         git log -5
     '''
+    bat '''
+        cd %WIN_COMMUNITY_ROOT%
+        git rm --cached tools/taos-tools 2>nul
+        git rm --cached tools/taosadapter 2>nul
+        git rm --cached tools/taosws-rs 2>nul
+        exit 0
+    '''
     bat '''
         cd %WIN_COMMUNITY_ROOT%
         git submodule update --init --recursive

cmake/cmake.options

@@ -84,6 +84,12 @@ ELSE ()
     ENDIF ()
 ENDIF ()
 
+option(
+    JEMALLOC_ENABLED
+    "If build with jemalloc"
+    OFF
+    )
+
 option(
     BUILD_SANITIZER
     "If build sanitizer"

cmake/jemalloc_CMakeLists.txt.in

+
+# jemalloc
+ExternalProject_Add(jemalloc
+        GIT_REPOSITORY https://github.com/jemalloc/jemalloc.git
+        GIT_TAG 5.3.0
+        SOURCE_DIR "${TD_CONTRIB_DIR}/jemalloc"
+        BINARY_DIR ""
+        CONFIGURE_COMMAND ""
+        BUILD_COMMAND ""
+        INSTALL_COMMAND ""
+        TEST_COMMAND ""
+        GIT_SHALLOW true
+        GIT_PROGRESS true
+        )

cmake/taostools_CMakeLists.txt.in

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

docs/en/07-develop/03-insert-data/03-opentsdb-telnet.mdx

@@ -23,7 +23,7 @@ A single line of text is used in OpenTSDB line protocol to represent one row of
 
 - `metric` will be used as the STable name.
 - `timestamp` is the timestamp of current row of data. The time precision will be determined automatically based on the length of the timestamp. Second and millisecond time precision are supported.
-- `value` is a metric which must be a numeric value, the corresponding column name is "value".
+- `value` is a metric which must be a numeric value, the corresponding column name is "_value".
 - The last part is the tag set separated by spaces, all tags will be converted to nchar type automatically.
 
 For example:
@@ -74,7 +74,7 @@ taos> show STables;
 Query OK, 2 row(s) in set (0.002544s)
 
 taos> select tbname, * from `meters.current`;
-             tbname             |           ts            |           value           | groupid |            location            |
+             tbname             |           _ts            |           _value           | groupid |            location            |
 ==================================================================================================================================
  t_0e7bcfa21a02331c06764f275... | 2022-03-28 09:56:51.249 |              10.800000000 | 3       | California.LoSangeles                |
  t_0e7bcfa21a02331c06764f275... | 2022-03-28 09:56:51.250 |              11.300000000 | 3       | California.LoSangeles                |

docs/en/07-develop/03-insert-data/04-opentsdb-json.mdx

@@ -91,7 +91,7 @@ taos> show STables;
 Query OK, 2 row(s) in set (0.001954s)
 
 taos> select * from `meters.current`;
-           ts            |           value           |          groupid          |            location            |
+           _ts            |           _value           |          groupid          |            location            |
 ===================================================================================================================
  2022-03-28 09:56:51.249 |              10.300000000 |               2.000000000 | California.SanFrancisco               |
  2022-03-28 09:56:51.250 |              12.600000000 |               2.000000000 | California.SanFrancisco               |

docs/zh/05-get-started/01-docker.md

@@ -111,7 +111,7 @@ TDengine REST API 详情请参考[官方文档](/reference/rest-api/)。
 
    这条命令很快完成 1 亿条记录的插入。具体时间取决于硬件性能。
 
-   taosBenchmark 命令本身带有很多选项，配置表的数目、记录条数等等，您可以设置不同参数进行体验，请执行 `taosBenchmark --help` 详细列出。taosBenchmark 详细使用方法请参照 [taosBenchmark 参考手册](../reference/taosbenchmark)。
+   taosBenchmark 命令本身带有很多选项，配置表的数目、记录条数等等，您可以设置不同参数进行体验，请执行 `taosBenchmark --help` 详细列出。taosBenchmark 详细使用方法请参照 [taosBenchmark 参考手册](../../reference/taosbenchmark)。
 
 ## 体验查询
 

docs/zh/05-get-started/03-package.md

@@ -245,7 +245,7 @@ select * from t;
 Query OK, 2 row(s) in set (0.003128s)
 ```
 
-除执行 SQL 语句外，系统管理员还可以从 TDengine CLI 进行检查系统运行状态、添加删除用户账号等操作。TDengine CLI 连同应用驱动也可以独立安装在 Linux 或 Windows 机器上运行，更多细节请参考 [这里](../reference/taos-shell/)
+除执行 SQL 语句外，系统管理员还可以从 TDengine CLI 进行检查系统运行状态、添加删除用户账号等操作。TDengine CLI 连同应用驱动也可以独立安装在 Linux 或 Windows 机器上运行，更多细节请参考 [这里](../../reference/taos-shell/)
 
 ## 使用 taosBenchmark 体验写入速度
 

docs/zh/07-develop/03-insert-data/03-opentsdb-telnet.mdx

@@ -23,7 +23,7 @@ OpenTSDB 行协议同样采用一行字符串来表示一行数据。OpenTSDB
 
 - metric 将作为超级表名。
 - timestamp 本行数据对应的时间戳。根据时间戳的长度自动识别时间精度。支持秒和毫秒两种时间精度
-- value 度量值，必须为一个数值。对应的列名也是 “value”。
+- value 度量值，必须为一个数值。对应的列名是 “_value”。
 - 最后一部分是标签集， 用空格分隔不同标签， 所有标签自动转化为 nchar 数据类型;
 
 例如：
@@ -74,7 +74,7 @@ taos> show stables;
 Query OK, 2 row(s) in set (0.002544s)
 
 taos> select tbname, * from `meters.current`;
-             tbname             |           ts            |           value           | groupid |            location            |
+             tbname             |           _ts            |           _value           | groupid |            location            |
 ==================================================================================================================================
  t_0e7bcfa21a02331c06764f275... | 2022-03-28 09:56:51.249 |              10.800000000 | 3       | California.LosAngeles                |
  t_0e7bcfa21a02331c06764f275... | 2022-03-28 09:56:51.250 |              11.300000000 | 3       | California.LosAngeles                |

docs/zh/07-develop/03-insert-data/04-opentsdb-json.mdx

@@ -91,7 +91,7 @@ taos> show stables;
 Query OK, 2 row(s) in set (0.001954s)
 
 taos> select * from `meters.current`;
-           ts            |           value           |          groupid          |            location            |
+           _ts            |           _value           |          groupid          |            location            |
 ===================================================================================================================
  2022-03-28 09:56:51.249 |              10.300000000 |               2.000000000 | California.SanFrancisco               |
  2022-03-28 09:56:51.250 |              12.600000000 |               2.000000000 | California.SanFrancisco               |

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

@@ -18,7 +18,7 @@ database_option: {
   | CACHESIZE value
   | COMP {0 | 1 | 2}
   | DURATION value
-  | FSYNC value
+  | WAL_FSYNC_PERIOD value
   | MAXROWS value
   | MINROWS value
   | KEEP value
@@ -28,7 +28,7 @@ database_option: {
   | REPLICA value
   | RETENTIONS ingestion_duration:keep_duration ...
   | STRICT {'off' | 'on'}
-  | WAL {1 | 2}
+  | WAL_LEVEL {1 | 2}
   | VGROUPS value
   | SINGLE_STABLE {0 | 1}
   | WAL_RETENTION_PERIOD value
@@ -46,7 +46,7 @@ database_option: {
   - last_row：表示缓存子表最近一行数据。这将显著改善 LAST_ROW 函数的性能表现。
   - last_value：表示缓存子表每一列的最近的非 NULL 值。这将显著改善无特殊影响（WHERE、ORDER BY、GROUP BY、INTERVAL）下的 LAST 函数的性能表现。
   - both：表示同时打开缓存最近行和列功能。
-- CACHESIZE：表示缓存子表最近数据的内存大小。默认为 1 ，范围是[1, 65536]，单位是 MB。
+- CACHESIZE：表示每个 vnode 中用于缓存子表最近数据的内存大小。默认为 1 ，范围是[1, 65536]，单位是 MB。
 - COMP：表示数据库文件压缩标志位，缺省值为 2，取值范围为 [0, 2]。
   - 0：表示不压缩。
   - 1：表示一阶段压缩。

docs/zh/12-taos-sql/06-select.md

@@ -218,7 +218,7 @@ GROUP BY 子句中的表达式可以包含表或视图中的任何列，这些
 
 PARTITION BY 子句是 TDengine 特色语法，按 part_list 对数据进行切分，在每个切分的分片中进行计算。
 
-详见 [TDengine 特色查询](./distinguished)
+详见 [TDengine 特色查询](../distinguished)
 
 ## ORDER BY
 

docs/zh/12-taos-sql/21-node.md

@@ -40,7 +40,6 @@ ALTER ALL DNODES dnode_option
 
 dnode_option: {
     'resetLog'
-  | 'resetQueryCache'
   | 'balance' value
   | 'monitor' value
   | 'debugFlag' value

docs/zh/12-taos-sql/26-udf.md

@@ -19,7 +19,7 @@ library_path：包含UDF函数实现的动态链接库的绝对路径，是在
 OUTPUTTYPE：标识此函数的返回类型。
 BUFSIZE：中间结果的缓冲区大小，单位是字节。不设置则默认为0。最大不可超过512字节。
 
-关于如何开发自定义函数，请参考 [UDF使用说明](../develop/udf)。
+关于如何开发自定义函数，请参考 [UDF使用说明](../../develop/udf)。
 
 ## 删除自定义函数
 

docs/zh/14-reference/14-taosx.md

+---
+sidebar_label: taosX
+title: 使用 taosX 在集群间复制数据
+---
\ No newline at end of file

docs/zh/14-reference/_category_.yml

-label: 参考指南
+label: 参考手册
\ No newline at end of file

docs/zh/14-reference/index.md

 ---
-sidebar_label: 参考手册
 title: 参考手册
 ---
 

docs/zh/17-operation/01-pkg-install.md

@@ -10,7 +10,7 @@ import TabItem from "@theme/TabItem";
 
 ## 安装
 
-关于安装，请参考 [使用安装包立即开始](../get-started/package)
+关于安装，请参考 [使用安装包立即开始](../../get-started/package)
 
 
 

docs/zh/17-operation/02-planning.mdx

@@ -5,33 +5,28 @@ title: 容量规划
 
 使用 TDengine 来搭建一个物联网大数据平台，计算资源、存储资源需要根据业务场景进行规划。下面分别讨论系统运行所需要的内存、CPU 以及硬盘空间。
 
-## 内存需求
+## 服务端内存需求
 
-每个 Database 可以创建固定数目的 vgroup，默认与 CPU 核数相同，可通过 maxVgroupsPerDb 配置；vgroup 中的每个副本会是一个 vnode；每个 vnode 会占用固定大小的内存（大小与数据库的配置参数 blocks 和 cache 有关)；每个 Table 会占用与标签总长度有关的内存；此外，系统会有一些固定的内存开销。因此，每个 DB 需要的系统内存可通过如下公式计算：
+每个 Database 可以创建固定数目的 vgroup，默认 2 个 vgroup，在创建数据库时可以通过`vgroups <num>`参数来指定，其副本数由参数`replica <num>`指定。vgroup 中的每个副本会是一个 vnode；所以每个数据库占用的内存由以下几个参数决定：
 
-```
-Database Memory Size = maxVgroupsPerDb * replica * (blocks * cache + 10MB) + numOfTables * (tagSizePerTable + 0.5KB)
-```
+- vgroups
+- replica
+- buffer
+- pages
+- pagesize
+- cachesize
 
-示例：假设 maxVgroupPerDB 是缺省值 64，cache 是缺省大小 16M, blocks 是缺省值 6，并且一个 DB 中有 10 万张表，单副本，标签总长度是 256 字节，则这个 DB 总的内存需求为：64 \* 1 \* (16 \* 6 + 10) + 100000 \* (0.25 + 0.5) / 1000 = 6792M。
+关于这些参数的详细说明请参考 [数据库管理](../../taos-sql/database)。
 
-在实际的系统运维中，我们通常会更关心 TDengine 服务进程（taosd）会占用的内存量。
+一个数据库所需要的内存大小等于
 
 ```
-taosd 内存总量 = vnode 内存 + mnode 内存 + 查询内存
+vgroups * replica * (buffer + pages * pagesize + cachesize)
 ```
 
-其中：
-
-1. “vnode 内存”指的是集群中所有的 Database 存储分摊到当前 taosd 节点上所占用的内存资源。可以按上文“Database Memory Size”计算公式估算每个 DB 的内存占用量进行加总，再按集群中总共的 TDengine 节点数做平均（如果设置为多副本，则还需要乘以对应的副本倍数）。
-2. “mnode 内存”指的是集群中管理节点所占用的资源。如果一个 taosd 节点上分布有 mnode 管理节点，则内存消耗还需要增加“0.2KB \* 集群中数据表总数”。
-3. “查询内存”指的是服务端处理查询请求时所需要占用的内存。单条查询语句至少会占用“0.2KB \* 查询涉及的数据表总数”的内存量。
+但要注意的是这些内存并不需要由单一服务器提供，而是由整个集群中所有数据节点共同负担，相当于由这些数据节点所在的服务器共同负担。如果集群中有不止一个数据库，则所需内存要累加。更复杂的场景是如果集群中的数据节点并非在最初就一次性全部建立，而是随着使用中系统负载的增加逐步增加服务器并增加数据节点，则新创建的数据库会导致新旧数据节点上的负载并不均衡，此时简单的理论计算并不能直接使用，要结合各数据节点的负载情况。
 
-注意：以上内存估算方法，主要讲解了系统的“必须内存需求”，而不是“内存总数上限”。在实际运行的生产环境中，由于操作系统缓存、资源管理调度等方面的原因，内存规划应当在估算结果的基础上保留一定冗余，以维持系统状态和系统性能的稳定性。并且，生产环境通常会配置系统资源的监控工具，以便及时发现硬件资源的紧缺情况。
-
-最后，如果内存充裕，可以考虑加大 Blocks 的配置，这样更多数据将保存在内存里，提高写入和查询速度。
-
-### 客户端内存需求
+## 客户端内存需求
 
 客户端应用采用 taosc 客户端驱动连接服务端，会有内存需求的开销。
 
@@ -56,7 +51,7 @@ CPU 的需求取决于如下两方面：
 - **数据插入** TDengine 单核每秒能至少处理一万个插入请求。每个插入请求可以带多条记录，一次插入一条记录与插入 10 条记录，消耗的计算资源差别很小。因此每次插入，条数越大，插入效率越高。如果一个插入请求带 200 条以上记录，单核就能达到每秒插入 100 万条记录的速度。但对前端数据采集的要求越高，因为需要缓存记录，然后一批插入。
 - **查询需求** TDengine 提供高效的查询，但是每个场景的查询差异很大，查询频次变化也很大，难以给出客观数字。需要用户针对自己的场景，写一些查询语句，才能确定。
 
-因此仅对数据插入而言，CPU 是可以估算出来的，但查询所耗的计算资源无法估算。在实际运营过程中，不建议 CPU 使用率超过 50%，超过后，需要增加新的节点，以获得更多计算资源。
+因此仅对数据插入而言，CPU 是可以估算出来的，但查询所耗的计算资源无法估算。在实际运行过程中，不建议 CPU 使用率超过 50%，超过后，需要增加新的节点，以获得更多计算资源。
 
 ## 存储需求
 
@@ -77,5 +72,3 @@ Raw DataSize = numOfTables * rowSizePerTable * rowsPerTable
 根据上面的内存、CPU、存储的预估，就可以知道整个系统需要多少核、多少内存、多少存储空间。如果数据副本数不为 1，总需求量需要再乘以副本数。
 
 因为 TDengine 具有很好的水平扩展能力，根据总量，再根据单个物理机或虚拟机的资源，就可以轻松决定需要购置多少台物理机或虚拟机了。
-
-**立即计算 CPU、内存、存储，请参见：[资源估算方法](https://www.taosdata.com/config/config.html)。**

docs/zh/17-operation/03-tolerance.md

@@ -4,25 +4,27 @@ title: 容错和灾备
 
 ## 容错
 
-TDengine 支持**WAL**（Write Ahead Log）机制，实现数据的容错能力，保证数据的高可用。
+TDengine 支持 **WAL**（Write Ahead Log）机制，实现数据的容错能力，保证数据的高可用。
 
 TDengine 接收到应用的请求数据包时，先将请求的原始数据包写入数据库日志文件，等数据成功写入数据库数据文件后，再删除相应的 WAL。这样保证了 TDengine 能够在断电等因素导致的服务重启时从数据库日志文件中恢复数据，避免数据的丢失。
 
 涉及的系统配置参数有两个：
 
-- walLevel：WAL 级别，0：不写 WAL; 1：写 WAL, 但不执行 fsync; 2：写 WAL, 而且执行 fsync。
-- fsync：当 walLevel 设置为 2 时，执行 fsync 的周期。设置为 0，表示每次写入，立即执行 fsync。
+- wal_level：WAL 级别，1：写 WAL，但不执行 fsync。2：写 WAL，而且执行 fsync。默认值为 1。
+- wal_fsync_period：当 wal_level 设置为 2 时，执行 fsync 的周期。设置为 0，表示每次写入，立即执行 fsync。
 
-如果要 100%的保证数据不丢失，需要将 walLevel 设置为 2，fsync 设置为 0。这时写入速度将会下降。但如果应用侧启动的写数据的线程数达到一定的数量(超过 50)，那么写入数据的性能也会很不错，只会比 fsync 设置为 3000 毫秒下降 30%左右。
+如果要 100%的保证数据不丢失，需要将 wal_level 设置为 2，wal_fsync_period 设置为 0。这时写入速度将会下降。但如果应用侧启动的写数据的线程数达到一定的数量(超过 50)，那么写入数据的性能也会很不错，只会比 wal_fsync_period 设置为 3000 毫秒下降 30%左右。
 
 ## 灾备
 
-TDengine 的集群通过多个副本的机制，来提供系统的高可用性，实现灾备能力。
+TDengine 的集群通过多个副本的机制，来提供系统的高可用性，同时具备一定的灾备能力。
 
-TDengine 集群是由 mnode 负责管理的，为保证 mnode 的高可靠，可以配置多个 mnode 副本，副本数由系统配置参数 numOfMnodes 决定，为了支持高可靠，需要设置大于 1。为保证元数据的强一致性，mnode 副本之间通过同步方式进行数据复制，保证了元数据的强一致性。
+TDengine 集群是由 mnode 负责管理的，为保证 mnode 的高可靠，可以配置 三个 mnode 副本。为保证元数据的强一致性，mnode 副本之间通过同步方式进行数据复制，保证了元数据的强一致性。
 
-TDengine 集群中的时序数据的副本数是与数据库关联的，一个集群里可以有多个数据库，每个数据库可以配置不同的副本数。创建数据库时，通过参数 replica 指定副本数。为了支持高可靠，需要设置副本数大于 1。
+TDengine 集群中的时序数据的副本数是与数据库关联的，一个集群里可以有多个数据库，每个数据库可以配置不同的副本数。创建数据库时，通过参数 replica 指定副本数。为了支持高可靠，需要设置副本数为 3。
 
 TDengine 集群的节点数必须大于等于副本数，否则创建表时将报错。
 
 当 TDengine 集群中的节点部署在不同的物理机上，并设置多个副本数时，就实现了系统的高可靠性，无需再使用其他软件或工具。TDengine 企业版还可以将副本部署在不同机房，从而实现异地容灾。
+
+另外一种灾备方式是通过 `taosX` 将一个 TDengine 集群的数据同步复制到物理上位于不同数据中心的另一个 TDengine 集群。其详细使用方法请参考 [taosX 参考手册](../../reference/taosX)

docs/zh/17-operation/06-admin.md

----
-title: 用户管理
----
-
-系统管理员可以在 CLI 界面里添加、删除用户，也可以修改密码。CLI 里 SQL 语法如下：
-
-```sql
-CREATE USER <user_name> PASS <'password'>;
-```
-
-创建用户，并指定用户名和密码，密码需要用单引号引起来，单引号为英文半角
-
-```sql
-DROP USER <user_name>;
-```
-
-删除用户，限 root 用户使用
-
-```sql
-ALTER USER <user_name> PASS <'password'>;
-```
-
-修改用户密码，为避免被转换为小写，密码需要用单引号引用，单引号为英文半角
-
-```sql
-ALTER USER <user_name> PRIVILEGE <write|read>;
-```
-
-修改用户权限为：write 或 read，不需要添加单引号
-
-说明：系统内共有 super/write/read 三种权限级别，但目前不允许通过 alter 指令把 super 权限赋予用户。
-
-```sql
-SHOW USERS;
-```
-
-显示所有用户
-
-:::note
-SQL 语法中，< >表示需要用户输入的部分，但请不要输入< >本身。
-
-:::

docs/zh/17-operation/09-status.md

----
-title: 系统连接、任务查询管理
----
-
-系统管理员可以从 CLI 查询系统的连接、正在进行的查询、流式计算，并且可以关闭连接、停止正在进行的查询和流式计算。
-
-## 显示数据库的连接
-
-```sql
-SHOW CONNECTIONS;
-```
-
-其结果中的一列显示 ip:port, 为连接的 IP 地址和端口号。
-
-## 强制关闭数据库连接
-
-```sql
-KILL CONNECTION <connection-id>;
-```
-
-其中的 connection-id 是 SHOW CONNECTIONS 中显示的第一列的数字。
-
-## 显示数据查询
-
-```sql
-SHOW QUERIES;
-```
-
-其中第一列显示的以冒号隔开的两个数字为 query-id，为发起该 query 应用连接的 connection-id 和查询次数。
-
-## 强制关闭数据查询
-
-```sql
-KILL QUERY <query-id>;
-```
-
-其中 query-id 是 SHOW QUERIES 中显示的 connection-id:query-no 字串，如“105:2”，拷贝粘贴即可。
-
-## 显示连续查询
-
-```sql
-SHOW STREAMS;
-```
-
-其中第一列显示的以冒号隔开的两个数字为 stream-id, 为启动该 stream 应用连接的 connection-id 和发起 stream 的次数。
-
-## 强制关闭连续查询
-
-```sql
-KILL STREAM <stream-id>;
-```
-
-其中的 stream-id 是 SHOW STREAMS 中显示的 connection-id:stream-no 字串，如 103:2，拷贝粘贴即可。

include/common/tglobal.h

@@ -41,10 +41,8 @@ extern int32_t tsCompressMsgSize;
 extern int32_t tsCompressColData;
 extern int32_t tsMaxNumOfDistinctResults;
 extern int32_t tsCompatibleModel;
-extern bool    tsEnableSlaveQuery;
 extern bool    tsPrintAuth;
 extern int64_t tsTickPerMin[3];
-
 extern int32_t tsCountAlwaysReturnValue;
 
 // multi-process
@@ -91,9 +89,6 @@ extern uint16_t tsTelemPort;
 // query buffer management
 extern int32_t tsQueryBufferSize;  // maximum allowed usage buffer size in MB for each data node during query processing
 extern int64_t tsQueryBufferSizeBytes;   // maximum allowed usage buffer size in byte for each data node
-extern bool    tsRetrieveBlockingModel;  // retrieve threads will be blocked
-extern bool    tsKeepOriginalColumnName;
-extern bool    tsDeadLockKillQuery;
 
 // query client
 extern int32_t tsQueryPolicy;
@@ -102,11 +97,6 @@ extern int32_t tsQuerySmaOptimize;
 // client
 extern int32_t tsMinSlidingTime;
 extern int32_t tsMinIntervalTime;
-extern int32_t tsMaxStreamComputDelay;
-extern int32_t tsStreamCompStartDelay;
-extern int32_t tsRetryStreamCompDelay;
-extern float   tsStreamComputDelayRatio;  // the delayed computing ration of the whole time window
-extern int64_t tsMaxRetentWindow;
 
 // build info
 extern char version[];

include/common/tmsg.h

@@ -3044,12 +3044,41 @@ typedef struct SDeleteRes {
   int64_t  skey;
   int64_t  ekey;
   int64_t  affectedRows;
-  char     tableFName[TSDB_TABLE_FNAME_LEN];
+  char     tableFName[TSDB_TABLE_NAME_LEN];
+  char     tsColName[TSDB_COL_NAME_LEN];
 } SDeleteRes;
 
 int32_t tEncodeDeleteRes(SEncoder* pCoder, const SDeleteRes* pRes);
 int32_t tDecodeDeleteRes(SDecoder* pCoder, SDeleteRes* pRes);
 
+typedef struct {
+  int32_t msgType;
+  int32_t msgLen;
+  void*   msg;
+} SBatchMsg;
+
+typedef struct {
+  SMsgHead  header;
+  int32_t   msgNum;
+  SBatchMsg msg[];
+} SBatchReq;
+
+typedef struct {
+  int32_t reqType;
+  int32_t msgLen;
+  int32_t rspCode;
+  void*   msg;
+} SBatchRsp;
+
+static FORCE_INLINE void tFreeSBatchRsp(void *p) {
+  if (NULL == p) {
+    return;
+  }
+
+  SBatchRsp* pRsp = (SBatchRsp*)p;
+  taosMemoryFree(pRsp->msg);
+}
+
 #pragma pack(pop)
 
 #ifdef __cplusplus

include/common/tmsgdef.h

@@ -136,6 +136,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_DROP_INDEX, "drop-index", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_GET_INDEX, "get-index", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_GET_TABLE_INDEX, "get-table-index", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_BATCH_META, "batch-meta", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_TABLE_CFG, "table-cfg", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TOPIC, "create-topic", SMCreateTopicReq, SMCreateTopicRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "alter-topic", NULL, NULL)
@@ -180,6 +181,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_TABLE_META, "vnode-table-meta", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TABLES_META, "vnode-tables-meta", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TABLE_CFG, "vnode-table-cfg", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_BATCH_META, "vnode-batch-meta", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_CREATE_STB, "vnode-create-stb", SVCreateStbReq, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_ALTER_STB, "vnode-alter-stb", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_DROP_STB, "vnode-drop-stb", SVDropStbReq, NULL)

include/libs/executor/dataSinkMgt.h

@@ -38,7 +38,8 @@ typedef struct SDeleterRes {
   int64_t  skey;
   int64_t  ekey;
   int64_t  affectedRows;
-  char     tableFName[TSDB_TABLE_FNAME_LEN];
+  char     tableName[TSDB_TABLE_NAME_LEN];
+  char     tsColName[TSDB_COL_NAME_LEN];
 } SDeleterRes;
 
 typedef struct SDeleterParam {

include/libs/executor/executor.h

@@ -64,17 +64,7 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers);
  * @param SReadHandle
  * @return
  */
-qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols,
-                                     SSchemaWrapper** pSchema);
-
-/**
- * Set the input data block for the stream scan.
- * @param tinfo
- * @param input
- * @param type
- * @return
- */
-int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type, bool assignUid);
+qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, SSchemaWrapper** pSchema);
 
 /**
  * Set multiple input data blocks for the stream scan.
@@ -84,7 +74,7 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type, bool
  * @param type
  * @return
  */
-int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type, bool assignUid);
+int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type);
 
 /**
  * Update the table id list, add or remove.

include/libs/nodes/plannodes.h

@@ -151,7 +151,8 @@ typedef struct SVnodeModifyLogicNode {
   uint64_t         tableId;
   uint64_t         stableId;
   int8_t           tableType;  // table type
-  char             tableFName[TSDB_TABLE_FNAME_LEN];
+  char             tableName[TSDB_TABLE_NAME_LEN];
+  char             tsColName[TSDB_COL_NAME_LEN];
   STimeWindow      deleteTimeRange;
   SVgroupsInfo*    pVgroupList;
   SNodeList*       pInsertCols;
@@ -494,7 +495,7 @@ typedef struct SQueryInserterNode {
   uint64_t      tableId;
   uint64_t      stableId;
   int8_t        tableType;  // table type
-  char          tableFName[TSDB_TABLE_FNAME_LEN];
+  char          tableName[TSDB_TABLE_NAME_LEN];
   int32_t       vgId;
   SEpSet        epSet;
 } SQueryInserterNode;
@@ -503,7 +504,7 @@ typedef struct SDataDeleterNode {
   SDataSinkNode sink;
   uint64_t      tableId;
   int8_t        tableType;  // table type
-  char          tableFName[TSDB_TABLE_FNAME_LEN];
+  char          tableFName[TSDB_TABLE_NAME_LEN];
   char          tsColName[TSDB_COL_NAME_LEN];
   STimeWindow   deleteTimeRange;
   SNode*        pAffectedRows;

include/libs/stream/tstream.h

@@ -55,7 +55,6 @@ enum {
   TASK_INPUT_STATUS__NORMAL = 1,
   TASK_INPUT_STATUS__BLOCKED,
   TASK_INPUT_STATUS__RECOVER,
-  TASK_INPUT_STATUS__PROCESSING,
   TASK_INPUT_STATUS__STOP,
   TASK_INPUT_STATUS__FAILED,
 };
@@ -270,7 +269,7 @@ typedef struct SStreamTask {
   int64_t startVer;
   int64_t checkpointVer;
   int64_t processedVer;
-  int32_t numOfVgroups;
+  // int32_t numOfVgroups;
 
   // children info
   SArray* childEpInfo;  // SArray<SStreamChildEpInfo*>
@@ -320,17 +319,6 @@ int32_t      tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask);
 void         tFreeSStreamTask(SStreamTask* pTask);
 
 static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem* pItem) {
-#if 0
-  while (1) {
-    int8_t inputStatus =
-        atomic_val_compare_exchange_8(&pTask->inputStatus, TASK_INPUT_STATUS__NORMAL, TASK_INPUT_STATUS__PROCESSING);
-    if (inputStatus == TASK_INPUT_STATUS__NORMAL) {
-      break;
-    }
-    ASSERT(0);
-  }
-#endif
-
   if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {
     SStreamDataSubmit* pSubmitClone = streamSubmitRefClone((SStreamDataSubmit*)pItem);
     if (pSubmitClone == NULL) {
@@ -443,13 +431,14 @@ typedef struct {
 typedef struct {
   int64_t streamId;
   int32_t taskId;
-  int32_t sourceTaskId;
-  int32_t sourceVg;
+  int32_t upstreamTaskId;
+  int32_t upstreamNodeId;
 } SStreamTaskRecoverReq;
 
 typedef struct {
   int64_t streamId;
-  int32_t taskId;
+  int32_t rspTaskId;
+  int32_t reqTaskId;
   int8_t  inputStatus;
 } SStreamTaskRecoverRsp;
 

include/libs/sync/sync.h

@@ -210,7 +210,7 @@ SyncGroupId syncGetVgId(int64_t rid);
 void        syncGetEpSet(int64_t rid, SEpSet* pEpSet);
 void        syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet);
 int32_t     syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak);
-int32_t     syncProposeBatch(int64_t rid, SRpcMsg* pMsgArr, bool* pIsWeakArr, int32_t arrSize);
+int32_t     syncProposeBatch(int64_t rid, SRpcMsg** pMsgPArr, bool* pIsWeakArr, int32_t arrSize);
 bool        syncEnvIsStart();
 const char* syncStr(ESyncState state);
 bool        syncIsRestoreFinish(int64_t rid);

include/libs/sync/syncTools.h

@@ -238,7 +238,7 @@ typedef struct SyncClientRequestBatch {
   char     data[];  // block2, block3
 } SyncClientRequestBatch;
 
-SyncClientRequestBatch* syncClientRequestBatchBuild(SRpcMsg* rpcMsgArr, SRaftMeta* raftArr, int32_t arrSize,
+SyncClientRequestBatch* syncClientRequestBatchBuild(SRpcMsg** rpcMsgPArr, SRaftMeta* raftArr, int32_t arrSize,
                                                     int32_t vgId);
 void                    syncClientRequestBatch2RpcMsg(const SyncClientRequestBatch* pSyncMsg, SRpcMsg* pRpcMsg);
 void                    syncClientRequestBatchDestroy(SyncClientRequestBatch* pMsg);

include/libs/wal/wal.h

@@ -77,11 +77,11 @@ typedef struct {
 } SWalSyncInfo;
 
 typedef struct {
-  int8_t  protoVer;
   int64_t version;
-  int16_t msgType;
+  int64_t ingestTs;
   int32_t bodyLen;
-  int64_t ingestTs;  // not implemented
+  int16_t msgType;
+  int8_t  protoVer;
 
   // sync meta
   SWalSyncInfo syncMeta;

include/util/taoserror.h

@@ -555,6 +555,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_ONLY_SUPPORT_SINGLE_TABLE TAOS_DEF_ERROR_CODE(0, 0x265F)
 #define TSDB_CODE_PAR_INVALID_SMA_INDEX         TAOS_DEF_ERROR_CODE(0, 0x2660)
 #define TSDB_CODE_PAR_INVALID_SELECTED_EXPR     TAOS_DEF_ERROR_CODE(0, 0x2661)
+#define TSDB_CODE_PAR_GET_META_ERROR            TAOS_DEF_ERROR_CODE(0, 0x2662)
 #define TSDB_CODE_PAR_INTERNAL_ERROR            TAOS_DEF_ERROR_CODE(0, 0x26FF)
 
 //planner

source/client/src/clientHb.c

@@ -353,6 +353,7 @@ int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) {
         desc.subDesc = NULL;
         desc.subPlanNum = 0;
       }
+      ASSERT(desc.subPlanNum == taosArrayGetSize(desc.subDesc));
     } else {
       desc.subDesc = NULL;
     }

source/client/src/clientImpl.c

@@ -591,7 +591,7 @@ int32_t buildAsyncExecNodeList(SRequestObj* pRequest, SArray** pNodeList, SArray
   return code;
 }
 
-void freeVgList(void *list) {
+void freeVgList(void* list) {
   SArray* pList = *(SArray**)list;
   taosArrayDestroy(pList);
 }
@@ -1278,8 +1278,8 @@ int32_t doProcessMsgFromServer(void* param) {
   char      tbuf[40] = {0};
   TRACE_TO_STR(trace, tbuf);
 
-  tscDebug("processMsgFromServer handle %p, message: %s, code: %s, gtid: %s", pMsg->info.handle, TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code),
-           tbuf);
+  tscDebug("processMsgFromServer handle %p, message: %s, code: %s, gtid: %s", pMsg->info.handle,
+           TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code), tbuf);
 
   if (pSendInfo->requestObjRefId != 0) {
     SRequestObj* pRequest = (SRequestObj*)taosAcquireRef(clientReqRefPool, pSendInfo->requestObjRefId);
@@ -2114,7 +2114,7 @@ TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly) {
     return NULL;
   }
 
-  TAOS_RES* pRes = execQuery(connId, sql, sqlLen, validateOnly);
+  TAOS_RES* pRes = execQuery(*(int64_t*)taos, sql, sqlLen, validateOnly);
   return pRes;
 #endif
 }

source/client/src/tmq.c

@@ -2823,35 +2823,35 @@ end:
 
 // delete from db.tabl where ..       -> delete from tabl where ..
 // delete from db    .tabl where ..   -> delete from tabl where ..
-static void getTbName(char *sql){
-  char *ch = sql;
-
-  bool inBackQuote = false;
-  int8_t dotIndex = 0;
-  while(*ch != '\0'){
-    if(!inBackQuote && *ch == '`'){
-      inBackQuote = true;
-      ch++;
-      continue;
-    }
-
-    if(inBackQuote && *ch == '`'){
-      inBackQuote = false;
-      ch++;
-
-      continue;
-    }
-
-    if(!inBackQuote && *ch == '.'){
-      dotIndex ++;
-      if(dotIndex == 2){
-        memmove(sql, ch + 1, strlen(ch + 1) + 1);
-        break;
-      }
-    }
-    ch++;
-  }
-}
+//static void getTbName(char *sql){
+//  char *ch = sql;
+//
+//  bool inBackQuote = false;
+//  int8_t dotIndex = 0;
+//  while(*ch != '\0'){
+//    if(!inBackQuote && *ch == '`'){
+//      inBackQuote = true;
+//      ch++;
+//      continue;
+//    }
+//
+//    if(inBackQuote && *ch == '`'){
+//      inBackQuote = false;
+//      ch++;
+//
+//      continue;
+//    }
+//
+//    if(!inBackQuote && *ch == '.'){
+//      dotIndex ++;
+//      if(dotIndex == 2){
+//        memmove(sql, ch + 1, strlen(ch + 1) + 1);
+//        break;
+//      }
+//    }
+//    ch++;
+//  }
+//}
 
 static int32_t taosDeleteData(TAOS* taos, void* meta, int32_t metaLen) {
   SDeleteRes     req = {0};
@@ -2867,9 +2867,9 @@ static int32_t taosDeleteData(TAOS* taos, void* meta, int32_t metaLen) {
     goto end;
   }
 
-  getTbName(req.tableFName);
+//  getTbName(req.tableFName);
   char sql[256] = {0};
-  sprintf(sql, "delete from `%s` where `%s` >= %" PRId64" and `%s` <= %" PRId64, req.tableFName, "ts", req.skey, "ts", req.ekey);
+  sprintf(sql, "delete from `%s` where `%s` >= %" PRId64" and `%s` <= %" PRId64, req.tableFName, req.tsColName, req.skey, req.tsColName, req.ekey);
   printf("delete sql:%s\n", sql);
 
   TAOS_RES* res = taos_query(taos, sql);

source/common/src/systable.c

@@ -257,14 +257,13 @@ static const SSysTableMeta infosMeta[] = {
     {TSDB_INS_TABLE_MNODES, mnodesSchema, tListLen(mnodesSchema)},
     {TSDB_INS_TABLE_MODULES, modulesSchema, tListLen(modulesSchema)},
     {TSDB_INS_TABLE_QNODES, qnodesSchema, tListLen(qnodesSchema)},
-    {TSDB_INS_TABLE_SNODES, snodesSchema, tListLen(snodesSchema)},
-    {TSDB_INS_TABLE_BNODES, bnodesSchema, tListLen(bnodesSchema)},
+//    {TSDB_INS_TABLE_SNODES, snodesSchema, tListLen(snodesSchema)},
+//    {TSDB_INS_TABLE_BNODES, bnodesSchema, tListLen(bnodesSchema)},
     {TSDB_INS_TABLE_CLUSTER, clusterSchema, tListLen(clusterSchema)},
     {TSDB_INS_TABLE_USER_DATABASES, userDBSchema, tListLen(userDBSchema)},
     {TSDB_INS_TABLE_USER_FUNCTIONS, userFuncSchema, tListLen(userFuncSchema)},
     {TSDB_INS_TABLE_USER_INDEXES, userIdxSchema, tListLen(userIdxSchema)},
     {TSDB_INS_TABLE_USER_STABLES, userStbsSchema, tListLen(userStbsSchema)},
-    {TSDB_PERFS_TABLE_STREAMS, streamSchema, tListLen(streamSchema)},
     {TSDB_INS_TABLE_USER_TABLES, userTblsSchema, tListLen(userTblsSchema)},
     {TSDB_INS_TABLE_USER_TAGS, userTagsSchema, tListLen(userTagsSchema)},
     // {TSDB_INS_TABLE_USER_TABLE_DISTRIBUTED, userTblDistSchema, tListLen(userTblDistSchema)},

source/common/src/tdatablock.c

@@ -1713,7 +1713,7 @@ void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag) {
     size_t       numOfCols = taosArrayGetSize(pDataBlock->pDataBlock);
 
     int32_t rows = pDataBlock->info.rows;
-    printf("%s |block type %d |child id %d|group id %zX\n", flag, (int32_t)pDataBlock->info.type,
+    printf("%s |block type %d |child id %d|group id %" PRIu64 "\n", flag, (int32_t)pDataBlock->info.type,
            pDataBlock->info.childId, pDataBlock->info.groupId);
     for (int32_t j = 0; j < rows; j++) {
       printf("%s |", flag);

source/common/src/tglobal.c

@@ -35,7 +35,6 @@ int32_t  tsNumOfSupportVnodes = 256;
 // common
 int32_t tsMaxShellConns = 50000;
 int32_t tsShellActivityTimer = 3;  // second
-bool    tsEnableSlaveQuery = true;
 bool    tsPrintAuth = false;
 
 // multi process
@@ -118,20 +117,6 @@ int32_t tsMaxNumOfDistinctResults = 1000 * 10000;
 // 1 database precision unit for interval time range, changed accordingly
 int32_t tsMinIntervalTime = 1;
 
-// 20sec, the maximum value of stream computing delay, changed accordingly
-int32_t tsMaxStreamComputDelay = 20000;
-
-// 10sec, the first stream computing delay time after system launched successfully, changed accordingly
-int32_t tsStreamCompStartDelay = 10000;
-
-// the stream computing delay time after executing failed, change accordingly
-int32_t tsRetryStreamCompDelay = 10 * 1000;
-
-// The delayed computing ration. 10% of the whole computing time window by default.
-float tsStreamComputDelayRatio = 0.1f;
-
-int64_t tsMaxRetentWindow = 24 * 3600L;  // maximum time window tolerance
-
 // the maximum allowed query buffer size during query processing for each data node.
 // -1 no limit (default)
 // 0  no query allowed, queries are disabled
@@ -139,15 +124,6 @@ int64_t tsMaxRetentWindow = 24 * 3600L;  // maximum time window tolerance
 int32_t tsQueryBufferSize = -1;
 int64_t tsQueryBufferSizeBytes = -1;
 
-// in retrieve blocking model, the retrieve threads will wait for the completion of the query processing.
-bool tsRetrieveBlockingModel = false;
-
-// last_row(*), first(*), last_row(ts, col1, col2) query, the result fields will be the original column name
-bool tsKeepOriginalColumnName = false;
-
-// kill long query
-bool tsDeadLockKillQuery = false;
-
 // tsdb config
 // For backward compatibility
 bool tsdbForceKeepFile = false;
@@ -317,6 +293,7 @@ static int32_t taosAddServerLogCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "smaDebugFlag", smaDebugFlag, 0, 255, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "idxDebugFlag", idxDebugFlag, 0, 255, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "tdbDebugFlag", tdbDebugFlag, 0, 255, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "metaDebugFlag", metaDebugFlag, 0, 255, 0) != 0) return -1;
   return 0;
 }
 
@@ -330,11 +307,10 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
   if (cfgAddString(pCfg, "fqdn", defaultFqdn, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "serverPort", defaultServerPort, 1, 65056, 1) != 0) return -1;
   if (cfgAddDir(pCfg, "tempDir", tsTempDir, 1) != 0) return -1;
-  if (cfgAddFloat(pCfg, "minimalTempDirGB", 1.0f, 0.001f, 10000000, 1) != 0) return -1;
+  if (cfgAddFloat(pCfg, "minimalTmpDirGB", 1.0f, 0.001f, 10000000, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "shellActivityTimer", tsShellActivityTimer, 1, 120, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "compressMsgSize", tsCompressMsgSize, -1, 100000000, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "compressColData", tsCompressColData, -1, 100000000, 1) != 0) return -1;
-  if (cfgAddBool(pCfg, "keepColumnName", tsKeepOriginalColumnName, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "queryPolicy", tsQueryPolicy, 1, 3, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "querySmaOptimize", tsQuerySmaOptimize, 0, 1, 1) != 0) return -1;
   if (cfgAddString(pCfg, "smlChildTableName", "", 1) != 0) return -1;
@@ -383,15 +359,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "minIntervalTime", tsMinIntervalTime, 1, 1000000, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "maxNumOfDistinctRes", tsMaxNumOfDistinctResults, 10 * 10000, 10000 * 10000, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "countAlwaysReturnValue", tsCountAlwaysReturnValue, 0, 1, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "maxStreamCompDelay", tsMaxStreamComputDelay, 10, 1000000000, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "maxFirstStreamCompDelay", tsStreamCompStartDelay, 1000, 1000000000, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "retryStreamCompDelay", tsRetryStreamCompDelay, 10, 1000000000, 0) != 0) return -1;
-  if (cfgAddFloat(pCfg, "streamCompDelayRatio", tsStreamComputDelayRatio, 0.1, 0.9, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "queryBufferSize", tsQueryBufferSize, -1, 500000000000, 0) != 0) return -1;
-  if (cfgAddBool(pCfg, "retrieveBlockingModel", tsRetrieveBlockingModel, 0) != 0) return -1;
   if (cfgAddBool(pCfg, "printAuth", tsPrintAuth, 0) != 0) return -1;
-  if (cfgAddBool(pCfg, "slaveQuery", tsEnableSlaveQuery, 0) != 0) return -1;
-  if (cfgAddBool(pCfg, "deadLockKillQuery", tsDeadLockKillQuery, 0) != 0) return -1;
 
   if (cfgAddInt32(pCfg, "multiProcess", tsMultiProcess, 0, 2, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "mnodeShmSize", tsMnodeShmSize, TSDB_MAX_MSG_SIZE * 2 + 1024, INT32_MAX, 0) != 0) return -1;
@@ -399,7 +368,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "qnodeShmSize", tsQnodeShmSize, TSDB_MAX_MSG_SIZE * 2 + 1024, INT32_MAX, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "snodeShmSize", tsSnodeShmSize, TSDB_MAX_MSG_SIZE * 2 + 1024, INT32_MAX, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "bnodeShmSize", tsBnodeShmSize, TSDB_MAX_MSG_SIZE * 2 + 1024, INT32_MAX, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "mumOfShmThreads", tsNumOfShmThreads, 1, 1024, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "numOfShmThreads", tsNumOfShmThreads, 1, 1024, 0) != 0) return -1;
 
   tsNumOfRpcThreads = tsNumOfCores / 2;
   tsNumOfRpcThreads = TRANGE(tsNumOfRpcThreads, 1, 4);
@@ -409,25 +378,21 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   tsNumOfCommitThreads = TRANGE(tsNumOfCommitThreads, 2, 4);
   if (cfgAddInt32(pCfg, "numOfCommitThreads", tsNumOfCommitThreads, 1, 1024, 0) != 0) return -1;
 
-  tsNumOfMnodeQueryThreads = tsNumOfCores * 2;
-  tsNumOfMnodeQueryThreads = TRANGE(tsNumOfMnodeQueryThreads, 4, 8);
-  if (cfgAddInt32(pCfg, "numOfMnodeQueryThreads", tsNumOfMnodeQueryThreads, 1, 1024, 0) != 0) return -1;
-
   tsNumOfMnodeReadThreads = tsNumOfCores / 8;
   tsNumOfMnodeReadThreads = TRANGE(tsNumOfMnodeReadThreads, 1, 4);
   if (cfgAddInt32(pCfg, "numOfMnodeReadThreads", tsNumOfMnodeReadThreads, 1, 1024, 0) != 0) return -1;
 
   tsNumOfVnodeQueryThreads = tsNumOfCores * 2;
   tsNumOfVnodeQueryThreads = TMAX(tsNumOfVnodeQueryThreads, 4);
-  if (cfgAddInt32(pCfg, "numOfVnodeQueryThreads", tsNumOfVnodeQueryThreads, 1, 1024, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "numOfVnodeQueryThreads", tsNumOfVnodeQueryThreads, 4, 1024, 0) != 0) return -1;
 
   tsNumOfVnodeStreamThreads = tsNumOfCores / 4;
   tsNumOfVnodeStreamThreads = TMAX(tsNumOfVnodeStreamThreads, 4);
-  if (cfgAddInt32(pCfg, "numOfVnodeStreamThreads", tsNumOfVnodeStreamThreads, 1, 1024, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "numOfVnodeStreamThreads", tsNumOfVnodeStreamThreads, 4, 1024, 0) != 0) return -1;
 
   tsNumOfVnodeFetchThreads = tsNumOfCores / 4;
   tsNumOfVnodeFetchThreads = TMAX(tsNumOfVnodeFetchThreads, 4);
-  if (cfgAddInt32(pCfg, "numOfVnodeFetchThreads", tsNumOfVnodeFetchThreads, 1, 1024, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "numOfVnodeFetchThreads", tsNumOfVnodeFetchThreads, 4, 1024, 0) != 0) return -1;
 
   tsNumOfVnodeWriteThreads = tsNumOfCores;
   tsNumOfVnodeWriteThreads = TMAX(tsNumOfVnodeWriteThreads, 1);
@@ -447,11 +412,11 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
 
   tsNumOfSnodeSharedThreads = tsNumOfCores / 4;
   tsNumOfSnodeSharedThreads = TRANGE(tsNumOfSnodeSharedThreads, 2, 4);
-  if (cfgAddInt32(pCfg, "numOfSnodeSharedThreads", tsNumOfSnodeSharedThreads, 1, 1024, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "numOfSnodeSharedThreads", tsNumOfSnodeSharedThreads, 2, 1024, 0) != 0) return -1;
 
   tsNumOfSnodeUniqueThreads = tsNumOfCores / 4;
   tsNumOfSnodeUniqueThreads = TRANGE(tsNumOfSnodeUniqueThreads, 2, 4);
-  if (cfgAddInt32(pCfg, "numOfSnodeUniqueThreads", tsNumOfSnodeUniqueThreads, 1, 1024, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "numOfSnodeUniqueThreads", tsNumOfSnodeUniqueThreads, 2, 1024, 0) != 0) return -1;
 
   tsRpcQueueMemoryAllowed = tsTotalMemoryKB * 1024 * 0.1;
   tsRpcQueueMemoryAllowed = TRANGE(tsRpcQueueMemoryAllowed, TSDB_MAX_MSG_SIZE * 10L, TSDB_MAX_MSG_SIZE * 10000L);
@@ -508,6 +473,7 @@ static void taosSetServerLogCfg(SConfig *pCfg) {
   smaDebugFlag = cfgGetItem(pCfg, "smaDebugFlag")->i32;
   idxDebugFlag = cfgGetItem(pCfg, "idxDebugFlag")->i32;
   tdbDebugFlag = cfgGetItem(pCfg, "tdbDebugFlag")->i32;
+  metaDebugFlag = cfgGetItem(pCfg, "metaDebugFlag")->i32;
 }
 
 static int32_t taosSetClientCfg(SConfig *pCfg) {
@@ -532,7 +498,7 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
 
   tstrncpy(tsTempDir, cfgGetItem(pCfg, "tempDir")->str, PATH_MAX);
   taosExpandDir(tsTempDir, tsTempDir, PATH_MAX);
-  tsTempSpace.reserved = cfgGetItem(pCfg, "minimalTempDirGB")->fval;
+  tsTempSpace.reserved = cfgGetItem(pCfg, "minimalTmpDirGB")->fval;
   if (taosMulMkDir(tsTempDir) != 0) {
     uError("failed to create tempDir:%s since %s", tsTempDir, terrstr());
     return -1;
@@ -545,7 +511,6 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
   tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32;
   tsCompressMsgSize = cfgGetItem(pCfg, "compressMsgSize")->i32;
   tsCompressColData = cfgGetItem(pCfg, "compressColData")->i32;
-  tsKeepOriginalColumnName = cfgGetItem(pCfg, "keepColumnName")->bval;
   tsNumOfTaskQueueThreads = cfgGetItem(pCfg, "numOfTaskQueueThreads")->i32;
   tsQueryPolicy = cfgGetItem(pCfg, "queryPolicy")->i32;
   tsQuerySmaOptimize = cfgGetItem(pCfg, "querySmaOptimize")->i32;
@@ -579,15 +544,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
   tsMinIntervalTime = cfgGetItem(pCfg, "minIntervalTime")->i32;
   tsMaxNumOfDistinctResults = cfgGetItem(pCfg, "maxNumOfDistinctRes")->i32;
   tsCountAlwaysReturnValue = cfgGetItem(pCfg, "countAlwaysReturnValue")->i32;
-  tsMaxStreamComputDelay = cfgGetItem(pCfg, "maxStreamCompDelay")->i32;
-  tsStreamCompStartDelay = cfgGetItem(pCfg, "maxFirstStreamCompDelay")->i32;
-  tsRetryStreamCompDelay = cfgGetItem(pCfg, "retryStreamCompDelay")->i32;
-  tsStreamComputDelayRatio = cfgGetItem(pCfg, "streamCompDelayRatio")->fval;
   tsQueryBufferSize = cfgGetItem(pCfg, "queryBufferSize")->i32;
-  tsRetrieveBlockingModel = cfgGetItem(pCfg, "retrieveBlockingModel")->bval;
   tsPrintAuth = cfgGetItem(pCfg, "printAuth")->bval;
-  tsEnableSlaveQuery = cfgGetItem(pCfg, "slaveQuery")->bval;
-  tsDeadLockKillQuery = cfgGetItem(pCfg, "deadLockKillQuery")->i32;
 
   tsMultiProcess = cfgGetItem(pCfg, "multiProcess")->bval;
   tsMnodeShmSize = cfgGetItem(pCfg, "mnodeShmSize")->i32;
@@ -598,7 +556,6 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
 
   tsNumOfRpcThreads = cfgGetItem(pCfg, "numOfRpcThreads")->i32;
   tsNumOfCommitThreads = cfgGetItem(pCfg, "numOfCommitThreads")->i32;
-  tsNumOfMnodeQueryThreads = cfgGetItem(pCfg, "numOfMnodeQueryThreads")->i32;
   tsNumOfMnodeReadThreads = cfgGetItem(pCfg, "numOfMnodeReadThreads")->i32;
   tsNumOfVnodeQueryThreads = cfgGetItem(pCfg, "numOfVnodeQueryThreads")->i32;
   tsNumOfVnodeStreamThreads = cfgGetItem(pCfg, "numOfVnodeStreamThreads")->i32;
@@ -673,9 +630,7 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
       break;
     }
     case 'd': {
-      if (strcasecmp("deadLockKillQuery", name) == 0) {
-        tsDeadLockKillQuery = cfgGetItem(pCfg, "deadLockKillQuery")->i32;
-      } else if (strcasecmp("dDebugFlag", name) == 0) {
+      if (strcasecmp("dDebugFlag", name) == 0) {
         dDebugFlag = cfgGetItem(pCfg, "dDebugFlag")->i32;
       }
       break;
@@ -732,9 +687,6 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
       break;
     }
     case 'k': {
-      if (strcasecmp("keepColumnName", name) == 0) {
-        tsKeepOriginalColumnName = cfgGetItem(pCfg, "keepColumnName")->bval;
-      }
       break;
     }
     case 'l': {
@@ -758,10 +710,6 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
             tsMaxShellConns = cfgGetItem(pCfg, "maxShellConns")->i32;
           } else if (strcasecmp("maxNumOfDistinctRes", name) == 0) {
             tsMaxNumOfDistinctResults = cfgGetItem(pCfg, "maxNumOfDistinctRes")->i32;
-          } else if (strcasecmp("maxStreamCompDelay", name) == 0) {
-            tsMaxStreamComputDelay = cfgGetItem(pCfg, "maxStreamCompDelay")->i32;
-          } else if (strcasecmp("maxFirstStreamCompDelay", name) == 0) {
-            tsStreamCompStartDelay = cfgGetItem(pCfg, "maxFirstStreamCompDelay")->i32;
           }
           break;
         }
@@ -772,8 +720,8 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
           break;
         }
         case 'i': {
-          if (strcasecmp("minimalTempDirGB", name) == 0) {
-            tsTempSpace.reserved = cfgGetItem(pCfg, "minimalTempDirGB")->fval;
+          if (strcasecmp("minimalTmpDirGB", name) == 0) {
+            tsTempSpace.reserved = cfgGetItem(pCfg, "minimalTmpDirGB")->fval;
           } else if (strcasecmp("minimalDataDirGB", name) == 0) {
             tsDataSpace.reserved = cfgGetItem(pCfg, "minimalDataDirGB")->fval;
           } else if (strcasecmp("minSlidingTime", name) == 0) {
@@ -834,8 +782,6 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
         tsNumOfRpcThreads = cfgGetItem(pCfg, "numOfRpcThreads")->i32;
       } else if (strcasecmp("numOfCommitThreads", name) == 0) {
         tsNumOfCommitThreads = cfgGetItem(pCfg, "numOfCommitThreads")->i32;
-      } else if (strcasecmp("numOfMnodeQueryThreads", name) == 0) {
-        tsNumOfMnodeQueryThreads = cfgGetItem(pCfg, "numOfMnodeQueryThreads")->i32;
       } else if (strcasecmp("numOfMnodeReadThreads", name) == 0) {
         tsNumOfMnodeReadThreads = cfgGetItem(pCfg, "numOfMnodeReadThreads")->i32;
       } else if (strcasecmp("numOfVnodeQueryThreads", name) == 0) {
@@ -883,11 +829,7 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
       break;
     }
     case 'r': {
-      if (strcasecmp("retryStreamCompDelay", name) == 0) {
-        tsRetryStreamCompDelay = cfgGetItem(pCfg, "retryStreamCompDelay")->i32;
-      } else if (strcasecmp("retrieveBlockingModel", name) == 0) {
-        tsRetrieveBlockingModel = cfgGetItem(pCfg, "retrieveBlockingModel")->bval;
-      } else if (strcasecmp("rpcQueueMemoryAllowed", name) == 0) {
+      if (strcasecmp("rpcQueueMemoryAllowed", name) == 0) {
         tsRpcQueueMemoryAllowed = cfgGetItem(pCfg, "rpcQueueMemoryAllowed")->i64;
       } else if (strcasecmp("rpcDebugFlag", name) == 0) {
         rpcDebugFlag = cfgGetItem(pCfg, "rpcDebugFlag")->i32;
@@ -913,10 +855,6 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
         tsNumOfSupportVnodes = cfgGetItem(pCfg, "supportVnodes")->i32;
       } else if (strcasecmp("statusInterval", name) == 0) {
         tsStatusInterval = cfgGetItem(pCfg, "statusInterval")->i32;
-      } else if (strcasecmp("streamCompDelayRatio", name) == 0) {
-        tsStreamComputDelayRatio = cfgGetItem(pCfg, "streamCompDelayRatio")->fval;
-      } else if (strcasecmp("slaveQuery", name) == 0) {
-        tsEnableSlaveQuery = cfgGetItem(pCfg, "slaveQuery")->bval;
       } else if (strcasecmp("snodeShmSize", name) == 0) {
         tsSnodeShmSize = cfgGetItem(pCfg, "snodeShmSize")->i32;
       } else if (strcasecmp("serverPort", name) == 0) {
@@ -1157,12 +1095,12 @@ void taosCfgDynamicOptions(const char *option, const char *value) {
   const char *options[] = {
       "dDebugFlag",   "vDebugFlag",  "mDebugFlag",   "wDebugFlag",   "sDebugFlag",   "tsdbDebugFlag",
       "tqDebugFlag",  "fsDebugFlag", "udfDebugFlag", "smaDebugFlag", "idxDebugFlag", "tdbDebugFlag",
-      "tmrDebugFlag", "uDebugFlag",  "smaDebugFlag", "rpcDebugFlag", "qDebugFlag",
+      "tmrDebugFlag", "uDebugFlag",  "smaDebugFlag", "rpcDebugFlag", "qDebugFlag", "metaDebugFlag",
   };
   int32_t *optionVars[] = {
       &dDebugFlag,   &vDebugFlag,  &mDebugFlag,   &wDebugFlag,   &sDebugFlag,   &tsdbDebugFlag,
       &tqDebugFlag,  &fsDebugFlag, &udfDebugFlag, &smaDebugFlag, &idxDebugFlag, &tdbDebugFlag,
-      &tmrDebugFlag, &uDebugFlag,  &smaDebugFlag, &rpcDebugFlag, &qDebugFlag,
+      &tmrDebugFlag, &uDebugFlag,  &smaDebugFlag, &rpcDebugFlag, &qDebugFlag, &metaDebugFlag,
   };
 
   int32_t optionSize = tListLen(options);
@@ -1209,5 +1147,6 @@ void taosSetAllDebugFlag(int32_t flag) {
   taosSetDebugFlag(&smaDebugFlag, "smaDebugFlag", flag);
   taosSetDebugFlag(&idxDebugFlag, "idxDebugFlag", flag);
   taosSetDebugFlag(&tdbDebugFlag, "tdbDebugFlag", flag);
+  taosSetDebugFlag(&metaDebugFlag, "metaDebugFlag", flag);
   uInfo("all debug flag are set to %d", flag);
 }

source/common/src/tmsg.c

@@ -306,6 +306,8 @@ static int32_t tDeserializeSClientHbReq(SDecoder *pDecoder, SClientHbReq *pReq)
             }
           }
           
+          ASSERT(desc.subPlanNum == taosArrayGetSize(desc.subDesc));
+
           taosArrayPush(pReq->query->queryDesc, &desc);
         }
       }
@@ -5682,6 +5684,7 @@ int32_t tEncodeDeleteRes(SEncoder *pCoder, const SDeleteRes *pRes) {
   if (tEncodeI64v(pCoder, pRes->affectedRows) < 0) return -1;
 
   if (tEncodeCStr(pCoder, pRes->tableFName) < 0) return -1;
+  if (tEncodeCStr(pCoder, pRes->tsColName) < 0) return -1;
   return 0;
 }
 
@@ -5700,6 +5703,7 @@ int32_t tDecodeDeleteRes(SDecoder *pCoder, SDeleteRes *pRes) {
   if (tDecodeI64v(pCoder, &pRes->affectedRows) < 0) return -1;
 
   if (tDecodeCStrTo(pCoder, pRes->tableFName) < 0) return -1;
+  if (tDecodeCStrTo(pCoder, pRes->tsColName) < 0) return -1;
   return 0;
 }
 int32_t tEncodeSMqDataRsp(SEncoder *pEncoder, const SMqDataRsp *pRsp) {

source/dnode/mgmt/CMakeLists.txt

@@ -14,4 +14,7 @@ target_include_directories(
     taosd
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/node_mgmt/inc"
 )
+IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
+    add_dependencies(taosd jemalloc)
+ENDIF ()
 target_link_libraries(taosd dnode)

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

@@ -184,6 +184,7 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_STB, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_STB, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_TABLE_META, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_BATCH_META, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_TABLE_CFG, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_SMA, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_SMA, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

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

@@ -343,6 +343,7 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_UPDATE_TAG_VAL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_CFG, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_BATCH_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_TABLES_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SCH_CANCEL_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SCH_DROP_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;

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

@@ -89,7 +89,7 @@ static void vmProcessStreamQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo);
   if (code != 0) {
     if (terrno != 0) code = terrno;
-    dGError("vgId:%d, msg:%p failed to stream since %s", pVnode->vgId, pMsg, terrstr());
+    dGError("vgId:%d, msg:%p failed to process stream since %s", pVnode->vgId, pMsg, terrstr());
     vmSendRsp(pMsg, code);
   }
 

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

@@ -788,9 +788,9 @@ _OVER:
 static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
   SMnode     *pMnode = pReq->info.node;
   const char *options[] = {
-      "debugFlag",     "dDebugFlag",   "vDebugFlag",  "mDebugFlag",   "wDebugFlag",   "sDebugFlag",
-      "tsdbDebugFlag", "tqDebugFlag",  "fsDebugFlag", "udfDebugFlag", "smaDebugFlag", "idxDebugFlag",
-      "tdbDebugFlag",  "tmrDebugFlag", "uDebugFlag",  "smaDebugFlag", "rpcDebugFlag", "qDebugFlag",
+      "debugFlag",   "dDebugFlag",   "vDebugFlag",   "mDebugFlag",   "wDebugFlag",    "sDebugFlag",   "tsdbDebugFlag",
+      "tqDebugFlag", "fsDebugFlag",  "udfDebugFlag", "smaDebugFlag", "idxDebugFlag",  "tdbDebugFlag", "tmrDebugFlag",
+      "uDebugFlag",  "smaDebugFlag", "rpcDebugFlag", "qDebugFlag",   "metaDebugFlag",
   };
   int32_t optionSize = tListLen(options);
 
@@ -805,14 +805,6 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
     return -1;
   }
 
-  SDnodeObj *pDnode = mndAcquireDnode(pMnode, cfgReq.dnodeId);
-  if (pDnode == NULL) {
-    mError("dnode:%d, failed to config since %s ", cfgReq.dnodeId, terrstr());
-    return -1;
-  }
-  SEpSet epSet = mndGetDnodeEpset(pDnode);
-  mndReleaseDnode(pMnode, pDnode);
-
   SDCfgDnodeReq dcfgReq = {0};
   if (strcasecmp(cfgReq.config, "resetlog") == 0) {
     strcpy(dcfgReq.config, "resetlog");
@@ -860,16 +852,36 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
     }
   }
 
+  int32_t code = -1;
+  SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
+  while (1) {
+    SDnodeObj *pDnode = NULL;
+    pIter = sdbFetch(pSdb, SDB_DNODE, pIter, (void **)&pDnode);
+    if (pIter == NULL) break;
+
+    if (pDnode->id == cfgReq.dnodeId || cfgReq.dnodeId == -1 || cfgReq.dnodeId == 0) {
+      SEpSet  epSet = mndGetDnodeEpset(pDnode);
       int32_t bufLen = tSerializeSDCfgDnodeReq(NULL, 0, &dcfgReq);
       void   *pBuf = rpcMallocCont(bufLen);
 
-  if (pBuf == NULL) return -1;
+      if (pBuf != NULL) {
         tSerializeSDCfgDnodeReq(pBuf, bufLen, &dcfgReq);
-
         mInfo("dnode:%d, send config req to dnode, app:%p config:%s value:%s", cfgReq.dnodeId, pReq->info.ahandle,
               dcfgReq.config, dcfgReq.value);
         SRpcMsg rpcMsg = {.msgType = TDMT_DND_CONFIG_DNODE, .pCont = pBuf, .contLen = bufLen};
-  return tmsgSendReq(&epSet, &rpcMsg);
+        tmsgSendReq(&epSet, &rpcMsg);
+        code = 0;
+      }
+    }
+
+    sdbRelease(pSdb, pDnode);
+  }
+  
+  if (code == -1) {
+    terrno = TSDB_CODE_MND_DNODE_NOT_EXIST;
+  }
+  return code;
 }
 
 static int32_t mndProcessConfigDnodeRsp(SRpcMsg *pRsp) {

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

@@ -63,6 +63,106 @@ int32_t mndProcessQueryMsg(SRpcMsg *pMsg) {
   return code;
 }
 
+int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
+  int32_t code = 0;
+  int32_t offset = 0;
+  int32_t rspSize = 0;
+  SBatchReq *batchReq = (SBatchReq*)pMsg->pCont;
+  int32_t msgNum = ntohl(batchReq->msgNum);
+  offset += sizeof(SBatchReq);
+  SBatchMsg req = {0};
+  SBatchRsp rsp = {0};
+  SRpcMsg reqMsg = *pMsg;
+  SRpcMsg rspMsg = {0};
+  void* pRsp = NULL;
+  SMnode *pMnode = pMsg->info.node;
+
+  SArray* batchRsp = taosArrayInit(msgNum, sizeof(SBatchRsp));
+  if (NULL == batchRsp) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _exit;
+  }
+  
+  for (int32_t i = 0; i < msgNum; ++i) {
+    req.msgType = ntohl(*(int32_t*)((char*)pMsg->pCont + offset));
+    offset += sizeof(req.msgType);
+
+    req.msgLen = ntohl(*(int32_t*)((char*)pMsg->pCont + offset));
+    offset += sizeof(req.msgLen);
+
+    req.msg = (char*)pMsg->pCont + offset;
+    offset += req.msgLen;
+
+    reqMsg.msgType = req.msgType;
+    reqMsg.pCont = req.msg;
+    reqMsg.contLen = req.msgLen;
+    reqMsg.info.rsp = NULL;
+    reqMsg.info.rspLen = 0;
+
+    MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(req.msgType)];
+    if (fp == NULL) {
+      mError("msg:%p, failed to get msg handle, app:%p type:%s", pMsg, pMsg->info.ahandle, TMSG_INFO(pMsg->msgType));
+      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+      return -1;
+    }
+
+    if ((*fp)(&reqMsg)) {
+      rsp.rspCode = terrno;
+    } else {
+      rsp.rspCode = 0;
+    }
+    rsp.reqType = reqMsg.msgType;
+    rsp.msgLen = reqMsg.info.rspLen;
+    rsp.msg = reqMsg.info.rsp;
+    
+    taosArrayPush(batchRsp, &rsp);
+
+    rspSize += sizeof(rsp) + rsp.msgLen - POINTER_BYTES;
+  }
+
+  rspSize += sizeof(int32_t);
+  offset = 0;
+  
+  pRsp = rpcMallocCont(rspSize);
+  if (pRsp == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _exit;
+  }
+
+  *(int32_t*)((char*)pRsp + offset) = htonl(msgNum);
+  offset += sizeof(msgNum);
+  for (int32_t i = 0; i < msgNum; ++i) {
+    SBatchRsp *p = taosArrayGet(batchRsp, i);
+    
+    *(int32_t*)((char*)pRsp + offset) = htonl(p->reqType);
+    offset += sizeof(p->reqType);
+    *(int32_t*)((char*)pRsp + offset) = htonl(p->msgLen);
+    offset += sizeof(p->msgLen);
+    *(int32_t*)((char*)pRsp + offset) = htonl(p->rspCode);
+    offset += sizeof(p->rspCode);
+    memcpy((char*)pRsp + offset, p->msg, p->msgLen);
+    offset += p->msgLen;
+
+    rpcFreeCont(p->msg);
+  }
+
+  taosArrayDestroy(batchRsp);
+  batchRsp = NULL;
+
+_exit:
+
+  pMsg->info.rsp = pRsp;
+  pMsg->info.rspLen = rspSize;
+
+  if (code) {
+    mError("mnd get batch meta failed cause of %s", tstrerror(code));
+  }
+
+  taosArrayDestroyEx(batchRsp, tFreeSBatchRsp);
+
+  return code;
+}
+
 int32_t mndInitQuery(SMnode *pMnode) {
   if (qWorkerInit(NODE_TYPE_MNODE, MNODE_HANDLE, NULL, (void **)&pMnode->pQuery, &pMnode->msgCb) != 0) {
     mError("failed to init qworker in mnode since %s", terrstr());
@@ -76,6 +176,7 @@ int32_t mndInitQuery(SMnode *pMnode) {
   mndSetMsgHandle(pMnode, TDMT_SCH_MERGE_FETCH, mndProcessQueryMsg);
   mndSetMsgHandle(pMnode, TDMT_SCH_DROP_TASK, mndProcessQueryMsg);
   mndSetMsgHandle(pMnode, TDMT_SCH_QUERY_HEARTBEAT, mndProcessQueryMsg);
+  mndSetMsgHandle(pMnode, TDMT_MND_BATCH_META, mndProcessBatchMetaMsg);
 
   return 0;
 }

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

@@ -391,10 +391,12 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
       // exec
       pInnerTask->execType = TASK_EXEC__PIPE;
 
+#if 0
       SDbObj* pSourceDb = mndAcquireDb(pMnode, pStream->sourceDb);
       ASSERT(pDbObj != NULL);
       sdbRelease(pSdb, pSourceDb);
       pInnerTask->numOfVgroups = pSourceDb->cfg.numOfVgroups;
+#endif
 
       if (tsSchedStreamToSnode) {
         SSnodeObj* pSnode = mndSchedFetchOneSnode(pMnode);

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

@@ -1287,6 +1287,7 @@ static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans) {
     mDebug("trans:%d, stage keep on redoAction since %s", pTrans->id, tstrerror(code));
     continueExec = false;
   } else {
+    pTrans->failedTimes++;
     pTrans->code = terrno;
     if (pTrans->policy == TRN_POLICY_ROLLBACK) {
       if (pTrans->lastAction != 0) {
@@ -1306,7 +1307,6 @@ static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans) {
       mError("trans:%d, stage from redoAction to rollback since %s", pTrans->id, terrstr());
       continueExec = true;
     } else {
-      pTrans->failedTimes++;
       mError("trans:%d, stage keep on redoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
       continueExec = false;
     }

source/dnode/snode/src/snode.c

@@ -179,7 +179,7 @@ static int32_t sndProcessTaskRecoverRsp(SSnode *pNode, SRpcMsg *pMsg) {
   SStreamMeta *pMeta = pNode->pMeta;
 
   SStreamTaskRecoverRsp *pRsp = pMsg->pCont;
-  int32_t                taskId = pRsp->taskId;
+  int32_t                taskId = pRsp->rspTaskId;
   SStreamTask           *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t));
   streamProcessRecoverRsp(pTask, pRsp);
   return 0;

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

@@ -78,8 +78,9 @@ void    vnodeBufPoolReset(SVBufPool* pPool);
 // vnodeQuery.c
 int32_t vnodeQueryOpen(SVnode* pVnode);
 void    vnodeQueryClose(SVnode* pVnode);
-int32_t vnodeGetTableMeta(SVnode* pVnode, SRpcMsg* pMsg);
-int     vnodeGetTableCfg(SVnode* pVnode, SRpcMsg* pMsg);
+int32_t vnodeGetTableMeta(SVnode* pVnode, SRpcMsg* pMsg, bool direct);
+int     vnodeGetTableCfg(SVnode* pVnode, SRpcMsg* pMsg, bool direct);
+int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg);
 
 // vnodeCommit.c
 int32_t vnodeBegin(SVnode* pVnode);

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

@@ -293,7 +293,10 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   tdbTbcOpen(pMeta->pUidIdx, &pUidIdxc, &pMeta->txn);
   ret = tdbTbcMoveTo(pUidIdxc, &pReq->suid, sizeof(tb_uid_t), &c);
   if (ret < 0 || c) {
-    ASSERT(0);
+    tdbTbcClose(pUidIdxc);
+
+    terrno = TSDB_CODE_TDB_STB_NOT_EXIST;
+    // ASSERT(0);
     return -1;
   }
 
@@ -980,6 +983,9 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
   tbDbKey.version = pME->version;
   tbDbKey.uid = pME->uid;
 
+  metaDebug("vgId:%d, start to save table version:%" PRId64 "uid: %" PRId64, TD_VID(pMeta->pVnode), pME->version,
+            pME->uid);
+
   pKey = &tbDbKey;
   kLen = sizeof(tbDbKey);
 
@@ -1012,6 +1018,9 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
   return 0;
 
 _err:
+  metaError("vgId:%d, failed to save table version:%" PRId64 "uid: %" PRId64 " %s", TD_VID(pMeta->pVnode), pME->version,
+            pME->uid, tstrerror(terrno));
+
   taosMemoryFree(pVal);
   return -1;
 }

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

@@ -599,20 +599,20 @@ static int32_t tdRSmaFetchAndSubmitResult(SRSmaInfoItem *pItem, STSchema *pTSche
       SSubmitReq *pReq = NULL;
       // TODO: the schema update should be handled
       if (buildSubmitReqFromDataBlock(&pReq, pResult, pTSchema, SMA_VID(pSma), suid) < 0) {
-        smaError("vgId:%d, build submit req for rsma table %" PRIi64 "l evel %" PRIi8 " failed since %s", SMA_VID(pSma),
+        smaError("vgId:%d, build submit req for rsma stable %" PRIi64 " level %" PRIi8 " failed since %s", SMA_VID(pSma),
                  suid, pItem->level, terrstr());
         goto _err;
       }
 
       if (pReq && tdProcessSubmitReq(sinkTsdb, output->info.version, pReq) < 0) {
         taosMemoryFreeClear(pReq);
-        smaError("vgId:%d, process submit req for rsma table %" PRIi64 " level %" PRIi8 " failed since %s",
+        smaError("vgId:%d, process submit req for rsma stable %" PRIi64 " level %" PRIi8 " failed since %s",
                  SMA_VID(pSma), suid, pItem->level, terrstr());
         goto _err;
       }
 
-      smaDebug("vgId:%d, process submit req for rsma table %" PRIi64 " level %" PRIi8 " version:%"PRIi64, SMA_VID(pSma),
-               suid, pItem->level, output->info.version);
+      smaDebug("vgId:%d, process submit req for rsma table %" PRIi64 " level %" PRIi8 " version:%" PRIi64,
+               SMA_VID(pSma), suid, pItem->level, output->info.version);
 
       taosMemoryFreeClear(pReq);
       taosArrayClear(pResult);
@@ -644,7 +644,7 @@ static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t inputType
   smaDebug("vgId:%d, execute rsma %" PRIi8 " task for qTaskInfo:%p suid:%" PRIu64, SMA_VID(pSma), level,
            pItem->taskInfo, suid);
 
-  if (qSetStreamInput(pItem->taskInfo, pMsg, inputType, true) < 0) {  // INPUT__DATA_SUBMIT
+  if (qSetMultiStreamInput(pItem->taskInfo, pMsg, 1, inputType) < 0) {  // INPUT__DATA_SUBMIT
     smaError("vgId:%d, rsma % " PRIi8 " qSetStreamInput failed since %s", SMA_VID(pSma), level, tstrerror(terrno));
     return TSDB_CODE_FAILED;
   }
@@ -1329,7 +1329,7 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
       tdRefRSmaInfo(pSma, pRSmaInfo);
 
       SSDataBlock dataBlock = {.info.type = STREAM_GET_ALL};
-      qSetStreamInput(pItem->taskInfo, &dataBlock, STREAM_INPUT__DATA_BLOCK, false);
+      qSetMultiStreamInput(pItem->taskInfo, &dataBlock, 1, STREAM_INPUT__DATA_BLOCK);
       tdRSmaFetchAndSubmitResult(pItem, pRSmaInfo->pTSchema, pRSmaInfo->suid, pStat, STREAM_INPUT__DATA_BLOCK);
 
       tdUnRefRSmaInfo(pSma, pRSmaInfo);

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

@@ -653,7 +653,7 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, char* msg, int32_t msgLen) {
     } else {
       SReadHandle mgHandle = {
           .vnode = NULL,
-          .numOfVgroups = pTask->numOfVgroups,
+          .numOfVgroups = (int32_t)taosArrayGetSize(pTask->childEpInfo),
       };
       pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &mgHandle);
     }
@@ -796,7 +796,7 @@ int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) {
 
 int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg) {
   SStreamTaskRecoverRsp* pRsp = pMsg->pCont;
-  int32_t                taskId = pRsp->taskId;
+  int32_t                taskId = pRsp->rspTaskId;
   SStreamTask**          ppTask = (SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
   if (ppTask) {
     streamProcessRecoverRsp(*ppTask, pRsp);

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

@@ -374,11 +374,12 @@ static int32_t tsdbGetOrCreateTbData(SMemTable *pMemTable, tb_uid_t suid, tb_uid
   p = taosArrayInsert(pMemTable->aTbData, idx, &pTbData);
   taosWUnLockLatch(&pMemTable->latch);
 
+  tsdbDebug("vgId:%d add table data %p at idx:%d", TD_VID(pMemTable->pTsdb->pVnode), pTbData, idx);
+
   if (p == NULL) {
     code = TSDB_CODE_OUT_OF_MEMORY;
     goto _err;
   }
-
 _exit:
   *ppTbData = pTbData;
   return code;

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

@@ -694,8 +694,8 @@ static int32_t tsdbSnapWriteTableDataImpl(STsdbSnapWriter* pWriter) {
     if (pWriter->bDataW.nRow < pWriter->maxRow * 4 / 5) continue;
 
   _write_block:
-    code = tsdbWriteBlockData(pWriter->pDataFWriter, &pWriter->bDataW, NULL, NULL, pWriter->pBlockIdx, &pWriter->blockW,
-                              pWriter->cmprAlg);
+    code = tsdbWriteBlockData(pWriter->pDataFWriter, &pWriter->bDataW, NULL, NULL, pWriter->pBlockIdxW,
+                              &pWriter->blockW, pWriter->cmprAlg);
     if (code) goto _err;
 
     code = tMapDataPutItem(&pWriter->mBlockW, &pWriter->blockW, tPutBlock);
@@ -756,7 +756,7 @@ static int32_t tsdbSnapWriteTableData(STsdbSnapWriter* pWriter, TABLEID id) {
     if (pWriter->iBlockIdx < taosArrayGetSize(pWriter->aBlockIdx)) {
       ASSERT(pWriter->pDataFReader);
 
-      SBlockIdx* pBlockIdx = (SBlockIdx*)taosArrayGet(pWriter->aBlockIdx, pWriter->iBlock);
+      SBlockIdx* pBlockIdx = (SBlockIdx*)taosArrayGet(pWriter->aBlockIdx, pWriter->iBlockIdx);
       int32_t    c = tTABLEIDCmprFn(pBlockIdx, &id);
 
       ASSERT(c >= 0);
@@ -833,7 +833,7 @@ static int32_t tsdbSnapWriteDataEnd(STsdbSnapWriter* pWriter) {
   }
 
 _exit:
-  tsdbError("vgId:%d vnode snapshot tsdb writer data end", TD_VID(pTsdb->pVnode));
+  tsdbInfo("vgId:%d vnode snapshot tsdb writer data end", TD_VID(pTsdb->pVnode));
   return code;
 
 _err:

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

@@ -288,7 +288,7 @@ int vnodeCommit(SVnode *pVnode) {
   // apply the commit (TODO)
   walEndSnapshot(pVnode->pWal);
 
-  vInfo("vgId:%d, commit over", TD_VID(pVnode));
+  vInfo("vgId:%d, commit end", TD_VID(pVnode));
 
   return 0;
 }

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

@@ -180,6 +180,7 @@ struct SVSnapWriter {
   SVnode *pVnode;
   int64_t sver;
   int64_t ever;
+  int64_t commitID;
   int64_t index;
   // meta
   SMetaSnapWriter *pMetaSnapWriter;
@@ -201,7 +202,16 @@ int32_t vnodeSnapWriterOpen(SVnode *pVnode, int64_t sver, int64_t ever, SVSnapWr
   pWriter->sver = sver;
   pWriter->ever = ever;
 
-  vInfo("vgId:%d vnode snapshot writer opened", TD_VID(pVnode));
+  // commit it
+  code = vnodeCommit(pVnode);
+  if (code) goto _err;
+
+  // inc commit ID
+  pVnode->state.commitID++;
+  pWriter->commitID = pVnode->state.commitID;
+
+  vInfo("vgId:%d vnode snapshot writer opened, sver:%" PRId64 " ever:%" PRId64 " commit id:%" PRId64, TD_VID(pVnode),
+        sver, ever, pWriter->commitID);
   *ppWriter = pWriter;
   return code;
 
@@ -244,6 +254,8 @@ int32_t vnodeSnapWriterClose(SVSnapWriter *pWriter, int8_t rollback, SSnapshot *
 
     code = vnodeCommitInfo(dir, &info);
     if (code) goto _err;
+
+    vnodeBegin(pVnode);
   } else {
     ASSERT(0);
   }

source/libs/catalog/inc/ctgRemote.h

@@ -20,12 +20,6 @@
 extern "C" {
 #endif
 
-typedef struct SCtgTaskCallbackParam {
-  uint64_t                queryId;
-  int64_t                 refId;
-  uint64_t                taskId;
-  int32_t                 reqType;
-} SCtgTaskCallbackParam;
 
 
 #ifdef __cplusplus