Merge branch '3.0' into test/TD-22889-3.0

2023-03-22 17:19:36 +08:00 · 2023-03-22 17:19:36 +08:00 · 2b938ffea3
parent 8ac1e7f02c a0a51187bf
commit 2b938ffea3
48 changed files with 4565 additions and 3548 deletions
--- a/cmake/cmake.version
+++ b/cmake/cmake.version
@ -2,7 +2,7 @@
 IF (DEFINED VERNUMBER)
  SET(TD_VER_NUMBER ${VERNUMBER})
 ELSE ()
-  SET(TD_VER_NUMBER "3.0.3.0")
+  SET(TD_VER_NUMBER "3.0.3.1")
 ENDIF ()

 IF (DEFINED VERCOMPATIBLE)
--- a/docs/en/07-develop/07-tmq.mdx
+++ b/docs/en/07-develop/07-tmq.mdx
@ -23,7 +23,7 @@ By subscribing to a topic, a consumer can obtain the latest data in that topic i

 To implement these features, TDengine indexes its write-ahead log (WAL) file for fast random access and provides configurable methods for replacing and retaining this file. You can define a retention period and size for this file. For information, see the CREATE DATABASE statement. In this way, the WAL file is transformed into a persistent storage engine that remembers the order in which events occur. However, note that configuring an overly long retention period for your WAL files makes database compression inefficient. TDengine then uses the WAL file instead of the time-series database as its storage engine for queries in the form of topics. TDengine reads the data from the WAL file; uses a unified query engine instance to perform filtering, transformations, and other operations; and finally pushes the data to consumers.

-
+Tips:The default data subscription is to consume data from the wal. If the wal is deleted, the consumed data will be incomplete. At this time, you can set the parameter experimental.snapshot.enable to true to obtain all data from the tsdb, but in this way, the consumption order of the data cannot be guaranteed. Therefore, it is recommended to set a reasonable retention policy for WAL based on your consumption situation to ensure that you can subscribe all data from WAL.

 ## Data Schema and API

@ -285,18 +285,17 @@ You configure the following parameters when creating a consumer:

 |            Parameter            |  Type   | Description                                                 | Remarks                                        |
 | :----------------------------: | :-----: | -------------------------------------------------------- | ------------------------------------------- |
-|        `td.connect.ip`         | string  | Used in establishing a connection; same as `taos_connect`                          |                                             |
-|        `td.connect.user`         | string  | Used in establishing a connection; same as `taos_connect`                          |                                             |
-|        `td.connect.pass`         | string  | Used in establishing a connection; same as `taos_connect`                          |                                             |
-|        `td.connect.port`         | string  | Used in establishing a connection; same as `taos_connect`                          |                                             |
+|        `td.connect.ip`         | string  | Used in establishing a connection; same as `taos_connect`                          |  Only valid for establishing native connection                                           |
+|        `td.connect.user`         | string  | Used in establishing a connection; same as `taos_connect`                          |   Only valid for establishing native connection                                          |
+|        `td.connect.pass`         | string  | Used in establishing a connection; same as `taos_connect`                          |     Only valid for establishing native connection                                        |
+|        `td.connect.port`         | string  | Used in establishing a connection; same as `taos_connect`                          |       Only valid for establishing native connection                                      |
 |           `group.id`           | string  | Consumer group ID; consumers with the same ID are in the same group                        | **Required**. Maximum length: 192.                 |
 |          `client.id`           | string  | Client ID                                                | Maximum length: 192.                             |
 |      `auto.offset.reset`       |  enum   | Initial offset for the consumer group                                     | Specify `earliest`, `latest`, or `none`(default) |
-|      `enable.auto.commit`      | boolean | Commit automatically                                             | Specify `true` or `false`.                   |
+|      `enable.auto.commit`      | boolean | Commit automatically; true: user application doesn't need to explicitly commit; false: user application need to handle commit by itself                                           | Default value is true                  |
 |   `auto.commit.interval.ms`    | integer | Interval for automatic commits, in milliseconds                           |
-| `enable.heartbeat.background`  | boolean | Backend heartbeat; if enabled, the consumer does not go offline even if it has not polled for a long time |                                             |
-| `experimental.snapshot.enable` | boolean | Specify whether to consume messages from TSDB                    |                                             |
-|     `msg.with.table.name`      | boolean | Specify whether to deserialize table names from messages                                 |
+| `experimental.snapshot.enable` | boolean | Specify whether to consume data in TSDB; true: both data in WAL and in TSDB can be consumed; false: only data in WAL can be consumed                   |     default value: false                                        |
+|     `msg.with.table.name`      | boolean | Specify whether to deserialize table names from messages                                 | default value: false

 The method of specifying these parameters depends on the language used:

--- a/docs/en/12-taos-sql/06-select.md
+++ b/docs/en/12-taos-sql/06-select.md
@ -248,13 +248,13 @@ You can also use the NULLS keyword to specify the position of null values. Ascen

 The LIMIT keyword controls the number of results that are displayed. You can also use the OFFSET keyword to specify the result to display first. `LIMIT` and `OFFSET` are executed after `ORDER BY` in the query execution. You can include an offset in a LIMIT clause. For example, LIMIT 5 OFFSET 2 can also be written LIMIT 2, 5. Both of these clauses display the third through the seventh results.

-In a statement that includes a PARTITION BY clause, the LIMIT keyword is performed on each partition, not on the entire set of results.
+In a statement that includes a PARTITION BY/GROUP BY clause, the LIMIT keyword is performed on each partition/group, not on the entire set of results.

 ## SLIMIT

-The SLIMIT keyword is used with a PARTITION BY clause to control the number of partitions that are displayed. You can include an offset in a SLIMIT clause. For example, SLIMIT 5 OFFSET 2 can also be written LIMIT 2, 5. Both of these clauses display the third through the seventh partitions.
+The SLIMIT keyword is used with a PARTITION BY/GROUP BY clause to control the number of partitions/groups that are displayed. You can include an offset in a SLIMIT clause. For example, SLIMIT 5 OFFSET 2 can also be written LIMIT 2, 5. Both of these clauses display the third through the seventh partitions/groups.

-Note: If you include an ORDER BY clause, only one partition can be displayed.
+Note: If you include an ORDER BY clause, only one partition/group can be displayed.

 ## Special Query

--- a/docs/en/14-reference/03-connector/07-python.mdx
+++ b/docs/en/14-reference/03-connector/07-python.mdx
@ -10,10 +10,11 @@ import TabItem from "@theme/TabItem";
 `taospy` is the official Python connector for TDengine. taospy provides a rich API that makes it easy for Python applications to use TDengine. `taospy` wraps both the [native interface](/reference/connector/cpp) and [REST interface](/reference/rest-api) of TDengine, which correspond to the `taos` and `taosrest` modules of the `taospy` package, respectively.
 In addition to wrapping the native and REST interfaces, `taospy` also provides a set of programming interfaces that conforms to the [Python Data Access Specification (PEP 249)](https://peps.python.org/pep-0249/). It is easy to integrate `taospy` with many third-party tools, such as [SQLAlchemy](https://www.sqlalchemy.org/) and [pandas](https://pandas.pydata.org/).

-The direct connection to the server using the native interface provided by the client driver is referred to hereinafter as a "native connection"; the connection to the server using the REST interface provided by taosAdapter is referred to hereinafter as a "REST connection".
+`taos-ws-py` is an optional package to enable using WebSocket to connect TDengine.
+
+The direct connection to the server using the native interface provided by the client driver is referred to hereinafter as a "native connection"; the connection to the server using the REST or WebSocket interface provided by taosAdapter is referred to hereinafter as a "REST connection" or "WebSocket connection".

 The source code for the Python connector is hosted on [GitHub](https://github.com/taosdata/taos-connector-python).
-
 ## Supported platforms

 - The [supported platforms](/reference/connector/#supported-platforms) for the native connection are the same as the ones supported by the TDengine client.
@ -114,6 +115,15 @@ For REST connections, verifying that the `taosrest` module can be imported succe
 import taosrest
 ```

+</TabItem>
+<TabItem  value="ws" label="WebSocket connection">
+
+For WebSocket connection, verifying that the `taosws` module can be imported successfully can be done in the Python Interactive Shell by typing.
+
+```python
+import taosws
+```
+
 </TabItem>
 </Tabs>

@ -182,6 +192,28 @@ If the test is successful, it will output the server version information, e.g.
 }
 ```

+</TabItem>
+<TabItem value="ws" label="WebSocket connection" groupId="connect">
+
+For WebSocket connection, make sure the cluster and taosAdapter component, are running. This can be testetd using the following `curl` command.
+
+```
+curl -i -N -d "show databases" -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -H "Connection: Upgrade" -H "Upgrade: websocket" -H "Host: <FQDN>:<PORT>" -H "Origin: http://<FQDN>:<PORT>" http://<FQDN>:<PORT>/rest/sql
+```
+
+The FQDN above is the FQDN of the machine running taosAdapter, PORT is the port taosAdapter listening, default is `6041`.
+
+If the test is successful, it will output the server version information, e.g.
+
+```json
+HTTP/1.1 200 OK
+Content-Type: application/json; charset=utf-8
+Date: Tue, 21 Mar 2023 09:29:17 GMT
+Transfer-Encoding: chunked
+
+{"status":"succ","head":["server_version()"],"column_meta":[["server_version()",8,8]],"data":[["2.6.0.27"]],"rows":1}
+```
+
 </TabItem>
 </Tabs>

--- a/docs/en/14-reference/03-connector/_linux_install.mdx
+++ b/docs/en/14-reference/03-connector/_linux_install.mdx
@ -14,7 +14,7 @@ import PkgListV3 from "/components/PkgListV3";

   Once the package is unzipped, you will see the following files in the directory:
   - _ install_client.sh_: install script
-   - _ taos.tar.gz_: client driver package
+   - _ package.tar.gz_: client driver package
   - _ driver_: TDengine client driver
   - _examples_: some example programs of different programming languages (C/C#/go/JDBC/MATLAB/python/R)
   You can run `install_client.sh`  to install it.
--- a/docs/en/14-reference/14-taosKeeper.md
+++ b/docs/en/14-reference/14-taosKeeper.md
@ -4,6 +4,9 @@ title: taosKeeper
 description: This document describes how to use taosKeeper, a tool for exporting TDengine monitoring metrics.
 ---

+import Tabs from "@theme/Tabs";
+import TabItem from "@theme/TabItem";
+
 ## Introduction

 taosKeeper is a tool for TDengine that exports monitoring metrics. With taosKeeper, you can easily monitor the operational status of your TDengine deployment. taosKeeper uses the TDengine REST API. It is not necessary to install TDengine Client to use taosKeeper.
@ -18,9 +21,9 @@ Methods of installing taosKeeper:
 - You can compile taosKeeper separately and install it. Please refer to the [taosKeeper](https://github.com/taosdata/taoskeeper) repository for details. -->
 You can compile taosKeeper separately and install it. Please refer to the [taosKeeper](https://github.com/taosdata/taoskeeper) repository for details.

-## Run
+## Configuration and Launch

-### Configuration and running methods
+### Configuration

 taosKeeper needs to be executed on the terminal of the operating system, it supports three configuration methods: [Command-line arguments](#command-line-arguments-in-detail), [environment variable](#environment-variable-in-detail) and [configuration file](#configuration-file-parameters-in-detail). The precedence of those is Command-line, environment variable and configuration file.

@ -33,26 +36,79 @@ monitorFqdn localhost # taoskeeper's FQDN

 For more information, see [TDengine Monitoring Configuration](../config/#monitoring).

-### Command-Line Parameters
+### Quick Launch

-You can use command-line parameters to run taosKeeper and control its behavior:
+<Tabs>
+<TabItem label="Linux" value="linux">

-```shell
-$ taosKeeper
-```
-### Environment variable
+After the installation is complete, run the following command to start the taoskeeper service:

-You can use Environment variable to run taosKeeper and control its behavior:
-
-```shell
-$ export TAOS_KEEPER_TDENGINE_HOST=192.168.64.3
- 
-$ taoskeeper
+```bash
+systemctl start taoskeeper
 ```

-you can run `taoskeeper -h` for more detail.
+Run the following command to confirm that taoskeeper is running normally:

-### Configuration File
+```bash
+systemctl status taoskeeper
+```
+
+Output similar to the following indicates that taoskeeper is running normally:
+
+```
+Active: active (running)
+```
+
+Output similar to the following indicates that taoskeeper has not started successfully:
+
+```
+Active: inactive (dead)
+```
+
+The following `systemctl` commands can help you manage taoskeeper service:
+
+- Start taoskeeper Server: `systemctl start taoskeeper`
+
+- Stop taoskeeper Server: `systemctl stop taoskeeper`
+
+- Restart taoskeeper Server: `systemctl restart taoskeeper`
+
+- Check taoskeeper Server status: `systemctl status taoskeeper`
+
+:::info
+
+- The `systemctl` command requires _root_ privileges. If you are not logged in as the _root_ user, use the `sudo` command.
+- The `systemctl stop taoskeeper` command will instantly stop taoskeeper Server.
+- If your system does not include `systemd`, you can run `/usr/local/taos/bin/taoskeeper` to start taoskeeper manually.
+
+:::
+</TabItem>
+
+<TabItem label="macOS" value="macos">
+
+After the installation is complete, run `launchctl start com.tdengine.taoskeeper` to start taoskeeper Server.
+
+The following `launchctl` commands can help you manage taoskeeper service:
+
+- Start taoskeeper Server: `sudo launchctl start com.tdengine.taoskeeper`
+
+- Stop taoskeeper Server: `sudo launchctl stop com.tdengine.taoskeeper`
+
+- Check taoskeeper Server status: `sudo launchctl list | grep taoskeeper`
+
+:::info
+- Please use `sudo` to run `launchctl` to manage _com.tdengine.taoskeeper_ with administrator privileges.
+- The administrator privilege is required for service management to enhance security.
+- Troubleshooting:
+- The first column returned by the command `launchctl list | grep taoskeeper` is the PID of the program. If it's `-`, that means the taoskeeper service is not running.
+- If the service is abnormal, please check the `launchd.log` file from the system log.
+
+:::
+
+</TabItem>
+</Tabs>
+
+#### Launch With Configuration File

 You can quickly launch taosKeeper with the following commands. If you do not specify a configuration file, `/etc/taos/keeper.toml` is used by default. If this file does not specify configurations, the default values are used. 

--- a/docs/en/28-releases/01-tdengine.md
+++ b/docs/en/28-releases/01-tdengine.md
@ -10,6 +10,10 @@ For TDengine 2.x installation packages by version, please visit [here](https://w

 import Release from "/components/ReleaseV3";

+## 3.0.3.1
+
+<Release type="tdengine" version="3.0.3.1" />
+
 ## 3.0.3.0

 <Release type="tdengine" version="3.0.3.0" />
--- a/docs/en/28-releases/02-tools.md
+++ b/docs/en/28-releases/02-tools.md
@ -10,6 +10,10 @@ For other historical version installers, please visit [here](https://www.taosdat

 import Release from "/components/ReleaseV3";

+## 2.4.10
+
+<Release type="tools" version="2.4.10" />
+
 ## 2.4.9

 <Release type="tools" version="2.4.9" />
--- a/docs/zh/07-develop/07-tmq.mdx
+++ b/docs/zh/07-develop/07-tmq.mdx
@ -25,6 +25,7 @@ import CDemo from "./_sub_c.mdx";

 本文档不对消息队列本身的基础知识做介绍，如果需要了解，请自行搜索。

+注意：默认是从wal消费数据，如果wal被删除，消费到的数据会不全，此时可以将参数 experimental.snapshot.enable 设置为true，从tsdb获取全部数据，但是这样的话就不能保证数据的消费顺序。所以建议根据自己的消费情况合理的设置wal的保留策略，保证可以从wal里订阅到全部数据。
 ## 主要数据结构和 API

 不同语言下， TMQ 订阅相关的 API 及数据结构如下：
@ -283,18 +284,17 @@ CREATE TOPIC topic_name AS DATABASE db_name;

 |            参数名称            |  类型   | 参数说明                                                 | 备注                                        |
 | :----------------------------: | :-----: | -------------------------------------------------------- | ------------------------------------------- |
-|        `td.connect.ip`         | string  | 用于创建连接，同 `taos_connect`                          |                                             |
-|       `td.connect.user`        | string  | 用于创建连接，同 `taos_connect`                          |                                             |
-|       `td.connect.pass`        | string  | 用于创建连接，同 `taos_connect`                          |                                             |
-|       `td.connect.port`        | integer | 用于创建连接，同 `taos_connect`                          |                                             |
+|        `td.connect.ip`         | string  | 用于创建连接，同 `taos_connect`                          |     仅用于建立原生连接                                        |
+|       `td.connect.user`        | string  | 用于创建连接，同 `taos_connect`                          |   仅用于建立原生连接                                          |
+|       `td.connect.pass`        | string  | 用于创建连接，同 `taos_connect`                          |  仅用于建立原生连接                                           |
+|       `td.connect.port`        | integer | 用于创建连接，同 `taos_connect`                          |    仅用于建立原生连接                                         |
 |           `group.id`           | string  | 消费组 ID，同一消费组共享消费进度                        | **必填项**。最大长度：192。                 |
 |          `client.id`           | string  | 客户端 ID                                                | 最大长度：192。                             |
 |      `auto.offset.reset`       |  enum   | 消费组订阅的初始位置                                     | <br />`earliest`: default;从头开始订阅; <br/>`latest`: 仅从最新数据开始订阅; <br/>`none`: 没有提交的 offset 无法订阅 |
-|      `enable.auto.commit`      | boolean | 是否启用消费位点自动提交                                 | 合法值：`true`, `false`。                   |
-|   `auto.commit.interval.ms`    | integer | 以毫秒为单位的消费记录自动提交消费位点时间间隔             | 默认 5000 m                                 |
-| `enable.heartbeat.background`  | boolean | 启用后台心跳，启用后即使长时间不 poll 消息也不会造成离线 | 默认开启                                    |
-| `experimental.snapshot.enable` | boolean | 是否允许从 TSDB 消费数据                                 | 实验功能，默认关闭                          |
-|     `msg.with.table.name`      | boolean | 是否允许从消息中解析表名, 不适用于列订阅（列订阅时可将 tbname 作为列写入 subquery 语句）               | |
+|      `enable.auto.commit`      | boolean | 是否启用消费位点自动提交，true: 自动提交，客户端应用无需commit；false：客户端应用需要自行commit     | 默认值为 true                   |
+|   `auto.commit.interval.ms`    | integer | 消费记录自动提交消费位点时间间隔，单位为毫秒           | 默认值为 5000                                |
+| `experimental.snapshot.enable` | boolean | 是否允许从 TSDB 消费数据。当其关闭时，只能消费依据 WAL 保留策略仍然在WAL中的数据；当其打开时，除WAL中的数据以外，也能够消费已经从WAL中删除但落盘到TSDB中的数据                              | 实验功能，默认关闭                          |
+|     `msg.with.table.name`      | boolean | 是否允许从消息中解析表名, 不适用于列订阅（列订阅时可将 tbname 作为列写入 subquery 语句）               |默认关闭 |

 对于不同编程语言，其设置方式如下：

--- a/docs/zh/08-connector/30-python.mdx
+++ b/docs/zh/08-connector/30-python.mdx
@ -10,7 +10,9 @@ import TabItem from "@theme/TabItem";
 `taospy` 是 TDengine 的官方 Python 连接器。`taospy` 提供了丰富的 API， 使得 Python 应用可以很方便地使用 TDengine。`taospy` 对 TDengine 的[原生接口](../cpp)和 [REST 接口](../rest-api)都进行了封装， 分别对应 `taospy` 包的 `taos` 模块 和 `taosrest` 模块。
 除了对原生接口和 REST 接口的封装，`taospy` 还提供了符合 [Python 数据访问规范(PEP 249)](https://peps.python.org/pep-0249/) 的编程接口。这使得 `taospy` 和很多第三方工具集成变得简单，比如 [SQLAlchemy](https://www.sqlalchemy.org/) 和 [pandas](https://pandas.pydata.org/)。

-使用客户端驱动提供的原生接口直接与服务端建立的连接的方式下文中称为“原生连接”；使用 taosAdapter 提供的 REST 接口与服务端建立的连接的方式下文中称为“REST 连接”。
+`taos-ws-py` 是使用 WebSocket 方式连接 TDengine 的 Python 连接器包。可以选装。
+
+使用客户端驱动提供的原生接口直接与服务端建立的连接的方式下文中称为“原生连接”；使用 taosAdapter 提供的 REST 接口或 WebSocket 接口与服务端建立的连接的方式下文中称为“REST 连接”或“WebSocket 连接”。

 Python 连接器的源码托管在 [GitHub](https://github.com/taosdata/taos-connector-python)。

@ -115,6 +117,15 @@ import taos
 import taosrest
 ```

+</TabItem>
+<TabItem  value="ws" label="WebSocket 连接">
+
+对于 WebSocket 连接，只需验证是否能成功导入 `taosws` 模块。可在 Python 交互式 Shell 中输入：
+
+```python
+import taosws
+```
+
 </TabItem>
 </Tabs>

@ -183,6 +194,27 @@ curl -u root:taosdata http://<FQDN>:<PORT>/rest/sql -d "select server_version()"
 }
 ```

+</TabItem>
+<TabItem value="ws" label="WebSocket 连接" groupId="connect">
+
+对于 WebSocket 连接， 除了确保集群已经启动，还要确保 taosAdapter 组件已经启动。可以使用如下 curl 命令测试：
+
+```
+curl -i -N -d "show databases" -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -H "Connection: Upgrade" -H "Upgrade: websocket" -H "Host: <FQDN>:<PORT>" -H "Origin: http://<FQDN>:<PORT>" http://<FQDN>:<PORT>/rest/sql
+```
+
+上面的 FQDN 为运行 taosAdapter 的机器的 FQDN， PORT 为 taosAdapter 配置的监听端口， 默认为 6041。
+如果测试成功，会输出服务器版本信息，比如：
+
+```json
+HTTP/1.1 200 OK
+Content-Type: application/json; charset=utf-8
+Date: Tue, 21 Mar 2023 09:29:17 GMT
+Transfer-Encoding: chunked
+
+{"status":"succ","head":["server_version()"],"column_meta":[["server_version()",8,8]],"data":[["2.6.0.27"]],"rows":1}
+```
+
 </TabItem>
 </Tabs>

--- a/docs/zh/08-connector/_linux_install.mdx
+++ b/docs/zh/08-connector/_linux_install.mdx
@ -14,7 +14,7 @@ import PkgListV3 from "/components/PkgListV3";

   解压软件包之后，会在解压目录下看到以下文件(目录)：
   - _ install_client.sh_：安装脚本，用于应用驱动程序
-   - _ taos.tar.gz_：应用驱动安装包
+   - _ package.tar.gz_：应用驱动安装包
   - _ driver_：TDengine 应用驱动 driver
   - _examples_: 各种编程语言的示例程序(c/C#/go/JDBC/MATLAB/python/R)
   运行 install_client.sh 进行安装。
--- a/docs/zh/12-taos-sql/06-select.md
+++ b/docs/zh/12-taos-sql/06-select.md
@ -248,11 +248,11 @@ NULLS 语法用来指定 NULL 值在排序中输出的位置。NULLS LAST 是升

 LIMIT 控制输出条数，OFFSET 指定从第几条之后开始输出。LIMIT/OFFSET 对结果集的执行顺序在 ORDER BY 之后。LIMIT 5 OFFSET 2 可以简写为 LIMIT 2, 5，都输出第 3 行到第 7 行数据。

-在有 PARTITION BY 子句时，LIMIT 控制的是每个切分的分片中的输出，而不是总的结果集输出。
+在有 PARTITION BY/GROUP BY 子句时，LIMIT 控制的是每个切分的分片中的输出，而不是总的结果集输出。

 ## SLIMIT

-SLIMIT 和 PARTITION BY 子句一起使用，用来控制输出的分片的数量。SLIMIT 5 SOFFSET 2 可以简写为 SLIMIT 2, 5，都表示输出第 3 个到第 7 个分片。
+SLIMIT 和 PARTITION BY/GROUP BY 子句一起使用，用来控制输出的分片的数量。SLIMIT 5 SOFFSET 2 可以简写为 SLIMIT 2, 5，都表示输出第 3 个到第 7 个分片。

 需要注意，如果有 ORDER BY 子句，则输出只有一个分片。

--- a/docs/zh/14-reference/14-taosKeeper.md
+++ b/docs/zh/14-reference/14-taosKeeper.md
@ -4,6 +4,9 @@ title: taosKeeper
 description: TDengine 3.0 版本监控指标的导出工具
 ---

+import Tabs from "@theme/Tabs";
+import TabItem from "@theme/TabItem";
+
 ## 简介

 taosKeeper 是 TDengine 3.0 版本监控指标的导出工具，通过简单的几项配置即可获取 TDengine 的运行状态。taosKeeper 使用 TDengine RESTful 接口，所以不需要安装 TDengine 客户端即可使用。
@ -18,9 +21,9 @@ taosKeeper 安装方式：
 <!-- - 单独编译 taosKeeper 并安装，详情请参考 [taosKeeper](https://github.com/taosdata/taoskeeper) 仓库。-->
 - 单独编译 taosKeeper 并安装，详情请参考 [taosKeeper](https://github.com/taosdata/taoskeeper) 仓库。

-## 运行
+## 配置和运行方式

-### 配置和运行方式
+### 配置

 taosKeeper 需要在操作系统终端执行，该工具支持三种配置方式：[命令行参数](#命令行参数启动)、[环境变量](#环境变量启动) 和 [配置文件](#配置文件启动)。优先级为：命令行参数、环境变量、配置文件参数。

@ -34,27 +37,80 @@ monitorFqdn localhost # taoskeeper 服务的 FQDN
 TDengine 监控配置相关，具体请参考：[TDengine 监控配置](../config/#监控相关)。


-### 命令行参数启动
+### 启动

-在使用命令行参数运行 taosKeeper 并控制其行为。
+<Tabs>
+<TabItem label="Linux" value="linux">

-```shell
-$ taosKeeper
+安装后，请使用 `systemctl` 命令来启动 taoskeeper 的服务进程。
+
+```bash
+systemctl start taoskeeper
 ```

-### 环境变量启动
+检查服务是否正常工作：

-通过设置环境变量达到控制启动参数的目的，通常在容器中运行时使用。
-
-```shell
-$ export TAOS_KEEPER_TDENGINE_HOST=192.168.64.3
- 
-$ taoskeeper
+```bash
+systemctl status taoskeeper
 ```

-具体参数列表请参照 `taoskeeper -h` 输入结果。
+如果服务进程处于活动状态，则 status 指令会显示如下的相关信息：

-### 配置文件启动
+```
+Active: active (running)
+```
+
+如果后台服务进程处于停止状态，则 status 指令会显示如下的相关信息：
+
+```
+Active: inactive (dead)
+```
+
+如下 `systemctl` 命令可以帮助你管理 taoskeeper 服务：
+
+- 启动服务进程：`systemctl start taoskeeper`
+
+- 停止服务进程：`systemctl stop taoskeeper`
+
+- 重启服务进程：`systemctl restart taoskeeper`
+
+- 查看服务状态：`systemctl status taoskeeper`
+
+:::info
+
+- `systemctl` 命令需要 _root_ 权限来运行，如果您非 _root_ 用户，请在命令前添加 `sudo`。
+- 如果系统中不支持 `systemd`，也可以用手动运行 `/usr/local/taos/bin/taoskeeper` 方式启动 taoskeeper 服务。
+- 故障排查：
+- 如果服务异常请查看系统日志获取更多信息。
+:::
+</TabItem>
+
+<TabItem label="macOS" value="macOS">
+
+安装后，可以运行 `sudo launchctl start com.tdengine.taoskeeper` 来启动 taoskeeper 服务进程。
+
+如下 `launchctl` 命令用于管理 taoskeeper 服务：
+
+- 启动服务进程：`sudo launchctl start com.tdengine.taoskeeper`
+
+- 停止服务进程：`sudo launchctl stop com.tdengine.taoskeeper`
+
+- 查看服务状态：`sudo launchctl list | grep taoskeeper`
+
+:::info
+
+- `launchctl` 命令管理`com.tdengine.taoskeeper`需要管理员权限，务必在前面加 `sudo` 来增强安全性。
+- `sudo launchctl list | grep taoskeeper` 指令返回的第一列是 `taoskeeper` 程序的 PID，若为 `-` 则说明 taoskeeper 服务未运行。
+- 故障排查：
+- 如果服务异常请查看系统日志获取更多信息。
+
+:::
+
+</TabItem>
+</Tabs>
+
+
+#### 配置文件启动

 执行以下命令即可快速体验 taosKeeper。当不指定 taosKeeper 配置文件时，优先使用 `/etc/taos/keeper.toml` 配置，否则将使用默认配置。 

--- a/docs/zh/28-releases/01-tdengine.md
+++ b/docs/zh/28-releases/01-tdengine.md
@ -10,6 +10,10 @@ TDengine 2.x 各版本安装包请访问[这里](https://www.taosdata.com/all-do

 import Release from "/components/ReleaseV3";

+## 3.0.3.1
+
+<Release type="tdengine" version="3.0.3.1" />
+
 ## 3.0.3.0

 <Release type="tdengine" version="3.0.3.0" />
--- a/docs/zh/28-releases/02-tools.md
+++ b/docs/zh/28-releases/02-tools.md
@ -10,6 +10,10 @@ taosTools 各版本安装包下载链接如下：

 import Release from "/components/ReleaseV3";

+## 2.4.10
+
+<Release type="tools" version="2.4.10" />
+
 ## 2.4.9

 <Release type="tools" version="2.4.9" />
--- a/include/common/tmsg.h
+++ b/include/common/tmsg.h
@ -846,6 +846,8 @@ typedef struct {
  int8_t  replications;
  int32_t sstTrigger;
  int32_t minRows;
+  int32_t walRetentionPeriod;
+  int32_t walRetentionSize;
 } SAlterDbReq;

 int32_t tSerializeSAlterDbReq(void* buf, int32_t bufLen, SAlterDbReq* pReq);
@ -1655,6 +1657,20 @@ typedef struct {
 int32_t tSerializeSRedistributeVgroupReq(void* buf, int32_t bufLen, SRedistributeVgroupReq* pReq);
 int32_t tDeserializeSRedistributeVgroupReq(void* buf, int32_t bufLen, SRedistributeVgroupReq* pReq);

+typedef struct {
+  int32_t useless;
+} SBalanceVgroupLeaderReq;
+
+int32_t tSerializeSBalanceVgroupLeaderReq(void* buf, int32_t bufLen, SBalanceVgroupLeaderReq* pReq);
+int32_t tDeserializeSBalanceVgroupLeaderReq(void* buf, int32_t bufLen, SBalanceVgroupLeaderReq* pReq);
+
+typedef struct {
+  int32_t  vgId;
+} SForceElectionReq;
+
+int32_t tSerializeSForceElectionReq(void* buf, int32_t bufLen, SForceElectionReq* pReq);
+int32_t tDeserializeSForceElectionReq(void* buf, int32_t bufLen, SForceElectionReq* pReq);
+
 typedef struct {
  int32_t vgId;
 } SSplitVgroupReq;
--- a/include/common/tmsgdef.h
+++ b/include/common/tmsgdef.h
@ -83,6 +83,7 @@ enum {
  TD_DEF_MSG_TYPE(TDMT_DND_CONFIG_DNODE, "config-dnode", NULL, NULL)
  TD_DEF_MSG_TYPE(TDMT_DND_SYSTABLE_RETRIEVE, "dnode-retrieve", NULL, NULL)
  TD_DEF_MSG_TYPE(TDMT_DND_MAX_MSG, "dnd-max", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_DND_FORCE_ELECTION, "balance-force-election", NULL, NULL)

  TD_NEW_MSG_SEG(TDMT_MND_MSG)
  TD_DEF_MSG_TYPE(TDMT_MND_CONNECT, "connect", NULL, NULL)
@ -165,6 +166,7 @@ enum {
  TD_DEF_MSG_TYPE(TDMT_MND_AUTH, "auth", NULL, NULL)
  TD_DEF_MSG_TYPE(TDMT_MND_APPLY_MSG, "mnode-apply", NULL, NULL)
  TD_DEF_MSG_TYPE(TDMT_MND_BALANCE_VGROUP, "balance-vgroup", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_BALANCE_VGROUP_LEADER, "balance-vgroup-leader", NULL, NULL)
  TD_DEF_MSG_TYPE(TDMT_MND_MERGE_VGROUP, "merge-vgroup", NULL, NULL)
  TD_DEF_MSG_TYPE(TDMT_MND_REDISTRIBUTE_VGROUP, "redistribute-vgroup", NULL, NULL)
  TD_DEF_MSG_TYPE(TDMT_MND_SPLIT_VGROUP, "split-vgroup", NULL, NULL)
--- a/include/common/ttokendef.h
+++ b/include/common/ttokendef.h
@ -223,127 +223,128 @@
 #define TK_TRANSACTION          205
 #define TK_BALANCE              206
 #define TK_VGROUP               207
-#define TK_MERGE                208
-#define TK_REDISTRIBUTE         209
-#define TK_SPLIT                210
-#define TK_DELETE               211
-#define TK_INSERT               212
-#define TK_NULL                 213
-#define TK_NK_QUESTION          214
-#define TK_NK_ARROW             215
-#define TK_ROWTS                216
-#define TK_QSTART               217
-#define TK_QEND                 218
-#define TK_QDURATION            219
-#define TK_WSTART               220
-#define TK_WEND                 221
-#define TK_WDURATION            222
-#define TK_IROWTS               223
-#define TK_ISFILLED             224
-#define TK_CAST                 225
-#define TK_NOW                  226
-#define TK_TODAY                227
-#define TK_TIMEZONE             228
-#define TK_CLIENT_VERSION       229
-#define TK_SERVER_VERSION       230
-#define TK_SERVER_STATUS        231
-#define TK_CURRENT_USER         232
-#define TK_CASE                 233
-#define TK_WHEN                 234
-#define TK_THEN                 235
-#define TK_ELSE                 236
-#define TK_BETWEEN              237
-#define TK_IS                   238
-#define TK_NK_LT                239
-#define TK_NK_GT                240
-#define TK_NK_LE                241
-#define TK_NK_GE                242
-#define TK_NK_NE                243
-#define TK_MATCH                244
-#define TK_NMATCH               245
-#define TK_CONTAINS             246
-#define TK_IN                   247
-#define TK_JOIN                 248
-#define TK_INNER                249
-#define TK_SELECT               250
-#define TK_DISTINCT             251
-#define TK_WHERE                252
-#define TK_PARTITION            253
-#define TK_BY                   254
-#define TK_SESSION              255
-#define TK_STATE_WINDOW         256
-#define TK_EVENT_WINDOW         257
-#define TK_SLIDING              258
-#define TK_FILL                 259
-#define TK_VALUE                260
-#define TK_VALUE_F              261
-#define TK_NONE                 262
-#define TK_PREV                 263
-#define TK_NULL_F               264
-#define TK_LINEAR               265
-#define TK_NEXT                 266
-#define TK_HAVING               267
-#define TK_RANGE                268
-#define TK_EVERY                269
-#define TK_ORDER                270
-#define TK_SLIMIT               271
-#define TK_SOFFSET              272
-#define TK_LIMIT                273
-#define TK_OFFSET               274
-#define TK_ASC                  275
-#define TK_NULLS                276
-#define TK_ABORT                277
-#define TK_AFTER                278
-#define TK_ATTACH               279
-#define TK_BEFORE               280
-#define TK_BEGIN                281
-#define TK_BITAND               282
-#define TK_BITNOT               283
-#define TK_BITOR                284
-#define TK_BLOCKS               285
-#define TK_CHANGE               286
-#define TK_COMMA                287
-#define TK_CONCAT               288
-#define TK_CONFLICT             289
-#define TK_COPY                 290
-#define TK_DEFERRED             291
-#define TK_DELIMITERS           292
-#define TK_DETACH               293
-#define TK_DIVIDE               294
-#define TK_DOT                  295
-#define TK_EACH                 296
-#define TK_FAIL                 297
-#define TK_FILE                 298
-#define TK_FOR                  299
-#define TK_GLOB                 300
-#define TK_ID                   301
-#define TK_IMMEDIATE            302
-#define TK_IMPORT               303
-#define TK_INITIALLY            304
-#define TK_INSTEAD              305
-#define TK_ISNULL               306
-#define TK_KEY                  307
-#define TK_MODULES              308
-#define TK_NK_BITNOT            309
-#define TK_NK_SEMI              310
-#define TK_NOTNULL              311
-#define TK_OF                   312
-#define TK_PLUS                 313
-#define TK_PRIVILEGE            314
-#define TK_RAISE                315
-#define TK_REPLACE              316
-#define TK_RESTRICT             317
-#define TK_ROW                  318
-#define TK_SEMI                 319
-#define TK_STAR                 320
-#define TK_STATEMENT            321
-#define TK_STRICT               322
-#define TK_STRING               323
-#define TK_TIMES                324
-#define TK_VALUES               325
-#define TK_VARIABLE             326
-#define TK_VIEW                 327
-#define TK_WAL                  328
+#define TK_LEADER               208
+#define TK_MERGE                209
+#define TK_REDISTRIBUTE         210
+#define TK_SPLIT                211
+#define TK_DELETE               212
+#define TK_INSERT               213
+#define TK_NULL                 214
+#define TK_NK_QUESTION          215
+#define TK_NK_ARROW             216
+#define TK_ROWTS                217
+#define TK_QSTART               218
+#define TK_QEND                 219
+#define TK_QDURATION            220
+#define TK_WSTART               221
+#define TK_WEND                 222
+#define TK_WDURATION            223
+#define TK_IROWTS               224
+#define TK_ISFILLED             225
+#define TK_CAST                 226
+#define TK_NOW                  227
+#define TK_TODAY                228
+#define TK_TIMEZONE             229
+#define TK_CLIENT_VERSION       230
+#define TK_SERVER_VERSION       231
+#define TK_SERVER_STATUS        232
+#define TK_CURRENT_USER         233
+#define TK_CASE                 234
+#define TK_WHEN                 235
+#define TK_THEN                 236
+#define TK_ELSE                 237
+#define TK_BETWEEN              238
+#define TK_IS                   239
+#define TK_NK_LT                240
+#define TK_NK_GT                241
+#define TK_NK_LE                242
+#define TK_NK_GE                243
+#define TK_NK_NE                244
+#define TK_MATCH                245
+#define TK_NMATCH               246
+#define TK_CONTAINS             247
+#define TK_IN                   248
+#define TK_JOIN                 249
+#define TK_INNER                250
+#define TK_SELECT               251
+#define TK_DISTINCT             252
+#define TK_WHERE                253
+#define TK_PARTITION            254
+#define TK_BY                   255
+#define TK_SESSION              256
+#define TK_STATE_WINDOW         257
+#define TK_EVENT_WINDOW         258
+#define TK_SLIDING              259
+#define TK_FILL                 260
+#define TK_VALUE                261
+#define TK_VALUE_F              262
+#define TK_NONE                 263
+#define TK_PREV                 264
+#define TK_NULL_F               265
+#define TK_LINEAR               266
+#define TK_NEXT                 267
+#define TK_HAVING               268
+#define TK_RANGE                269
+#define TK_EVERY                270
+#define TK_ORDER                271
+#define TK_SLIMIT               272
+#define TK_SOFFSET              273
+#define TK_LIMIT                274
+#define TK_OFFSET               275
+#define TK_ASC                  276
+#define TK_NULLS                277
+#define TK_ABORT                278
+#define TK_AFTER                279
+#define TK_ATTACH               280
+#define TK_BEFORE               281
+#define TK_BEGIN                282
+#define TK_BITAND               283
+#define TK_BITNOT               284
+#define TK_BITOR                285
+#define TK_BLOCKS               286
+#define TK_CHANGE               287
+#define TK_COMMA                288
+#define TK_CONCAT               289
+#define TK_CONFLICT             290
+#define TK_COPY                 291
+#define TK_DEFERRED             292
+#define TK_DELIMITERS           293
+#define TK_DETACH               294
+#define TK_DIVIDE               295
+#define TK_DOT                  296
+#define TK_EACH                 297
+#define TK_FAIL                 298
+#define TK_FILE                 299
+#define TK_FOR                  300
+#define TK_GLOB                 301
+#define TK_ID                   302
+#define TK_IMMEDIATE            303
+#define TK_IMPORT               304
+#define TK_INITIALLY            305
+#define TK_INSTEAD              306
+#define TK_ISNULL               307
+#define TK_KEY                  308
+#define TK_MODULES              309
+#define TK_NK_BITNOT            310
+#define TK_NK_SEMI              311
+#define TK_NOTNULL              312
+#define TK_OF                   313
+#define TK_PLUS                 314
+#define TK_PRIVILEGE            315
+#define TK_RAISE                316
+#define TK_REPLACE              317
+#define TK_RESTRICT             318
+#define TK_ROW                  319
+#define TK_SEMI                 320
+#define TK_STAR                 321
+#define TK_STATEMENT            322
+#define TK_STRICT               323
+#define TK_STRING               324
+#define TK_TIMES                325
+#define TK_VALUES               326
+#define TK_VARIABLE             327
+#define TK_VIEW                 328
+#define TK_WAL                  329

 #define TK_NK_SPACE   600
 #define TK_NK_COMMENT 601
--- a/include/libs/nodes/cmdnodes.h
+++ b/include/libs/nodes/cmdnodes.h
@ -466,6 +466,10 @@ typedef struct SBalanceVgroupStmt {
  ENodeType type;
 } SBalanceVgroupStmt;

+typedef struct SBalanceVgroupLeaderStmt {
+  ENodeType type;
+} SBalanceVgroupLeaderStmt;
+
 typedef struct SMergeVgroupStmt {
  ENodeType type;
  int32_t   vgId1;
--- a/include/libs/nodes/nodes.h
+++ b/include/libs/nodes/nodes.h
@ -209,7 +209,8 @@ typedef enum ENodeType {
  QUERY_NODE_INSERT_STMT,
  QUERY_NODE_QUERY,
  QUERY_NODE_SHOW_DB_ALIVE_STMT,
-  QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT,  
+  QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT,
+  QUERY_NODE_BALANCE_VGROUP_LEADER_STMT,

  // logic plan node
  QUERY_NODE_LOGIC_PLAN_SCAN = 1000,
--- a/include/libs/sync/sync.h
+++ b/include/libs/sync/sync.h
@ -245,6 +245,7 @@ bool    syncIsReadyForRead(int64_t rid);
 bool    syncSnapshotSending(int64_t rid);
 bool    syncSnapshotRecving(int64_t rid);
 int32_t syncSendTimeoutRsp(int64_t rid, int64_t seq);
+int32_t syncLeaderForceElection(int64_t rid);

 SSyncState  syncGetState(int64_t rid);
 void        syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet);
--- a/packaging/tools/makepkg.sh
+++ b/packaging/tools/makepkg.sh
@ -53,7 +53,7 @@ if [ -d ${top_dir}/tools/taos-tools/packaging/deb ]; then
  cd ${top_dir}/tools/taos-tools/packaging/deb
  [ -z "$taos_tools_ver" ] && taos_tools_ver="0.1.0"

-  taostools_ver=$(git tag |grep -v taos | sort | tail -1)
+  taostools_ver=$(git for-each-ref --sort=taggerdate --format '%(tag)' refs/tags|grep -v taos | tail -1)
  taostools_install_dir="${release_dir}/${clientName2}Tools-${taostools_ver}"

  cd ${curr_dir}
--- a/source/client/jni/com_taosdata_jdbc_TSDBJNIConnector.h
+++ b/source/client/jni/com_taosdata_jdbc_TSDBJNIConnector.h
@ -75,6 +75,14 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_connectImp(JNIEn
 */
 JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_executeQueryImp(JNIEnv *, jobject, jbyteArray, jlong);

+/*
+ * Class:     com_taosdata_jdbc_TSDBJNIConnector
+ * Method:    executeQueryWithReqId
+ * Signature: ([BJJ)I
+ */
+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_executeQueryWithReqId(JNIEnv *, jobject, jbyteArray,
+                                                                                      jlong, jlong);
+
 /*
 * Class:     com_taosdata_jdbc_TSDBJNIConnector
 * Method:    getErrCodeImp
@ -186,6 +194,14 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_validateCreateTab
 */
 JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_prepareStmtImp(JNIEnv *, jobject, jbyteArray, jlong);

+/*
+ * Class:     com_taosdata_jdbc_TSDBJNIConnector
+ * Method:    prepareStmtWithReqId
+ * Signature: ([BJJ)I
+ */
+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_prepareStmtWithReqId(JNIEnv *, jobject, jbyteArray,
+                                                                                     jlong, jlong);
+
 /*
 * Class:     com_taosdata_jdbc_TSDBJNIConnector
 * Method:    setBindTableNameImp
@ -260,6 +276,32 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_insertLinesImp(JN
 JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertImp(JNIEnv *, jobject, jobjectArray,
                                                                                    jlong, jint, jint);

+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertWithReqId(JNIEnv *, jobject, jlong,
+                                                                                          jobjectArray, jint, jint,
+                                                                                          jlong);
+
+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertWithTtl(JNIEnv *, jobject, jlong,
+                                                                                        jobjectArray, jint, jint, jint);
+
+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertWithTtlAndReqId(JNIEnv *, jobject,
+                                                                                                jlong, jobjectArray,
+                                                                                                jint, jint, jint,
+                                                                                                jlong);
+
+JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertRaw(JNIEnv *, jobject, jlong, jstring,
+                                                                                      jint, jint);
+
+JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertRawWithReqId(JNIEnv *, jobject, jlong,
+                                                                                               jstring, jint, jint,
+                                                                                               jlong);
+
+JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertRawWithTtl(JNIEnv *, jobject, jlong,
+                                                                                             jstring, jint, jint, jint);
+
+JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertRawWithTtlAndReqId(JNIEnv *, jobject,
+                                                                                                     jlong, jstring,
+                                                                                                     jint, jint, jint,
+                                                                                                     jlong);
 /**
 * Class:     com_taosdata_jdbc_TSDBJNIConnector
 * Method:    getTableVgID
--- a/source/client/src/clientJniConnector.c
+++ b/source/client/src/clientJniConnector.c
@ -55,7 +55,7 @@ jclass    g_tmqClass;
 jmethodID g_createConsumerErrorCallback;
 jmethodID g_topicListCallback;

-jclass    g_consumerClass;
+jclass g_consumerClass;
 // deprecated
 jmethodID g_commitCallback;

@ -331,13 +331,58 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_executeQueryImp(
  int32_t   code = taos_errno(tres);

  if (code != TSDB_CODE_SUCCESS) {
-    jniError("jobj:%p, conn:%p, code:%s, msg:%s", jobj, tscon, tstrerror(code), taos_errstr(tres));
+    jniError("jobj:%p, conn:%p, code:0x%x, msg:%s", jobj, tscon, code, taos_errstr(tres));
  } else {
    if (taos_is_update_query(tres)) {
      int32_t affectRows = taos_affected_rows(tres);
-      jniDebug("jobj:%p, conn:%p, code:%s, affect rows:%d", jobj, tscon, tstrerror(code), affectRows);
+      jniDebug("jobj:%p, conn:%p, code:0x%x, affect rows:%d", jobj, tscon, code, affectRows);
    } else {
-      jniDebug("jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code));
+      jniDebug("jobj:%p, conn:%p, code:0x%x", jobj, tscon, code);
+    }
+  }
+
+  taosMemoryFreeClear(str);
+  return (jlong)tres;
+}
+
+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_executeQueryWithReqId(JNIEnv *env, jobject jobj,
+                                                                                      jbyteArray jsql, jlong con,
+                                                                                      jlong reqId) {
+  TAOS *tscon = (TAOS *)con;
+  if (tscon == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    return JNI_CONNECTION_NULL;
+  }
+
+  if (jsql == NULL) {
+    jniError("jobj:%p, conn:%p, empty sql string", jobj, tscon);
+    return JNI_SQL_NULL;
+  }
+
+  jsize len = (*env)->GetArrayLength(env, jsql);
+
+  char *str = (char *)taosMemoryCalloc(1, sizeof(char) * (len + 1));
+  if (str == NULL) {
+    jniError("jobj:%p, conn:%p, alloc memory failed", jobj, tscon);
+    return JNI_OUT_OF_MEMORY;
+  }
+
+  (*env)->GetByteArrayRegion(env, jsql, 0, len, (jbyte *)str);
+  if ((*env)->ExceptionCheck(env)) {
+    // todo handle error
+  }
+
+  TAOS_RES *tres = taos_query_with_reqid(tscon, str, reqId);
+  int32_t   code = taos_errno(tres);
+
+  if (code != TSDB_CODE_SUCCESS) {
+    jniError("jobj:%p, conn:%p, code:0x%x, msg:%s", jobj, tscon, code, taos_errstr(tres));
+  } else {
+    if (taos_is_update_query(tres)) {
+      int32_t affectRows = taos_affected_rows(tres);
+      jniDebug("jobj:%p, conn:%p, code:0x%x, affect rows:%d", jobj, tscon, code, affectRows);
+    } else {
+      jniDebug("jobj:%p, conn:%p, code:0x%x", jobj, tscon, code);
    }
  }

@ -489,7 +534,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_fetchRowImp(JNIEn
               numOfFields);
      return JNI_FETCH_END;
    } else {
-      jniDebug("jobj:%p, conn:%p, interrupted query. fetch row error code: %d, msg:%s", jobj, tscon, code,
+      jniDebug("jobj:%p, conn:%p, interrupted query. fetch row error code: 0x%x, msg:%s", jobj, tscon, code,
               taos_errstr(result));
      return JNI_RESULT_SET_NULL;
    }
@ -575,9 +620,9 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_fetchBlockImp(JNI
  TAOS_RES *tres = (TAOS_RES *)res;

  int32_t numOfFields = taos_num_fields(tres);
-  if(numOfFields <= 0){
-    jniError("jobj:%p, conn:%p, query interrupted. taos_num_fields error code:%d, msg:%s", jobj, tscon, numOfFields,
-             taos_errstr(tres));
+  if (numOfFields <= 0) {
+    jniError("jobj:%p, conn:%p, query interrupted. taos_num_fields error code: 0x%x, msg:%s", jobj, tscon,
+             taos_errno(tres), taos_errstr(tres));
    return JNI_RESULT_SET_NULL;
  }

@ -589,7 +634,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_fetchBlockImp(JNI
      jniDebug("jobj:%p, conn:%p, resultset:%p, no data to retrieve", jobj, tscon, (void *)res);
      return JNI_FETCH_END;
    } else {
-      jniError("jobj:%p, conn:%p, query interrupted. fetch block error code:%d, msg:%s", jobj, tscon, error_code,
+      jniError("jobj:%p, conn:%p, query interrupted. fetch block error code: 0x%x, msg:%s", jobj, tscon, error_code,
               taos_errstr(tres));
      return JNI_RESULT_SET_NULL;
    }
@ -639,7 +684,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_validateCreateTab
  }

  int code = taos_validate_sql(tscon, str);
-  jniDebug("jobj:%p, conn:%p, code is %d", jobj, tscon, code);
+  jniDebug("jobj:%p, conn:%p, code: 0x%x", jobj, tscon, code);

  taosMemoryFreeClear(str);
  return code;
@ -704,7 +749,45 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_prepareStmtImp(J
  int32_t    code = taos_stmt_prepare(pStmt, str, len);
  taosMemoryFreeClear(str);
  if (code != TSDB_CODE_SUCCESS) {
-    jniError("prepareStmt jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code));
+    jniError("prepareStmt jobj:%p, conn:%p, code: 0x%x", jobj, tscon, code);
+    return JNI_TDENGINE_ERROR;
+  }
+
+  return (jlong)pStmt;
+}
+
+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_prepareStmtWithReqId(JNIEnv *env, jobject jobj,
+                                                                                     jbyteArray jsql, jlong con,
+                                                                                     jlong reqId) {
+  TAOS *tscon = (TAOS *)con;
+  if (tscon == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    return JNI_CONNECTION_NULL;
+  }
+
+  if (jsql == NULL) {
+    jniError("jobj:%p, conn:%p, empty sql string", jobj, tscon);
+    return JNI_SQL_NULL;
+  }
+
+  jsize len = (*env)->GetArrayLength(env, jsql);
+
+  char *str = (char *)taosMemoryCalloc(1, sizeof(char) * (len + 1));
+  if (str == NULL) {
+    jniError("jobj:%p, conn:%p, alloc memory failed", jobj, tscon);
+    return JNI_OUT_OF_MEMORY;
+  }
+
+  (*env)->GetByteArrayRegion(env, jsql, 0, len, (jbyte *)str);
+  if ((*env)->ExceptionCheck(env)) {
+    // todo handle error
+  }
+
+  TAOS_STMT *pStmt = taos_stmt_init_with_reqid(tscon, reqId);
+  int32_t    code = taos_stmt_prepare(pStmt, str, len);
+  taosMemoryFreeClear(str);
+  if (code != TSDB_CODE_SUCCESS) {
+    jniError("prepareStmtWithReqId jobj:%p, conn:%p, code: 0x%x", jobj, tscon, code);
    return JNI_TDENGINE_ERROR;
  }

@ -732,7 +815,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_setBindTableNameI
  if (code != TSDB_CODE_SUCCESS) {
    (*env)->ReleaseStringUTFChars(env, jname, name);

-    jniError("bindTableName jobj:%p, conn:%p, code:%s", jobj, tsconn, tstrerror(code));
+    jniError("bindTableName jobj:%p, conn:%p, code: 0x%x", jobj, tsconn, code);
    return JNI_TDENGINE_ERROR;
  }

@ -807,7 +890,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_setTableNameTagsI
  (*env)->ReleaseStringUTFChars(env, tableName, name);

  if (code != TSDB_CODE_SUCCESS) {
-    jniError("tableNameTags jobj:%p, conn:%p, code:%s", jobj, tsconn, tstrerror(code));
+    jniError("tableNameTags jobj:%p, conn:%p, code: 0x%x", jobj, tsconn, code);
    return JNI_TDENGINE_ERROR;
  }
  return JNI_SUCCESS;
@ -873,7 +956,7 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_bindColDataImp(
  taosMemoryFreeClear(b);

  if (code != TSDB_CODE_SUCCESS) {
-    jniError("bindColData jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code));
+    jniError("bindColData jobj:%p, conn:%p, code: 0x%x", jobj, tscon, code);
    return JNI_TDENGINE_ERROR;
  }

@ -896,7 +979,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_addBatchImp(JNIEn

  int32_t code = taos_stmt_add_batch(pStmt);
  if (code != TSDB_CODE_SUCCESS) {
-    jniError("add batch jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code));
+    jniError("add batch jobj:%p, conn:%p, code: 0x%x", jobj, tscon, code);
    return JNI_TDENGINE_ERROR;
  }

@ -920,7 +1003,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_executeBatchImp(J

  int32_t code = taos_stmt_execute(pStmt);
  if (code != TSDB_CODE_SUCCESS) {
-    jniError("excute batch jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code));
+    jniError("excute batch jobj:%p, conn:%p, code: 0x%x", jobj, tscon, code);
    return JNI_TDENGINE_ERROR;
  }

@ -944,7 +1027,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_closeStmt(JNIEnv

  int32_t code = taos_stmt_close(pStmt);
  if (code != TSDB_CODE_SUCCESS) {
-    jniError("close stmt jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code));
+    jniError("close stmt jobj:%p, conn:%p, code: 0x%x", jobj, tscon, code);
    return JNI_TDENGINE_ERROR;
  }

@ -1006,12 +1089,9 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_insertLinesImp(JN

  TAOS_RES *tres = schemalessInsert(env, jobj, lines, taos, protocol, precision);

-  if (tres == NULL) {
-    return JNI_OUT_OF_MEMORY;
-  }
  int code = taos_errno(tres);
  if (code != TSDB_CODE_SUCCESS) {
-    jniError("jobj:%p, conn:%p, code:%s, msg:%s", jobj, taos, tstrerror(code), taos_errstr(tres));
+    jniError("jobj:%p, conn:%p, code: 0x%x, msg:%s", jobj, taos, code, taos_errstr(tres));
    taos_free_result(tres);
    return JNI_TDENGINE_ERROR;
  }
@ -1030,12 +1110,247 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsert
  }

  TAOS_RES *tres = schemalessInsert(env, jobj, lines, taos, protocol, precision);
-  if (tres == NULL) {
+
+  return (jlong)tres;
+}
+
+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertWithReqId(
+    JNIEnv *env, jobject jobj, jlong conn, jobjectArray lines, jint protocol, jint precision, jlong reqId) {
+  TAOS *taos = (TAOS *)conn;
+  if (taos == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    return JNI_CONNECTION_NULL;
+  }
+
+  int    numLines = (*env)->GetArrayLength(env, lines);
+  char **c_lines = taosMemoryCalloc(numLines, sizeof(char *));
+  if (c_lines == NULL) {
+    jniError("c_lines:%p, alloc memory failed", c_lines);
    return JNI_OUT_OF_MEMORY;
  }
+
+  for (int i = 0; i < numLines; ++i) {
+    jstring line = (jstring)((*env)->GetObjectArrayElement(env, lines, i));
+    c_lines[i] = (char *)(*env)->GetStringUTFChars(env, line, 0);
+  }
+
+  TAOS_RES *tres = taos_schemaless_insert_with_reqid(taos, c_lines, numLines, protocol, precision, reqId);
+
+  for (int i = 0; i < numLines; ++i) {
+    jstring line = (jstring)((*env)->GetObjectArrayElement(env, lines, i));
+    (*env)->ReleaseStringUTFChars(env, line, c_lines[i]);
+  }
+
+  taosMemoryFreeClear(c_lines);
+
  return (jlong)tres;
 }

+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertWithTtl(JNIEnv *env, jobject jobj,
+                                                                                        jlong conn, jobjectArray lines,
+                                                                                        jint protocol, jint precision,
+                                                                                        jint ttl) {
+  TAOS *taos = (TAOS *)conn;
+  if (taos == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    return JNI_CONNECTION_NULL;
+  }
+
+  int    numLines = (*env)->GetArrayLength(env, lines);
+  char **c_lines = taosMemoryCalloc(numLines, sizeof(char *));
+  if (c_lines == NULL) {
+    jniError("c_lines:%p, alloc memory failed", c_lines);
+    return JNI_OUT_OF_MEMORY;
+  }
+
+  for (int i = 0; i < numLines; ++i) {
+    jstring line = (jstring)((*env)->GetObjectArrayElement(env, lines, i));
+    c_lines[i] = (char *)(*env)->GetStringUTFChars(env, line, 0);
+  }
+
+  TAOS_RES *tres = taos_schemaless_insert_ttl(taos, c_lines, numLines, protocol, precision, ttl);
+
+  for (int i = 0; i < numLines; ++i) {
+    jstring line = (jstring)((*env)->GetObjectArrayElement(env, lines, i));
+    (*env)->ReleaseStringUTFChars(env, line, c_lines[i]);
+  }
+
+  taosMemoryFreeClear(c_lines);
+
+  return (jlong)tres;
+}
+
+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertWithTtlAndReqId(
+    JNIEnv *env, jobject jobj, jlong conn, jobjectArray lines, jint protocol, jint precision, jint ttl, jlong reqId) {
+  TAOS *taos = (TAOS *)conn;
+  if (taos == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    return JNI_CONNECTION_NULL;
+  }
+
+  int    numLines = (*env)->GetArrayLength(env, lines);
+  char **c_lines = taosMemoryCalloc(numLines, sizeof(char *));
+  if (c_lines == NULL) {
+    jniError("c_lines:%p, alloc memory failed", c_lines);
+    return JNI_OUT_OF_MEMORY;
+  }
+
+  for (int i = 0; i < numLines; ++i) {
+    jstring line = (jstring)((*env)->GetObjectArrayElement(env, lines, i));
+    c_lines[i] = (char *)(*env)->GetStringUTFChars(env, line, 0);
+  }
+
+  TAOS_RES *tres = taos_schemaless_insert_ttl_with_reqid(taos, c_lines, numLines, protocol, precision, ttl, reqId);
+
+  for (int i = 0; i < numLines; ++i) {
+    jstring line = (jstring)((*env)->GetObjectArrayElement(env, lines, i));
+    (*env)->ReleaseStringUTFChars(env, line, c_lines[i]);
+  }
+
+  taosMemoryFreeClear(c_lines);
+
+  return (jlong)tres;
+}
+
+JNIEXPORT jobject createSchemalessResp(JNIEnv *env, int totalRows, int code, const char *msg) {
+  // find class
+  jclass schemaless_clazz = (*env)->FindClass(env, "com/taosdata/jdbc/SchemalessResp");
+  // find methods
+  jmethodID init_method = (*env)->GetMethodID(env, schemaless_clazz, "<init>", "()V");
+  jmethodID setCode_method = (*env)->GetMethodID(env, schemaless_clazz, "setCode", "(I)V");
+  jmethodID setMsg_method = (*env)->GetMethodID(env, schemaless_clazz, "setMsg", "(Ljava/lang/String;)V");
+  jmethodID setTotalRows_method = (*env)->GetMethodID(env, schemaless_clazz, "setTotalRows", "(I)V");
+  // new schemaless
+  jobject schemaless_obj = (*env)->NewObject(env, schemaless_clazz, init_method);
+  // set code
+  (*env)->CallVoidMethod(env, schemaless_obj, setCode_method, code);
+  // set totalRows
+  (*env)->CallVoidMethod(env, schemaless_obj, setTotalRows_method, totalRows);
+  // set message
+  jstring message = (*env)->NewStringUTF(env, msg);
+  (*env)->CallVoidMethod(env, schemaless_obj, setMsg_method, message);
+
+  return schemaless_obj;
+}
+
+JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertRaw(JNIEnv *env, jobject jobj,
+                                                                                      jlong conn, jstring data,
+                                                                                      jint protocol, jint precision) {
+  TAOS *taos = (TAOS *)conn;
+  if (taos == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    char *msg = "JNI connection is NULL";
+    return createSchemalessResp(env, 0, JNI_CONNECTION_NULL, msg);
+  }
+
+  char     *line = (char *)(*env)->GetStringUTFChars(env, data, NULL);
+  jint      len = (*env)->GetStringUTFLength(env, data);
+  int32_t   totalRows;
+  TAOS_RES *tres = taos_schemaless_insert_raw(taos, line, len, &totalRows, protocol, precision);
+
+  (*env)->ReleaseStringUTFChars(env, data, line);
+
+  // if (tres == NULL) {
+  //   jniError("jobj:%p, schemaless raw insert failed", jobj);
+  //   char *msg = "JNI schemaless raw insert return null";
+  //   return createSchemalessResp(env, 0, JNI_TDENGINE_ERROR, msg);
+  // }
+
+  int code = taos_errno(tres);
+  if (code != TSDB_CODE_SUCCESS) {
+    jniError("jobj:%p, conn:%p, code: 0x%x, msg:%s", jobj, taos, code, taos_errstr(tres));
+    taos_free_result(tres);
+    return createSchemalessResp(env, 0, code, taos_errstr(tres));
+  }
+  taos_free_result(tres);
+
+  return createSchemalessResp(env, totalRows, JNI_SUCCESS, NULL);
+}
+
+JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertRawWithReqId(
+    JNIEnv *env, jobject jobj, jlong conn, jstring data, jint protocol, jint precision, jlong reqId) {
+  TAOS *taos = (TAOS *)conn;
+  if (taos == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    char *msg = "JNI connection is NULL";
+    return createSchemalessResp(env, 0, JNI_CONNECTION_NULL, msg);
+  }
+
+  char     *line = (char *)(*env)->GetStringUTFChars(env, data, NULL);
+  jint      len = (*env)->GetStringUTFLength(env, data);
+  int32_t   totalRows;
+  TAOS_RES *tres = taos_schemaless_insert_raw_with_reqid(taos, line, len, &totalRows, protocol, precision, reqId);
+
+  (*env)->ReleaseStringUTFChars(env, data, line);
+
+  int code = taos_errno(tres);
+  if (code != TSDB_CODE_SUCCESS) {
+    jniError("jobj:%p, conn:%p, code: 0x%x, msg:%s", jobj, taos, code, taos_errstr(tres));
+    taos_free_result(tres);
+    return createSchemalessResp(env, 0, code, taos_errstr(tres));
+  }
+  taos_free_result(tres);
+
+  return createSchemalessResp(env, totalRows, JNI_SUCCESS, NULL);
+}
+
+JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertRawWithTtl(JNIEnv *env, jobject jobj,
+                                                                                             jlong conn, jstring data,
+                                                                                             jint protocol,
+                                                                                             jint precision, jint ttl) {
+  TAOS *taos = (TAOS *)conn;
+  if (taos == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    char *msg = "JNI connection is NULL";
+    return createSchemalessResp(env, 0, JNI_CONNECTION_NULL, msg);
+  }
+
+  char     *line = (char *)(*env)->GetStringUTFChars(env, data, NULL);
+  jint      len = (*env)->GetStringUTFLength(env, data);
+  int32_t   totalRows;
+  TAOS_RES *tres = taos_schemaless_insert_raw_ttl(taos, line, len, &totalRows, protocol, precision, ttl);
+
+  (*env)->ReleaseStringUTFChars(env, data, line);
+
+  int code = taos_errno(tres);
+  if (code != TSDB_CODE_SUCCESS) {
+    jniError("jobj:%p, conn:%p, code: 0x%x, msg:%s", jobj, taos, code, taos_errstr(tres));
+    taos_free_result(tres);
+    return createSchemalessResp(env, 0, code, taos_errstr(tres));
+  }
+  taos_free_result(tres);
+
+  return createSchemalessResp(env, totalRows, JNI_SUCCESS, NULL);
+}
+
+JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsertRawWithTtlAndReqId(
+    JNIEnv *env, jobject jobj, jlong conn, jstring data, jint protocol, jint precision, jint ttl, jlong reqId) {
+  TAOS *taos = (TAOS *)conn;
+  if (taos == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    char *msg = "JNI connection is NULL";
+    return createSchemalessResp(env, 0, JNI_CONNECTION_NULL, msg);
+  }
+
+  char     *line = (char *)(*env)->GetStringUTFChars(env, data, NULL);
+  jint      len = (*env)->GetStringUTFLength(env, data);
+  int32_t   totalRows;
+  TAOS_RES *tres =
+      taos_schemaless_insert_raw_ttl_with_reqid(taos, line, len, &totalRows, protocol, precision, ttl, reqId);
+
+  (*env)->ReleaseStringUTFChars(env, data, line);
+
+  int code = taos_errno(tres);
+  if (code != TSDB_CODE_SUCCESS) {
+    jniError("jobj:%p, conn:%p, code: 0x%x, msg:%s", jobj, taos, code, taos_errstr(tres));
+    taos_free_result(tres);
+    return createSchemalessResp(env, 0, code, taos_errstr(tres));
+  }
+  taos_free_result(tres);
+
+  return createSchemalessResp(env, totalRows, JNI_SUCCESS, NULL);
+}
+
 // TABLE_VG_ID_FID_CACHE cache resp object for getTableVgID
 typedef struct TABLE_VG_ID_FIELD_CACHE {
  int      cached;
--- a/source/common/src/tmsg.c
+++ b/source/common/src/tmsg.c
@ -2219,6 +2219,8 @@ int32_t tSerializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
  if (tEncodeI8(&encoder, pReq->cacheLast) < 0) return -1;
  if (tEncodeI8(&encoder, pReq->replications) < 0) return -1;
  if (tEncodeI32(&encoder, pReq->sstTrigger) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->walRetentionPeriod) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->walRetentionSize) < 0) return -1;

  // 1st modification
  if (tEncodeI32(&encoder, pReq->minRows) < 0) return -1;
@ -2250,6 +2252,13 @@ int32_t tDeserializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
  if (tDecodeI8(&decoder, &pReq->cacheLast) < 0) return -1;
  if (tDecodeI8(&decoder, &pReq->replications) < 0) return -1;
  if (tDecodeI32(&decoder, &pReq->sstTrigger) < 0) return -1;
+  if (!tDecodeIsEnd(&decoder)) {
+    if (tDecodeI32(&decoder, &pReq->walRetentionPeriod) < 0) return -1;
+    if (tDecodeI32(&decoder, &pReq->walRetentionSize) < 0) return -1;
+  } else {
+    pReq->walRetentionPeriod = -1;
+    pReq->walRetentionSize = -1;
+  }

  // 1st modification
  if (!tDecodeIsEnd(&decoder)) {
@ -4438,6 +4447,31 @@ int32_t tDeserializeSBalanceVgroupReq(void *buf, int32_t bufLen, SBalanceVgroupR
  return 0;
 }

+int32_t tSerializeSBalanceVgroupLeaderReq(void *buf, int32_t bufLen, SBalanceVgroupLeaderReq *pReq) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->useless) < 0) return -1;
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSBalanceVgroupLeaderReq(void *buf, int32_t bufLen, SBalanceVgroupLeaderReq *pReq) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->useless) < 0) return -1;
+  tEndDecode(&decoder);
+
+  tDecoderClear(&decoder);
+  return 0;
+}
+
 int32_t tSerializeSMergeVgroupReq(void *buf, int32_t bufLen, SMergeVgroupReq *pReq) {
  SEncoder encoder = {0};
  tEncoderInit(&encoder, buf, bufLen);
@ -4521,6 +4555,31 @@ int32_t tDeserializeSSplitVgroupReq(void *buf, int32_t bufLen, SSplitVgroupReq *
  return 0;
 }

+int32_t tSerializeSForceElectionReq(void *buf, int32_t bufLen, SForceElectionReq *pReq) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->vgId) < 0) return -1;
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSForceElectionReq(void *buf, int32_t bufLen, SForceElectionReq *pReq) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->vgId) < 0) return -1;
+  tEndDecode(&decoder);
+
+  tDecoderClear(&decoder);
+  return 0;
+}
+
 int32_t tSerializeSDCreateMnodeReq(void *buf, int32_t bufLen, SDCreateMnodeReq *pReq) {
  SEncoder encoder = {0};
  tEncoderInit(&encoder, buf, bufLen);
--- a/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
+++ b/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
@ -92,6 +92,7 @@ SArray *mmGetMsgHandles() {
  if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_CONFIG_DNODE_RSP, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_DND_FORCE_ELECTION_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

  if (dmSetMgmtHandle(pArray, TDMT_MND_CONNECT, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_ACCT, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
@ -126,6 +127,7 @@ SArray *mmGetMsgHandles() {
  if (dmSetMgmtHandle(pArray, TDMT_MND_MERGE_VGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_MND_SPLIT_VGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_MND_BALANCE_VGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_BALANCE_VGROUP_LEADER, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_FUNC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_MND_RETRIEVE_FUNC, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_FUNC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
--- a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
+++ b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
@ -86,6 +86,7 @@ void       vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemo
 // vmHandle.c
 SArray *vmGetMsgHandles();
 int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmProcessForceElectionReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmProcessAlterVnodeReplicaReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmProcessDisableVnodeWriteReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
--- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
@ -304,6 +304,26 @@ int32_t vmProcessDisableVnodeWriteReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
  return 0;
 }

+int32_t vmProcessForceElectionReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg){
+  SForceElectionReq req = {0};
+
+  if (tDeserializeSForceElectionReq(pMsg->pCont, pMsg->contLen, &req) != 0) {
+    terrno = TSDB_CODE_INVALID_MSG;
+    return -1;
+  }
+
+  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId);
+  if (pVnode == NULL) {
+    dError("vgId:%d, failed to alter hashrange since %s", req.vgId, terrstr());
+    terrno = TSDB_CODE_VND_NOT_EXIST;
+    return -1;
+  }
+
+  vnodeForceElection(pVnode->pImpl);
+
+  return 0;
+}
+
 int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
  SAlterVnodeHashRangeReq req = {0};
  if (tDeserializeSAlterVnodeHashRangeReq(pMsg->pCont, pMsg->contLen, &req) != 0) {
@ -548,6 +568,7 @@ SArray *vmGetMsgHandles() {
  if (dmSetMgmtHandle(pArray, TDMT_VND_TRIM, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_DND_FORCE_ELECTION, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;

  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
--- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
@ -37,6 +37,9 @@ static void vmProcessMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
    case TDMT_DND_CREATE_VNODE:
      code = vmProcessCreateVnodeReq(pMgmt, pMsg);
      break;
+    case TDMT_DND_FORCE_ELECTION:
+      code = vmProcessForceElectionReq(pMgmt, pMsg);
+      break;
    case TDMT_DND_DROP_VNODE:
      code = vmProcessDropVnodeReq(pMgmt, pMsg);
      break;
--- a/source/dnode/mnode/impl/src/mndConsumer.c
+++ b/source/dnode/mnode/impl/src/mndConsumer.c
@ -77,7 +77,7 @@ void mndCleanupConsumer(SMnode *pMnode) {}

 bool mndRebTryStart() {
  int32_t old = atomic_val_compare_exchange_32(&mqRebInExecCnt, 0, 1);
-  mInfo("tq timer, rebalance counter old val:%d", old);
+  mDebug("tq timer, rebalance counter old val:%d", old);
  return old == 0;
 }

@ -101,7 +101,7 @@ void mndRebCntDec() {
    int32_t newVal = val - 1;
    int32_t oldVal = atomic_val_compare_exchange_32(&mqRebInExecCnt, val, newVal);
    if (oldVal == val) {
-      mInfo("rebalance trans end, rebalance counter:%d", newVal);
+      mDebug("rebalance trans end, rebalance counter:%d", newVal);
      break;
    }
  }
@ -356,7 +356,7 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) {
  } else {
    taosHashCleanup(pRebMsg->rebSubHash);
    rpcFreeCont(pRebMsg);
-    mInfo("mq rebalance finished, no modification");
+    mDebug("mq rebalance finished, no modification");
    mndRebEnd();
  }
  return 0;
--- a/source/dnode/mnode/impl/src/mndDb.c
+++ b/source/dnode/mnode/impl/src/mndDb.c
@ -737,6 +737,20 @@ static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
    terrno = 0;
  }

+  if (pAlter->walRetentionPeriod > TSDB_DB_MIN_WAL_RETENTION_PERIOD &&
+      pAlter->walRetentionPeriod != pDb->cfg.walRetentionPeriod) {
+    pDb->cfg.walRetentionPeriod = pAlter->walRetentionPeriod;
+    pDb->vgVersion++;
+    terrno = 0;
+  }
+
+  if (pAlter->walRetentionSize > TSDB_DB_MIN_WAL_RETENTION_SIZE &&
+      pAlter->walRetentionSize != pDb->cfg.walRetentionSize) {
+    pDb->cfg.walRetentionSize = pAlter->walRetentionSize;
+    pDb->vgVersion++;
+    terrno = 0;
+  }
+
  return terrno;
 }

--- a/source/dnode/mnode/impl/src/mndVgroup.c
+++ b/source/dnode/mnode/impl/src/mndVgroup.c
@ -40,6 +40,7 @@ static void    mndCancelGetNextVnode(SMnode *pMnode, void *pIter);
 static int32_t mndProcessRedistributeVgroupMsg(SRpcMsg *pReq);
 static int32_t mndProcessSplitVgroupMsg(SRpcMsg *pReq);
 static int32_t mndProcessBalanceVgroupMsg(SRpcMsg *pReq);
+static int32_t mndProcessVgroupBalanceLeaderMsg(SRpcMsg *pReq);

 int32_t mndInitVgroup(SMnode *pMnode) {
  SSdbTable table = {
@ -60,10 +61,13 @@ int32_t mndInitVgroup(SMnode *pMnode) {
  mndSetMsgHandle(pMnode, TDMT_DND_DROP_VNODE_RSP, mndTransProcessRsp);
  mndSetMsgHandle(pMnode, TDMT_VND_COMPACT_RSP, mndTransProcessRsp);
  mndSetMsgHandle(pMnode, TDMT_VND_DISABLE_WRITE_RSP, mndTransProcessRsp);
+  mndSetMsgHandle(pMnode, TDMT_DND_FORCE_ELECTION_RSP, mndTransProcessRsp);

  mndSetMsgHandle(pMnode, TDMT_MND_REDISTRIBUTE_VGROUP, mndProcessRedistributeVgroupMsg);
  mndSetMsgHandle(pMnode, TDMT_MND_SPLIT_VGROUP, mndProcessSplitVgroupMsg);
+  //mndSetMsgHandle(pMnode, TDMT_MND_BALANCE_VGROUP, mndProcessVgroupBalanceLeaderMsg);
  mndSetMsgHandle(pMnode, TDMT_MND_BALANCE_VGROUP, mndProcessBalanceVgroupMsg);
+  mndSetMsgHandle(pMnode, TDMT_MND_BALANCE_VGROUP_LEADER, mndProcessVgroupBalanceLeaderMsg);

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_VGROUP, mndRetrieveVgroups);
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_VGROUP, mndCancelGetNextVgroup);
@ -1770,6 +1774,157 @@ _OVER:
  return code;
 }

+static void *mndBuildSForceElectionReq(SMnode *pMnode, SVgObj *pVgroup, int32_t dnodeId,
+                                          int32_t *pContLen) {
+  SForceElectionReq balanceReq = {
+      .vgId = pVgroup->vgId,
+  };
+
+  int32_t contLen = tSerializeSForceElectionReq(NULL, 0, &balanceReq);
+  if (contLen < 0) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return NULL;
+  }
+
+  void *pReq = taosMemoryMalloc(contLen);
+  if (pReq == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return NULL;
+  }
+
+  tSerializeSForceElectionReq((char *)pReq, contLen, &balanceReq);
+  *pContLen = contLen;
+  return pReq;                                    
+}
+
+int32_t mndAddBalanceVgroupLeaderAction(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup, int32_t dnodeId) {
+  SDnodeObj *pDnode = mndAcquireDnode(pMnode, dnodeId);
+  if (pDnode == NULL) return -1;
+
+  STransAction action = {0};
+  action.epSet = mndGetDnodeEpset(pDnode);
+  mndReleaseDnode(pMnode, pDnode);
+
+  int32_t contLen = 0;
+  void   *pReq = mndBuildSForceElectionReq(pMnode, pVgroup, dnodeId, &contLen);
+  if (pReq == NULL) return -1;
+
+  action.pCont = pReq;
+  action.contLen = contLen;
+  action.msgType = TDMT_DND_FORCE_ELECTION;
+
+  if (mndTransAppendRedoAction(pTrans, &action) != 0) {
+    taosMemoryFree(pReq);
+    return -1;
+  }
+
+  return 0;
+}
+
+int32_t mndAddVgroupBalanceToTrans(SMnode *pMnode, SVgObj *pVgroup, STrans *pTrans){
+  SSdb *pSdb = pMnode->pSdb;
+
+  int32_t vgid = pVgroup->vgId;
+  int8_t replica = pVgroup->replica;
+
+  if(pVgroup->replica <= 1) {
+    mInfo("trans:%d, vgid:%d no need to balance, replica:%d", pTrans->id, vgid, replica);
+    return -1;
+  }
+
+  int32_t index = vgid%replica;
+  int32_t dnodeId = pVgroup->vnodeGid[index].dnodeId;
+
+  bool       exist = false;
+  bool       online = false;
+  int64_t curMs = taosGetTimestampMs();
+  SDnodeObj *pDnode = mndAcquireDnode(pMnode, dnodeId);
+  if (pDnode != NULL) {
+    exist = true;
+    online = mndIsDnodeOnline(pDnode, curMs);
+    mndReleaseDnode(pMnode, pDnode);
+  }
+
+  if(exist && online)
+  {
+    mInfo("trans:%d, vgid:%d leader to dnode:%d", pTrans->id, vgid, dnodeId);
+
+    if (mndAddBalanceVgroupLeaderAction(pMnode, pTrans, pVgroup, dnodeId) != 0) {
+      mError("trans:%d, vgid:%d failed to be balanced to dnode:%d", pTrans->id, vgid, dnodeId);
+      return -1;
+    }
+
+    SSdbRaw *pRaw = mndVgroupActionEncode(pVgroup);
+    if (pRaw == NULL) {
+      mError("trans:%d, vgid:%d failed to encode action to dnode:%d", pTrans->id, vgid, dnodeId);
+      return -1;
+    }
+    if (mndTransAppendCommitlog(pTrans, pRaw) != 0) {
+      sdbFreeRaw(pRaw);
+      mError("trans:%d, vgid:%d failed to append commit log dnode:%d", pTrans->id, vgid, dnodeId);
+      return -1;
+    }
+    (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
+  }
+  else
+  {
+    mInfo("trans:%d, vgid:%d cant be balanced to dnode:%d, exist:%d, online:%d", pTrans->id, vgid, dnodeId, exist, online);
+  }
+
+  return 0;
+}
+
+int32_t mndProcessVgroupBalanceLeaderMsg(SRpcMsg *pReq) {
+  int32_t code = -1;
+  
+  SBalanceVgroupLeaderReq req = {0};
+  if (tDeserializeSBalanceVgroupLeaderReq(pReq->pCont, pReq->contLen, &req) != 0) {
+    terrno = TSDB_CODE_INVALID_MSG;
+    return code;
+  }
+
+  SMnode *pMnode = pReq->info.node;
+  SSdb *pSdb = pMnode->pSdb;
+
+  int32_t total = sdbGetSize(pSdb, SDB_VGROUP);
+  if(total <= 0) {
+    terrno = TSDB_CODE_TSC_INVALID_OPERATION;
+    return code;
+  }
+  
+  STrans *pTrans = NULL;
+  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pReq, "bal-vg-leader");
+  if (pTrans == NULL) goto _OVER;
+  mndTransSetSerial(pTrans);
+  mInfo("trans:%d, used to balance vgroup leader", pTrans->id);
+
+  void *pIter = NULL;
+  int32_t count = 0;
+  while (1) {
+    SVgObj *pVgroup = NULL;
+    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
+    if (pIter == NULL) break;
+
+    if(mndAddVgroupBalanceToTrans(pMnode, pVgroup, pTrans) == 0){
+      count++;
+    }
+
+    sdbRelease(pSdb, pVgroup);
+  }
+  
+  if(count == 0) {
+    terrno = TSDB_CODE_TSC_INVALID_OPERATION;
+    goto _OVER;
+  }
+
+  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
+  code = 0;
+
+_OVER:
+  mndTransDrop(pTrans);
+  return code;
+}
+
 static int32_t mndCheckDnodeMemory(SMnode *pMnode, SDbObj *pOldDb, SDbObj *pNewDb, SVgObj *pOldVgroup,
                                   SVgObj *pNewVgroup, SArray *pArray) {
  for (int32_t i = 0; i < (int32_t)taosArrayGetSize(pArray); ++i) {
--- a/source/dnode/vnode/inc/vnode.h
+++ b/source/dnode/vnode/inc/vnode.h
@ -56,6 +56,7 @@ void    vnodeDestroy(const char *path, STfs *pTfs);
 SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb);
 void    vnodePreClose(SVnode *pVnode);
 void    vnodePostClose(SVnode *pVnode);
+void    vnodeForceElection(SVnode *pVnode);
 void    vnodeSyncCheckTimeout(SVnode *pVnode);
 void    vnodeClose(SVnode *pVnode);
 int32_t vnodeSyncCommit(SVnode *pVnode);
--- a/source/dnode/vnode/src/vnd/vnodeOpen.c
+++ b/source/dnode/vnode/src/vnd/vnodeOpen.c
@ -380,6 +380,10 @@ void vnodePreClose(SVnode *pVnode) {

 void vnodePostClose(SVnode *pVnode) { vnodeSyncPostClose(pVnode); }

+void vnodeForceElection(SVnode *pVnode) {
+  syncLeaderForceElection(pVnode->sync);
+}
+
 void vnodeClose(SVnode *pVnode) {
  if (pVnode) {
    tsem_wait(&pVnode->canCommit);
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@ -32,9 +32,6 @@
 #define TAIL_MAX_POINTS_NUM    100
 #define TAIL_MAX_OFFSET        100

-#define UNIQUE_MAX_RESULT_SIZE (1024 * 1024 * 10)
-#define MODE_MAX_RESULT_SIZE   UNIQUE_MAX_RESULT_SIZE
-
 #define HLL_BUCKET_BITS 14  // The bits of the bucket
 #define HLL_DATA_BITS   (64 - HLL_BUCKET_BITS)
 #define HLL_BUCKETS     (1 << HLL_BUCKET_BITS)
@ -244,12 +241,11 @@ typedef struct SUniqueInfo {

 typedef struct SModeItem {
  int64_t   count;
+  STuplePos dataPos;
  STuplePos tuplePos;
-  char      data[];
 } SModeItem;

 typedef struct SModeInfo {
-  int32_t   numOfPoints;
  uint8_t   colType;
  int16_t   colBytes;
  SHashObj* pHash;
@ -257,7 +253,7 @@ typedef struct SModeInfo {
  STuplePos nullTuplePos;
  bool      nullTupleSaved;

-  char pItems[];
+  char*     buf; // serialize data buffer
 } SModeInfo;

 typedef struct SDerivInfo {
@ -3113,7 +3109,7 @@ void* serializeTupleData(const SSDataBlock* pSrcBlock, int32_t rowIndex, SSubsid
  return buf;
 }

-static int32_t doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf, size_t length, STupleKey key,
+static int32_t doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf, size_t length, STupleKey* key,
                               STuplePos* pPos) {
  STuplePos p = {0};
  if (pHandle->pBuf != NULL) {
@ -3149,8 +3145,8 @@ static int32_t doSaveTupleData(SSerializeDataHandle* pHandle, const void* pBuf,
    releaseBufPage(pHandle->pBuf, pPage);
  } else {
    // other tuple save policy
-    if (streamStateFuncPut(pHandle->pState, &key, pBuf, length) >= 0) {
-      p.streamTupleKey = key;
+    if (streamStateFuncPut(pHandle->pState, key, pBuf, length) >= 0) {
+      p.streamTupleKey = *key;
    }
  }

@ -3174,7 +3170,7 @@ int32_t saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock*
  }

  char* buf = serializeTupleData(pSrcBlock, rowIndex, &pCtx->subsidiaries, pCtx->subsidiaries.buf);
-  return doSaveTupleData(&pCtx->saveHandle, buf, pCtx->subsidiaries.rowLen, key, pPos);
+  return doSaveTupleData(&pCtx->saveHandle, buf, pCtx->subsidiaries.rowLen, &key, pPos);
 }

 static int32_t doUpdateTupleData(SSerializeDataHandle* pHandle, const void* pBuf, size_t length, STuplePos* pPos) {
@ -4949,7 +4945,7 @@ int32_t uniqueFunction(SqlFunctionCtx* pCtx) {
 }

 bool getModeFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
-  pEnv->calcMemSize = sizeof(SModeInfo) + MODE_MAX_RESULT_SIZE;
+  pEnv->calcMemSize = sizeof(SModeInfo);
  return true;
 }

@ -4959,7 +4955,6 @@ bool modeFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
  }

  SModeInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
-  pInfo->numOfPoints = 0;
  pInfo->colType = pCtx->resDataInfo.type;
  pInfo->colBytes = pCtx->resDataInfo.bytes;
  if (pInfo->pHash != NULL) {
@ -4970,38 +4965,60 @@ bool modeFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
  pInfo->nullTupleSaved = false;
  pInfo->nullTuplePos.pageId = -1;

+  pInfo->buf = taosMemoryMalloc(pInfo->colBytes);
+
  return true;
 }

+static void modeFunctionCleanup(SModeInfo * pInfo) {
+  taosHashCleanup(pInfo->pHash);
+  taosMemoryFreeClear(pInfo->buf);
+}
+
+static int32_t saveModeTupleData(SqlFunctionCtx* pCtx, char* data, SModeInfo *pInfo, STuplePos* pPos) {
+  if (IS_VAR_DATA_TYPE(pInfo->colType)) {
+    memcpy(pInfo->buf, data, varDataTLen(data));
+  } else {
+    memcpy(pInfo->buf, data, pInfo->colBytes);
+  }
+
+  return doSaveTupleData(&pCtx->saveHandle, pInfo->buf, pInfo->colBytes, NULL, pPos);
+}
+
 static int32_t doModeAdd(SModeInfo* pInfo, int32_t rowIndex, SqlFunctionCtx* pCtx, char* data) {
-  int32_t     hashKeyBytes = IS_STR_DATA_TYPE(pInfo->colType) ? varDataTLen(data) : pInfo->colBytes;
-  SModeItem** pHashItem = taosHashGet(pInfo->pHash, data, hashKeyBytes);
+  int32_t code = TSDB_CODE_SUCCESS;
+  int32_t hashKeyBytes = IS_STR_DATA_TYPE(pInfo->colType) ? varDataTLen(data) : pInfo->colBytes;
+
+  SModeItem* pHashItem = (SModeItem *)taosHashGet(pInfo->pHash, data, hashKeyBytes);
  if (pHashItem == NULL) {
-    int32_t    size = sizeof(SModeItem) + pInfo->colBytes;
-    SModeItem* pItem = (SModeItem*)(pInfo->pItems + pInfo->numOfPoints * size);
-    memcpy(pItem->data, data, hashKeyBytes);
-    pItem->count += 1;
+    int32_t    size = sizeof(SModeItem);
+    SModeItem  item = {0};
+
+    item.count += 1;
+    code = saveModeTupleData(pCtx, data, pInfo, &item.dataPos);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }

    if (pCtx->subsidiaries.num > 0) {
-      int32_t code = saveTupleData(pCtx, rowIndex, pCtx->pSrcBlock, &pItem->tuplePos);
+      code = saveTupleData(pCtx, rowIndex, pCtx->pSrcBlock, &item.tuplePos);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
    }

-    taosHashPut(pInfo->pHash, data, hashKeyBytes, &pItem, sizeof(SModeItem*));
-    pInfo->numOfPoints++;
+    taosHashPut(pInfo->pHash, data, hashKeyBytes, &item, sizeof(SModeItem));
  } else {
-    (*pHashItem)->count += 1;
+    pHashItem->count += 1;
    if (pCtx->subsidiaries.num > 0) {
-      int32_t code = updateTupleData(pCtx, rowIndex, pCtx->pSrcBlock, &((*pHashItem)->tuplePos));
+      int32_t code = updateTupleData(pCtx, rowIndex, pCtx->pSrcBlock, &pHashItem->tuplePos);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
    }
  }

-  return TSDB_CODE_SUCCESS;
+  return code;
 }

 int32_t modeFunction(SqlFunctionCtx* pCtx) {
@ -5024,18 +5041,15 @@ int32_t modeFunction(SqlFunctionCtx* pCtx) {
    char*   data = colDataGetData(pInputCol, i);
    int32_t code = doModeAdd(pInfo, i, pCtx, data);
    if (code != TSDB_CODE_SUCCESS) {
+      modeFunctionCleanup(pInfo);
      return code;
    }
-
-    if (sizeof(SModeInfo) + pInfo->numOfPoints * (sizeof(SModeItem) + pInfo->colBytes) >= MODE_MAX_RESULT_SIZE) {
-      taosHashCleanup(pInfo->pHash);
-      return TSDB_CODE_OUT_OF_MEMORY;
-    }
  }

  if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pInfo->nullTupleSaved) {
    int32_t code = saveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pInfo->nullTuplePos);
    if (code != TSDB_CODE_SUCCESS) {
+      modeFunctionCleanup(pInfo);
      return code;
    }
    pInfo->nullTupleSaved = true;
@ -5054,26 +5068,37 @@ int32_t modeFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
  SColumnInfoData*     pCol = taosArrayGet(pBlock->pDataBlock, slotId);
  int32_t              currentRow = pBlock->info.rows;

-  int32_t resIndex = -1;
+  STuplePos resDataPos, resTuplePos;
  int32_t maxCount = 0;
-  for (int32_t i = 0; i < pInfo->numOfPoints; ++i) {
-    SModeItem* pItem = (SModeItem*)(pInfo->pItems + i * (sizeof(SModeItem) + pInfo->colBytes));
+
+  void *pIter = taosHashIterate(pInfo->pHash, NULL);
+  while (pIter != NULL) {
+    SModeItem *pItem = (SModeItem *)pIter;
    if (pItem->count >= maxCount) {
      maxCount = pItem->count;
-      resIndex = i;
+      resDataPos = pItem->dataPos;
+      resTuplePos = pItem->tuplePos;
    }
+
+    pIter = taosHashIterate(pInfo->pHash, pIter);
  }

  if (maxCount != 0) {
-    SModeItem* pResItem = (SModeItem*)(pInfo->pItems + resIndex * (sizeof(SModeItem) + pInfo->colBytes));
-    colDataSetVal(pCol, currentRow, pResItem->data, false);
-    code = setSelectivityValue(pCtx, pBlock, &pResItem->tuplePos, currentRow);
+    const char* pData = loadTupleData(pCtx, &resDataPos);
+    if (pData == NULL) {
+      code = TSDB_CODE_NO_AVAIL_DISK;
+      modeFunctionCleanup(pInfo);
+      return code;
+    }
+
+    colDataSetVal(pCol, currentRow, pData, false);
+    code = setSelectivityValue(pCtx, pBlock, &resTuplePos, currentRow);
  } else {
    colDataSetNULL(pCol, currentRow);
    code = setSelectivityValue(pCtx, pBlock, &pInfo->nullTuplePos, currentRow);
  }

-  taosHashCleanup(pInfo->pHash);
+  modeFunctionCleanup(pInfo);

  return code;
 }
--- a/source/libs/function/src/udfd.c
+++ b/source/libs/function/src/udfd.c
@ -554,6 +554,8 @@ int32_t udfdInitUdf(char *udfName, SUdf *udf) {
    fnError("udf name %s init failed. error %d", udfName, err);
    return err;
  }
+
+  fnInfo("udf init succeeded. name %s type %d context %p", udf->name, udf->scriptType, (void*)udf->scriptUdfCtx);
  return 0;
 }

@ -763,6 +765,7 @@ void udfdProcessTeardownRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
  }
  uv_mutex_unlock(&global.udfsMutex);
  if (unloadUdf) {
+    fnInfo("udf teardown. udf name: %s type %d: context %p", udf->name, udf->scriptType, (void*)(udf->scriptUdfCtx));
    uv_cond_destroy(&udf->condReady);
    uv_mutex_destroy(&udf->lock);
    code = udf->scriptPlugin->udfDestroyFunc(udf->scriptUdfCtx);
--- a/source/libs/nodes/src/nodesCodeFuncs.c
+++ b/source/libs/nodes/src/nodesCodeFuncs.c
@ -173,6 +173,8 @@ const char* nodesNodeName(ENodeType type) {
      return "DropStreamStmt";
    case QUERY_NODE_BALANCE_VGROUP_STMT:
      return "BalanceVgroupStmt";
+    case QUERY_NODE_BALANCE_VGROUP_LEADER_STMT:
+      return "BalanceVgroupLeaderStmt";
    case QUERY_NODE_MERGE_VGROUP_STMT:
      return "MergeVgroupStmt";
    case QUERY_NODE_SHOW_DB_ALIVE_STMT:
@ -6433,6 +6435,8 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
      return dropStreamStmtToJson(pObj, pJson);
    case QUERY_NODE_BALANCE_VGROUP_STMT:
      return TSDB_CODE_SUCCESS;  // SBalanceVgroupStmt has no fields to serialize.
+    case QUERY_NODE_BALANCE_VGROUP_LEADER_STMT:
+      return TSDB_CODE_SUCCESS;  // SBalanceVgroupLeaderStmt has no fields to serialize.
    case QUERY_NODE_MERGE_VGROUP_STMT:
      return mergeVgroupStmtToJson(pObj, pJson);
    case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT:
@ -6741,6 +6745,8 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
      return jsonToDropStreamStmt(pJson, pObj);
    case QUERY_NODE_BALANCE_VGROUP_STMT:
      return TSDB_CODE_SUCCESS;  // SBalanceVgroupStmt has no fields to deserialize.
+    case QUERY_NODE_BALANCE_VGROUP_LEADER_STMT:
+      return TSDB_CODE_SUCCESS;  // SBalanceVgroupLeaderStmt has no fields to deserialize.
    case QUERY_NODE_MERGE_VGROUP_STMT:
      return jsonToMergeVgroupStmt(pJson, pObj);
    case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT:
--- a/source/libs/nodes/src/nodesUtilFuncs.c
+++ b/source/libs/nodes/src/nodesUtilFuncs.c
@ -386,6 +386,8 @@ SNode* nodesMakeNode(ENodeType type) {
      return makeNode(type, sizeof(SDropStreamStmt));
    case QUERY_NODE_BALANCE_VGROUP_STMT:
      return makeNode(type, sizeof(SBalanceVgroupStmt));
+    case QUERY_NODE_BALANCE_VGROUP_LEADER_STMT:
+      return makeNode(type, sizeof(SBalanceVgroupLeaderStmt));
    case QUERY_NODE_MERGE_VGROUP_STMT:
      return makeNode(type, sizeof(SMergeVgroupStmt));
    case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT:
@ -923,9 +925,14 @@ void nodesDestroyNode(SNode* pNode) {
      taosMemoryFree(((SDescribeStmt*)pNode)->pMeta);
      break;
    case QUERY_NODE_RESET_QUERY_CACHE_STMT:  // no pointer field
-    case QUERY_NODE_COMPACT_DATABASE_STMT:   // no pointer field
-    case QUERY_NODE_CREATE_FUNCTION_STMT:    // no pointer field
-    case QUERY_NODE_DROP_FUNCTION_STMT:      // no pointer field
+    case QUERY_NODE_COMPACT_DATABASE_STMT: {
+      SCompactDatabaseStmt* pStmt = (SCompactDatabaseStmt*)pNode;
+      nodesDestroyNode(pStmt->pStart);
+      nodesDestroyNode(pStmt->pEnd);
+      break;
+    }
+    case QUERY_NODE_CREATE_FUNCTION_STMT:  // no pointer field
+    case QUERY_NODE_DROP_FUNCTION_STMT:    // no pointer field
      break;
    case QUERY_NODE_CREATE_STREAM_STMT: {
      SCreateStreamStmt* pStmt = (SCreateStreamStmt*)pNode;
@ -937,6 +944,7 @@ void nodesDestroyNode(SNode* pNode) {
    }
    case QUERY_NODE_DROP_STREAM_STMT:     // no pointer field
    case QUERY_NODE_BALANCE_VGROUP_STMT:  // no pointer field
+    case QUERY_NODE_BALANCE_VGROUP_LEADER_STMT:  // no pointer field
    case QUERY_NODE_MERGE_VGROUP_STMT:    // no pointer field
      break;
    case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT:
--- a/source/libs/parser/inc/parAst.h
+++ b/source/libs/parser/inc/parAst.h
@ -223,6 +223,7 @@ SNode* createDropStreamStmt(SAstCreateContext* pCxt, bool ignoreNotExists, SToke
 SNode* createKillStmt(SAstCreateContext* pCxt, ENodeType type, const SToken* pId);
 SNode* createKillQueryStmt(SAstCreateContext* pCxt, const SToken* pQueryId);
 SNode* createBalanceVgroupStmt(SAstCreateContext* pCxt);
+SNode* createBalanceVgroupLeaderStmt(SAstCreateContext* pCxt);
 SNode* createMergeVgroupStmt(SAstCreateContext* pCxt, const SToken* pVgId1, const SToken* pVgId2);
 SNode* createRedistributeVgroupStmt(SAstCreateContext* pCxt, const SToken* pVgId, SNodeList* pDnodes);
 SNode* createSplitVgroupStmt(SAstCreateContext* pCxt, const SToken* pVgId);
--- a/source/libs/parser/inc/sql.y
+++ b/source/libs/parser/inc/sql.y
@ -237,6 +237,18 @@ alter_db_option(A) ::= REPLICA NK_INTEGER(B).
 alter_db_option(A) ::= WAL_LEVEL NK_INTEGER(B).                                   { A.type = DB_OPTION_WAL; A.val = B; }
 alter_db_option(A) ::= STT_TRIGGER NK_INTEGER(B).                                 { A.type = DB_OPTION_STT_TRIGGER; A.val = B; }
 alter_db_option(A) ::= MINROWS NK_INTEGER(B).                                     { A.type = DB_OPTION_MINROWS; A.val = B; }
+alter_db_option(A) ::= WAL_RETENTION_PERIOD NK_INTEGER(B).                        { A.type = DB_OPTION_WAL_RETENTION_PERIOD; A.val = B; }
+alter_db_option(A) ::= WAL_RETENTION_PERIOD NK_MINUS(B) NK_INTEGER(C).            { 
+                                                                                    SToken t = B;
+                                                                                    t.n = (C.z + C.n) - B.z;
+                                                                                    A.type = DB_OPTION_WAL_RETENTION_PERIOD; A.val = t;
+                                                                                  }
+alter_db_option(A) ::= WAL_RETENTION_SIZE NK_INTEGER(B).                          { A.type = DB_OPTION_WAL_RETENTION_SIZE; A.val = B; }
+alter_db_option(A) ::= WAL_RETENTION_SIZE NK_MINUS(B) NK_INTEGER(C).              { 
+                                                                                    SToken t = B;
+                                                                                    t.n = (C.z + C.n) - B.z;
+                                                                                    A.type = DB_OPTION_WAL_RETENTION_SIZE; A.val = t;
+                                                                                  }

 %type integer_list                                                                { SNodeList* }
 %destructor integer_list                                                          { nodesDestroyList($$); }
@ -597,6 +609,7 @@ cmd ::= KILL TRANSACTION NK_INTEGER(A).

 /************************************************ merge/redistribute/ vgroup ******************************************/
 cmd ::= BALANCE VGROUP.                                                           { pCxt->pRootNode = createBalanceVgroupStmt(pCxt); }
+cmd ::= BALANCE VGROUP LEADER.                                                    { pCxt->pRootNode = createBalanceVgroupLeaderStmt(pCxt); }
 cmd ::= MERGE VGROUP NK_INTEGER(A) NK_INTEGER(B).                                 { pCxt->pRootNode = createMergeVgroupStmt(pCxt, &A, &B); }
 cmd ::= REDISTRIBUTE VGROUP NK_INTEGER(A) dnode_list(B).                          { pCxt->pRootNode = createRedistributeVgroupStmt(pCxt, &A, B); }
 cmd ::= SPLIT VGROUP NK_INTEGER(A).                                               { pCxt->pRootNode = createSplitVgroupStmt(pCxt, &A); }
--- a/source/libs/parser/src/parAstCreater.c
+++ b/source/libs/parser/src/parAstCreater.c
@ -925,8 +925,8 @@ SNode* createAlterDatabaseOptions(SAstCreateContext* pCxt) {
  pOptions->numOfVgroups = -1;
  pOptions->singleStable = -1;
  pOptions->schemaless = -1;
-  pOptions->walRetentionPeriod = -1;
-  pOptions->walRetentionSize = -1;
+  pOptions->walRetentionPeriod = -2;  // -1 is a valid value
+  pOptions->walRetentionSize = -2;    // -1 is a valid value
  pOptions->walRollPeriod = -1;
  pOptions->walSegmentSize = -1;
  pOptions->sstTrigger = -1;
@ -935,7 +935,8 @@ SNode* createAlterDatabaseOptions(SAstCreateContext* pCxt) {
  return (SNode*)pOptions;
 }

-SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOptionType type, void* pVal) {
+static SNode* setDatabaseOptionImpl(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOptionType type, void* pVal,
+                                    bool alter) {
  CHECK_PARSER_STATUS(pCxt);
  SDatabaseOptions* pDbOptions = (SDatabaseOptions*)pOptions;
  switch (type) {
@ -986,7 +987,9 @@ SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOpti
      break;
    case DB_OPTION_REPLICA:
      pDbOptions->replica = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
-      updateWalOptionsDefault(pDbOptions);
+      if (!alter) {
+        updateWalOptionsDefault(pDbOptions);
+      }
      break;
    case DB_OPTION_STRICT:
      COPY_STRING_FORM_STR_TOKEN(pDbOptions->strictStr, (SToken*)pVal);
@ -1033,16 +1036,20 @@ SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOpti
  return pOptions;
 }

+SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOptionType type, void* pVal) {
+  return setDatabaseOptionImpl(pCxt, pOptions, type, pVal, false);
+}
+
 SNode* setAlterDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, SAlterOption* pAlterOption) {
  CHECK_PARSER_STATUS(pCxt);
  switch (pAlterOption->type) {
    case DB_OPTION_KEEP:
    case DB_OPTION_RETENTIONS:
-      return setDatabaseOption(pCxt, pOptions, pAlterOption->type, pAlterOption->pList);
+      return setDatabaseOptionImpl(pCxt, pOptions, pAlterOption->type, pAlterOption->pList, true);
    default:
      break;
  }
-  return setDatabaseOption(pCxt, pOptions, pAlterOption->type, &pAlterOption->val);
+  return setDatabaseOptionImpl(pCxt, pOptions, pAlterOption->type, &pAlterOption->val, true);
 }

 SNode* createCreateDatabaseStmt(SAstCreateContext* pCxt, bool ignoreExists, SToken* pDbName, SNode* pOptions) {
@ -1945,6 +1952,13 @@ SNode* createBalanceVgroupStmt(SAstCreateContext* pCxt) {
  return (SNode*)pStmt;
 }

+SNode* createBalanceVgroupLeaderStmt(SAstCreateContext* pCxt) {
+  CHECK_PARSER_STATUS(pCxt);
+  SBalanceVgroupLeaderStmt* pStmt = (SBalanceVgroupLeaderStmt*)nodesMakeNode(QUERY_NODE_BALANCE_VGROUP_LEADER_STMT);
+  CHECK_OUT_OF_MEM(pStmt);
+  return (SNode*)pStmt;
+}
+
 SNode* createMergeVgroupStmt(SAstCreateContext* pCxt, const SToken* pVgId1, const SToken* pVgId2) {
  CHECK_PARSER_STATUS(pCxt);
  SMergeVgroupStmt* pStmt = (SMergeVgroupStmt*)nodesMakeNode(QUERY_NODE_MERGE_VGROUP_STMT);
--- a/source/libs/parser/src/parTokenizer.c
+++ b/source/libs/parser/src/parTokenizer.c
@ -127,6 +127,7 @@ static SKeyword keywordTable[] = {
    {"LANGUAGE",             TK_LANGUAGE},
    {"LAST",                 TK_LAST},
    {"LAST_ROW",             TK_LAST_ROW},
+    {"LEADER",               TK_LEADER},
    {"LICENCES",             TK_LICENCES},
    {"LIKE",                 TK_LIKE},
    {"LIMIT",                TK_LIMIT},
--- a/source/libs/parser/src/parTranslater.c
+++ b/source/libs/parser/src/parTranslater.c
@ -4254,6 +4254,8 @@ static void buildAlterDbReq(STranslateContext* pCxt, SAlterDatabaseStmt* pStmt,
  pReq->replications = pStmt->pOptions->replica;
  pReq->sstTrigger = pStmt->pOptions->sstTrigger;
  pReq->minRows = pStmt->pOptions->minRowsPerBlock;
+  pReq->walRetentionPeriod = pStmt->pOptions->walRetentionPeriod;
+  pReq->walRetentionSize = pStmt->pOptions->walRetentionSize;
  return;
 }

@ -6393,6 +6395,15 @@ static int32_t translateCreateFunction(STranslateContext* pCxt, SCreateFunctionS
  if (fmIsBuiltinFunc(pStmt->funcName)) {
    return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_FUNCTION_NAME);
  }
+
+  if (TSDB_DATA_TYPE_JSON == pStmt->outputDt.type ||
+      TSDB_DATA_TYPE_VARBINARY == pStmt->outputDt.type ||
+      TSDB_DATA_TYPE_DECIMAL == pStmt->outputDt.type ||
+      TSDB_DATA_TYPE_BLOB == pStmt->outputDt.type ||
+      TSDB_DATA_TYPE_MEDIUMBLOB == pStmt->outputDt.type) {
+    return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_SYNTAX_ERROR, "Unsupported output type for UDF");
+  }
+
  SCreateFuncReq req = {0};
  strcpy(req.name, pStmt->funcName);
  req.igExists = pStmt->ignoreExists;
@ -6458,6 +6469,11 @@ static int32_t translateBalanceVgroup(STranslateContext* pCxt, SBalanceVgroupStm
  return buildCmdMsg(pCxt, TDMT_MND_BALANCE_VGROUP, (FSerializeFunc)tSerializeSBalanceVgroupReq, &req);
 }

+static int32_t translateBalanceVgroupLeader(STranslateContext* pCxt, SBalanceVgroupLeaderStmt* pStmt) {
+  SBalanceVgroupLeaderReq req = {0};
+  return buildCmdMsg(pCxt, TDMT_MND_BALANCE_VGROUP_LEADER, (FSerializeFunc)tSerializeSBalanceVgroupLeaderReq, &req);
+}
+
 static int32_t translateMergeVgroup(STranslateContext* pCxt, SMergeVgroupStmt* pStmt) {
  SMergeVgroupReq req = {.vgId1 = pStmt->vgId1, .vgId2 = pStmt->vgId2};
  return buildCmdMsg(pCxt, TDMT_MND_MERGE_VGROUP, (FSerializeFunc)tSerializeSMergeVgroupReq, &req);
@ -6669,6 +6685,9 @@ static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode) {
    case QUERY_NODE_BALANCE_VGROUP_STMT:
      code = translateBalanceVgroup(pCxt, (SBalanceVgroupStmt*)pNode);
      break;
+    case QUERY_NODE_BALANCE_VGROUP_LEADER_STMT:
+      code = translateBalanceVgroupLeader(pCxt, (SBalanceVgroupLeaderStmt*)pNode);
+      break;
    case QUERY_NODE_MERGE_VGROUP_STMT:
      code = translateMergeVgroup(pCxt, (SMergeVgroupStmt*)pNode);
      break;
--- a/source/libs/parser/src/sql.c
+++ b/source/libs/parser/src/sql.c
--- a/source/libs/parser/test/parAlterToBalanceTest.cpp
+++ b/source/libs/parser/test/parAlterToBalanceTest.cpp
@ -109,7 +109,8 @@ TEST_F(ParserInitialATest, alterDnode) {
 *   | WAL_LEVEL int_value                                       -- enum 1, 2, default 1
 *   | STT_TRIGGER int_value                                     -- rang [1, 16], default 8
 *   | MINROWS int_value                                         -- rang [10, 1000], default 100
- * }
+ *   | WAL_RETENTION_PERIOD int_value                            -- rang [-1, INT32_MAX], default 0
+ *   | WAL_RETENTION_SIZE int_value                              -- rang [-1, INT32_MAX], default 0
 */
 TEST_F(ParserInitialATest, alterDatabase) {
  useDb("root", "test");
@ -135,6 +136,8 @@ TEST_F(ParserInitialATest, alterDatabase) {
    expect.replications = -1;
    expect.sstTrigger = -1;
    expect.minRows = -1;
+    expect.walRetentionPeriod = -2;
+    expect.walRetentionSize = -2;
  };
  auto setAlterDbBuffer = [&](int32_t buffer) { expect.buffer = buffer; };
  auto setAlterDbPageSize = [&](int32_t pageSize) { expect.pageSize = pageSize; };
@ -153,6 +156,10 @@ TEST_F(ParserInitialATest, alterDatabase) {
  auto setAlterDbReplica = [&](int8_t replications) { expect.replications = replications; };
  auto setAlterDbSttTrigger = [&](int8_t sstTrigger) { expect.sstTrigger = sstTrigger; };
  auto setAlterDbMinRows = [&](int32_t minRows) { expect.minRows = minRows; };
+  auto setAlterDbWalRetentionPeriod = [&](int32_t walRetentionPeriod) {
+    expect.walRetentionPeriod = walRetentionPeriod;
+  };
+  auto setAlterDbWalRetentionSize = [&](int32_t walRetentionSize) { expect.walRetentionSize = walRetentionSize; };

  setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
    ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_ALTER_DATABASE_STMT);
@ -174,6 +181,8 @@ TEST_F(ParserInitialATest, alterDatabase) {
    ASSERT_EQ(req.replications, expect.replications);
    ASSERT_EQ(req.sstTrigger, expect.sstTrigger);
    ASSERT_EQ(req.minRows, expect.minRows);
+    ASSERT_EQ(req.walRetentionPeriod, expect.walRetentionPeriod);
+    ASSERT_EQ(req.walRetentionSize, expect.walRetentionSize);
  });

  const int32_t MINUTE_PER_DAY = MILLISECOND_PER_DAY / MILLISECOND_PER_MINUTE;
@ -189,8 +198,10 @@ TEST_F(ParserInitialATest, alterDatabase) {
  setAlterDbBuffer(16);
  setAlterDbPages(128);
  setAlterDbReplica(3);
+  setAlterDbWalRetentionPeriod(10);
+  setAlterDbWalRetentionSize(20);
  run("ALTER DATABASE test BUFFER 16 CACHEMODEL 'last_row' CACHESIZE 32 WAL_FSYNC_PERIOD 200 KEEP 10 PAGES 128 "
-      "REPLICA 3 WAL_LEVEL 1 STT_TRIGGER 16");
+      "REPLICA 3 WAL_LEVEL 1 STT_TRIGGER 16 WAL_RETENTION_PERIOD 10 WAL_RETENTION_SIZE 20");
  clearAlterDbReq();

  initAlterDb("test");
@ -290,6 +301,20 @@ TEST_F(ParserInitialATest, alterDatabase) {
  setAlterDbMinRows(1000);
  run("ALTER DATABASE test MINROWS 1000");
  clearAlterDbReq();
+
+  initAlterDb("test");
+  setAlterDbWalRetentionPeriod(-1);
+  run("ALTER DATABASE test WAL_RETENTION_PERIOD -1");
+  setAlterDbWalRetentionPeriod(50);
+  run("ALTER DATABASE test WAL_RETENTION_PERIOD 50");
+  clearAlterDbReq();
+
+  initAlterDb("test");
+  setAlterDbWalRetentionSize(-1);
+  run("ALTER DATABASE test WAL_RETENTION_SIZE -1");
+  setAlterDbWalRetentionSize(50);
+  run("ALTER DATABASE test WAL_RETENTION_SIZE 50");
+  clearAlterDbReq();
 }

 TEST_F(ParserInitialATest, alterDatabaseSemanticCheck) {
@ -612,7 +637,9 @@ TEST_F(ParserInitialATest, alterTable) {
      }
      ASSERT_EQ(req.isNull, expect.isNull);
      ASSERT_EQ(req.nTagVal, expect.nTagVal);
-      ASSERT_EQ(memcmp(req.pTagVal, expect.pTagVal, expect.nTagVal), 0);
+      if (nullptr != req.pTagVal) {
+        ASSERT_EQ(memcmp(req.pTagVal, expect.pTagVal, expect.nTagVal), 0);
+      }
      ASSERT_EQ(req.updateTTL, expect.updateTTL);
      ASSERT_EQ(req.newTTL, expect.newTTL);
      if (nullptr != expect.newComment) {
@ -831,4 +858,21 @@ TEST_F(ParserInitialATest, balanceVgroup) {
  run("BALANCE VGROUP");
 }

+/*
+ * BALANCE VGROUP LEADER
+ */
+TEST_F(ParserInitialATest, balanceVgroupLeader) {
+  useDb("root", "test");
+
+  setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
+    ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_BALANCE_VGROUP_LEADER_STMT);
+    ASSERT_EQ(pQuery->pCmdMsg->msgType, TDMT_MND_BALANCE_VGROUP_LEADER);
+    SBalanceVgroupLeaderReq req = {0};
+    ASSERT_EQ(tDeserializeSBalanceVgroupLeaderReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req),
+              TSDB_CODE_SUCCESS);
+  });
+
+  run("BALANCE VGROUP LEADER");
+}
+
 }  // namespace ParserTest
--- a/source/libs/sync/src/syncMain.c
+++ b/source/libs/sync/src/syncMain.c
@ -228,6 +228,15 @@ int32_t syncLeaderTransfer(int64_t rid) {
  return ret;
 }

+int32_t syncLeaderForceElection(int64_t rid) {
+  SSyncNode* pSyncNode = syncNodeAcquire(rid);
+  if (pSyncNode == NULL) return -1;
+
+  int32_t ret = syncNodeElect(pSyncNode);
+  syncNodeRelease(pSyncNode);
+  return ret;
+}
+
 int32_t syncSendTimeoutRsp(int64_t rid, int64_t seq) {
  SSyncNode* pNode = syncNodeAcquire(rid);
  if (pNode == NULL) return -1;
--- a/tests/script/tsim/query/udf.sim
+++ b/tests/script/tsim/query/udf.sim
@ -29,6 +29,9 @@ else
  sql create function bit_and as '/tmp/udf/libbitand.so' outputtype int bufSize 8;
  sql create aggregate function l2norm as '/tmp/udf/libl2norm.so' outputtype double bufSize 8;
 endi
+
+sql_error create function bit_and as '/tmp/udf/libbitand.so' oputtype json;
+
 sql show functions;
 if $rows != 2 then
  return -1
--- a/tests/system-test/2-query/mode.py
+++ b/tests/system-test/2-query/mode.py
@ -53,6 +53,7 @@ class TDTestCase:
            ( '2022-02-01 01:01:20.000', 6, 66666, 666, 66, 6.66, 66.66, 1, "binary6", "nchar6", now()+6a )
            ( '2022-10-28 01:01:26.000', 7, 00000, 000, 00, 0.00, 00.00, 1, "binary7", "nchar7", "1970-01-01 08:00:00.000" )
            ( '2022-12-01 01:01:30.000', 8, -88888, -888, -88, -8.88, -88.88, 0, "binary8", "nchar8", "1969-01-01 01:00:00.000" )
+            ( '2022-12-31 01:01:32.000', 9, -99999999999999999, -999, -99, -9.99, -999999999999999999999.99, 1, "binary9", "nchar9", "1900-01-01 00:00:00.000" )
            ( '2022-12-31 01:01:36.000', 9, -99999999999999999, -999, -99, -9.99, -999999999999999999999.99, 1, "binary9", "nchar9", "1900-01-01 00:00:00.000" )
            ( '2023-02-21 01:01:01.000', NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL )
            '''