Merge remote-tracking branch 'origin/3.0' into feat/TD-24700

.gitignore

@@ -16,7 +16,6 @@ debug/
 release/
 target/
 debs/
-deps/
 rpms/
 mac/
 *.pyc

README-CN.md

@@ -52,7 +52,7 @@ TDengine 还提供一组辅助工具软件 taosTools，目前它包含 taosBench
 ### Ubuntu 18.04 及以上版本 & Debian：
 
 ```bash
-sudo apt-get install -y gcc cmake build-essential git libssl-dev libgflags2.2 libgflags-dev libgeos-dev
+sudo apt-get install -y gcc cmake build-essential git libssl-dev libgflags2.2 libgflags-dev
 ```
 
 #### 为 taos-tools 安装编译需要的软件
@@ -68,14 +68,14 @@ sudo apt install build-essential libjansson-dev libsnappy-dev liblzma-dev libz-d
 ```bash
 sudo yum install epel-release
 sudo yum update
-sudo yum install -y gcc gcc-c++ make cmake3 git openssl-devel geos geos-devel
+sudo yum install -y gcc gcc-c++ make cmake3 git openssl-devel
 sudo ln -sf /usr/bin/cmake3 /usr/bin/cmake
 ```
 
 ### CentOS 8/Fedora/Rocky Linux
 
 ```bash
-sudo dnf install -y gcc gcc-c++ make cmake epel-release git openssl-devel geos geos-devel
+sudo dnf install -y gcc gcc-c++ make cmake epel-release git openssl-devel
 ```
 
 #### 在 CentOS 上构建 taosTools 安装依赖软件
@@ -117,7 +117,7 @@ scl enable devtoolset-9 -- bash
 ### macOS
 
 ```
-brew install argp-standalone pkgconfig geos
+brew install argp-standalone pkgconfig
 ```
 
 ### 设置 golang 开发环境

README.md

@@ -60,7 +60,7 @@ To build TDengine, use [CMake](https://cmake.org/) 3.0.2 or higher versions in t
 ### Ubuntu 18.04 and above or Debian
 
 ```bash
-sudo apt-get install -y gcc cmake build-essential git libssl-dev libgflags2.2 libgflags-dev libgeos-dev
+sudo apt-get install -y gcc cmake build-essential git libssl-dev libgflags2.2 libgflags-dev
 ```
 
 #### Install build dependencies for taosTools
@@ -76,14 +76,14 @@ sudo apt install build-essential libjansson-dev libsnappy-dev liblzma-dev libz-d
 ```bash
 sudo yum install epel-release
 sudo yum update
-sudo yum install -y gcc gcc-c++ make cmake3 git openssl-devel geos geos-devel
+sudo yum install -y gcc gcc-c++ make cmake3 git openssl-devel
 sudo ln -sf /usr/bin/cmake3 /usr/bin/cmake
 ```
 
 ### CentOS 8/Fedora/Rocky Linux
 
 ```bash
-sudo dnf install -y gcc gcc-c++ make cmake epel-release git openssl-devel geos geos-devel
+sudo dnf install -y gcc gcc-c++ make cmake epel-release git openssl-devel
 ```
 
 #### Install build dependencies for taosTools on CentOS
@@ -124,7 +124,7 @@ scl enable devtoolset-9 -- bash
 ### macOS
 
 ```
-brew install argp-standalone pkgconfig geos
+brew install argp-standalone pkgconfig
 ```
 
 ### Setup golang environment

cmake/cmake.define

 cmake_minimum_required(VERSION 3.0)
 
-set(CMAKE_VERBOSE_MAKEFILE OFF)
+set(CMAKE_VERBOSE_MAKEFILE ON)
 set(TD_BUILD_TAOSA_INTERNAL FALSE)
 
 #set output directory
@@ -115,18 +115,6 @@ ELSE ()
         SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${GCC_COVERAGE_COMPILE_FLAGS} ${GCC_COVERAGE_LINK_FLAGS}")
     ENDIF ()
 
-    IF (${BUILD_SANITIZER})
-        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS}     -Werror -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=shift-base -fno-sanitize=alignment -g3 -Wformat=0")
-        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=shift-base -fno-sanitize=alignment -g3 -Wformat=0")
-	      MESSAGE(STATUS "Compile with Address Sanitizer!")
-    ELSEIF (${BUILD_RELEASE})
-        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -Werror=return-type -fPIC -O3 -Wformat=2 -Wno-format-nonliteral -Wno-format-truncation -Wno-format-y2k")
-        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -Wno-reserved-user-defined-literal -Wno-literal-suffix -Werror=return-type -fPIC -O3 -Wformat=2 -Wno-format-nonliteral -Wno-format-truncation -Wno-format-y2k")
-    ELSE ()
-        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -Werror=return-type -fPIC -g3 -gdwarf-2 -Wformat=2 -Wno-format-nonliteral -Wno-format-truncation -Wno-format-y2k")
-        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-reserved-user-defined-literal -g3 -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -Wformat=2 -Wno-format-nonliteral -Wno-format-truncation -Wno-format-y2k")
-    ENDIF ()
-
     # disable all assert
     IF ((${DISABLE_ASSERT} MATCHES "true") OR (${DISABLE_ASSERTS} MATCHES "true"))
         ADD_DEFINITIONS(-DDISABLE_ASSERT)
@@ -168,4 +156,20 @@ ELSE ()
         MESSAGE(STATUS "SIMD instructions (FMA/AVX/AVX2) is ACTIVATED")
     ENDIF()
 
+    # build mode
+    SET(CMAKE_C_FLAGS_REL "${CMAKE_C_FLAGS} -Werror -Werror=return-type -fPIC -O3 -Wformat=2 -Wno-format-nonliteral -Wno-format-truncation -Wno-format-y2k")
+    SET(CMAKE_CXX_FLAGS_REL "${CMAKE_CXX_FLAGS} -Werror -Wno-reserved-user-defined-literal -Wno-literal-suffix -Werror=return-type -fPIC -O3 -Wformat=2 -Wno-format-nonliteral -Wno-format-truncation -Wno-format-y2k")
+
+    IF (${BUILD_SANITIZER})
+        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS}     -Werror -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=shift-base -fno-sanitize=alignment -g3 -Wformat=0")
+        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=shift-base -fno-sanitize=alignment -g3 -Wformat=0")
+	      MESSAGE(STATUS "Compile with Address Sanitizer!")
+    ELSEIF (${BUILD_RELEASE})
+        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS_REL}")
+        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS_REL}")
+    ELSE ()
+        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -Werror=return-type -fPIC -g3 -gdwarf-2 -Wformat=2 -Wno-format-nonliteral -Wno-format-truncation -Wno-format-y2k")
+        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-reserved-user-defined-literal -g3 -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -Wformat=2 -Wno-format-nonliteral -Wno-format-truncation -Wno-format-y2k")
+    ENDIF ()
+
 ENDIF ()

cmake/cmake.platform

@@ -172,5 +172,15 @@ ENDIF()
 
 MESSAGE(STATUS "Platform arch:" ${PLATFORM_ARCH_STR})
 
+set(TD_DEPS_DIR "x86")
+if (TD_LINUX)
+  IF (TD_ARM_64 OR TD_ARM_32)
+    set(TD_DEPS_DIR "arm")
+  ELSE() 
+    set(TD_DEPS_DIR "x86")
+  ENDIF()
+endif()
+MESSAGE(STATUS "DEPS_DIR: " ${TD_DEPS_DIR})
+
 MESSAGE("C Compiler: ${CMAKE_C_COMPILER} (${CMAKE_C_COMPILER_ID}, ${CMAKE_C_COMPILER_VERSION})")
 MESSAGE("CXX Compiler: ${CMAKE_CXX_COMPILER} (${CMAKE_C_COMPILER_ID}, ${CMAKE_CXX_COMPILER_VERSION})")

cmake/cmake.version

@@ -2,7 +2,7 @@
 IF (DEFINED VERNUMBER)
   SET(TD_VER_NUMBER ${VERNUMBER})
 ELSE ()
-  SET(TD_VER_NUMBER "3.0.5.0")
+  SET(TD_VER_NUMBER "3.0.6.0.alpha")
 ENDIF ()
 
 IF (DEFINED VERCOMPATIBLE)

cmake/rocksdb_CMakeLists.txt.in

 
 # rocksdb
+IF (NOT ${TD_LINUX})
 ExternalProject_Add(rocksdb
         GIT_REPOSITORY https://github.com/facebook/rocksdb.git
         GIT_TAG v8.1.1
@@ -9,3 +10,5 @@ ExternalProject_Add(rocksdb
         INSTALL_COMMAND ""
         TEST_COMMAND ""
         )
+ENDIF(NOT ${TD_LINUX})
+

contrib/CMakeLists.txt

@@ -78,10 +78,18 @@ if(${BUILD_WITH_LEVELDB})
 endif(${BUILD_WITH_LEVELDB})
 
 # rocksdb
+IF (NOT ${TD_LINUX})
 if(${BUILD_WITH_ROCKSDB})
     cat("${TD_SUPPORT_DIR}/rocksdb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
     add_definitions(-DUSE_ROCKSDB)
 endif(${BUILD_WITH_ROCKSDB})
+ELSE()
+if(${BUILD_WITH_ROCKSDB})
+    #cat("${TD_SUPPORT_DIR}/rocksdb_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+    add_definitions(-DUSE_ROCKSDB)
+endif(${BUILD_WITH_ROCKSDB})
+    
+ENDIF(NOT ${TD_LINUX})
 
 # canonical-raft
 if(${BUILD_WITH_CRAFT})
@@ -227,9 +235,14 @@ endif(${BUILD_WITH_LEVELDB})
 
 # rocksdb
 # To support rocksdb build on ubuntu: sudo apt-get install libgflags-dev
+IF (NOT ${TD_LINUX})
+
 if(${BUILD_WITH_ROCKSDB})
     if(${TD_LINUX})
-        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error=maybe-uninitialized -Wno-error=unused-but-set-variable -Wno-error=unused-variable -Wno-error=unused-function -Wno-errno=unused-private-field -Wno-error=unused-result")
+        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS_REL} -Wno-error=maybe-uninitialized -Wno-error=unused-but-set-variable -Wno-error=unused-variable -Wno-error=unused-function -Wno-errno=unused-private-field -Wno-error=unused-result")
+        IF ("${CMAKE_BUILD_TYPE}" STREQUAL "")
+            SET(CMAKE_BUILD_TYPE Release)
+        endif()
     endif(${TD_LINUX})
     MESSAGE(STATUS "CXXXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS})
 
@@ -265,7 +278,7 @@ if(${BUILD_WITH_ROCKSDB})
     option(WITH_FALLOCATE "" OFF)
     option(WITH_JEMALLOC "" OFF)
     option(WITH_GFLAGS "" OFF)
-    option(PORTABLE "" OFF)
+    option(PORTABLE "" ON)
     option(WITH_LIBURING "" OFF)
     option(FAIL_ON_WARNINGS OFF)
 
@@ -285,6 +298,7 @@ if(${BUILD_WITH_ROCKSDB})
     )
 endif(${BUILD_WITH_ROCKSDB})
 
+ENDIF(NOT ${TD_LINUX})
 # lucene
 # To support build on ubuntu: sudo apt-get install libboost-all-dev
 if(${BUILD_WITH_LUCENE})
@@ -485,8 +499,16 @@ endif(${BUILD_ADDR2LINE})
 
 # geos
 if(${BUILD_GEOS})
+    if(${TD_LINUX})
+        set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS_REL}")
+        set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS_REL}")
+        IF ("${CMAKE_BUILD_TYPE}" STREQUAL "")
+            SET(CMAKE_BUILD_TYPE Release)
+        endif()
+    endif(${TD_LINUX})
     option(BUILD_SHARED_LIBS "Build GEOS with shared libraries" OFF)
     add_subdirectory(geos EXCLUDE_FROM_ALL)
+    unset(CMAKE_CXX_STANDARD CACHE) # undo libgeos's setting of global CMAKE_CXX_STANDARD
     target_include_directories(
         geos_c
         PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/geos/include>

docs/en/14-reference/03-connector/07-python.mdx

@@ -667,6 +667,137 @@ Insert with req_id argument
 </TabItem>
 </Tabs>
 
+### Parameter Binding
+
+The Python connector provides a parameter binding api for inserting data. Similar to most databases, TDengine currently only supports the question mark `?` to indicate the parameters to be bound.
+
+<Tabs>
+<TabItem value="native" label="native connection">
+
+#### Create Stmt
+
+Call the `statement` method in `Connection` to create the `stmt` for parameter binding.
+
+```
+import taos
+
+conn = taos.connect()
+stmt = conn.statement("insert into log values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
+```
+
+#### parameter binding
+
+Call the `new_multi_binds` function to create the parameter list for parameter bindings.
+
+```
+params = new_multi_binds(16)
+params[0].timestamp((1626861392589, 1626861392590, 1626861392591))
+params[1].bool((True, None, False))
+params[2].tinyint([-128, -128, None])  # -128 is tinyint null
+params[3].tinyint([0, 127, None])
+params[4].smallint([3, None, 2])
+params[5].int([3, 4, None])
+params[6].bigint([3, 4, None])
+params[7].tinyint_unsigned([3, 4, None])
+params[8].smallint_unsigned([3, 4, None])
+params[9].int_unsigned([3, 4, None])
+params[10].bigint_unsigned([3, 4, None])
+params[11].float([3, None, 1])
+params[12].double([3, None, 1.2])
+params[13].binary(["abc", "dddafadfadfadfadfa", None])
+params[14].nchar(["涛思数据", None, "a long string with 中文字符"])
+params[15].timestamp([None, None, 1626861392591])
+```
+
+Call the `bind_param` (for a single row) method or the `bind_param_batch` (for multiple rows) method to set the values.
+
+```
+stmt.bind_param_batch(params)
+```
+
+#### execute sql
+
+Call `execute` method to execute sql.
+
+```
+stmt.execute()
+```
+
+#### Close Stmt
+
+```
+stmt.close()
+```
+
+#### Example
+
+```python
+{{#include docs/examples/python/stmt_example.py}}
+```
+</TabItem>
+
+<TabItem value="websocket" label="WebSocket connection">
+
+#### Create Stmt
+
+Call the `statement` method in `Connection` to create the `stmt` for parameter binding.
+
+```
+import taosws
+
+conn = taosws.connect('taosws://localhost:6041/test')
+stmt = conn.statement()
+```
+
+#### Prepare sql
+
+Call `prepare` method in stmt to prepare sql.
+
+```
+stmt.prepare("insert into t1 values (?, ?, ?, ?)")
+```
+
+#### parameter binding
+
+Call the `bind_param` method to bind parameters.
+
+```
+stmt.bind_param([
+    taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
+    taosws.ints_to_column([1, 2, 3, 4]),
+    taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
+    taosws.varchar_to_column(['a', 'b', 'c', 'd']),
+])
+```
+
+Call the `add_batch` method to add parameters to the batch.
+
+```
+stmt.add_batch()
+```
+
+#### execute sql
+
+Call `execute` method to execute sql.
+
+```
+stmt.execute()
+```
+
+#### Close Stmt
+
+```
+stmt.close()
+```
+
+#### Example
+
+```python
+{{#include docs/examples/python/stmt_websocket_example.py}}
+```
+</TabItem>
+</Tabs>
+
 ### Other sample programs
 
 | Example program links | Example program content |

docs/en/14-reference/06-taosdump.md

@@ -79,8 +79,6 @@ Usage: taosdump [OPTION...] dbname [tbname ...]
   -e, --escape-character     Use escaped character for database name
   -N, --without-property     Dump database without its properties.
   -s, --schemaonly           Only dump table schemas.
-  -y, --answer-yes           Input yes for prompt. It will skip data file
-                             checking!
   -d, --avro-codec=snappy    Choose an avro codec among null, deflate, snappy,
                              and lzma.
   -S, --start-time=START_TIME   Start time to dump. Either epoch or

docs/en/14-reference/12-config/index.md

@@ -365,6 +365,16 @@ The charset that takes effect is UTF-8.
 | Unit          | GB                                                                                                |
 | Default Value | 2.0                                                                                               |
 
+### metaCacheMaxSize
+
+| Attribute     | Description                                                                                       |
+| ------------- | ------------------------------------------------------------------------------------------------- |
+| Applicable    | Client Only                                                                                       |
+| Meaning       | Maximum meta cache size in single client process                                                  |
+| Unit          | MB                                                                                                |
+| Default Value | -1 (No limitation)                                                                                |
+
+
 ## Cluster Parameters
 
 ### supportVnodes
@@ -433,6 +443,26 @@ The charset that takes effect is UTF-8.
 | Default Value | 0                                                                                                                                           |
 | Note          | When it's bigger than 0, the log file would be renamed to "taosdlog.xxx" in which "xxx" is the timestamp when the file is changed last time |
 
+### slowLogThreshold
+
+| Attribute     | Description                                                                                              |
+| ------------- | -------------------------------------------------------------------------------------------------------- |
+| Applicable    | Client only                                                                                              |
+| Meaning       | When an operation execution time exceeds this threshold, the operation will be logged in slow log file   |
+| Unit          | second                                                                                                   |
+| Default Value | 3                                                                                                        |
+| Note          | All slow operations will be logged in file "taosSlowLog" in the log directory                            |
+
+### slowLogScope
+
+| Attribute       | Description                                                             |
+| --------------- | ----------------------------------------------------------------------- |
+| Applicable      | Client only                                                             |
+| Meaning         | Slow log type to be logged                                              |
+| Optional Values | ALL, QUERY, INSERT, OTHERS, NONE                                        |
+| Default Value   | ALL                                                                     |
+| Note            | All slow operations will be logged by default, one option could be set  |
+
 ### debugFlag
 
 | Attribute     | Description                                               |

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.5.1
+
+<Release type="tdengine" version="3.0.5.1" />
+
 ## 3.0.5.0
 
 <Release type="tdengine" version="3.0.5.0" />

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

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

docs/examples/python/stmt_example.py

+#!
+
+import taosws
+
+import taos
+
+db_name = 'test_ws_stmt'
+
+
+def before():
+    taos_conn = taos.connect()
+    taos_conn.execute("drop database if exists %s" % db_name)
+    taos_conn.execute("create database %s" % db_name)
+    taos_conn.select_db(db_name)
+    taos_conn.execute("create table t1 (ts timestamp, a int, b float, c varchar(10))")
+    taos_conn.execute(
+        "create table stb1 (ts timestamp, a int, b float, c varchar(10)) tags (t1 int, t2 binary(10))")
+    taos_conn.close()
+
+
+def stmt_insert():
+    before()
+
+    conn = taosws.connect('taosws://root:taosdata@localhost:6041/%s' % db_name)
+
+    while True:
+        try:
+            stmt = conn.statement()
+            stmt.prepare("insert into t1 values (?, ?, ?, ?)")
+
+            stmt.bind_param([
+                taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
+                taosws.ints_to_column([1, 2, 3, 4]),
+                taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
+                taosws.varchar_to_column(['a', 'b', 'c', 'd']),
+            ])
+
+            stmt.add_batch()
+            rows = stmt.execute()
+            print(rows)
+            stmt.close()
+        except Exception as e:
+            if 'Retry needed' in e.args[0]:  # deal with [0x0125] Retry needed
+                continue
+            else:
+                raise e
+
+        break
+
+
+def stmt_insert_into_stable():
+    before()
+
+    conn = taosws.connect("taosws://root:taosdata@localhost:6041/%s" % db_name)
+
+    while True:
+        try:
+            stmt = conn.statement()
+            stmt.prepare("insert into ? using stb1 tags (?, ?) values (?, ?, ?, ?)")
+            stmt.set_tbname('stb1_1')
+            stmt.set_tags([
+                taosws.int_to_tag(1),
+                taosws.varchar_to_tag('aaa'),
+            ])
+            stmt.bind_param([
+                taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
+                taosws.ints_to_column([1, 2, 3, 4]),
+                taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
+                taosws.varchar_to_column(['a', 'b', 'c', 'd']),
+            ])
+
+            stmt.add_batch()
+            rows = stmt.execute()
+            print(rows)
+            stmt.close()
+        except Exception as e:
+            if 'Retry needed' in e.args[0]:  # deal with [0x0125] Retry needed
+                continue
+            else:
+                raise e
+
+        break

docs/examples/python/stmt_websocket_example.py

+#!
+import time
+
+import taosws
+
+import taos
+
+
+def before_test(db_name):
+    taos_conn = taos.connect()
+    taos_conn.execute("drop database if exists %s" % db_name)
+    taos_conn.execute("create database %s" % db_name)
+    taos_conn.select_db(db_name)
+    taos_conn.execute("create table t1 (ts timestamp, a int, b float, c varchar(10))")
+    taos_conn.execute(
+        "create table stb1 (ts timestamp, a int, b float, c varchar(10)) tags (t1 int, t2 binary(10))")
+    taos_conn.close()
+
+
+def after_test(db_name):
+    taos_conn = taos.connect()
+    taos_conn.execute("drop database if exists %s" % db_name)
+    taos_conn.close()
+
+
+def stmt_insert():
+    db_name = 'test_ws_stmt_{}'.format(int(time.time()))
+    before_test(db_name)
+
+    conn = taosws.connect('taosws://root:taosdata@localhost:6041/%s' % db_name)
+
+    stmt = conn.statement()
+    stmt.prepare("insert into t1 values (?, ?, ?, ?)")
+
+    stmt.bind_param([
+        taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
+        taosws.ints_to_column([1, 2, 3, 4]),
+        taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
+        taosws.varchar_to_column(['a', 'b', 'c', 'd']),
+    ])
+
+    stmt.add_batch()
+    rows = stmt.execute()
+    assert rows == 4
+    stmt.close()
+    after_test(db_name)
+
+
+def stmt_insert_into_stable():
+    db_name = 'test_ws_stmt_{}'.format(int(time.time()))
+    before_test(db_name)
+
+    conn = taosws.connect("taosws://root:taosdata@localhost:6041/%s" % db_name)
+
+    stmt = conn.statement()
+    stmt.prepare("insert into ? using stb1 tags (?, ?) values (?, ?, ?, ?)")
+    stmt.set_tbname('stb1_1')
+    stmt.set_tags([
+        taosws.int_to_tag(1),
+        taosws.varchar_to_tag('aaa'),
+    ])
+    stmt.bind_param([
+        taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
+        taosws.ints_to_column([1, 2, 3, 4]),
+        taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
+        taosws.varchar_to_column(['a', 'b', 'c', 'd']),
+    ])
+
+    stmt.add_batch()
+    rows = stmt.execute()
+    assert rows == 4
+    stmt.close()
+    after_test(db_name)
+
+
+if __name__ == '__main__':
+    stmt_insert()
+    stmt_insert_into_stable()

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

@@ -299,7 +299,7 @@ SELECT COUNT(*) FROM test.meters WHERE location = "California.SanFrancisco";
 SELECT AVG(current), MAX(voltage), MIN(phase) FROM test.meters WHERE groupId = 10;
 ```
 
-对表 `d10` 按 10 每秒进行平均值、最大值和最小值聚合统计：
+对表 `d10` 按每 10 秒进行平均值、最大值和最小值聚合统计：
 
 ```sql
 SELECT FIRST(ts), AVG(current), MAX(voltage), MIN(phase) FROM test.d10 INTERVAL(10s);

docs/zh/08-connector/30-python.mdx

@@ -672,6 +672,141 @@ consumer.close()
 </TabItem>
 </Tabs>
 
+### 通过参数绑定写入数据
+
+TDengine 的 Python 连接器支持参数绑定风格的 Prepare API 方式写入数据，和大多数数据库类似，目前仅支持用 `?` 来代表待绑定的参数。
+
+<Tabs>
+<TabItem value="native" label="原生连接">
+
+#### 创建 stmt
+
+Python 连接器的 `Connection` 提供了 `statement` 方法用于创建参数绑定对象 stmt，该方法接收 sql 字符串作为参数，sql 字符串目前仅支持用 `?` 来代表绑定的参数。
+
+```
+import taos
+
+conn = taos.connect()
+stmt = conn.statement("insert into log values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
+```
+
+#### 参数绑定
+
+调用 `new_multi_binds` 函数创建 params 列表，用于参数绑定。
+
+```
+params = new_multi_binds(16)
+params[0].timestamp((1626861392589, 1626861392590, 1626861392591))
+params[1].bool((True, None, False))
+params[2].tinyint([-128, -128, None])  # -128 is tinyint null
+params[3].tinyint([0, 127, None])
+params[4].smallint([3, None, 2])
+params[5].int([3, 4, None])
+params[6].bigint([3, 4, None])
+params[7].tinyint_unsigned([3, 4, None])
+params[8].smallint_unsigned([3, 4, None])
+params[9].int_unsigned([3, 4, None])
+params[10].bigint_unsigned([3, 4, None])
+params[11].float([3, None, 1])
+params[12].double([3, None, 1.2])
+params[13].binary(["abc", "dddafadfadfadfadfa", None])
+params[14].nchar(["涛思数据", None, "a long string with 中文字符"])
+params[15].timestamp([None, None, 1626861392591])
+```
+
+调用 stmt 的 `bind_param` 以单行的方式设置 values 或 `bind_param_batch` 以多行的方式设置 values 方法绑定参数。
+
+```
+stmt.bind_param_batch(params)
+```
+
+#### 执行 sql
+
+调用 stmt 的 `execute` 方法执行 sql
+
+```
+stmt.execute()
+```
+
+#### 关闭 stmt
+
+最后需要关闭 stmt。
+
+```
+stmt.close()
+```
+
+#### 示例代码
+
+```python
+{{#include docs/examples/python/stmt_example.py}}
+```
+</TabItem>
+
+<TabItem value="websocket" label="WebSocket 连接">
+
+#### 创建 stmt
+
+Python WebSocket 连接器的 `Connection` 提供了 `statement` 方法用于创建参数绑定对象 stmt，该方法接收 sql 字符串作为参数，sql 字符串目前仅支持用 `?` 来代表绑定的参数。
+
+```
+import taosws
+
+conn = taosws.connect('taosws://localhost:6041/test')
+stmt = conn.statement()
+```
+
+#### 解析 sql
+
+调用 stmt 的 `prepare` 方法来解析 insert 语句。
+
+```
+stmt.prepare("insert into t1 values (?, ?, ?, ?)")
+```
+
+#### 参数绑定
+
+调用 stmt 的 `bind_param` 方法绑定参数。
+
+```
+stmt.bind_param([
+    taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
+    taosws.ints_to_column([1, 2, 3, 4]),
+    taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
+    taosws.varchar_to_column(['a', 'b', 'c', 'd']),
+])
+```
+
+调用 stmt 的 `add_batch` 方法，将参数加入批处理。
+
+```
+stmt.add_batch()
+```
+
+#### 执行 sql
+
+调用 stmt 的 `execute` 方法执行 sql
+
+```
+stmt.execute()
+```
+
+#### 关闭 stmt
+
+最后需要关闭 stmt。
+
+```
+stmt.close()
+```
+
+#### 示例代码
+
+```python
+{{#include docs/examples/python/stmt_websocket_example.py}}
+```
+</TabItem>
+</Tabs>
+
 ### 其它示例程序
 
 | 示例程序链接                                                                                                  | 示例程序内容            |

docs/zh/14-reference/06-taosdump.md

@@ -82,8 +82,6 @@ Usage: taosdump [OPTION...] dbname [tbname ...]
   -e, --escape-character     Use escaped character for database name
   -N, --without-property     Dump database without its properties.
   -s, --schemaonly           Only dump tables' schema.
-  -y, --answer-yes           Input yes for prompt. It will skip data file
-                             checking!
   -d, --avro-codec=snappy    Choose an avro codec among null, deflate, snappy,
                              and lzma.
   -S, --start-time=START_TIME   Start time to dump. Either epoch or

docs/zh/14-reference/12-config/index.md

@@ -384,6 +384,15 @@ charset 的有效值是 UTF-8。
 | 单位     | GB                                             |
 | 缺省值   | 2.0                                            |
 
+### metaCacheMaxSize
+
+| 属性     | 说明                                           |
+| -------- | ---------------------------------------------- |
+| 适用范围 | 仅客户端适用                                   |
+| 含义     | 指定单个客户端元数据缓存大小的最大值           |
+| 单位     | MB                                             |
+| 缺省值   | -1 (无限制)                                    |
+
 ## 集群相关
 
 ### supportVnodes
@@ -452,6 +461,26 @@ charset 的有效值是 UTF-8。
 | 缺省值   | 0                                                                                   |
 | 补充说明 | 大于 0 时，日志文件会被重命名为 taosdlog.xxx，其中 xxx 为日志文件最后修改的时间戳。 |
 
+### slowLogThreshold
+
+| 属性     | 说明                                                          |
+| -------- | ------------------------------------------------------------- |
+| 适用范围 | 仅客户端适用                                                  |
+| 含义     | 指定慢查询门限值，大于等于门限值认为是慢查询                  |
+| 单位     | 秒                                                            |
+| 缺省值   | 3                                                             |
+| 补充说明 | 每个客户端中所有慢查询会被记录在日志目录下的taosSlowLog文件中 |
+
+### slowLogScope
+
+| 属性     | 说明                                                          |
+| -------- | --------------------------------------------------------------|
+| 适用范围 | 仅客户端适用                                                  |
+| 含义     | 指定启动记录哪些类型的慢查询                                  |
+| 可选值   | ALL, QUERY, INSERT, OTHERS, NONE                              |
+| 缺省值   | ALL                                                           |
+| 补充说明 | 默认记录所有类型的慢查询，可通过配置只记录某一类型的慢查询    |
+
 ### debugFlag
 
 | 属性     | 说明                                                                                              |

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.5.1
+
+<Release type="tdengine" version="3.0.5.1" />
+
 ## 3.0.5.0
 
 <Release type="tdengine" version="3.0.5.0" />

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

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

examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcDemo.java

@@ -51,27 +51,27 @@ public class JdbcDemo {
 
     private void createDatabase() {
         String sql = "create database if not exists " + dbName;
-        exuete(sql);
+        execute(sql);
     }
 
     private void useDatabase() {
         String sql = "use " + dbName;
-        exuete(sql);
+        execute(sql);
     }
 
     private void dropTable() {
         final String sql = "drop table if exists " + dbName + "." + tbName + "";
-        exuete(sql);
+        execute(sql);
     }
 
     private void createTable() {
         final String sql = "create table if not exists " + dbName + "." + tbName + " (ts timestamp, temperature float, humidity int)";
-        exuete(sql);
+        execute(sql);
     }
 
     private void insert() {
         final String sql = "insert into " + dbName + "." + tbName + " (ts, temperature, humidity) values(now, 20.5, 34)";
-        exuete(sql);
+        execute(sql);
     }
 
     private void select() {
@@ -120,7 +120,7 @@ public class JdbcDemo {
         System.out.println("[ " + (succeed ? "OK" : "ERROR!") + " ] time cost: " + cost + " ms, execute statement ====> " + sql);
     }
 
-    private void exuete(String sql) {
+    private void execute(String sql) {
         long start = System.currentTimeMillis();
         try (Statement statement = connection.createStatement()) {
             boolean execute = statement.execute(sql);

examples/JDBC/consumer-demo/pom.xml

@@ -22,7 +22,7 @@
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
-            <version>30.1.1-jre</version>
+            <version>32.0.0-jre</version>
         </dependency>
     </dependencies>
 

include/common/tcommon.h

@@ -198,6 +198,7 @@ typedef struct SDataBlockInfo {
   SBlockID    id;
   int16_t     hasVarCol;
   int16_t     dataLoad;  // denote if the data is loaded or not
+  uint8_t     scanFlag;
 
   // TODO: optimize and remove following
   int64_t     version;    // used for stream, and need serialization

include/common/tglobal.h

@@ -82,6 +82,7 @@ extern int64_t tsVndCommitMaxIntervalMs;
 // mnode
 extern int64_t tsMndSdbWriteDelta;
 extern int64_t tsMndLogRetention;
+extern int8_t  tsGrant;
 extern bool    tsMndSkipGrant;
 
 // monitor
@@ -119,6 +120,7 @@ extern bool    tsQueryUseNodeAllocator;
 extern bool    tsKeepColumnName;
 extern bool    tsEnableQueryHb;
 extern bool    tsEnableScience;
+extern bool    tsTtlChangeOnWrite;
 extern int32_t tsRedirectPeriod;
 extern int32_t tsRedirectFactor;
 extern int32_t tsRedirectMaxPeriod;
@@ -198,6 +200,7 @@ void    taosSetAllDebugFlag(int32_t flag, bool rewrite);
 void    taosSetDebugFlag(int32_t *pFlagPtr, const char *flagName, int32_t flagVal, bool rewrite);
 int32_t taosApplyLocalCfg(SConfig *pCfg, char *name);
 void    taosLocalCfgForbiddenToChange(char *name, bool *forbidden);
+int8_t  taosGranted();
 
 #ifdef __cplusplus
 }

include/common/tmsg.h

@@ -945,7 +945,7 @@ int32_t tSerializeSVTrimDbReq(void* buf, int32_t bufLen, SVTrimDbReq* pReq);
 int32_t tDeserializeSVTrimDbReq(void* buf, int32_t bufLen, SVTrimDbReq* pReq);
 
 typedef struct {
-  int32_t timestamp;
+  int32_t timestampSec;
 } SVDropTtlTableReq;
 
 int32_t tSerializeSVDropTtlTableReq(void* buf, int32_t bufLen, SVDropTtlTableReq* pReq);
@@ -1975,6 +1975,7 @@ typedef struct {
   SArray*  fillNullCols;  // array of SColLocation
   int64_t  deleteMark;
   int8_t   igUpdate;
+  int64_t  lastTs;
 } SCMCreateStreamReq;
 
 typedef struct {
@@ -2033,6 +2034,11 @@ typedef struct {
   char    cgroup[TSDB_CGROUP_LEN];
   char    clientId[256];
   SArray* topicNames;  // SArray<char**>
+
+  int8_t         withTbName;
+  int8_t         autoCommit;
+  int32_t        autoCommitInterval;
+  int8_t         resetOffsetCfg;
 } SCMSubscribeReq;
 
 static FORCE_INLINE int32_t tSerializeSCMSubscribeReq(void** buf, const SCMSubscribeReq* pReq) {
@@ -2047,6 +2053,12 @@ static FORCE_INLINE int32_t tSerializeSCMSubscribeReq(void** buf, const SCMSubsc
   for (int32_t i = 0; i < topicNum; i++) {
     tlen += taosEncodeString(buf, (char*)taosArrayGetP(pReq->topicNames, i));
   }
+
+  tlen += taosEncodeFixedI8(buf, pReq->withTbName);
+  tlen += taosEncodeFixedI8(buf, pReq->autoCommit);
+  tlen += taosEncodeFixedI32(buf, pReq->autoCommitInterval);
+  tlen += taosEncodeFixedI8(buf, pReq->resetOffsetCfg);
+
   return tlen;
 }
 
@@ -2064,6 +2076,11 @@ static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq
     buf = taosDecodeString(buf, &name);
     taosArrayPush(pReq->topicNames, &name);
   }
+
+  buf = taosDecodeFixedI8(buf, &pReq->withTbName);
+  buf = taosDecodeFixedI8(buf, &pReq->autoCommit);
+  buf = taosDecodeFixedI32(buf, &pReq->autoCommitInterval);
+  buf = taosDecodeFixedI8(buf, &pReq->resetOffsetCfg);
   return buf;
 }
 
@@ -2261,7 +2278,7 @@ typedef struct SVCreateTbReq {
   int32_t  flags;
   char*    name;
   tb_uid_t uid;
-  int64_t  ctime;
+  int64_t  btime;
   int32_t  ttl;
   int32_t  commentLen;
   char*    comment;
@@ -2398,10 +2415,12 @@ typedef struct {
   int32_t newTTL;
   int32_t newCommentLen;
   char*   newComment;
+  int64_t ctimeMs; // fill by vnode
 } SVAlterTbReq;
 
 int32_t tEncodeSVAlterTbReq(SEncoder* pEncoder, const SVAlterTbReq* pReq);
 int32_t tDecodeSVAlterTbReq(SDecoder* pDecoder, SVAlterTbReq* pReq);
+int32_t tDecodeSVAlterTbReqSetCtime(SDecoder* pDecoder, SVAlterTbReq* pReq, int64_t ctimeMs);
 
 typedef struct {
   int32_t        code;
@@ -2455,15 +2474,6 @@ typedef struct {
   char    cgroup[TSDB_CGROUP_LEN];
 } SMqAskEpReq;
 
-typedef struct {
-  int64_t consumerId;
-  int32_t epoch;
-} SMqHbReq;
-
-typedef struct {
-  int8_t reserved;
-} SMqHbRsp;
-
 typedef struct {
   int32_t key;
   int32_t valueLen;
@@ -2487,6 +2497,7 @@ typedef struct {
   int64_t  stime;  // timestamp precision ms
   int64_t  reqRid;
   bool     stableQuery;
+  bool     isSubQuery;
   char     fqdn[TSDB_FQDN_LEN];
   int32_t  subPlanNum;
   SArray*  subDesc;  // SArray<SQuerySubDesc>
@@ -2891,7 +2902,7 @@ int32_t tDecodeSMqCMCommitOffsetReq(SDecoder* decoder, SMqCMCommitOffsetReq* pRe
 // tqOffset
 enum {
   TMQ_OFFSET__RESET_NONE = -3,
-  TMQ_OFFSET__RESET_EARLIEAST = -2,
+  TMQ_OFFSET__RESET_EARLIEST = -2,
   TMQ_OFFSET__RESET_LATEST = -1,
   TMQ_OFFSET__LOG = 1,
   TMQ_OFFSET__SNAPSHOT_DATA = 2,
@@ -3354,6 +3365,28 @@ static FORCE_INLINE void tDeleteSMqAskEpRsp(SMqAskEpRsp* pRsp) {
   taosArrayDestroyEx(pRsp->topics, (FDelete)tDeleteMqSubTopicEp);
 }
 
+typedef struct {
+  int32_t      vgId;
+  STqOffsetVal offset;
+  int64_t      rows;
+}OffsetRows;
+
+typedef struct{
+  char       topicName[TSDB_TOPIC_FNAME_LEN];
+  SArray*    offsetRows;
+}TopicOffsetRows;
+
+typedef struct {
+  int64_t consumerId;
+  int32_t epoch;
+  SArray* topics;
+} SMqHbReq;
+
+typedef struct {
+  int8_t reserved;
+} SMqHbRsp;
+
+
 #define TD_AUTO_CREATE_TABLE 0x1
 typedef struct {
   int64_t       suid;
@@ -3405,6 +3438,7 @@ typedef struct SDeleteRes {
   int64_t  affectedRows;
   char     tableFName[TSDB_TABLE_NAME_LEN];
   char     tsColName[TSDB_COL_NAME_LEN];
+  int64_t  ctimeMs; // fill by vnode
 } SDeleteRes;
 
 int32_t tEncodeDeleteRes(SEncoder* pCoder, const SDeleteRes* pRes);
@@ -3423,10 +3457,12 @@ int32_t tDecodeSSingleDeleteReq(SDecoder* pCoder, SSingleDeleteReq* pReq);
 typedef struct {
   int64_t suid;
   SArray* deleteReqs;  // SArray<SSingleDeleteReq>
+  int64_t ctimeMs;     // fill by vnode
 } SBatchDeleteReq;
 
 int32_t tEncodeSBatchDeleteReq(SEncoder* pCoder, const SBatchDeleteReq* pReq);
 int32_t tDecodeSBatchDeleteReq(SDecoder* pCoder, SBatchDeleteReq* pReq);
+int32_t tDecodeSBatchDeleteReqSetCtime(SDecoder* pDecoder, SBatchDeleteReq* pReq, int64_t ctimeMs);
 
 typedef struct {
   int32_t msgIdx;
@@ -3478,10 +3514,8 @@ int32_t tSerializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
 int32_t tDeserializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
 int32_t tSerializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
 int32_t tDeserializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
-int32_t tSerializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
-int32_t tDeserializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
-int32_t tSerializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
-int32_t tDeserializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
+int32_t tDeatroySMqHbReq(SMqHbReq* pReq);
+
 
 #define SUBMIT_REQ_AUTO_CREATE_TABLE  0x1
 #define SUBMIT_REQ_COLUMN_DATA_FORMAT 0x2
@@ -3496,6 +3530,7 @@ typedef struct {
     SArray* aRowP;
     SArray* aCol;
   };
+  int64_t        ctimeMs;
 } SSubmitTbData;
 
 typedef struct {

include/common/tmsgdef.h

@@ -150,7 +150,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_TMQ_HB, "consumer-hb", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_TMQ_DO_REBALANCE, "do-rebalance", SMqDoRebalanceMsg, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_TMQ_DROP_CGROUP, "drop-cgroup", SMqDropCGroupReq, SMqDropCGroupRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_UNUSED2, "unused2", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_VG, "create-vg", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_TMQ_TIMER, "tmq-tmr", SMTimerReq, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_TELEM_TIMER, "telem-tmr", SMTimerReq, SMTimerReq)
   TD_DEF_MSG_TYPE(TDMT_MND_TRANS_TIMER, "trans-tmr", NULL, NULL)
@@ -310,6 +310,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_ADD_CHECKINFO, "vnode-tmq-add-checkinfo", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_DEL_CHECKINFO, "vnode-del-checkinfo", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_CONSUME, "vnode-tmq-consume", SMqPollReq, SMqDataBlkRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_CONSUME_PUSH, "vnode-tmq-consume-push", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_VG_WALINFO, "vnode-tmq-vg-walinfo", SMqPollReq, SMqDataBlkRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_MAX_MSG, "vnd-tmq-max", NULL, NULL)
 

include/libs/catalog/catalog.h

@@ -87,6 +87,7 @@ typedef struct SCatalogReq {
   bool    dNodeRequired;  // valid dnode
   bool    svrVerRequired;
   bool    forceUpdate;
+  bool    cloned;
 } SCatalogReq;
 
 typedef struct SMetaRes {

include/libs/executor/storageapi.h

@@ -54,7 +54,7 @@ typedef struct SMetaEntry {
       SRSmaParam     rsmaParam;
     } stbEntry;
     struct {
-      int64_t  ctime;
+      int64_t  btime;
       int32_t  ttlDays;
       int32_t  commentLen;
       char*    comment;
@@ -62,7 +62,7 @@ typedef struct SMetaEntry {
       uint8_t* pTags;
     } ctbEntry;
     struct {
-      int64_t        ctime;
+      int64_t        btime;
       int32_t        ttlDays;
       int32_t        commentLen;
       char*          comment;

include/libs/function/functionMgt.h

@@ -233,6 +233,7 @@ bool fmIsGroupKeyFunc(int32_t funcId);
 bool fmIsBlockDistFunc(int32_t funcId);
 
 void getLastCacheDataType(SDataType* pType);
+SFunctionNode* createFunction(const char* pName, SNodeList* pParameterList);
 
 int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);
 

include/libs/nodes/cmdnodes.h

@@ -432,9 +432,11 @@ typedef struct SCreateStreamStmt {
   bool                ignoreExists;
   SStreamOptions*     pOptions;
   SNode*              pQuery;
+  SNode*              pPrevQuery;
   SNodeList*          pTags;
   SNode*              pSubtable;
   SNodeList*          pCols;
+  SCMCreateStreamReq* pReq;
 } SCreateStreamStmt;
 
 typedef struct SDropStreamStmt {

include/libs/nodes/plannodes.h

@@ -53,6 +53,8 @@ typedef struct SLogicNode {
   EDataOrderLevel    requireDataOrder;  // requirements for input data
   EDataOrderLevel    resultDataOrder;   // properties of the output data
   EGroupAction       groupAction;
+  EOrder             inputTsOrder;
+  EOrder             outputTsOrder;
 } SLogicNode;
 
 typedef enum EScanType {
@@ -111,7 +113,6 @@ typedef struct SJoinLogicNode {
   SNode*     pMergeCondition;
   SNode*     pOnConditions;
   bool       isSingleTableJoin;
-  EOrder     inputTsOrder;
   SNode*     pColEqualOnConditions;
 } SJoinLogicNode;
 
@@ -229,8 +230,6 @@ typedef struct SWindowLogicNode {
   int8_t           igExpired;
   int8_t           igCheckUpdate;
   EWindowAlgorithm windowAlgo;
-  EOrder           inputTsOrder;
-  EOrder           outputTsOrder;
 } SWindowLogicNode;
 
 typedef struct SFillLogicNode {
@@ -241,7 +240,6 @@ typedef struct SFillLogicNode {
   SNode*      pWStartTs;
   SNode*      pValues;  // SNodeListNode
   STimeWindow timeRange;
-  EOrder      inputTsOrder;
 } SFillLogicNode;
 
 typedef struct SSortLogicNode {
@@ -310,6 +308,8 @@ typedef struct SDataBlockDescNode {
 
 typedef struct SPhysiNode {
   ENodeType           type;
+  EOrder              inputTsOrder;
+  EOrder              outputTsOrder;
   SDataBlockDescNode* pOutputDataBlockDesc;
   SNode*              pConditions;
   SNodeList*          pChildren;
@@ -406,7 +406,6 @@ typedef struct SSortMergeJoinPhysiNode {
   SNode*     pMergeCondition;
   SNode*     pOnConditions;
   SNodeList* pTargets;
-  EOrder     inputTsOrder;
   SNode*     pColEqualOnConditions;
 } SSortMergeJoinPhysiNode;
 
@@ -470,8 +469,6 @@ typedef struct SWindowPhysiNode {
   int64_t    watermark;
   int64_t    deleteMark;
   int8_t     igExpired;
-  EOrder     inputTsOrder;
-  EOrder     outputTsOrder;
   bool       mergeDataBlock;
 } SWindowPhysiNode;
 
@@ -498,7 +495,6 @@ typedef struct SFillPhysiNode {
   SNode*      pWStartTs;  // SColumnNode
   SNode*      pValues;    // SNodeListNode
   STimeWindow timeRange;
-  EOrder      inputTsOrder;
 } SFillPhysiNode;
 
 typedef SFillPhysiNode SStreamFillPhysiNode;
@@ -627,6 +623,7 @@ typedef struct SQueryPlan {
   int32_t      numOfSubplans;
   SNodeList*   pSubplans;  // Element is SNodeListNode. The execution level of subplan, starting from 0.
   SExplainInfo explainInfo;
+  void*        pPostPlan;
 } SQueryPlan;
 
 const char* dataOrderStr(EDataOrderLevel order);

include/libs/nodes/querynodes.h

@@ -69,6 +69,7 @@ typedef struct SColumnNode {
   uint64_t    tableId;
   int8_t      tableType;
   col_id_t    colId;
+  uint16_t    projIdx;  // the idx in project list, start from 1
   EColumnType colType;  // column or tag
   bool        hasIndex;
   char        dbName[TSDB_DB_NAME_LEN];
@@ -445,7 +446,9 @@ typedef struct SQuery {
   EQueryExecStage execStage;
   EQueryExecMode  execMode;
   bool            haveResultSet;
+  SNode*          pPrevRoot;
   SNode*          pRoot;
+  SNode*          pPostRoot;
   int32_t         numOfResCols;
   SSchema*        pResSchema;
   int8_t          precision;

include/libs/parser/parser.h

@@ -74,6 +74,7 @@ int32_t qAnalyseSqlSemantic(SParseContext* pCxt, const struct SCatalogReq* pCata
                             const struct SMetaData* pMetaData, SQuery* pQuery);
 int32_t qContinueParseSql(SParseContext* pCxt, struct SCatalogReq* pCatalogReq, const struct SMetaData* pMetaData,
                           SQuery* pQuery);
+int32_t qContinueParsePostQuery(SParseContext* pCxt, SQuery* pQuery, void** pResRow);
 
 void qDestroyParseContext(SParseContext* pCxt);
 

include/libs/planner/planner.h

@@ -52,6 +52,7 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
 // @groupId id of a group of datasource subplans of this @pSubplan
 // @pSource one execution location of this group of datasource subplans
 int32_t qSetSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId, SDownstreamSourceNode* pSource);
+int32_t qContinuePlanPostQuery(void *pPostPlan);
 
 void qClearSubplanExecutionNode(SSubplan* pSubplan);
 

include/libs/qcom/query.h

@@ -286,7 +286,8 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
    (_code) == TSDB_CODE_PAR_INVALID_DROP_COL || ((_code) == TSDB_CODE_TDB_INVALID_TABLE_ID))
 #define NEED_CLIENT_REFRESH_VG_ERROR(_code) \
   ((_code) == TSDB_CODE_VND_HASH_MISMATCH || (_code) == TSDB_CODE_VND_INVALID_VGROUP_ID)
-#define NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code) ((_code) == TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER)
+#define NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code) \
+  ((_code) == TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER || (_code) == TSDB_CODE_MND_INVALID_SCHEMA_VER)
 #define NEED_CLIENT_HANDLE_ERROR(_code)                                          \
   (NEED_CLIENT_RM_TBLMETA_ERROR(_code) || NEED_CLIENT_REFRESH_VG_ERROR(_code) || \
    NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code))

include/libs/sync/sync.h

@@ -154,14 +154,14 @@ typedef struct SSnapshotMeta {
 typedef struct SSyncFSM {
   void* data;
 
-  int32_t (*FpCommitCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta);
+  int32_t (*FpCommitCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, SFsmCbMeta* pMeta);
   SyncIndex (*FpAppliedIndexCb)(const struct SSyncFSM* pFsm);
-  int32_t (*FpPreCommitCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta);
-  void (*FpRollBackCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta);
+  int32_t (*FpPreCommitCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, SFsmCbMeta* pMeta);
+  void (*FpRollBackCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, SFsmCbMeta* pMeta);
 
   void (*FpRestoreFinishCb)(const struct SSyncFSM* pFsm, const SyncIndex commitIdx);
-  void (*FpReConfigCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SReConfigCbMeta* pMeta);
-  void (*FpLeaderTransferCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, const SFsmCbMeta* pMeta);
+  void (*FpReConfigCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, SReConfigCbMeta* pMeta);
+  void (*FpLeaderTransferCb)(const struct SSyncFSM* pFsm, SRpcMsg* pMsg, SFsmCbMeta* pMeta);
   bool (*FpApplyQueueEmptyCb)(const struct SSyncFSM* pFsm);
   int32_t (*FpApplyQueueItems)(const struct SSyncFSM* pFsm);
 

include/libs/wal/wal.h

@@ -214,7 +214,7 @@ int32_t walSkipFetchBody(SWalReader *pRead, const SWalCkHead *pHead);
 
 void walRefFirstVer(SWal *, SWalRef *);
 void walRefLastVer(SWal *, SWalRef *);
-SWalRef *walRefCommittedVer(SWal *);
+void walRefCommitVer(SWal *, SWalRef *);
 
 SWalRef *walOpenRef(SWal *);
 void     walCloseRef(SWal *pWal, int64_t refId);

include/util/talgo.h

@@ -31,7 +31,7 @@ typedef void *(*__array_item_dup_fn_t)(void *);
 
 typedef void (*FDelete)(void *);
 typedef int32_t (*FEncode)(void **buf, const void *dst);
-typedef void *(*FDecode)(const void *buf, void *dst);
+typedef void *(*FDecode)(const void *buf, void *dst, int8_t sver);
 
 #define TD_EQ 0x1
 #define TD_GT 0x2

include/util/taoserror.h

@@ -345,7 +345,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_TRANS_CLOG_IS_NULL        TAOS_DEF_ERROR_CODE(0, 0x03D4)
 #define TSDB_CODE_MND_TRANS_NETWORK_UNAVAILL    TAOS_DEF_ERROR_CODE(0, 0x03D5)
 #define TSDB_CODE_MND_LAST_TRANS_NOT_FINISHED   TAOS_DEF_ERROR_CODE(0, 0x03D6) //internal
-#define TSDB_CODE_MND_TRNAS_SYNC_TIMEOUT        TAOS_DEF_ERROR_CODE(0, 0x03D7)
+#define TSDB_CODE_MND_TRANS_SYNC_TIMEOUT        TAOS_DEF_ERROR_CODE(0, 0x03D7)
 #define TSDB_CODE_MND_TRANS_UNKNOW_ERROR        TAOS_DEF_ERROR_CODE(0, 0x03DF)
 
 // mnode-mq

include/util/tarray.h

@@ -244,7 +244,7 @@ int32_t taosArraySearchIdx(const SArray* pArray, const void* key, __compar_fn_t
 void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* param);
 
 int32_t taosEncodeArray(void** buf, const SArray* pArray, FEncode encode);
-void*   taosDecodeArray(const void* buf, SArray** pArray, FDecode decode, int32_t dataSz);
+void*   taosDecodeArray(const void* buf, SArray** pArray, FDecode decode, int32_t dataSz, int8_t sver);
 
 #ifdef __cplusplus
 }

include/util/tdef.h

@@ -195,6 +195,7 @@ typedef enum ELogicConditionType {
 #define TSDB_TABLE_NAME_LEN  193                                // it is a null-terminated string
 #define TSDB_TOPIC_NAME_LEN  193                                // it is a null-terminated string
 #define TSDB_CGROUP_LEN      193                                // it is a null-terminated string
+#define TSDB_OFFSET_LEN      64                                // it is a null-terminated string
 #define TSDB_USER_CGROUP_LEN (TSDB_USER_LEN + TSDB_CGROUP_LEN)  // it is a null-terminated string
 #define TSDB_STREAM_NAME_LEN 193                                // it is a null-terminated string
 #define TSDB_DB_NAME_LEN     65

packaging/checkPackageRuning.py

@@ -42,8 +42,8 @@ else:
 # os.system("rm -rf /var/lib/taos/*")
 # os.system("systemctl restart taosd ")
 
-# wait a moment ,at least 5 seconds
-time.sleep(5)
+# wait a moment ,at least 10 seconds
+time.sleep(10)
 
 # prepare data by taosBenchmark 
 

packaging/deb/makedeb.sh

@@ -124,12 +124,12 @@ if [ -f ${compile_dir}/build/bin/jemalloc-config ]; then
         cp ${compile_dir}/build/lib/libjemalloc.so.2 ${pkg_dir}${install_user_local_path}/lib/
         ln -sf libjemalloc.so.2 ${pkg_dir}${install_user_local_path}/lib/libjemalloc.so
     fi
-    if [ -f ${compile_dir}/build/lib/libjemalloc.a ]; then
-        cp ${compile_dir}/build/lib/libjemalloc.a ${pkg_dir}${install_user_local_path}/lib/
-    fi
-    if [ -f ${compile_dir}/build/lib/libjemalloc_pic.a ]; then
-        cp ${compile_dir}/build/lib/libjemalloc_pic.a ${pkg_dir}${install_user_local_path}/lib/
-    fi
+    # if [ -f ${compile_dir}/build/lib/libjemalloc.a ]; then
+    #     cp ${compile_dir}/build/lib/libjemalloc.a ${pkg_dir}${install_user_local_path}/lib/
+    # fi
+    # if [ -f ${compile_dir}/build/lib/libjemalloc_pic.a ]; then
+    #     cp ${compile_dir}/build/lib/libjemalloc_pic.a ${pkg_dir}${install_user_local_path}/lib/
+    # fi
     if [ -f ${compile_dir}/build/lib/pkgconfig/jemalloc.pc ]; then
         cp ${compile_dir}/build/lib/pkgconfig/jemalloc.pc ${pkg_dir}${install_user_local_path}/lib/pkgconfig/
     fi

packaging/rpm/tdengine.spec

@@ -123,12 +123,12 @@ if [ -f %{_compiledir}/build/bin/jemalloc-config ]; then
         cp %{_compiledir}/build/lib/libjemalloc.so.2 %{buildroot}%{homepath}/jemalloc/lib
         ln -sf libjemalloc.so.2 %{buildroot}%{homepath}/jemalloc/lib/libjemalloc.so
     fi
-    if [ -f %{_compiledir}/build/lib/libjemalloc.a ]; then
-        cp %{_compiledir}/build/lib/libjemalloc.a %{buildroot}%{homepath}/jemalloc/lib
-    fi
-    if [ -f %{_compiledir}/build/lib/libjemalloc_pic.a ]; then
-        cp %{_compiledir}/build/lib/libjemalloc_pic.a %{buildroot}%{homepath}/jemalloc/lib
-    fi
+#    if [ -f %{_compiledir}/build/lib/libjemalloc.a ]; then
+#        cp %{_compiledir}/build/lib/libjemalloc.a %{buildroot}%{homepath}/jemalloc/lib
+#    fi
+#    if [ -f %{_compiledir}/build/lib/libjemalloc_pic.a ]; then
+#        cp %{_compiledir}/build/lib/libjemalloc_pic.a %{buildroot}%{homepath}/jemalloc/lib
+#    fi
     if [ -f %{_compiledir}/build/lib/pkgconfig/jemalloc.pc ]; then
         cp %{_compiledir}/build/lib/pkgconfig/jemalloc.pc %{buildroot}%{homepath}/jemalloc/lib/pkgconfig
     fi

packaging/tools/install.sh

@@ -315,13 +315,13 @@ function install_jemalloc() {
       ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.so.2 /usr/local/lib
       ${csudo}ln -sf libjemalloc.so.2 /usr/local/lib/libjemalloc.so
       ${csudo}/usr/bin/install -c -d /usr/local/lib
-      if [ -f ${jemalloc_dir}/lib/libjemalloc.a ]; then
-        ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.a /usr/local/lib
-      fi
-      if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
-        ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc_pic.a /usr/local/lib
-      fi
-      if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
+      # if [ -f ${jemalloc_dir}/lib/libjemalloc.a ]; then
+      #   ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.a /usr/local/lib
+      # fi
+      # if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
+      #   ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc_pic.a /usr/local/lib
+      # fi
+      if [ -f ${jemalloc_dir}/lib/pkgconfig/jemalloc.pc ]; then
         ${csudo}/usr/bin/install -c -d /usr/local/lib/pkgconfig
         ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/lib/pkgconfig/jemalloc.pc /usr/local/lib/pkgconfig
       fi

packaging/tools/install_client.sh

@@ -214,13 +214,13 @@ function install_jemalloc() {
             ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.so.2 /usr/local/lib
             ${csudo}ln -sf libjemalloc.so.2 /usr/local/lib/libjemalloc.so
             ${csudo}/usr/bin/install -c -d /usr/local/lib
-            if [ -f ${jemalloc_dir}/lib/libjemalloc.a ]; then
-                ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.a /usr/local/lib
-            fi
-            if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
-                ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc_pic.a /usr/local/lib
-            fi
-            if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
+            # if [ -f ${jemalloc_dir}/lib/libjemalloc.a ]; then
+            #     ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.a /usr/local/lib
+            # fi
+            # if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
+            #     ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc_pic.a /usr/local/lib
+            # fi
+            if [ -f ${jemalloc_dir}/lib/pkgconfig/jemalloc.pc ]; then
                 ${csudo}/usr/bin/install -c -d /usr/local/lib/pkgconfig
                 ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/lib/pkgconfig/jemalloc.pc /usr/local/lib/pkgconfig
             fi

packaging/tools/make_install.sh

@@ -241,10 +241,10 @@ function install_jemalloc() {
       ${csudo}/usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc.so.2 /usr/local/lib
       ${csudo}ln -sf libjemalloc.so.2 /usr/local/lib/libjemalloc.so > /dev/null 2>&1
       ${csudo}/usr/bin/install -c -d /usr/local/lib
-      [ -f ${binary_dir}/build/lib/libjemalloc.a ] &&
-        ${csudo}/usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc.a /usr/local/lib
-      [ -f ${binary_dir}/build/lib/libjemalloc_pic.a ] &&
-        ${csudo}/usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc_pic.a /usr/local/lib
+      # [ -f ${binary_dir}/build/lib/libjemalloc.a ] &&
+      #   ${csudo}/usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc.a /usr/local/lib
+      # [ -f ${binary_dir}/build/lib/libjemalloc_pic.a ] &&
+      #   ${csudo}/usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc_pic.a /usr/local/lib
       if [ -f "${binary_dir}/build/lib/pkgconfig/jemalloc.pc" ]; then
         ${csudo}/usr/bin/install -c -d /usr/local/lib/pkgconfig
         ${csudo}/usr/bin/install -c -m 644 ${binary_dir}/build/lib/pkgconfig/jemalloc.pc \

packaging/tools/makeclient.sh

@@ -118,12 +118,12 @@ if [ -f ${build_dir}/bin/jemalloc-config ]; then
     cp ${build_dir}/lib/libjemalloc.so.2 ${install_dir}/jemalloc/lib
     ln -sf libjemalloc.so.2 ${install_dir}/jemalloc/lib/libjemalloc.so
   fi
-  if [ -f ${build_dir}/lib/libjemalloc.a ]; then
-    cp ${build_dir}/lib/libjemalloc.a ${install_dir}/jemalloc/lib
-  fi
-  if [ -f ${build_dir}/lib/libjemalloc_pic.a ]; then
-    cp ${build_dir}/lib/libjemalloc_pic.a ${install_dir}/jemalloc/lib
-  fi
+  # if [ -f ${build_dir}/lib/libjemalloc.a ]; then
+  #   cp ${build_dir}/lib/libjemalloc.a ${install_dir}/jemalloc/lib
+  # fi
+  # if [ -f ${build_dir}/lib/libjemalloc_pic.a ]; then
+  #   cp ${build_dir}/lib/libjemalloc_pic.a ${install_dir}/jemalloc/lib
+  # fi
   if [ -f ${build_dir}/lib/pkgconfig/jemalloc.pc ]; then
     cp ${build_dir}/lib/pkgconfig/jemalloc.pc ${install_dir}/jemalloc/lib/pkgconfig
   fi

packaging/tools/makepkg.sh

@@ -217,12 +217,12 @@ if [ -f ${build_dir}/bin/jemalloc-config ]; then
     cp ${build_dir}/lib/libjemalloc.so.2 ${install_dir}/jemalloc/lib
     ln -sf libjemalloc.so.2 ${install_dir}/jemalloc/lib/libjemalloc.so
   fi
-  if [ -f ${build_dir}/lib/libjemalloc.a ]; then
-    cp ${build_dir}/lib/libjemalloc.a ${install_dir}/jemalloc/lib
-  fi
-  if [ -f ${build_dir}/lib/libjemalloc_pic.a ]; then
-    cp ${build_dir}/lib/libjemalloc_pic.a ${install_dir}/jemalloc/lib
-  fi
+  # if [ -f ${build_dir}/lib/libjemalloc.a ]; then
+  #   cp ${build_dir}/lib/libjemalloc.a ${install_dir}/jemalloc/lib
+  # fi
+  # if [ -f ${build_dir}/lib/libjemalloc_pic.a ]; then
+  #   cp ${build_dir}/lib/libjemalloc_pic.a ${install_dir}/jemalloc/lib
+  # fi
   if [ -f ${build_dir}/lib/pkgconfig/jemalloc.pc ]; then
     cp ${build_dir}/lib/pkgconfig/jemalloc.pc ${install_dir}/jemalloc/lib/pkgconfig
   fi

packaging/tools/post.sh

@@ -169,13 +169,13 @@ function install_jemalloc() {
       ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.so.2 /usr/local/lib
       ${csudo}ln -sf libjemalloc.so.2 /usr/local/lib/libjemalloc.so
       ${csudo}/usr/bin/install -c -d /usr/local/lib
-      if [ -f ${jemalloc_dir}/lib/libjemalloc.a ]; then
-        ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.a /usr/local/lib
-      fi
-      if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
-        ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc_pic.a /usr/local/lib
-      fi
-      if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
+      # if [ -f ${jemalloc_dir}/lib/libjemalloc.a ]; then
+      #   ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc.a /usr/local/lib
+      # fi
+      # if [ -f ${jemalloc_dir}/lib/libjemalloc_pic.a ]; then
+      #   ${csudo}/usr/bin/install -c -m 755 ${jemalloc_dir}/lib/libjemalloc_pic.a /usr/local/lib
+      # fi
+      if [ -f ${jemalloc_dir}/lib/pkgconfig/jemalloc.pc ]; then
         ${csudo}/usr/bin/install -c -d /usr/local/lib/pkgconfig
         ${csudo}/usr/bin/install -c -m 644 ${jemalloc_dir}/lib/pkgconfig/jemalloc.pc /usr/local/lib/pkgconfig
       fi

source/client/inc/clientInt.h

@@ -227,6 +227,12 @@ typedef struct {
   STaosxRsp      rsp;
 } SMqTaosxRspObj;
 
+typedef struct SReqRelInfo {
+  uint64_t userRefId;
+  uint64_t prevRefId;
+  uint64_t nextRefId;
+} SReqRelInfo;
+
 typedef struct SRequestObj {
   int8_t               resType;  // query or tmq
   uint64_t             requestId;
@@ -250,10 +256,14 @@ typedef struct SRequestObj {
   bool                 validateOnly;  // todo refactor
   bool                 killed;
   bool                 inRetry;
+  bool                 isSubReq;
   uint32_t             prevCode;  // previous error code: todo refactor, add update flag for catalog
   uint32_t             retry;
   int64_t              allocatorRefId;
   SQuery*              pQuery;
+  void*                pPostPlan;
+  SReqRelInfo          relation;
+  void*                pWrapper;
 } SRequestObj;
 
 typedef struct SSyncQueryParam {
@@ -279,6 +289,7 @@ TAOS_RES* taosQueryImplWithReqid(TAOS* taos, const char* sql, bool validateOnly,
 void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly);
 void taosAsyncQueryImplWithReqid(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly,
                                  int64_t reqid);
+void taosAsyncFetchImpl(SRequestObj *pRequest, __taos_async_fn_t fp, void *param);
 
 int32_t getVersion1BlockMetaSize(const char* p, int32_t numOfCols);
 
@@ -368,6 +379,7 @@ typedef struct SSqlCallbackWrapper {
   SParseContext* pParseCtx;
   SCatalogReq*   pCatalogReq;
   SRequestObj*   pRequest;
+  void*          pPlanInfo;
 } SSqlCallbackWrapper;
 
 SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQuery, void** res);
@@ -382,6 +394,12 @@ int32_t handleCreateTbExecRes(void* res, SCatalog* pCatalog);
 bool    qnodeRequired(SRequestObj* pRequest);
 void    continueInsertFromCsv(SSqlCallbackWrapper* pWrapper, SRequestObj* pRequest);
 void    destorySqlCallbackWrapper(SSqlCallbackWrapper* pWrapper);
+void    handleQueryAnslyseRes(SSqlCallbackWrapper *pWrapper, SMetaData *pResultMeta, int32_t code);
+void    restartAsyncQuery(SRequestObj *pRequest, int32_t code);
+int32_t buildPreviousRequest(SRequestObj *pRequest, const char* sql, SRequestObj** pNewRequest);
+int32_t prepareAndParseSqlSyntax(SSqlCallbackWrapper **ppWrapper, SRequestObj *pRequest, bool updateMetaForce);
+void    returnToUser(SRequestObj* pRequest);
+void    stopAllQueries(SRequestObj *pRequest);
 
 #ifdef __cplusplus
 }

source/client/src/clientEnv.c

@@ -358,6 +358,49 @@ int32_t releaseRequest(int64_t rid) { return taosReleaseRef(clientReqRefPool, ri
 
 int32_t removeRequest(int64_t rid) { return taosRemoveRef(clientReqRefPool, rid); }
 
+
+void destroySubRequests(SRequestObj *pRequest) {
+  int32_t reqIdx = -1;
+  SRequestObj *pReqList[16] = {NULL};
+  uint64_t tmpRefId = 0;
+
+  if (pRequest->relation.userRefId && pRequest->relation.userRefId != pRequest->self) {
+    return;
+  }
+  
+  SRequestObj* pTmp = pRequest;
+  while (pTmp->relation.prevRefId) {
+    tmpRefId = pTmp->relation.prevRefId;
+    pTmp = acquireRequest(tmpRefId);
+    if (pTmp) {
+      pReqList[++reqIdx] = pTmp;
+      releaseRequest(tmpRefId);
+    } else {
+      tscError("0x%" PRIx64 ", prev req ref 0x%" PRIx64 " is not there, reqId:0x%" PRIx64, pTmp->self,
+               tmpRefId, pTmp->requestId);
+      break;         
+    }
+  }
+
+  for (int32_t i = reqIdx; i >= 0; i--) {
+    removeRequest(pReqList[i]->self);
+  }
+
+  tmpRefId = pRequest->relation.nextRefId;
+  while (tmpRefId) {
+    pTmp = acquireRequest(tmpRefId);
+    if (pTmp) {
+      tmpRefId = pTmp->relation.nextRefId;
+      removeRequest(pTmp->self);      
+      releaseRequest(pTmp->self);
+    } else {
+      tscError("0x%" PRIx64 " is not there", tmpRefId);
+      break;         
+    }
+  }
+}
+
+
 void doDestroyRequest(void *p) {
   if (NULL == p) {
     return;
@@ -368,10 +411,14 @@ void doDestroyRequest(void *p) {
   uint64_t reqId = pRequest->requestId;
   tscTrace("begin to destroy request %" PRIx64 " p:%p", reqId, pRequest);
 
+  destroySubRequests(pRequest);
+  
   taosHashRemove(pRequest->pTscObj->pRequests, &pRequest->self, sizeof(pRequest->self));
 
   schedulerFreeJob(&pRequest->body.queryJob, 0);
 
+  destorySqlCallbackWrapper(pRequest->pWrapper);
+
   taosMemoryFreeClear(pRequest->msgBuf);
   taosMemoryFreeClear(pRequest->pDb);
 
@@ -412,6 +459,63 @@ void destroyRequest(SRequestObj *pRequest) {
   removeRequest(pRequest->self);
 }
 
+void taosStopQueryImpl(SRequestObj *pRequest) {
+  pRequest->killed = true;
+
+  // It is not a query, no need to stop.
+  if (NULL == pRequest->pQuery || QUERY_EXEC_MODE_SCHEDULE != pRequest->pQuery->execMode) {
+    tscDebug("request 0x%" PRIx64 " no need to be killed since not query", pRequest->requestId);
+    return;
+  }
+
+  schedulerFreeJob(&pRequest->body.queryJob, TSDB_CODE_TSC_QUERY_KILLED);
+  tscDebug("request %" PRIx64 " killed", pRequest->requestId);
+}
+
+void stopAllQueries(SRequestObj *pRequest) {
+  int32_t reqIdx = -1;
+  SRequestObj *pReqList[16] = {NULL};
+  uint64_t tmpRefId = 0;
+
+  if (pRequest->relation.userRefId && pRequest->relation.userRefId != pRequest->self) {
+    return;
+  }
+  
+  SRequestObj* pTmp = pRequest;
+  while (pTmp->relation.prevRefId) {
+    tmpRefId = pTmp->relation.prevRefId;
+    pTmp = acquireRequest(tmpRefId);
+    if (pTmp) {
+      pReqList[++reqIdx] = pTmp;
+      releaseRequest(tmpRefId);
+    } else {
+      tscError("0x%" PRIx64 ", prev req ref 0x%" PRIx64 " is not there, reqId:0x%" PRIx64, pTmp->self,
+               tmpRefId, pTmp->requestId);
+      break;         
+    }
+  }
+
+  for (int32_t i = reqIdx; i >= 0; i--) {
+    taosStopQueryImpl(pReqList[i]);
+  }
+
+  taosStopQueryImpl(pRequest);
+
+  tmpRefId = pRequest->relation.nextRefId;
+  while (tmpRefId) {
+    pTmp = acquireRequest(tmpRefId);
+    if (pTmp) {
+      tmpRefId = pTmp->relation.nextRefId;
+      taosStopQueryImpl(pTmp);
+      releaseRequest(pTmp->self);
+    } else {
+      tscError("0x%" PRIx64 " is not there", tmpRefId);
+      break;         
+    }
+  }
+}
+
+
 void crashReportThreadFuncUnexpectedStopped(void) { atomic_store_32(&clientStop, -1); }
 
 static void *tscCrashReportThreadFp(void *param) {

source/client/src/clientHb.c

@@ -464,6 +464,7 @@ int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) {
     desc.useconds = now - pRequest->metric.start;
     desc.reqRid = pRequest->self;
     desc.stableQuery = pRequest->stableQuery;
+    desc.isSubQuery = pRequest->isSubReq;
     taosGetFqdn(desc.fqdn);
     desc.subPlanNum = pRequest->body.subplanNum;
 

source/client/src/clientImpl.c

@@ -237,6 +237,17 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param,
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t buildPreviousRequest(SRequestObj *pRequest, const char* sql, SRequestObj** pNewRequest) {
+  int32_t code = buildRequest(pRequest->pTscObj->id, sql, strlen(sql), pRequest, pRequest->validateOnly, pNewRequest, 0);
+  if (TSDB_CODE_SUCCESS == code) {
+    pRequest->relation.prevRefId = (*pNewRequest)->self;
+    (*pNewRequest)->relation.nextRefId = pRequest->self;
+    (*pNewRequest)->relation.userRefId = pRequest->self;
+    (*pNewRequest)->isSubReq = true;
+  }
+  return code;
+}
+
 int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtCallback* pStmtCb) {
   STscObj* pTscObj = pRequest->pTscObj;
 
@@ -878,6 +889,81 @@ static bool incompletaFileParsing(SNode* pStmt) {
   return QUERY_NODE_VNODE_MODIFY_STMT != nodeType(pStmt) ? false : ((SVnodeModifyOpStmt*)pStmt)->fileProcessing;
 }
 
+void continuePostSubQuery(SRequestObj* pRequest, TAOS_ROW row) {
+  SSqlCallbackWrapper* pWrapper = pRequest->pWrapper;
+  int32_t code = nodesAcquireAllocator(pWrapper->pParseCtx->allocatorId);
+  if (TSDB_CODE_SUCCESS == code) {
+    int64_t analyseStart = taosGetTimestampUs();
+    code = qContinueParsePostQuery(pWrapper->pParseCtx, pRequest->pQuery, (void**)row);
+    pRequest->metric.analyseCostUs += taosGetTimestampUs() - analyseStart;
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = qContinuePlanPostQuery(pRequest->pPostPlan);
+  }
+  nodesReleaseAllocator(pWrapper->pParseCtx->allocatorId);
+
+  handleQueryAnslyseRes(pWrapper, NULL, code);
+}
+
+void returnToUser(SRequestObj* pRequest) {
+  if (pRequest->relation.userRefId == pRequest->self || 0 == pRequest->relation.userRefId) {
+    // return to client
+    pRequest->body.queryFp(pRequest->body.param, pRequest, pRequest->code);
+    return;
+  }
+
+  SRequestObj* pUserReq = acquireRequest(pRequest->relation.userRefId);
+  if (pUserReq) {
+    pUserReq->code = pRequest->code;
+    // return to client
+    pUserReq->body.queryFp(pUserReq->body.param, pUserReq, pUserReq->code);
+    releaseRequest(pRequest->relation.userRefId);
+    return;
+  } else {
+    tscError("0x%" PRIx64 ", user ref 0x%" PRIx64 " is not there, reqId:0x%" PRIx64, pRequest->self,
+             pRequest->relation.userRefId, pRequest->requestId);
+  }
+}
+
+void postSubQueryFetchCb(void* param, TAOS_RES* res, int32_t rowNum) {
+  SRequestObj* pRequest = (SRequestObj*)res;
+  if (pRequest->code) {
+    returnToUser(pRequest);
+    return;
+  }
+
+  TAOS_ROW row = NULL;
+  if (rowNum > 0) {
+    row = taos_fetch_row(res); // for single row only now
+  }
+
+  SRequestObj* pNextReq = acquireRequest(pRequest->relation.nextRefId);
+  if (pNextReq) {
+    continuePostSubQuery(pNextReq, row);
+    releaseRequest(pRequest->relation.nextRefId);
+  } else {
+    tscError("0x%" PRIx64 ", next req ref 0x%" PRIx64 " is not there, reqId:0x%" PRIx64, pRequest->self,
+             pRequest->relation.nextRefId, pRequest->requestId);
+  }
+}
+
+void handlePostSubQuery(SSqlCallbackWrapper* pWrapper) {
+  SRequestObj* pRequest = pWrapper->pRequest;
+  if (TD_RES_QUERY(pRequest)) {
+    taosAsyncFetchImpl(pRequest, postSubQueryFetchCb, pWrapper);
+    return;
+  }
+
+  SRequestObj* pNextReq = acquireRequest(pRequest->relation.nextRefId);
+  if (pNextReq) {
+    continuePostSubQuery(pNextReq, NULL);
+    releaseRequest(pRequest->relation.nextRefId);
+  } else {
+    tscError("0x%" PRIx64 ", next req ref 0x%" PRIx64 " is not there, reqId:0x%" PRIx64, pRequest->self,
+             pRequest->relation.nextRefId, pRequest->requestId);
+  }
+}
+
 // todo refacto the error code  mgmt
 void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) {
   SSqlCallbackWrapper* pWrapper = param;
@@ -912,12 +998,7 @@ void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) {
   if (code != TSDB_CODE_SUCCESS && NEED_CLIENT_HANDLE_ERROR(code) && pRequest->sqlstr != NULL) {
     tscDebug("0x%" PRIx64 " client retry to handle the error, code:%s, tryCount:%d, reqId:0x%" PRIx64, pRequest->self,
              tstrerror(code), pRequest->retry, pRequest->requestId);
-    pRequest->prevCode = code;
-    schedulerFreeJob(&pRequest->body.queryJob, 0);
-    qDestroyQuery(pRequest->pQuery);
-    pRequest->pQuery = NULL;
-    destorySqlCallbackWrapper(pWrapper);
-    doAsyncQuery(pRequest, true);
+    restartAsyncQuery(pRequest, code);
     return;
   }
 
@@ -938,10 +1019,15 @@ void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) {
     return;
   }
 
+  if (pRequest->relation.nextRefId) {
+    handlePostSubQuery(pWrapper);
+  } else {
     destorySqlCallbackWrapper(pWrapper);
+    pRequest->pWrapper = NULL;
 
     // return to client
     pRequest->body.queryFp(pRequest->body.param, pRequest, code);
+  }
 }
 
 SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQuery, void** res) {
@@ -1049,6 +1135,7 @@ static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaDat
              pRequest->requestId);
   } else {
     pRequest->body.subplanNum = pDag->numOfSubplans;
+    TSWAP(pRequest->pPostPlan, pDag->pPostPlan);
   }
 
   pRequest->metric.execStart = taosGetTimestampUs();
@@ -1084,6 +1171,7 @@ static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaDat
     tscDebug("0x%" PRIx64 " plan not executed, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
              pRequest->requestId);
     destorySqlCallbackWrapper(pWrapper);
+    pRequest->pWrapper = NULL;
     if (TSDB_CODE_SUCCESS != code) {
       pRequest->code = terrno;
     }
@@ -1103,6 +1191,7 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
   pRequest->body.execMode = pQuery->execMode;
   if (QUERY_EXEC_MODE_SCHEDULE != pRequest->body.execMode) {
     destorySqlCallbackWrapper(pWrapper);
+    pRequest->pWrapper = NULL;
   }
 
   if (pQuery->pRoot && !pRequest->inRetry) {
@@ -2402,3 +2491,90 @@ TAOS_RES* taosQueryImplWithReqid(TAOS* taos, const char* sql, bool validateOnly,
 
   return pRequest;
 }
+
+
+static void fetchCallback(void *pResult, void *param, int32_t code) {
+  SRequestObj *pRequest = (SRequestObj *)param;
+
+  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
+
+  tscDebug("0x%" PRIx64 " enter scheduler fetch cb, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code,
+           tstrerror(code), pRequest->requestId);
+
+  pResultInfo->pData = pResult;
+  pResultInfo->numOfRows = 0;
+
+  if (code != TSDB_CODE_SUCCESS) {
+    pRequest->code = code;
+    taosMemoryFreeClear(pResultInfo->pData);
+    pRequest->body.fetchFp(pRequest->body.param, pRequest, 0);
+    return;
+  }
+
+  if (pRequest->code != TSDB_CODE_SUCCESS) {
+    taosMemoryFreeClear(pResultInfo->pData);
+    pRequest->body.fetchFp(pRequest->body.param, pRequest, 0);
+    return;
+  }
+
+  pRequest->code =
+      setQueryResultFromRsp(pResultInfo, (const SRetrieveTableRsp *)pResultInfo->pData, pResultInfo->convertUcs4, true);
+  if (pRequest->code != TSDB_CODE_SUCCESS) {
+    pResultInfo->numOfRows = 0;
+    pRequest->code = code;
+    tscError("0x%" PRIx64 " fetch results failed, code:%s, reqId:0x%" PRIx64, pRequest->self, tstrerror(code),
+             pRequest->requestId);
+  } else {
+    tscDebug("0x%" PRIx64 " fetch results, numOfRows:%" PRId64 " total Rows:%" PRId64 ", complete:%d, reqId:0x%" PRIx64,
+             pRequest->self, pResultInfo->numOfRows, pResultInfo->totalRows, pResultInfo->completed,
+             pRequest->requestId);
+
+    STscObj            *pTscObj = pRequest->pTscObj;
+    SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary;
+    atomic_add_fetch_64((int64_t *)&pActivity->fetchBytes, pRequest->body.resInfo.payloadLen);
+  }
+
+  pRequest->body.fetchFp(pRequest->body.param, pRequest, pResultInfo->numOfRows);
+}
+
+void taosAsyncFetchImpl(SRequestObj *pRequest, __taos_async_fn_t fp, void *param) {
+  pRequest->body.fetchFp = fp;
+  pRequest->body.param = param;
+
+  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
+
+  // this query has no results or error exists, return directly
+  if (taos_num_fields(pRequest) == 0 || pRequest->code != TSDB_CODE_SUCCESS) {
+    pResultInfo->numOfRows = 0;
+    pRequest->body.fetchFp(param, pRequest, pResultInfo->numOfRows);
+    return;
+  }
+
+  // all data has returned to App already, no need to try again
+  if (pResultInfo->completed) {
+    // it is a local executed query, no need to do async fetch
+    if (QUERY_EXEC_MODE_SCHEDULE != pRequest->body.execMode) {
+      if (pResultInfo->localResultFetched) {
+        pResultInfo->numOfRows = 0;
+        pResultInfo->current = 0;
+      } else {
+        pResultInfo->localResultFetched = true;
+      }
+    } else {
+      pResultInfo->numOfRows = 0;
+    }
+
+    pRequest->body.fetchFp(param, pRequest, pResultInfo->numOfRows);
+    return;
+  }
+
+  SSchedulerReq req = {
+      .syncReq = false,
+      .fetchFp = fetchCallback,
+      .cbParam = pRequest,
+  };
+
+  schedulerFetchRows(pRequest->body.queryJob, &req);
+}
+
+

source/client/src/clientMsgHandler.c

@@ -77,6 +77,7 @@ int32_t processConnectRsp(void* param, SDataBuf* pMsg, int32_t code) {
   }
 
   if ((code = taosCheckVersionCompatibleFromStr(version, connectRsp.sVer, 3)) != 0) {
+    tscError("version not compatible. client version: %s, server version: %s", version, connectRsp.sVer);
     setErrno(pRequest, code);
     tsem_post(&pRequest->body.rspSem);
     goto End;

source/client/src/clientSml.c

@@ -1553,17 +1553,8 @@ static int32_t smlParseLine(SSmlHandle *info, char *lines[], char *rawLine, char
       }
     }
 
-    char cTmp = 0;  // for print tmp if is raw
-    if (info->isRawLine) {
-      cTmp = tmp[len];
-      tmp[len] = '\0';
-    }
-
     uDebug("SML:0x%" PRIx64 " smlParseLine israw:%d, numLines:%d, protocol:%d, len:%d, sql:%s", info->id,
-           info->isRawLine, numLines, info->protocol, len, tmp);
-    if (info->isRawLine) {
-      tmp[len] = cTmp;
-    }
+           info->isRawLine, numLines, info->protocol, len, info->isRawLine ? "rawdata" : tmp);
 
     if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
       if (info->dataFormat) {
@@ -1584,8 +1575,7 @@ static int32_t smlParseLine(SSmlHandle *info, char *lines[], char *rawLine, char
       code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
     }
     if (code != TSDB_CODE_SUCCESS) {
-      tmp[len] = '\0';
-      uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, tmp);
+      uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, info->isRawLine ? "rawdata" : tmp);
       return code;
     }
     if (info->reRun) {
@@ -1756,9 +1746,8 @@ TAOS_RES *taos_schemaless_insert_inner(TAOS *taos, char *lines[], char *rawLine,
     request->code = code;
     info->cost.endTime = taosGetTimestampUs();
     info->cost.code = code;
-    if (code == TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER || code == TSDB_CODE_SDB_OBJ_CREATING ||
-        code == TSDB_CODE_PAR_VALUE_TOO_LONG || code == TSDB_CODE_MND_TRANS_CONFLICT ||
-        code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {
+    if (NEED_CLIENT_HANDLE_ERROR(code) || code == TSDB_CODE_SDB_OBJ_CREATING ||
+        code == TSDB_CODE_PAR_VALUE_TOO_LONG || code == TSDB_CODE_MND_TRANS_CONFLICT) {
       if (cnt++ >= 10) {
         uInfo("SML:%" PRIx64 " retry:%d/10 end code:%d, msg:%s", info->id, cnt, code, tstrerror(code));
         break;

source/common/src/systable.c

@@ -280,7 +280,7 @@ static const SSysDbTableSchema topicSchema[] = {
     {.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = false},
     {.name = "sql", .bytes = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
-    {.name = "schema", .bytes = TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
+    {.name = "schema", .bytes = TSDB_MAX_BINARY_LEN, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
     {.name = "meta", .bytes = 4 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
     {.name = "type", .bytes = 8 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
 };
@@ -291,6 +291,8 @@ static const SSysDbTableSchema subscriptionSchema[] = {
     {.name = "consumer_group", .bytes = TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
     {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
     {.name = "consumer_id", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = false},
+    {.name = "offset", .bytes = TSDB_OFFSET_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
+    {.name = "rows", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = false},
 };
 
 static const SSysDbTableSchema vnodesSchema[] = {
@@ -359,6 +361,7 @@ static const SSysDbTableSchema consumerSchema[] = {
     {.name = "up_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = false},
     {.name = "subscribe_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = false},
     {.name = "rebalance_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = false},
+    {.name = "parameters", .bytes = 64 + TSDB_OFFSET_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
 };
 
 static const SSysDbTableSchema offsetSchema[] = {
@@ -381,6 +384,7 @@ static const SSysDbTableSchema querySchema[] = {
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = false},
     {.name = "exec_usec", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = false},
     {.name = "stable_query", .bytes = 1, .type = TSDB_DATA_TYPE_BOOL, .sysInfo = false},
+    {.name = "sub_query", .bytes = 1, .type = TSDB_DATA_TYPE_BOOL, .sysInfo = false},
     {.name = "sub_num", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
     {.name = "sub_status", .bytes = TSDB_SHOW_SUBQUERY_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
     {.name = "sql", .bytes = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},

source/common/src/tdatablock.c

@@ -841,8 +841,8 @@ size_t blockDataGetRowSize(SSDataBlock* pBlock) {
  * @return
  */
 size_t blockDataGetSerialMetaSize(uint32_t numOfCols) {
-  // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column
-  // length |
+  // | version | total length | total rows | total columns | flag seg| block group id | column schema
+  // | each column length |
   return sizeof(int32_t) + sizeof(int32_t) + sizeof(int32_t) + sizeof(int32_t) + sizeof(int32_t) + sizeof(uint64_t) +
          numOfCols * (sizeof(int8_t) + sizeof(int32_t)) + numOfCols * sizeof(int32_t);
 }
@@ -1640,18 +1640,35 @@ size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize, int
   int32_t nRows = payloadSize / rowSize;
   ASSERT(nRows >= 1);
 
-  // the true value must be less than the value of nRows
-  int32_t additional = 0;
+  int32_t numVarCols = 0;
+  int32_t numFixCols = 0;
   for (int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i);
     if (IS_VAR_DATA_TYPE(pCol->info.type)) {
-      additional += nRows * sizeof(int32_t);
+      ++numVarCols;
+    } else {
+      ++numFixCols;
+    }
+  }
+
+  // find the data payload whose size is greater than payloadSize
+  int result = -1;
+  int start = 1;
+  int end = nRows;
+  while (start <= end) {
+    int mid = start + (end - start) / 2;
+    //data size + var data type columns offset + fixed data type columns bitmap len 
+    int midSize = rowSize * mid + numVarCols * sizeof(int32_t) * mid + numFixCols * BitmapLen(mid); 
+    if (midSize > payloadSize) {
+      result = mid;
+      end = mid - 1;
     } else {
-      additional += BitmapLen(nRows);
+      start = mid + 1;
     }
   }
 
-  int32_t newRows = (payloadSize - additional) / rowSize;
+  int32_t newRows = (result != -1) ? result - 1 : nRows;
+  // the true value must be less than the value of nRows
   ASSERT(newRows <= nRows && newRows >= 1);
 
   return newRows;

source/common/src/tglobal.c

@@ -14,6 +14,7 @@
  */
 
 #define _DEFAULT_SOURCE
+#include "os.h"
 #include "tglobal.h"
 #include "tconfig.h"
 #include "tgrant.h"
@@ -73,6 +74,7 @@ int64_t tsVndCommitMaxIntervalMs = 600 * 1000;
 // mnode
 int64_t tsMndSdbWriteDelta = 200;
 int64_t tsMndLogRetention = 2000;
+int8_t  tsGrant = 1;
 bool    tsMndSkipGrant = false;
 
 // monitor
@@ -108,6 +110,7 @@ int32_t tsQueryRspPolicy = 0;
 int64_t tsQueryMaxConcurrentTables = 200;  // unit is TSDB_TABLE_NUM_UNIT
 bool    tsEnableQueryHb = false;
 bool    tsEnableScience = false;  // on taos-cli show float and doulbe with scientific notation if true
+bool    tsTtlChangeOnWrite = false; // ttl delete time changes on last write if true
 int32_t tsQuerySmaOptimize = 0;
 int32_t tsQueryRsmaTolerance = 1000;  // the tolerance time (ms) to judge from which level to query rsma data.
 bool    tsQueryPlannerTrace = false;
@@ -511,6 +514,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "mqRebalanceInterval", tsMqRebalanceInterval, 1, 10000, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "ttlUnit", tsTtlUnit, 1, 86400 * 365, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "ttlPushInterval", tsTtlPushInterval, 1, 100000, 1) != 0) return -1;
+  if (cfgAddBool(pCfg, "ttlChangeOnWrite", tsTtlChangeOnWrite, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "uptimeInterval", tsUptimeInterval, 1, 100000, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "queryRsmaTolerance", tsQueryRsmaTolerance, 0, 900000, 0) != 0) return -1;
 
@@ -871,6 +875,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
 
   tsEnableTelem = cfgGetItem(pCfg, "telemetryReporting")->bval;
   tsEnableCrashReport = cfgGetItem(pCfg, "crashReporting")->bval;
+  tsTtlChangeOnWrite = cfgGetItem(pCfg, "ttlChangeOnWrite")->bval;
   tsTelemInterval = cfgGetItem(pCfg, "telemetryInterval")->i32;
   tstrncpy(tsTelemServer, cfgGetItem(pCfg, "telemetryServer")->str, TSDB_FQDN_LEN);
   tsTelemPort = (uint16_t)cfgGetItem(pCfg, "telemetryPort")->i32;
@@ -976,6 +981,8 @@ int32_t taosApplyLocalCfg(SConfig *pCfg, char *name) {
         taosSetCoreDump(enableCore);
       } else if (strcasecmp("enableQueryHb", name) == 0) {
         tsEnableQueryHb = cfgGetItem(pCfg, "enableQueryHb")->bval;
+      }  else if (strcasecmp("ttlChangeOnWrite", name) == 0) {
+        tsTtlChangeOnWrite = cfgGetItem(pCfg, "ttlChangeOnWrite")->bval;
       }
       break;
     }
@@ -1525,3 +1532,5 @@ void taosSetAllDebugFlag(int32_t flag, bool rewrite) {
   taosSetDebugFlag(&metaDebugFlag, "metaDebugFlag", flag, rewrite);
   uInfo("all debug flag are set to %d", flag);
 }
+
+int8_t taosGranted() { return atomic_load_8(&tsGrant); }

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

@@ -163,7 +163,7 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_DROP_TOPIC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_SUBSCRIBE, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_ASK_EP, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_HB, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_HB, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_DROP_CGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_DROP_CGROUP_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_KILL_TRANS, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

source/dnode/mgmt/mgmt_vnode/inc/vmInt.h

@@ -46,6 +46,7 @@ typedef struct {
   int32_t vgId;
   int32_t vgVersion;
   int8_t  dropped;
+  int32_t toVgId;
   char    path[PATH_MAX + 20];
 } SWrapperCfg;
 
@@ -55,6 +56,7 @@ typedef struct {
   int32_t       refCount;
   int8_t        dropped;
   int8_t        disable;
+  int32_t       toVgId;
   char         *path;
   SVnode       *pImpl;
   SMultiWorker  pWriteW;
@@ -70,6 +72,7 @@ typedef struct {
   int32_t      vnodeNum;
   int32_t      opened;
   int32_t      failed;
+  bool         updateVnodesList;
   int32_t      threadIndex;
   TdThread     thread;
   SVnodeMgmt  *pMgmt;

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

@@ -71,6 +71,8 @@ static int32_t vmDecodeVnodeList(SJson *pJson, SVnodeMgmt *pMgmt, SWrapperCfg **
     if (code < 0) goto _OVER;
     tjsonGetInt32ValueFromDouble(vnode, "vgVersion", pCfg->vgVersion, code);
     if (code < 0) goto _OVER;
+    tjsonGetInt32ValueFromDouble(vnode, "toVgId", pCfg->toVgId, code);
+    if (code < 0) goto _OVER;
 
     snprintf(pCfg->path, sizeof(pCfg->path), "%s%svnode%d", pMgmt->path, TD_DIRSEP, pCfg->vgId);
   }
@@ -165,6 +167,7 @@ static int32_t vmEncodeVnodeList(SJson *pJson, SVnodeObj **ppVnodes, int32_t num
     if (tjsonAddDoubleToObject(vnode, "vgId", pVnode->vgId) < 0) return -1;
     if (tjsonAddDoubleToObject(vnode, "dropped", pVnode->dropped) < 0) return -1;
     if (tjsonAddDoubleToObject(vnode, "vgVersion", pVnode->vgVersion) < 0) return -1;
+    if (pVnode->toVgId && tjsonAddDoubleToObject(vnode, "toVgId", pVnode->toVgId) < 0) return -1;
     if (tjsonAddItemToArray(vnodes, vnode) < 0) return -1;
   }
 
@@ -179,7 +182,7 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
   SVnodeObj **ppVnodes = NULL;
   char        file[PATH_MAX] = {0};
   char        realfile[PATH_MAX] = {0};
-  snprintf(file, sizeof(file), "%s%svnodes.json.bak", pMgmt->path, TD_DIRSEP);
+  snprintf(file, sizeof(file), "%s%svnodes_tmp.json", pMgmt->path, TD_DIRSEP);
   snprintf(realfile, sizeof(realfile), "%s%svnodes.json", pMgmt->path, TD_DIRSEP);
 
   int32_t numOfVnodes = 0;

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

@@ -265,6 +265,7 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId);
 
+#if 0
   if (pMgmt->pTfs) {
     if (tfsDirExistAt(pMgmt->pTfs, path, (SDiskID){0})) {
       terrno = TSDB_CODE_VND_DIR_ALREADY_EXIST;
@@ -278,8 +279,9 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
       return -1;
     }
   }
+#endif
 
-if (vnodeCreate(path, &vnodeCfg, pMgmt->pTfs) < 0) {
+  if (vnodeCreate(path, &vnodeCfg, pMgmt->pTfs) < 0) {
     tFreeSCreateVnodeReq(&req);
     dError("vgId:%d, failed to create vnode since %s", req.vgId, terrstr());
     code = terrno;
@@ -482,10 +484,18 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 
   int32_t srcVgId = req.srcVgId;
   int32_t dstVgId = req.dstVgId;
+
+  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, dstVgId);
+  if (pVnode != NULL) {
+    dError("vgId:%d, vnode already exist", dstVgId);
+    vmReleaseVnode(pMgmt, pVnode);
+    terrno = TSDB_CODE_VND_ALREADY_EXIST;
+    return -1;
+  }
+
   dInfo("vgId:%d, start to alter vnode hashrange:[%u, %u], dstVgId:%d", req.srcVgId, req.hashBegin, req.hashEnd,
         req.dstVgId);
-
-  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, srcVgId);
+  pVnode = vmAcquireVnode(pMgmt, srcVgId);
   if (pVnode == NULL) {
     dError("vgId:%d, failed to alter hashrange since %s", srcVgId, terrstr());
     terrno = TSDB_CODE_VND_NOT_EXIST;
@@ -499,6 +509,13 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   };
   tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
 
+  // prepare alter
+  pVnode->toVgId = dstVgId;
+  if (vmWriteVnodeListToFile(pMgmt) != 0) {
+    dError("vgId:%d, failed to write vnode list since %s", dstVgId, terrstr());
+    return -1;
+  }
+
   dInfo("vgId:%d, close vnode", srcVgId);
   vmCloseVnode(pMgmt, pVnode, true);
 
@@ -530,6 +547,7 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     return -1;
   }
 
+  // complete alter
   if (vmWriteVnodeListToFile(pMgmt) != 0) {
     dError("vgId:%d, failed to write vnode list since %s", dstVgId, terrstr());
     return -1;
@@ -712,6 +730,7 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_ADD_CHECKINFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_DEL_CHECKINFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_CONSUME, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_CONSUME_PUSH, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_VG_WALINFO, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_BATCH_DEL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

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

@@ -218,6 +218,7 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
       if (pMsg->msgType != TDMT_VND_ALTER_CONFIRM && pVnode->disable) {
         dDebug("vgId:%d, msg:%p put into vnode-write queue failed since its disable", pVnode->vgId, pMsg);
         terrno = TSDB_CODE_VND_STOPPED;
+        code = terrno;
         break;
       }
       dGTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg);

source/dnode/mgmt/node_mgmt/src/dmTransport.c

@@ -23,10 +23,6 @@ static inline void dmBuildMnodeRedirectRsp(SDnode *pDnode, SRpcMsg *pMsg) {
   SEpSet epSet = {0};
   dmGetMnodeEpSetForRedirect(&pDnode->data, pMsg, &epSet);
 
-  if (epSet.numOfEps == 1) {
-    return;
-  }
-
   const int32_t contLen = tSerializeSEpSet(NULL, 0, &epSet);
   pMsg->pCont = rpcMallocCont(contLen);
   if (pMsg->pCont == NULL) {

source/dnode/mgmt/node_util/inc/dmUtil.h

@@ -105,6 +105,7 @@ typedef struct {
   SHashObj      *dnodeHash;
   TdThreadRwlock lock;
   SMsgCb         msgCb;
+  bool           validMnodeEps;
 } SDnodeData;
 
 typedef struct {