Merge branch '3.0' into feature/3.0_interval_hash_optimize

2022-08-19 17:20:08 +08:00 · 2022-08-19 17:20:08 +08:00 · 59739e0cb7
parent 441ce21677 43881c1469
commit 59739e0cb7
83 changed files with 938 additions and 749 deletions
--- a/cmake/cmake.install
+++ b/cmake/cmake.install
@ -1,38 +1,8 @@
-IF (EXISTS /var/lib/taos/dnode/dnodeCfg.json)
-  INSTALL(CODE "MESSAGE(\"The default data directory /var/lib/taos contains old data of tdengine 2.x, please clear it before installing!\")")
-ELSEIF (EXISTS C:/TDengine/data/dnode/dnodeCfg.json)
-  INSTALL(CODE "MESSAGE(\"The default data directory C:/TDengine/data contains old data of tdengine 2.x, please clear it before installing!\")")
-ELSEIF (TD_LINUX)
+IF (TD_LINUX)
  SET(TD_MAKE_INSTALL_SH "${TD_SOURCE_DIR}/packaging/tools/make_install.sh")
  INSTALL(CODE "MESSAGE(\"make install script: ${TD_MAKE_INSTALL_SH}\")")
  INSTALL(CODE "execute_process(COMMAND bash ${TD_MAKE_INSTALL_SH} ${TD_SOURCE_DIR} ${PROJECT_BINARY_DIR} Linux ${TD_VER_NUMBER})")
 ELSEIF (TD_WINDOWS)
-  SET(CMAKE_INSTALL_PREFIX C:/TDengine)
-
-  # INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/go     DESTINATION connector)
-  # INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/nodejs DESTINATION connector)
-  # INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/python DESTINATION connector)
-  # INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/C\# DESTINATION connector)
-  # INSTALL(DIRECTORY ${TD_SOURCE_DIR}/examples DESTINATION .)
-  INSTALL(CODE "IF (NOT EXISTS ${CMAKE_INSTALL_PREFIX}/cfg/taos.cfg)
-    execute_process(COMMAND ${CMAKE_COMMAND} -E copy ${TD_SOURCE_DIR}/packaging/cfg/taos.cfg ${CMAKE_INSTALL_PREFIX}/cfg/taos.cfg)
-  ENDIF ()")
-  INSTALL(FILES ${TD_SOURCE_DIR}/include/client/taos.h DESTINATION include)
-  INSTALL(FILES ${TD_SOURCE_DIR}/include/util/taoserror.h DESTINATION include)
-  INSTALL(FILES ${TD_SOURCE_DIR}/include/libs/function/taosudf.h DESTINATION include)
-  INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos.lib DESTINATION driver)
-  INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos_static.lib DESTINATION driver)
-  INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos.dll DESTINATION driver)
-  INSTALL(FILES ${EXECUTABLE_OUTPUT_PATH}/taos.exe DESTINATION .)
-  INSTALL(FILES ${EXECUTABLE_OUTPUT_PATH}/taosd.exe DESTINATION .)
-  INSTALL(FILES ${EXECUTABLE_OUTPUT_PATH}/udfd.exe DESTINATION .)
-  IF (BUILD_TOOLS)
-    INSTALL(FILES ${EXECUTABLE_OUTPUT_PATH}/taosBenchmark.exe DESTINATION .)
-  ENDIF ()
-
-  IF (TD_MVN_INSTALLED)
-    INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos-jdbcdriver-2.0.38-dist.jar DESTINATION connector/jdbc)
-  ENDIF ()
  SET(TD_MAKE_INSTALL_SH "${TD_SOURCE_DIR}/packaging/tools/make_install.bat")
  INSTALL(CODE "MESSAGE(\"make install script: ${TD_MAKE_INSTALL_SH}\")")
  INSTALL(CODE "execute_process(COMMAND ${TD_MAKE_INSTALL_SH} :needAdmin ${TD_SOURCE_DIR} ${PROJECT_BINARY_DIR} Windows ${TD_VER_NUMBER})")
--- a/cmake/taosadapter_CMakeLists.txt.in
+++ b/cmake/taosadapter_CMakeLists.txt.in
@ -2,7 +2,7 @@
 # taosadapter
 ExternalProject_Add(taosadapter
        GIT_REPOSITORY https://github.com/taosdata/taosadapter.git
-        GIT_TAG 3d21433
+        GIT_TAG abed566
        SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosadapter"
        BINARY_DIR ""
        #BUILD_IN_SOURCE TRUE
--- a/contrib/CMakeLists.txt
+++ b/contrib/CMakeLists.txt
@ -136,24 +136,6 @@ execute_process(COMMAND "${CMAKE_COMMAND}" -G "${CMAKE_GENERATOR}" .
 execute_process(COMMAND "${CMAKE_COMMAND}" --build .
        WORKING_DIRECTORY "${TD_CONTRIB_DIR}/deps-download")

-# clear submodule
-execute_process(COMMAND git submodule deinit -f tools/taos-tools
-        WORKING_DIRECTORY "${TD_SOURCE_DIR}")
-execute_process(COMMAND git rm --cached tools/taos-tools
-        WORKING_DIRECTORY "${TD_SOURCE_DIR}")
-execute_process(COMMAND git submodule deinit -f tools/taosadapter
-        WORKING_DIRECTORY "${TD_SOURCE_DIR}")
-execute_process(COMMAND git rm --cached tools/taosadapter
-        WORKING_DIRECTORY "${TD_SOURCE_DIR}")
-execute_process(COMMAND git submodule deinit -f tools/taosws-rs
-        WORKING_DIRECTORY "${TD_SOURCE_DIR}")
-execute_process(COMMAND git rm --cached tools/taosws-rs
-        WORKING_DIRECTORY "${TD_SOURCE_DIR}")
-execute_process(COMMAND git submodule deinit -f examples/rust
-        WORKING_DIRECTORY "${TD_SOURCE_DIR}")
-execute_process(COMMAND git rm --cached examples/rust
-        WORKING_DIRECTORY "${TD_SOURCE_DIR}")
-
 # ================================================================================================
 # Build
 # ================================================================================================
--- a/docs/en/07-develop/02-model/index.mdx
+++ b/docs/en/07-develop/02-model/index.mdx
@ -9,7 +9,7 @@ The data model employed by TDengine is similar to that of a relational database.
 The [characteristics of time-series data](https://www.taosdata.com/blog/2019/07/09/86.html) from different data collection points may be different. Characteristics include collection frequency, retention policy and others which determine how you create and configure the database. For e.g. days to keep, number of replicas, data block size, whether data updates are allowed and other configurable parameters would be determined by the characteristics of your data and your business requirements. For TDengine to operate with the best performance, we strongly recommend that you create and configure different databases for data with different characteristics. This allows you, for example, to set up different storage and retention policies. When creating a database, there are a lot of parameters that can be configured such as, the days to keep data, the number of replicas, the number of memory blocks, time precision, the minimum and maximum number of rows in each data block, whether compression is enabled, the time range of the data in single data file and so on. Below is an example of the SQL statement to create a database.

 ```sql
-CREATE DATABASE power KEEP 365 DURATION 10 BUFFER 16 VGROUPS 100 WAL 1;
+CREATE DATABASE power KEEP 365 DURATION 10 BUFFER 16 WAL_LEVEL 1;
 ```

 In the above SQL statement:
--- a/docs/en/07-develop/04-query-data/index.mdx
+++ b/docs/en/07-develop/04-query-data/index.mdx
@ -61,20 +61,20 @@ In summary, records across subtables can be aggregated by a simple query on thei
 In TDengine CLI `taos`, use the SQL below to get the average voltage of all the meters in California grouped by location.

 ```
-taos> SELECT AVG(voltage) FROM meters GROUP BY location;
+taos> SELECT AVG(voltage), location FROM meters GROUP BY location;
       avg(voltage)        |                             location                             |
-=============================================================
-             222.000000000 | California.LosAngeles                |
+===============================================================================================
             219.200000000 | California.SanFrancisco                                          |
-Query OK, 2 row(s) in set (0.002136s)
+             221.666666667 | California.LosAngeles                                            |
+Query OK, 2 rows in database (0.005995s)
 ```

 ### Example 2

-In TDengine CLI `taos`, use the SQL below to get the number of rows and the maximum current in the past 24 hours from meters whose groupId is 2.
+In TDengine CLI `taos`, use the SQL below to get the number of rows and the maximum current from meters whose groupId is 2.

 ```
-taos> SELECT count(*), max(current) FROM meters where groupId = 2 and ts > now - 24h;
+taos> SELECT count(*), max(current) FROM meters where groupId = 2;
     count(*)  |    max(current)  |
 ==================================
            5 |             13.4 |
@ -88,40 +88,41 @@ Join queries are only allowed between subtables of the same STable. In [Select](
 In IoT use cases, down sampling is widely used to aggregate data by time range. The `INTERVAL` keyword in TDengine can be used to simplify the query by time window. For example, the SQL statement below can be used to get the sum of current every 10 seconds from meters table d1001.

 ```
-taos> SELECT sum(current) FROM d1001 INTERVAL(10s);
-           ts            |       sum(current)        |
+taos> SELECT _wstart, sum(current) FROM d1001 INTERVAL(10s);
+         _wstart         |       sum(current)        |
 ======================================================
 2018-10-03 14:38:00.000 |              10.300000191 |
 2018-10-03 14:38:10.000 |              24.900000572 |
-Query OK, 2 row(s) in set (0.000883s)
+Query OK, 2 rows in database (0.003139s)
 ```

 Down sampling can also be used for STable. For example, the below SQL statement can be used to get the sum of current from all meters in California.

 ```
-taos> SELECT SUM(current) FROM meters where location like "California%" INTERVAL(1s);
-           ts            |       sum(current)        |
+taos> SELECT _wstart, SUM(current) FROM meters where location like "California%" INTERVAL(1s);
+         _wstart         |       sum(current)        |
 ======================================================
 2018-10-03 14:38:04.000 |              10.199999809 |
- 2018-10-03 14:38:05.000 |              32.900000572 |
+ 2018-10-03 14:38:05.000 |              23.699999809 |
 2018-10-03 14:38:06.000 |              11.500000000 |
 2018-10-03 14:38:15.000 |              12.600000381 |
- 2018-10-03 14:38:16.000 |              36.000000000 |
-Query OK, 5 row(s) in set (0.001538s)
+ 2018-10-03 14:38:16.000 |              34.400000572 |
+Query OK, 5 rows in database (0.007413s)
 ```

 Down sampling also supports time offset. For example, the below SQL statement can be used to get the sum of current from all meters but each time window must start at the boundary of 500 milliseconds.

 ```
-taos> SELECT SUM(current) FROM meters INTERVAL(1s, 500a);
-           ts            |       sum(current)        |
+taos> SELECT _wstart, SUM(current) FROM meters INTERVAL(1s, 500a);
+         _wstart         |       sum(current)        |
 ======================================================
- 2018-10-03 14:38:04.500 |              11.189999809 |
- 2018-10-03 14:38:05.500 |              31.900000572 |
- 2018-10-03 14:38:06.500 |              11.600000000 |
- 2018-10-03 14:38:15.500 |              12.300000381 |
- 2018-10-03 14:38:16.500 |              35.000000000 |
-Query OK, 5 row(s) in set (0.001521s)
+ 2018-10-03 14:38:03.500 |              10.199999809 |
+ 2018-10-03 14:38:04.500 |              10.300000191 |
+ 2018-10-03 14:38:05.500 |              13.399999619 |
+ 2018-10-03 14:38:06.500 |              11.500000000 |
+ 2018-10-03 14:38:14.500 |              12.600000381 |
+ 2018-10-03 14:38:16.500 |              34.400000572 |
+Query OK, 6 rows in database (0.005515s)
 ```

 In many use cases, it's hard to align the timestamp of the data collected by each collection point. However, a lot of algorithms like FFT require the data to be aligned with same time interval and application programs have to handle this by themselves. In TDengine, it's easy to achieve the alignment using down sampling.
--- a/docs/examples/node/nativeexample/subscribe_demo.js
+++ b/docs/examples/node/nativeexample/subscribe_demo.js
@ -28,8 +28,7 @@ function runConsumer() {
        console.log(msg.topicPartition);
        console.log(msg.block);
        console.log(msg.fields)
-        // fixme(@xiaolei): commented temp, should be fixed.
-        //consumer.commit(msg);
+        consumer.commit(msg);
        console.log(`=======consumer ${i} done`)
    }

--- a/docs/examples/node/package.json
+++ b/docs/examples/node/package.json
@ -4,7 +4,7 @@
  "main": "index.js",
  "license": "MIT",
  "dependencies": {
-    "@tdengine/client": "^3.0.0",
+    "@tdengine/client": "^3.0.1",
    "@tdengine/rest": "^3.0.0"
  }
 }
--- a/docs/zh/07-develop/02-model/index.mdx
+++ b/docs/zh/07-develop/02-model/index.mdx
@ -11,10 +11,10 @@ TDengine 采用类关系型数据模型，需要建库、建表。因此对于
 不同类型的数据采集点往往具有不同的数据特征，包括数据采集频率的高低，数据保留时间的长短，副本的数目，数据块的大小，是否允许更新数据等等。为了在各种场景下 TDengine 都能最大效率的工作，TDengine 建议将不同数据特征的表创建在不同的库里，因为每个库可以配置不同的存储策略。创建一个库时，除 SQL 标准的选项外，还可以指定保留时长、副本数、缓存大小、时间精度、文件块里最大最小记录条数、是否压缩、一个数据文件覆盖的天数等多种参数。比如：

 ```sql
-CREATE DATABASE power KEEP 365 DURATION 10 BUFFER 16 VGROUPS 100 WAL 1;
+CREATE DATABASE power KEEP 365 DURATION 10 BUFFER 16 WAL_LEVEL 1;
 ```

-上述语句将创建一个名为 power 的库，这个库的数据将保留 365 天（超过 365 天将被自动删除），每 10 天一个数据文件，每个 VNODE 的写入内存池的大小为 16 MB，数据库的 VGROUPS 数量，对该数据库入会写 WAL 但不执行 FSYNC。详细的语法及参数请见 [数据库管理](/taos-sql/database) 章节。
+上述语句将创建一个名为 power 的库，这个库的数据将保留 365 天（超过 365 天将被自动删除），每 10 天一个数据文件，每个 VNODE 的写入内存池的大小为 16 MB，对该数据库入会写 WAL 但不执行 FSYNC。详细的语法及参数请见 [数据库管理](/taos-sql/database) 章节。

 创建库之后，需要使用 SQL 命令 `USE` 将当前库切换过来，例如：

--- a/docs/zh/07-develop/04-query-data/index.mdx
+++ b/docs/zh/07-develop/04-query-data/index.mdx
@ -54,20 +54,20 @@ Query OK, 2 row(s) in set (0.001100s)
 在 TAOS Shell，查找加利福尼亚州所有智能电表采集的电压平均值，并按照 location 分组。

 ```
-taos> SELECT AVG(voltage) FROM meters GROUP BY location;
+taos> SELECT AVG(voltage), location FROM meters GROUP BY location;
       avg(voltage)        |                             location                             |
-=============================================================
-             222.000000000 | California.LosAngeles                |
+===============================================================================================
             219.200000000 | California.SanFrancisco                                          |
-Query OK, 2 row(s) in set (0.002136s)
+             221.666666667 | California.LosAngeles                                            |
+Query OK, 2 rows in database (0.005995s)
 ```

 ### 示例二

-在 TAOS shell, 查找 groupId 为 2 的所有智能电表过去 24 小时的记录条数，电流的最大值。
+在 TAOS shell, 查找 groupId 为 2 的所有智能电表的记录条数，电流的最大值。

 ```
-taos> SELECT count(*), max(current) FROM meters where groupId = 2 and ts > now - 24h;
+taos> SELECT count(*), max(current) FROM meters where groupId = 2;
     cunt(*)  |    max(current)  |
 ==================================
            5 |             13.4 |
@ -81,40 +81,41 @@ Query OK, 1 row(s) in set (0.002136s)
 物联网场景里，经常需要通过降采样（down sampling）将采集的数据按时间段进行聚合。TDengine 提供了一个简便的关键词 interval 让按照时间窗口的查询操作变得极为简单。比如，将智能电表 d1001 采集的电流值每 10 秒钟求和

 ```
-taos> SELECT sum(current) FROM d1001 INTERVAL(10s);
-           ts            |       sum(current)        |
+taos> SELECT _wstart, sum(current) FROM d1001 INTERVAL(10s);
+         _wstart         |       sum(current)        |
 ======================================================
 2018-10-03 14:38:00.000 |              10.300000191 |
 2018-10-03 14:38:10.000 |              24.900000572 |
-Query OK, 2 row(s) in set (0.000883s)
+Query OK, 2 rows in database (0.003139s)
 ```

 降采样操作也适用于超级表，比如：将加利福尼亚州所有智能电表采集的电流值每秒钟求和

 ```
-taos> SELECT SUM(current) FROM meters where location like "California%" INTERVAL(1s);
-           ts            |       sum(current)        |
+taos> SELECT _wstart, SUM(current) FROM meters where location like "California%" INTERVAL(1s);
+         _wstart         |       sum(current)        |
 ======================================================
 2018-10-03 14:38:04.000 |              10.199999809 |
- 2018-10-03 14:38:05.000 |              32.900000572 |
+ 2018-10-03 14:38:05.000 |              23.699999809 |
 2018-10-03 14:38:06.000 |              11.500000000 |
 2018-10-03 14:38:15.000 |              12.600000381 |
- 2018-10-03 14:38:16.000 |              36.000000000 |
-Query OK, 5 row(s) in set (0.001538s)
+ 2018-10-03 14:38:16.000 |              34.400000572 |
+Query OK, 5 rows in database (0.007413s)
 ```

 降采样操作也支持时间偏移，比如：将所有智能电表采集的电流值每秒钟求和，但要求每个时间窗口从 500 毫秒开始

 ```
-taos> SELECT SUM(current) FROM meters INTERVAL(1s, 500a);
-           ts            |       sum(current)        |
+taos> SELECT _wstart, SUM(current) FROM meters INTERVAL(1s, 500a);
+         _wstart         |       sum(current)        |
 ======================================================
- 2018-10-03 14:38:04.500 |              11.189999809 |
- 2018-10-03 14:38:05.500 |              31.900000572 |
- 2018-10-03 14:38:06.500 |              11.600000000 |
- 2018-10-03 14:38:15.500 |              12.300000381 |
- 2018-10-03 14:38:16.500 |              35.000000000 |
-Query OK, 5 row(s) in set (0.001521s)
+ 2018-10-03 14:38:03.500 |              10.199999809 |
+ 2018-10-03 14:38:04.500 |              10.300000191 |
+ 2018-10-03 14:38:05.500 |              13.399999619 |
+ 2018-10-03 14:38:06.500 |              11.500000000 |
+ 2018-10-03 14:38:14.500 |              12.600000381 |
+ 2018-10-03 14:38:16.500 |              34.400000572 |
+Query OK, 6 rows in database (0.005515s)
 ```

 物联网场景里，每个数据采集点采集数据的时间是难同步的，但很多分析算法(比如 FFT)需要把采集的数据严格按照时间等间隔的对齐，在很多系统里，需要应用自己写程序来处理，但使用 TDengine 的降采样操作就轻松解决。
--- a/docs/zh/17-operation/03-tolerance.md
+++ b/docs/zh/17-operation/03-tolerance.md
@ -26,5 +26,3 @@ TDengine 集群中的时序数据的副本数是与数据库关联的，一个
 TDengine 集群的节点数必须大于等于副本数，否则创建表时将报错。

 当 TDengine 集群中的节点部署在不同的物理机上，并设置多个副本数时，就实现了系统的高可靠性，无需再使用其他软件或工具。TDengine 企业版还可以将副本部署在不同机房，从而实现异地容灾。
-
-另外一种灾备方式是通过 `taosX` 将一个 TDengine 集群的数据同步复制到物理上位于不同数据中心的另一个 TDengine 集群。其详细使用方法请参考 [taosX 参考手册](../../reference/taosX)
--- a/include/libs/nodes/nodes.h
+++ b/include/libs/nodes/nodes.h
@ -105,7 +105,7 @@ typedef enum ENodeType {
  QUERY_NODE_COLUMN_REF,

  // Statement nodes are used in parser and planner module.
-  QUERY_NODE_SET_OPERATOR,
+  QUERY_NODE_SET_OPERATOR = 100,
  QUERY_NODE_SELECT_STMT,
  QUERY_NODE_VNODE_MODIF_STMT,
  QUERY_NODE_CREATE_DATABASE_STMT,
@ -198,7 +198,7 @@ typedef enum ENodeType {
  QUERY_NODE_QUERY,

  // logic plan node
-  QUERY_NODE_LOGIC_PLAN_SCAN,
+  QUERY_NODE_LOGIC_PLAN_SCAN = 1000,
  QUERY_NODE_LOGIC_PLAN_JOIN,
  QUERY_NODE_LOGIC_PLAN_AGG,
  QUERY_NODE_LOGIC_PLAN_PROJECT,
@ -215,7 +215,7 @@ typedef enum ENodeType {
  QUERY_NODE_LOGIC_PLAN,

  // physical plan node
-  QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN,
+  QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN = 1100,
  QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN,
  QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN,
  QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN,
--- a/include/libs/nodes/querynodes.h
+++ b/include/libs/nodes/querynodes.h
@ -428,6 +428,9 @@ void    nodesValueNodeToVariant(const SValueNode* pNode, SVariant* pVal);
 char*   nodesGetFillModeString(EFillMode mode);
 int32_t nodesMergeConds(SNode** pDst, SNodeList** pSrc);

+const char* operatorTypeStr(EOperatorType type);
+const char* logicConditionTypeStr(ELogicConditionType type);
+
 #ifdef __cplusplus
 }
 #endif
--- a/include/libs/sync/sync.h
+++ b/include/libs/sync/sync.h
@ -33,7 +33,7 @@ extern bool gRaftDetailLog;
 #define SYNC_MAX_READ_RANGE          2
 #define SYNC_MAX_PROGRESS_WAIT_MS    4000
 #define SYNC_MAX_START_TIME_RANGE_MS (1000 * 20)
-#define SYNC_MAX_RECV_TIME_RANGE_MS  1000
+#define SYNC_MAX_RECV_TIME_RANGE_MS  1200
 #define SYNC_ADD_QUORUM_COUNT        3

 #define SYNC_MAX_BATCH_SIZE 1
--- a/include/util/tdef.h
+++ b/include/util/tdef.h
@ -132,15 +132,14 @@ typedef enum EOperatorType {
  OP_TYPE_DIV,
  OP_TYPE_REM,
  // unary arithmetic operator
-  OP_TYPE_MINUS,
-  OP_TYPE_ASSIGN,
+  OP_TYPE_MINUS = 20,

  // bitwise operator
-  OP_TYPE_BIT_AND,
+  OP_TYPE_BIT_AND = 30,
  OP_TYPE_BIT_OR,

  // binary comparison operator
-  OP_TYPE_GREATER_THAN,
+  OP_TYPE_GREATER_THAN = 40,
  OP_TYPE_GREATER_EQUAL,
  OP_TYPE_LOWER_THAN,
  OP_TYPE_LOWER_EQUAL,
@ -153,7 +152,7 @@ typedef enum EOperatorType {
  OP_TYPE_MATCH,
  OP_TYPE_NMATCH,
  // unary comparison operator
-  OP_TYPE_IS_NULL,
+  OP_TYPE_IS_NULL = 100,
  OP_TYPE_IS_NOT_NULL,
  OP_TYPE_IS_TRUE,
  OP_TYPE_IS_FALSE,
@ -163,8 +162,11 @@ typedef enum EOperatorType {
  OP_TYPE_IS_NOT_UNKNOWN,

  // json operator
-  OP_TYPE_JSON_GET_VALUE,
-  OP_TYPE_JSON_CONTAINS
+  OP_TYPE_JSON_GET_VALUE = 150,
+  OP_TYPE_JSON_CONTAINS,
+
+  // internal operator
+  OP_TYPE_ASSIGN = 200
 } EOperatorType;

 #define OP_TYPE_CALC_MAX OP_TYPE_BIT_OR
--- a/packaging/tools/make_install.bat
+++ b/packaging/tools/make_install.bat
@ -1,7 +1,57 @@
@echo off
 goto %1
 :needAdmin
+
+if exist C:\\TDengine\\data\\dnode\\dnodeCfg.json (
+  echo The default data directory C:/TDengine/data contains old data of tdengine 2.x, please clear it before installing!
+)
+set source_dir=%2
+set source_dir=%source_dir:/=\\%
+set binary_dir=%3
+set binary_dir=%binary_dir:/=\\%
+set osType=%4
+set verNumber=%5
+set tagert_dir=C:\\TDengine
+
+if not exist %tagert_dir% (
+    mkdir %tagert_dir%
+)
+if not exist %tagert_dir%\\cfg (
+    mkdir %tagert_dir%\\cfg
+)
+if not exist %tagert_dir%\\include (
+    mkdir %tagert_dir%\\include
+)
+if not exist %tagert_dir%\\driver (
+    mkdir %tagert_dir%\\driver
+)
+if not exist C:\\TDengine\\cfg\\taos.cfg (
+    copy %source_dir%\\packaging\\cfg\\taos.cfg %tagert_dir%\\cfg\\taos.cfg > nul
+)
+
+if exist %binary_dir%\\test\\cfg\\taosadapter.toml (
+    if not exist %tagert_dir%\\cfg\\taosadapter.toml (
+        copy %binary_dir%\\test\\cfg\\taosadapter.toml %tagert_dir%\\cfg\\taosadapter.toml > nul
+    )
+)
+
+copy %source_dir%\\include\\client\\taos.h %tagert_dir%\\include > nul
+copy %source_dir%\\include\\util\\taoserror.h %tagert_dir%\\include > nul
+copy %source_dir%\\include\\libs\\function\\taosudf.h %tagert_dir%\\include > nul
+copy %binary_dir%\\build\\lib\\taos.lib %tagert_dir%\\driver > nul
+copy %binary_dir%\\build\\lib\\taos_static.lib %tagert_dir%\\driver > nul
+copy %binary_dir%\\build\\lib\\taos.dll %tagert_dir%\\driver > nul
+copy %binary_dir%\\build\\bin\\taos.exe %tagert_dir% > nul
+copy %binary_dir%\\build\\bin\\taosd.exe %tagert_dir% > nul
+copy %binary_dir%\\build\\bin\\udfd.exe %tagert_dir% > nul
+if exist %binary_dir%\\build\\bin\\taosBenchmark.exe (
+    copy %binary_dir%\\build\\bin\\taosBenchmark.exe %tagert_dir% > nul
+)
+if exist %binary_dir%\\build\\bin\\taosadapter.exe (
+    copy %binary_dir%\\build\\bin\\taosadapter.exe %tagert_dir% > nul
+)
+
 mshta vbscript:createobject("shell.application").shellexecute("%~s0",":hasAdmin","","runas",1)(window.close)&& echo To start/stop TDengine with administrator privileges: sc start/stop taosd &goto :eof
 :hasAdmin
-cp -f C:\\TDengine\\driver\\taos.dll C:\\Windows\\System32
+copy /y C:\\TDengine\\driver\\taos.dll C:\\Windows\\System32 > nul
 sc query "taosd" >nul || sc create "taosd" binPath= "C:\\TDengine\\taosd.exe --win_service" start= DEMAND
--- a/packaging/tools/make_install.sh
+++ b/packaging/tools/make_install.sh
@ -664,7 +664,9 @@ function install_TDengine() {
 ## ==============================Main program starts from here============================
 echo source directory: $1
 echo binary directory: $2
-if [ "$osType" != "Darwin" ]; then
+if [ -x ${data_dir}/dnode/dnodeCfg.json ]; then
+  echo -e "\033[44;31;5mThe default data directory ${data_dir} contains old data of tdengine 2.x, please clear it before installing!\033[0m"
+elif [ "$osType" != "Darwin" ]; then
  if [ -x ${bin_dir}/${clientName} ]; then
    update_TDengine
  else
--- a/source/common/src/systable.c
+++ b/source/common/src/systable.c
@ -88,7 +88,7 @@ static const SSysDbTableSchema userDBSchema[] = {
    {.name = "comp", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT},
    {.name = "precision", .bytes = 2 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
    {.name = "status", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
-    {.name = "retention", .bytes = 60 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "retentions", .bytes = 60 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
    {.name = "single_stable", .bytes = 1, .type = TSDB_DATA_TYPE_BOOL},
    {.name = "cachemodel", .bytes = TSDB_CACHE_MODEL_STR_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
    {.name = "cachesize", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
--- a/source/common/src/tglobal.c
+++ b/source/common/src/tglobal.c
@ -75,7 +75,7 @@ int32_t  tsMonitorMaxLogs = 100;
 bool     tsMonitorComp = false;

 // telem
-bool     tsEnableTelem = false;
+bool     tsEnableTelem = true;
 int32_t  tsTelemInterval = 86400;
 char     tsTelemServer[TSDB_FQDN_LEN] = "telemetry.taosdata.com";
 uint16_t tsTelemPort = 80;
--- a/source/dnode/mnode/impl/src/mndSma.c
+++ b/source/dnode/mnode/impl/src/mndSma.c
@ -489,7 +489,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
  smaObj.uid = mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);
  ASSERT(smaObj.uid != 0);
  char resultTbName[TSDB_TABLE_FNAME_LEN + 16] = {0};
-  snprintf(resultTbName, TSDB_TABLE_FNAME_LEN + 16, "%s_td_tsma_rst_tb",pCreate->name);
+  snprintf(resultTbName, TSDB_TABLE_FNAME_LEN + 16, "%s_td_tsma_rst_tb", pCreate->name);
  memcpy(smaObj.dstTbName, resultTbName, TSDB_TABLE_FNAME_LEN);
  smaObj.dstTbUid = mndGenerateUid(smaObj.dstTbName, TSDB_TABLE_FNAME_LEN);
  smaObj.stbUid = pStb->uid;
@ -530,7 +530,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
  streamObj.sourceDbUid = pDb->uid;
  streamObj.targetDbUid = pDb->uid;
  streamObj.version = 1;
-  streamObj.sql = pCreate->sql;
+  streamObj.sql = strdup(pCreate->sql);
  streamObj.smaId = smaObj.uid;
  streamObj.watermark = pCreate->watermark;
  streamObj.trigger = STREAM_TRIGGER_WINDOW_CLOSE;
@ -585,6 +585,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
    return -1;
  }
  if (pAst != NULL) nodesDestroyNode(pAst);
+  nodesDestroyNode((SNode *)pPlan);

  int32_t code = -1;
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq);
@ -609,6 +610,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
  code = 0;

 _OVER:
+  tFreeStreamObj(&streamObj);
  mndDestroySmaObj(&smaObj);
  mndTransDrop(pTrans);
  return code;
--- a/source/dnode/mnode/impl/src/mndVgroup.c
+++ b/source/dnode/mnode/impl/src/mndVgroup.c
@ -509,6 +509,7 @@ int32_t mndAllocSmaVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup) {
  pVgroup->replica = 1;

  if (mndGetAvailableDnode(pMnode, pDb, pVgroup, pArray) != 0) return -1;
+  taosArrayDestroy(pArray);

  mInfo("db:%s, sma vgId:%d is alloced", pDb->name, pVgroup->vgId);
  return 0;
--- a/source/dnode/vnode/inc/vnode.h
+++ b/source/dnode/vnode/inc/vnode.h
@ -96,6 +96,7 @@ int32_t     metaGetTableTags(SMeta *pMeta, uint64_t suid, SArray *uidList, SHash
 int32_t     metaReadNext(SMetaReader *pReader);
 const void *metaGetTableTagVal(void *tag, int16_t type, STagVal *tagVal);
 int         metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName);
+bool        metaIsTableExist(SMeta  *pMeta, tb_uid_t uid);

 typedef struct SMetaFltParam {
  tb_uid_t suid;
--- a/source/dnode/vnode/src/inc/tsdb.h
+++ b/source/dnode/vnode/src/inc/tsdb.h
@ -260,7 +260,7 @@ void    tsdbUntakeReadSnap(STsdb *pTsdb, STsdbReadSnap *pSnap);

 // tsdbCache
 int32_t tsdbOpenCache(STsdb *pTsdb);
-void    tsdbCloseCache(SLRUCache *pCache);
+void    tsdbCloseCache(STsdb *pTsdb);
 int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, STSRow *row, STsdb *pTsdb);
 int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, STSRow *row, bool dup);
 int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUHandle **h);
@ -298,6 +298,7 @@ struct STsdb {
  SMemTable     *imem;
  STsdbFS        fs;
  SLRUCache     *lruCache;
+  TdThreadMutex  lruMutex;
 };

 struct TSDBKEY {
--- a/source/dnode/vnode/src/meta/metaQuery.c
+++ b/source/dnode/vnode/src/meta/metaQuery.c
@ -127,6 +127,15 @@ _err:
 //   return 0;
 // }

+bool metaIsTableExist(SMeta  *pMeta, tb_uid_t uid) {
+  // query uid.idx
+  if (tdbTbGet(pMeta->pUidIdx, &uid, sizeof(uid), NULL, NULL) < 0) {
+    return false;
+  }
+
+  return true;
+}
+
 int metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) {
  SMeta  *pMeta = pReader->pMeta;
  int64_t version;
--- a/source/dnode/vnode/src/tq/tqRead.c
+++ b/source/dnode/vnode/src/tq/tqRead.c
@ -341,7 +341,7 @@ FAIL:
  return -1;
 }

-void tqReaderSetColIdList(STqReader* pReadHandle, SArray* pColIdList) { pReadHandle->pColIdList = pColIdList; }
+void tqReaderSetColIdList(STqReader* pReader, SArray* pColIdList) { pReader->pColIdList = pColIdList; }

 int tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList) {
  if (pReader->tbIdHash) {
--- a/source/dnode/vnode/src/tsdb/tsdbCache.c
+++ b/source/dnode/vnode/src/tsdb/tsdbCache.c
@ -33,16 +33,21 @@ int32_t tsdbOpenCache(STsdb *pTsdb) {

  taosLRUCacheSetStrictCapacity(pCache, true);

+  taosThreadMutexInit(&pTsdb->lruMutex, NULL);
+
 _err:
  pTsdb->lruCache = pCache;
  return code;
 }

-void tsdbCloseCache(SLRUCache *pCache) {
+void tsdbCloseCache(STsdb *pTsdb) {
+  SLRUCache *pCache = pTsdb->lruCache;
  if (pCache) {
    taosLRUCacheEraseUnrefEntries(pCache);

    taosLRUCacheCleanup(pCache);
+
+    taosThreadMutexDestroy(&pTsdb->lruMutex);
  }
 }

@ -1100,8 +1105,11 @@ int32_t tsdbCacheGetLastrowH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUH
  //  getTableCacheKeyS(uid, "lr", key, &keyLen);
  getTableCacheKey(uid, 0, key, &keyLen);
  LRUHandle *h = taosLRUCacheLookup(pCache, key, keyLen);
-  if (h) {
-  } else {
+  if (!h) {
+    taosThreadMutexLock(&pTsdb->lruMutex);
+
+    h = taosLRUCacheLookup(pCache, key, keyLen);
+    if (!h) {
      STSRow *pRow = NULL;
      bool    dup = false;  // which is always false for now
      code = mergeLastRow(uid, pTsdb, &dup, &pRow);
@ -1111,7 +1119,10 @@ int32_t tsdbCacheGetLastrowH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUH
          taosMemoryFree(pRow);
        }

+        taosThreadMutexUnlock(&pTsdb->lruMutex);
+
        *handle = NULL;
+
        return 0;
      }

@ -1122,7 +1133,12 @@ int32_t tsdbCacheGetLastrowH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUH
        code = -1;
      }

+      taosThreadMutexUnlock(&pTsdb->lruMutex);
+
      h = taosLRUCacheLookup(pCache, key, keyLen);
+    } else {
+      taosThreadMutexUnlock(&pTsdb->lruMutex);
+    }
  }

  *handle = h;
--- a/source/dnode/vnode/src/tsdb/tsdbOpen.c
+++ b/source/dnode/vnode/src/tsdb/tsdbOpen.c
@ -86,7 +86,7 @@ int tsdbClose(STsdb **pTsdb) {
  if (*pTsdb) {
    taosThreadRwlockDestroy(&(*pTsdb)->rwLock);
    tsdbFSClose(*pTsdb);
-    tsdbCloseCache((*pTsdb)->lruCache);
+    tsdbCloseCache(*pTsdb);
    taosMemoryFreeClear(*pTsdb);
  }
  return 0;
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@ -205,7 +205,7 @@ typedef struct SExprSupp {
 } SExprSupp;

 typedef struct SOperatorInfo {
-  uint8_t                operatorType;
+  uint16_t               operatorType;
  bool                   blocking;  // block operator or not
  uint8_t                status;    // denote if current operator is completed
  char*                  name;      // name, for debug purpose
@ -434,7 +434,7 @@ typedef struct SStreamAggSupporter {
 typedef struct SessionWindowSupporter {
  SStreamAggSupporter* pStreamAggSup;
  int64_t              gap;
-  uint8_t              parentType;
+  uint16_t             parentType;
  SAggSupporter*       pIntervalAggSup;
 } SessionWindowSupporter;

--- a/source/libs/executor/src/executil.c
+++ b/source/libs/executor/src/executil.c
@ -533,8 +533,10 @@ int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode,
      vnodeGetCtbIdList(pVnode, pScanNode->suid, res);
    }
  } else {  // Create one table group.
+    if(metaIsTableExist(metaHandle, tableUid)){
      taosArrayPush(res, &tableUid);
    }
+  }

  if (pTagCond) {
    SColumnInfoData* pColInfoData = getColInfoResult(metaHandle, pListInfo->suid, res, pTagCond);
@ -599,7 +601,10 @@ size_t getTableTagsBufLen(const SNodeList* pGroups) {
 int32_t getGroupIdFromTagsVal(void* pMeta, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId) {
  SMetaReader mr = {0};
  metaReaderInit(&mr, pMeta, 0);
-  metaGetTableEntryByUid(&mr, uid);
+  if(metaGetTableEntryByUid(&mr, uid) != 0){    // table not exist
+    metaReaderClear(&mr);
+    return TSDB_CODE_PAR_TABLE_NOT_EXIST;
+  }

  SNodeList* groupNew = nodesCloneList(pGroupNode);

--- a/source/libs/executor/src/executor.c
+++ b/source/libs/executor/src/executor.c
@ -348,7 +348,7 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
  taosThreadOnce(&initPoolOnce, initRefPool);
  atexit(cleanupRefPool);

-  qDebug("start to create subplan task, TID:0x%"PRIx64 " QID:0x%"PRIx64, taskId, pSubplan->id.queryId);
+  qDebug("start to create subplan task, TID:0x%" PRIx64 " QID:0x%" PRIx64, taskId, pSubplan->id.queryId);

  int32_t code = createExecTaskInfoImpl(pSubplan, pTask, readHandle, taskId, sql, model);
  if (code != TSDB_CODE_SUCCESS) {
@ -374,7 +374,7 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
    }
  }

-  qDebug("subplan task create completed, TID:0x%"PRIx64 " QID:0x%"PRIx64, taskId, pSubplan->id.queryId);
+  qDebug("subplan task create completed, TID:0x%" PRIx64 " QID:0x%" PRIx64, taskId, pSubplan->id.queryId);

 _error:
  // if failed to add ref for all tables in this query, abort current query
@ -427,7 +427,7 @@ int waitMoment(SQInfo* pQInfo) {
 #endif

 static void freeBlock(void* param) {
-  SSDataBlock* pBlock = *(SSDataBlock**) param;
+  SSDataBlock* pBlock = *(SSDataBlock**)param;
  blockDataDestroy(pBlock);
 }

@ -472,7 +472,7 @@ int32_t qExecTaskOpt(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds) {

  int64_t st = taosGetTimestampUs();

-  while((pRes = pTaskInfo->pRoot->fpSet.getNextFn(pTaskInfo->pRoot)) != NULL) {
+  while ((pRes = pTaskInfo->pRoot->fpSet.getNextFn(pTaskInfo->pRoot)) != NULL) {
    SSDataBlock* p = createOneDataBlock(pRes, true);
    current += p->info.rows;
    ASSERT(p->info.rows > 0);
@ -494,7 +494,7 @@ int32_t qExecTaskOpt(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds) {
  uint64_t total = pTaskInfo->pRoot->resultInfo.totalRows;

  qDebug("%s task suspended, %d rows in %d blocks returned, total:%" PRId64 " rows, in sinkNode:%d, elapsed:%.2f ms",
-         GET_TASKID(pTaskInfo), current, (int32_t) taosArrayGetSize(pResList), total, 0, el / 1000.0);
+         GET_TASKID(pTaskInfo), current, (int32_t)taosArrayGetSize(pResList), total, 0, el / 1000.0);

  atomic_store_64(&pTaskInfo->owner, 0);
  return pTaskInfo->code;
@ -632,7 +632,7 @@ int32_t qExtractStreamScanner(qTaskInfo_t tinfo, void** scanner) {
  SOperatorInfo* pOperator = pTaskInfo->pRoot;

  while (1) {
-    uint8_t type = pOperator->operatorType;
+    uint16_t type = pOperator->operatorType;
    if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
      *scanner = pOperator->info;
      return 0;
@ -691,7 +691,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) {
  pTaskInfo->streamInfo.prepareStatus = *pOffset;
  if (!tOffsetEqual(pOffset, &pTaskInfo->streamInfo.lastStatus)) {
    while (1) {
-      uint8_t type = pOperator->operatorType;
+      uint16_t type = pOperator->operatorType;
      pOperator->status = OP_OPENED;
      // TODO add more check
      if (type != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@ -3218,8 +3218,8 @@ int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDa
 }

 static void doApplyScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t order, int32_t scanFlag);
-static void doHandleRemainBlockForNewGroupImpl(SOperatorInfo *pOperator, SFillOperatorInfo* pInfo, SResultInfo* pResultInfo,
-                                               SExecTaskInfo* pTaskInfo) {
+static void doHandleRemainBlockForNewGroupImpl(SOperatorInfo* pOperator, SFillOperatorInfo* pInfo,
+                                               SResultInfo* pResultInfo, SExecTaskInfo* pTaskInfo) {
  pInfo->totalInputRows = pInfo->existNewGroupBlock->info.rows;
  SSDataBlock* pResBlock = pInfo->pFinalRes;

@ -3243,8 +3243,8 @@ static void doHandleRemainBlockForNewGroupImpl(SOperatorInfo *pOperator, SFillOp
  pInfo->existNewGroupBlock = NULL;
 }

-static void doHandleRemainBlockFromNewGroup(SOperatorInfo* pOperator, SFillOperatorInfo* pInfo, SResultInfo* pResultInfo,
-                                            SExecTaskInfo* pTaskInfo) {
+static void doHandleRemainBlockFromNewGroup(SOperatorInfo* pOperator, SFillOperatorInfo* pInfo,
+                                            SResultInfo* pResultInfo, SExecTaskInfo* pTaskInfo) {
  if (taosFillHasMoreResults(pInfo->pFillInfo)) {
    int32_t numOfResultRows = pResultInfo->capacity - pInfo->pFinalRes->info.rows;
    taosFillResultDataBlock(pInfo->pFillInfo, pInfo->pFinalRes, numOfResultRows);
@ -3271,7 +3271,7 @@ static void doApplyScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pBlo
  SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, pInfo->primarySrcSlotId);
  colDataAssign(pDst, pSrc, pInfo->pRes->info.rows, &pResBlock->info);

-  for(int32_t i = 0; i < pInfo->numOfNotFillExpr; ++i) {
+  for (int32_t i = 0; i < pInfo->numOfNotFillExpr; ++i) {
    SFillColInfo* pCol = &pInfo->pFillInfo->pFillCol[i + pInfo->numOfExpr];
    ASSERT(pCol->notFillCol);

@ -3665,7 +3665,7 @@ void destroyExchangeOperatorInfo(void* param, int32_t numOfOutput) {
  taosRemoveRef(exchangeObjRefPool, pExInfo->self);
 }

-void freeSourceDataInfo(void *p) {
+void freeSourceDataInfo(void* p) {
  SSourceDataInfo* pInfo = (SSourceDataInfo*)p;
  taosMemoryFreeClear(pInfo->pRsp);
 }
@ -3695,8 +3695,8 @@ static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t
  STimeWindow w = getAlignQueryTimeWindow(pInterval, pInterval->precision, win.skey);
  w = getFirstQualifiedTimeWindow(win.skey, &w, pInterval, TSDB_ORDER_ASC);

-  pInfo->pFillInfo =
-      taosCreateFillInfo(w.skey, numOfCols, numOfNotFillCols, capacity, pInterval, fillType, pColInfo, pInfo->primaryTsCol, order, id);
+  pInfo->pFillInfo = taosCreateFillInfo(w.skey, numOfCols, numOfNotFillCols, capacity, pInterval, fillType, pColInfo,
+                                        pInfo->primaryTsCol, order, id);

  pInfo->win = win;
  pInfo->p = taosMemoryCalloc(numOfCols, POINTER_BYTES);
@ -3742,9 +3742,9 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode*
  SArray* pColMatchColInfo = extractColMatchInfo(pPhyFillNode->pFillExprs, pPhyFillNode->node.pOutputDataBlockDesc,
                                                 &numOfOutputCols, COL_MATCH_FROM_SLOT_ID);

-  int32_t code =
-      initFillInfo(pInfo, pExprInfo, pInfo->numOfExpr, pInfo->pNotFillExprInfo, pInfo->numOfNotFillExpr, (SNodeListNode*)pPhyFillNode->pValues,
-                   pPhyFillNode->timeRange, pResultInfo->capacity, pTaskInfo->id.str, pInterval, type, order);
+  int32_t code = initFillInfo(pInfo, pExprInfo, pInfo->numOfExpr, pInfo->pNotFillExprInfo, pInfo->numOfNotFillExpr,
+                              (SNodeListNode*)pPhyFillNode->pValues, pPhyFillNode->timeRange, pResultInfo->capacity,
+                              pTaskInfo->id.str, pInterval, type, order);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@ -139,7 +139,7 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
      }

      assert(w.ekey > pBlockInfo->window.ekey);
-      if (w.skey <= pBlockInfo->window.ekey && w.skey > pBlockInfo->window.skey) {
+      if (TMAX(w.skey, pBlockInfo->window.skey) <= pBlockInfo->window.ekey) {
        return true;
      }
    }
@ -147,7 +147,7 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
    w = getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.ekey);
    assert(w.skey <= pBlockInfo->window.ekey);

-    if (w.skey > pBlockInfo->window.skey) {
+    if (TMAX(w.skey, pBlockInfo->window.skey) <= TMIN(w.ekey, pBlockInfo->window.ekey)) {
      return true;
    }

@ -158,7 +158,7 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
      }

      assert(w.skey < pBlockInfo->window.skey);
-      if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) {
+      if (pBlockInfo->window.skey <= TMIN(w.ekey, pBlockInfo->window.ekey)) {
        return true;
      }
    }
@ -1649,6 +1649,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
    }
    taosArrayDestroy(tableIdList);
    memcpy(&pTaskInfo->streamInfo.tableCond, &pTSInfo->cond, sizeof(SQueryTableDataCond));
+  } else {
+    taosArrayDestroy(pColIds);
  }

  // create the pseduo columns info
@ -2038,9 +2040,33 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
    metaReaderClear(&smr);

    if (numOfRows >= pOperator->resultInfo.capacity) {
+      p->info.rows = numOfRows;
+      pInfo->pRes->info.rows = numOfRows;
+
+      relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
+      doFilterResult(pInfo);
+
+      blockDataCleanup(p);
+      numOfRows = 0;
+
+      if (pInfo->pRes->info.rows > 0) {
        break;
      }
    }
+  }
+
+  if (numOfRows > 0) {
+    p->info.rows = numOfRows;
+    pInfo->pRes->info.rows = numOfRows;
+
+    relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
+    doFilterResult(pInfo);
+
+    blockDataCleanup(p);
+    numOfRows = 0;
+  }
+
+  blockDataDestroy(p);

  // todo temporarily free the cursor here, the true reason why the free is not valid needs to be found
  if (ret != 0) {
@ -2049,14 +2075,6 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
    doSetOperatorCompleted(pOperator);
  }

-  p->info.rows = numOfRows;
-  pInfo->pRes->info.rows = numOfRows;
-
-  relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
-  doFilterResult(pInfo);
-
-  blockDataDestroy(p);
-
  pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
  return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
 }
@ -2213,9 +2231,33 @@ static SSDataBlock* sysTableScanUserTables(SOperatorInfo* pOperator) {
      colDataAppend(pColInfoData, numOfRows, n, false);

      if (++numOfRows >= pOperator->resultInfo.capacity) {
+        p->info.rows = numOfRows;
+        pInfo->pRes->info.rows = numOfRows;
+
+        relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
+        doFilterResult(pInfo);
+
+        blockDataCleanup(p);
+        numOfRows = 0;
+
+        if (pInfo->pRes->info.rows > 0) {
          break;
        }
      }
+    }
+
+    if (numOfRows > 0) {
+      p->info.rows = numOfRows;
+      pInfo->pRes->info.rows = numOfRows;
+
+      relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
+      doFilterResult(pInfo);
+
+      blockDataCleanup(p);
+      numOfRows = 0;
+    }
+
+    blockDataDestroy(p);

    // todo temporarily free the cursor here, the true reason why the free is not valid needs to be found
    if (ret != 0) {
@ -2224,14 +2266,6 @@ static SSDataBlock* sysTableScanUserTables(SOperatorInfo* pOperator) {
      doSetOperatorCompleted(pOperator);
    }

-    p->info.rows = numOfRows;
-    pInfo->pRes->info.rows = numOfRows;
-
-    relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
-    doFilterResult(pInfo);
-
-    blockDataDestroy(p);
-
    pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
    return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
  }
--- a/source/libs/executor/src/timewindowoperator.c
+++ b/source/libs/executor/src/timewindowoperator.c
@ -910,10 +910,10 @@ int32_t compareWinRes(void* pKey, void* data, int32_t index) {
 }

 static void removeDeleteResults(SHashObj* pUpdatedMap, SArray* pDelWins) {
-  if (!pUpdatedMap || taosHashGetSize(pUpdatedMap) == 0) {
+  int32_t delSize = taosArrayGetSize(pDelWins);
+  if (taosHashGetSize(pUpdatedMap) == 0 || delSize == 0) {
    return;
  }
-  int32_t delSize = taosArrayGetSize(pDelWins);
  void* pIte = NULL;
  while ((pIte = taosHashIterate(pUpdatedMap, pIte)) != NULL) {
    SResKeyPos* pResKey = (SResKeyPos*)pIte;
@ -1706,14 +1706,19 @@ void destroyStreamFinalIntervalOperatorInfo(void* param, int32_t numOfOutput) {
  blockDataDestroy(pInfo->pPullDataRes);
  taosArrayDestroy(pInfo->pRecycledPages);
  blockDataDestroy(pInfo->pUpdateRes);
+  taosArrayDestroy(pInfo->pDelWins);
+  blockDataDestroy(pInfo->pDelRes);

  if (pInfo->pChildren) {
    int32_t size = taosArrayGetSize(pInfo->pChildren);
    for (int32_t i = 0; i < size; i++) {
      SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, i);
      destroyStreamFinalIntervalOperatorInfo(pChildOp->info, numOfOutput);
+      taosMemoryFree(pChildOp->pDownstream);
+      cleanupExprSupp(&pChildOp->exprSupp);
      taosMemoryFreeClear(pChildOp);
    }
+    taosArrayDestroy(pInfo->pChildren);
  }
  nodesDestroyNode((SNode*)pInfo->pPhyNode);
  colDataDestroy(&pInfo->twAggSup.timeWindowData);
@ -1780,7 +1785,7 @@ void increaseTs(SqlFunctionCtx* pCtx) {
  }
 }

-void initIntervalDownStream(SOperatorInfo* downstream, uint8_t type, SAggSupporter* pSup) {
+void initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SAggSupporter* pSup) {
  if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
    // Todo(liuyao) support partition by column
    return;
@ -2154,7 +2159,7 @@ static void doKeepLinearInfo(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlo

 static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp* pExprSup, SSDataBlock* pResBlock) {
  int32_t rows = pResBlock->info.rows;
-
+  blockDataEnsureCapacity(pResBlock, rows + 1);
  // todo set the correct primary timestamp column

  // output the result
@ -2408,12 +2413,12 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
            break;
          }
        }
+      }

      if (pSliceInfo->current > pSliceInfo->win.ekey) {
        doSetOperatorCompleted(pOperator);
        break;
      }
-      }

      if (ts == pSliceInfo->current) {
        for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {
@ -2644,7 +2649,6 @@ void destroyTimeSliceOperatorInfo(void* param, int32_t numOfOutput) {
  taosMemoryFreeClear(param);
 }

-
 SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo) {
  STimeSliceOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STimeSliceOperatorInfo));
  SOperatorInfo*          pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
@ -3542,7 +3546,7 @@ void initDummyFunction(SqlFunctionCtx* pDummy, SqlFunctionCtx* pCtx, int32_t num
 }

 void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, int64_t gap, int64_t waterMark,
-                    uint8_t type) {
+                    uint16_t type) {
  ASSERT(downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN);
  SStreamScanInfo* pScanInfo = downstream->info;
  pScanInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = pAggSup, .gap = gap, .parentType = type};
--- a/source/libs/function/src/builtins.c
+++ b/source/libs/function/src/builtins.c
@ -1503,11 +1503,17 @@ static int32_t translateInterp(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
    return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
  }

+  uint8_t nodeType = nodeType(nodesListGetNode(pFunc->pParameterList, 0));
+  uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
+  if (!IS_NUMERIC_TYPE(paraType) || QUERY_NODE_VALUE == nodeType) {
+    return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
  if (3 <= numOfParams) {
    int64_t timeVal[2] = {0};
    for (int32_t i = 1; i < 3; ++i) {
-      uint8_t nodeType = nodeType(nodesListGetNode(pFunc->pParameterList, i));
-      uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, i))->resType.type;
+      nodeType = nodeType(nodesListGetNode(pFunc->pParameterList, i));
+      paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, i))->resType.type;
      if (!IS_VAR_DATA_TYPE(paraType) || QUERY_NODE_VALUE != nodeType) {
        return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
      }
@ -1525,8 +1531,8 @@ static int32_t translateInterp(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
  }

  if (4 == numOfParams) {
-    uint8_t nodeType = nodeType(nodesListGetNode(pFunc->pParameterList, 3));
-    uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 3))->resType.type;
+    nodeType = nodeType(nodesListGetNode(pFunc->pParameterList, 3));
+    paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 3))->resType.type;
    if (!IS_INTEGER_TYPE(paraType) || QUERY_NODE_VALUE != nodeType) {
      return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
    }
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@ -3970,16 +3970,16 @@ int32_t elapsedFunction(SqlFunctionCtx* pCtx) {
    TSKEY*  ptsList = (int64_t*)colDataGetData(pCol, 0);
    if (pCtx->order == TSDB_ORDER_DESC) {
      if (pCtx->start.key == INT64_MIN) {
-        pInfo->max =
-            (pInfo->max < ptsList[start + pInput->numOfRows - 1]) ? ptsList[start + pInput->numOfRows - 1] : pInfo->max;
+        pInfo->max = (pInfo->max < ptsList[start]) ? ptsList[start] : pInfo->max;
      } else {
        pInfo->max = pCtx->start.key + 1;
      }

-      if (pCtx->end.key != INT64_MIN) {
-        pInfo->min = pCtx->end.key;
+      if (pCtx->end.key == INT64_MIN) {
+        pInfo->min = (pInfo->min > ptsList[start + pInput->numOfRows - 1]) ?
+                     ptsList[start + pInput->numOfRows - 1] : pInfo->min;
      } else {
-        pInfo->min = ptsList[start];
+        pInfo->min = pCtx->end.key;
      }
    } else {
      if (pCtx->start.key == INT64_MIN) {
@ -3988,10 +3988,11 @@ int32_t elapsedFunction(SqlFunctionCtx* pCtx) {
        pInfo->min = pCtx->start.key;
      }

-      if (pCtx->end.key != INT64_MIN) {
-        pInfo->max = pCtx->end.key + 1;
+      if (pCtx->end.key == INT64_MIN) {
+        pInfo->max = (pInfo->max < ptsList[start + pInput->numOfRows - 1]) ?
+                     ptsList[start + pInput->numOfRows - 1] : pInfo->max;
      } else {
-        pInfo->max = ptsList[start + pInput->numOfRows - 1];
+        pInfo->max = pCtx->end.key + 1;
      }
    }
  }
--- a/source/libs/nodes/src/nodesCodeFuncs.c
+++ b/source/libs/nodes/src/nodesCodeFuncs.c
@ -4673,7 +4673,6 @@ static int32_t jsonToNode(const SJson* pJson, void* pObj) {

  int32_t code;
  tjsonGetNumberValue(pJson, jkNodeType, pNode->type, code);
-  ;
  if (TSDB_CODE_SUCCESS == code) {
    code = tjsonToObject(pJson, nodesNodeName(pNode->type), jsonToSpecificNode, pNode);
    if (TSDB_CODE_SUCCESS != code) {
--- a/source/libs/nodes/src/nodesToSQLFuncs.c
+++ b/source/libs/nodes/src/nodesToSQLFuncs.c
@ -21,36 +21,89 @@
 #include "taoserror.h"
 #include "thash.h"

-char *gOperatorStr[] = {NULL,
-                        "+",
-                        "-",
-                        "*",
-                        "/",
-                        "%",
-                        "-",
-                        "&",
-                        "|",
-                        ">",
-                        ">=",
-                        "<",
-                        "<=",
-                        "=",
-                        "<>",
-                        "IN",
-                        "NOT IN",
-                        "LIKE",
-                        "NOT LIKE",
-                        "MATCH",
-                        "NMATCH",
-                        "IS NULL",
-                        "IS NOT NULL",
-                        "IS TRUE",
-                        "IS FALSE",
-                        "IS UNKNOWN",
-                        "IS NOT TRUE",
-                        "IS NOT FALSE",
-                        "IS NOT UNKNOWN"};
-char *gLogicConditionStr[] = {"AND", "OR", "NOT"};
+const char *operatorTypeStr(EOperatorType type) {
+  switch (type) {
+    case OP_TYPE_ADD:
+      return "+";
+    case OP_TYPE_SUB:
+      return "-";
+    case OP_TYPE_MULTI:
+      return "*";
+    case OP_TYPE_DIV:
+      return "/";
+    case OP_TYPE_REM:
+      return "%";
+    case OP_TYPE_MINUS:
+      return "-";
+    case OP_TYPE_BIT_AND:
+      return "&";
+    case OP_TYPE_BIT_OR:
+      return "|";
+    case OP_TYPE_GREATER_THAN:
+      return ">";
+    case OP_TYPE_GREATER_EQUAL:
+      return ">=";
+    case OP_TYPE_LOWER_THAN:
+      return "<";
+    case OP_TYPE_LOWER_EQUAL:
+      return "<=";
+    case OP_TYPE_EQUAL:
+      return "=";
+    case OP_TYPE_NOT_EQUAL:
+      return "<>";
+    case OP_TYPE_IN:
+      return "IN";
+    case OP_TYPE_NOT_IN:
+      return "NOT IN";
+    case OP_TYPE_LIKE:
+      return "LIKE";
+    case OP_TYPE_NOT_LIKE:
+      return "NOT LIKE";
+    case OP_TYPE_MATCH:
+      return "MATCH";
+    case OP_TYPE_NMATCH:
+      return "NMATCH";
+    case OP_TYPE_IS_NULL:
+      return "IS NULL";
+    case OP_TYPE_IS_NOT_NULL:
+      return "IS NOT NULL";
+    case OP_TYPE_IS_TRUE:
+      return "IS TRUE";
+    case OP_TYPE_IS_FALSE:
+      return "IS FALSE";
+    case OP_TYPE_IS_UNKNOWN:
+      return "IS UNKNOWN";
+    case OP_TYPE_IS_NOT_TRUE:
+      return "IS NOT TRUE";
+    case OP_TYPE_IS_NOT_FALSE:
+      return "IS NOT FALSE";
+    case OP_TYPE_IS_NOT_UNKNOWN:
+      return "IS NOT UNKNOWN";
+    case OP_TYPE_JSON_GET_VALUE:
+      return "=>";
+    case OP_TYPE_JSON_CONTAINS:
+      return "CONTAINS";
+    case OP_TYPE_ASSIGN:
+      return "=";
+    default:
+      break;
+  }
+  return "UNKNOWN";
+}
+
+const char *logicConditionTypeStr(ELogicConditionType type) {
+  switch (type) {
+    case LOGIC_COND_TYPE_AND:
+      return "AND";
+    case LOGIC_COND_TYPE_OR:
+      return "OR";
+    case LOGIC_COND_TYPE_NOT:
+      return "NOT";
+    default:
+      break;
+  }
+  return "UNKNOWN";
+}

 int32_t nodesNodeToSQL(SNode *pNode, char *buf, int32_t bufSize, int32_t *len) {
  switch (pNode->type) {
@ -94,12 +147,7 @@ int32_t nodesNodeToSQL(SNode *pNode, char *buf, int32_t bufSize, int32_t *len) {
        NODES_ERR_RET(nodesNodeToSQL(pOpNode->pLeft, buf, bufSize, len));
      }

-      if (pOpNode->opType >= (sizeof(gOperatorStr) / sizeof(gOperatorStr[0]))) {
-        nodesError("unknown operation type:%d", pOpNode->opType);
-        NODES_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
-      }
-
-      *len += snprintf(buf + *len, bufSize - *len, " %s ", gOperatorStr[pOpNode->opType]);
+      *len += snprintf(buf + *len, bufSize - *len, " %s ", operatorTypeStr(pOpNode->opType));

      if (pOpNode->pRight) {
        NODES_ERR_RET(nodesNodeToSQL(pOpNode->pRight, buf, bufSize, len));
@ -118,7 +166,7 @@ int32_t nodesNodeToSQL(SNode *pNode, char *buf, int32_t bufSize, int32_t *len) {

      FOREACH(node, pLogicNode->pParameterList) {
        if (!first) {
-          *len += snprintf(buf + *len, bufSize - *len, " %s ", gLogicConditionStr[pLogicNode->condType]);
+          *len += snprintf(buf + *len, bufSize - *len, " %s ", logicConditionTypeStr(pLogicNode->condType));
        }
        NODES_ERR_RET(nodesNodeToSQL(node, buf, bufSize, len));
        first = false;
--- a/source/libs/parser/src/parInsert.c
+++ b/source/libs/parser/src/parInsert.c
@ -143,9 +143,9 @@ static int32_t createSName(SName* pName, SToken* pTableName, int32_t acctId, con
    }
    char name[TSDB_DB_FNAME_LEN] = {0};
    strncpy(name, pTableName->z, dbLen);
-    dbLen = strdequote(name);
+    int32_t actualDbLen = strdequote(name);

-    code = tNameSetDbName(pName, acctId, name, dbLen);
+    code = tNameSetDbName(pName, acctId, name, actualDbLen);
    if (code != TSDB_CODE_SUCCESS) {
      return buildInvalidOperationMsg(pMsgBuf, msg1);
    }
--- a/source/libs/scalar/inc/filterInt.h
+++ b/source/libs/scalar/inc/filterInt.h
@ -350,7 +350,6 @@ struct SFilterInfo {

 extern bool filterDoCompare(__compar_fn_t func, uint8_t optr, void *left, void *right);
 extern __compar_fn_t filterGetCompFunc(int32_t type, int32_t optr);
-extern OptrStr gOptrStr[];

 #ifdef __cplusplus
 }
--- a/source/libs/scalar/src/filter.c
+++ b/source/libs/scalar/src/filter.c
@ -24,46 +24,6 @@
 #include "ttime.h"
 #include "functionMgt.h"

-OptrStr gOptrStr[] = {
-  {0,                                      "invalid"},
-  {OP_TYPE_ADD,                            "+"},
-  {OP_TYPE_SUB,                            "-"},
-  {OP_TYPE_MULTI,                          "*"},
-  {OP_TYPE_DIV,                            "/"},
-  {OP_TYPE_REM,                            "%"},
-  {OP_TYPE_MINUS,                          "minus"},
-  {OP_TYPE_ASSIGN,                         "assign"},
-  // bit operator
-  {OP_TYPE_BIT_AND,                        "&"},
-  {OP_TYPE_BIT_OR,                         "|"},
-
-  // comparison operator
-  {OP_TYPE_GREATER_THAN,                   ">"},
-  {OP_TYPE_GREATER_EQUAL,                  ">="},
-  {OP_TYPE_LOWER_THAN,                     "<"},
-  {OP_TYPE_LOWER_EQUAL,                    "<="},
-  {OP_TYPE_EQUAL,                          "=="},
-  {OP_TYPE_NOT_EQUAL,                      "!="},
-  {OP_TYPE_IN,                             "in"},
-  {OP_TYPE_NOT_IN,                         "not in"},
-  {OP_TYPE_LIKE,                           "like"},
-  {OP_TYPE_NOT_LIKE,                       "not like"},
-  {OP_TYPE_MATCH,                          "match"},
-  {OP_TYPE_NMATCH,                         "nmatch"},
-  {OP_TYPE_IS_NULL,                        "is null"},
-  {OP_TYPE_IS_NOT_NULL,                    "not null"},
-  {OP_TYPE_IS_TRUE,                        "is true"},
-  {OP_TYPE_IS_FALSE,                       "is false"},
-  {OP_TYPE_IS_UNKNOWN,                     "is unknown"},
-  {OP_TYPE_IS_NOT_TRUE,                    "not true"},
-  {OP_TYPE_IS_NOT_FALSE,                   "not false"},
-  {OP_TYPE_IS_NOT_UNKNOWN,                 "not unknown"},
-
-  // json operator
-  {OP_TYPE_JSON_GET_VALUE,                 "->"},
-  {OP_TYPE_JSON_CONTAINS,                  "json contains"}
-};
-
 bool filterRangeCompGi (const void *minv, const void *maxv, const void *minr, const void *maxr, __compar_fn_t cfunc) {
  int32_t result = cfunc(maxv, minr);
  return result >= 0;
@ -986,7 +946,7 @@ int32_t filterAddUnit(SFilterInfo *info, uint8_t optr, SFilterFieldId *left, SFi
  } else {
    int32_t paramNum = scalarGetOperatorParamNum(optr);
    if (1 != paramNum) {
-      fltError("invalid right field in unit, operator:%s, rightType:%d", gOptrStr[optr].str, u->right.type);
+      fltError("invalid right field in unit, operator:%s, rightType:%d", operatorTypeStr(optr), u->right.type);
      return TSDB_CODE_QRY_APP_ERROR;
    }
  }
@ -1517,7 +1477,7 @@ void filterDumpInfoToString(SFilterInfo *info, const char *msg, int32_t options)
        SFilterField *left = FILTER_UNIT_LEFT_FIELD(info, unit);
        SColumnNode *refNode = (SColumnNode *)left->desc;
        if (unit->compare.optr >= 0 && unit->compare.optr <= OP_TYPE_JSON_CONTAINS){
-          len = sprintf(str, "UNIT[%d] => [%d][%d]  %s  [", i, refNode->dataBlockId, refNode->slotId, gOptrStr[unit->compare.optr].str);
+          len = sprintf(str, "UNIT[%d] => [%d][%d]  %s  [", i, refNode->dataBlockId, refNode->slotId, operatorTypeStr(unit->compare.optr));
        }

        if (unit->right.type == FLD_TYPE_VALUE && FILTER_UNIT_OPTR(unit) != OP_TYPE_IN) {
@ -1536,7 +1496,7 @@ void filterDumpInfoToString(SFilterInfo *info, const char *msg, int32_t options)
        if (unit->compare.optr2) {
          strcat(str, " && ");
          if (unit->compare.optr2 >= 0 && unit->compare.optr2 <= OP_TYPE_JSON_CONTAINS){
-            sprintf(str + strlen(str), "[%d][%d]  %s  [", refNode->dataBlockId, refNode->slotId, gOptrStr[unit->compare.optr2].str);
+            sprintf(str + strlen(str), "[%d][%d]  %s  [", refNode->dataBlockId, refNode->slotId, operatorTypeStr(unit->compare.optr2));
          }

          if (unit->right2.type == FLD_TYPE_VALUE && FILTER_UNIT_OPTR(unit) != OP_TYPE_IN) {
--- a/source/libs/scalar/test/scalar/scalarTests.cpp
+++ b/source/libs/scalar/test/scalar/scalarTests.cpp
@ -1089,16 +1089,16 @@ void makeCalculate(void *json, void *key, int32_t rightType, void *rightData, do

 }else if(opType == OP_TYPE_ADD || opType == OP_TYPE_SUB || opType == OP_TYPE_MULTI || opType == OP_TYPE_DIV ||
            opType == OP_TYPE_REM || opType == OP_TYPE_MINUS){
-   printf("op:%s,1result:%f,except:%f\n", gOptrStr[opType].str, *((double *)colDataGetData(column, 0)), exceptValue);
+   printf("op:%s,1result:%f,except:%f\n", operatorTypeStr(opType), *((double *)colDataGetData(column, 0)), exceptValue);
   ASSERT_TRUE(fabs(*((double *)colDataGetData(column, 0)) - exceptValue) < 0.0001);
 }else if(opType == OP_TYPE_BIT_AND || opType == OP_TYPE_BIT_OR){
-   printf("op:%s,2result:%" PRId64 ",except:%f\n", gOptrStr[opType].str, *((int64_t *)colDataGetData(column, 0)), exceptValue);
+   printf("op:%s,2result:%" PRId64 ",except:%f\n", operatorTypeStr(opType), *((int64_t *)colDataGetData(column, 0)), exceptValue);
   ASSERT_EQ(*((int64_t *)colDataGetData(column, 0)), exceptValue);
 }else if(opType == OP_TYPE_GREATER_THAN || opType == OP_TYPE_GREATER_EQUAL || opType == OP_TYPE_LOWER_THAN ||
            opType == OP_TYPE_LOWER_EQUAL || opType == OP_TYPE_EQUAL || opType == OP_TYPE_NOT_EQUAL ||
            opType == OP_TYPE_IS_NULL || opType == OP_TYPE_IS_NOT_NULL || opType == OP_TYPE_IS_TRUE ||
            opType == OP_TYPE_LIKE || opType == OP_TYPE_NOT_LIKE || opType == OP_TYPE_MATCH || opType == OP_TYPE_NMATCH){
-   printf("op:%s,3result:%d,except:%f\n", gOptrStr[opType].str, *((bool *)colDataGetData(column, 0)), exceptValue);
+   printf("op:%s,3result:%d,except:%f\n", operatorTypeStr(opType), *((bool *)colDataGetData(column, 0)), exceptValue);
   ASSERT_EQ(*((bool *)colDataGetData(column, 0)), exceptValue);
 }

--- a/source/libs/scheduler/src/schRemote.c
+++ b/source/libs/scheduler/src/schRemote.c
@ -20,7 +20,7 @@
 #include "tmsg.h"
 #include "tref.h"
 #include "trpc.h"
-
+// clang-format off
 int32_t schValidateRspMsgType(SSchJob *pJob, SSchTask *pTask, int32_t msgType) {
  int32_t lastMsgType = pTask->lastMsgType;
  int32_t taskStatus = SCH_GET_TASK_STATUS(pTask);
@ -1104,7 +1104,7 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,

 #if 1
  SSchTrans trans = {.pTrans = pJob->conn.pTrans, .pHandle = SCH_GET_TASK_HANDLE(pTask)};
-  schAsyncSendMsg(pJob, pTask, &trans, addr, msgType, msg, msgSize, persistHandle, (rpcCtx.args ? &rpcCtx : NULL));
+  code = schAsyncSendMsg(pJob, pTask, &trans, addr, msgType, msg, msgSize, persistHandle, (rpcCtx.args ? &rpcCtx : NULL));
  msg = NULL;
  SCH_ERR_JRET(code);

@ -1114,7 +1114,7 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,
 #else
  if (TDMT_VND_SUBMIT != msgType) {
    SSchTrans trans = {.pTrans = pJob->conn.pTrans, .pHandle = SCH_GET_TASK_HANDLE(pTask)};
-    schAsyncSendMsg(pJob, pTask, &trans, addr, msgType, msg, msgSize, persistHandle, (rpcCtx.args ? &rpcCtx : NULL));
+    code = schAsyncSendMsg(pJob, pTask, &trans, addr, msgType, msg, msgSize, persistHandle, (rpcCtx.args ? &rpcCtx : NULL));
    msg = NULL;
    SCH_ERR_JRET(code);

@ -1136,3 +1136,4 @@ _return:
  taosMemoryFreeClear(msg);
  SCH_RET(code);
 }
+// clang-format on
--- a/source/libs/stream/src/streamTask.c
+++ b/source/libs/stream/src/streamTask.c
@ -152,11 +152,12 @@ int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
 }

 void tFreeSStreamTask(SStreamTask* pTask) {
+  qDebug("free stream task %d", pTask->taskId);
  if (pTask->inputQueue) streamQueueClose(pTask->inputQueue);
  if (pTask->outputQueue) streamQueueClose(pTask->outputQueue);
  if (pTask->exec.qmsg) taosMemoryFree(pTask->exec.qmsg);
  if (pTask->exec.executor) qDestroyTask(pTask->exec.executor);
-  taosArrayDestroy(pTask->childEpInfo);
+  taosArrayDestroyP(pTask->childEpInfo, taosMemoryFree);
  if (pTask->outputType == TASK_OUTPUT__TABLE) {
    tDeleteSSchemaWrapper(pTask->tbSink.pSchemaWrapper);
    taosMemoryFree(pTask->tbSink.pTSchema);
--- a/source/libs/transport/inc/transComm.h
+++ b/source/libs/transport/inc/transComm.h
@ -98,6 +98,17 @@ typedef void* queue[2];
 #define TRANS_RETRY_INTERVAL    15    // retry interval (ms)
 #define TRANS_CONN_TIMEOUT      3     // connect timeout (s)
 #define TRANS_READ_TIMEOUT      3000  // read timeout  (ms)
+#define TRANS_PACKET_LIMIT      1024 * 1024 * 512
+
+#define TRANS_MAGIC_NUM 0x5f375a86
+
+#define TRANS_NOVALID_PACKET(src) ((src) != TRANS_MAGIC_NUM ? 1 : 0)
+
+#define TRANS_PACKET_LIMIT 1024 * 1024 * 512
+
+#define TRANS_MAGIC_NUM 0x5f375a86
+
+#define TRANS_NOVALID_PACKET(src) ((src) != TRANS_MAGIC_NUM ? 1 : 0)

 typedef SRpcMsg      STransMsg;
 typedef SRpcCtx      STransCtx;
@ -151,6 +162,7 @@ typedef struct {
  char hasEpSet : 2;  // contain epset or not, 0(default): no epset, 1: contain epset

  char     user[TSDB_UNI_LEN];
+  uint32_t magicNum;
  STraceId traceId;
  uint64_t ahandle;  // ahandle assigned by client
  uint32_t code;     // del later
@ -203,6 +215,7 @@ typedef struct SConnBuffer {
  int   cap;
  int   left;
  int   total;
+  int   invalid;
 } SConnBuffer;

 typedef void (*AsyncCB)(uv_async_t* handle);
--- a/source/libs/transport/src/thttp.c
+++ b/source/libs/transport/src/thttp.c
@ -14,15 +14,22 @@
 */

 #define _DEFAULT_SOURCE
-#ifdef USE_UV
-#include <uv.h>
-#endif
 // clang-format off
+#include <uv.h>
 #include "zlib.h"
 #include "thttp.h"
 #include "taoserror.h"
 #include "tlog.h"

+typedef struct SHttpClient {
+  uv_connect_t conn;
+  uv_tcp_t     tcp;
+  uv_write_t   req;
+  uv_buf_t*    buf;
+  char*        addr;
+  uint16_t     port;
+} SHttpClient;
+
 static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pHead, int32_t headLen,
                                   EHttpCompFlag flag) {
  if (flag == HTTP_FLAT) {
@ -45,7 +52,7 @@ static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pH
  }
 }

-int32_t taosCompressHttpRport(char* pSrc, int32_t srcLen) {
+static int32_t taosCompressHttpRport(char* pSrc, int32_t srcLen) {
  int32_t code = -1;
  int32_t destLen = srcLen;
  void*   pDest = taosMemoryMalloc(destLen);
@ -114,84 +121,53 @@ _OVER:
  return code;
 }

-#ifdef USE_UV
-static void clientConnCb(uv_connect_t* req, int32_t status) {
-  if (status < 0) {
+static void destroyHttpClient(SHttpClient* cli) {
+  taosMemoryFree(cli->buf);
+  taosMemoryFree(cli->addr);
+  taosMemoryFree(cli);
+}
+static void clientCloseCb(uv_handle_t* handle) {
+  SHttpClient* cli = handle->data;
+  destroyHttpClient(cli);
+}
+static void clientSentCb(uv_write_t* req, int32_t status) {
+  SHttpClient* cli = req->data;
+  if (status != 0) {
    terrno = TAOS_SYSTEM_ERROR(status);
-    uError("connection error %s", uv_strerror(status));
-    uv_close((uv_handle_t*)req->handle, NULL);
+    uError("http-report failed to send data %s", uv_strerror(status));
+  } else {
+    uInfo("http-report succ to send data");
+  }
+  uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
+}
+static void clientConnCb(uv_connect_t* req, int32_t status) {
+  SHttpClient* cli = req->data;
+  if (status != 0) {
+    terrno = TAOS_SYSTEM_ERROR(status);
+    uError("http-report failed to conn to server, reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
+    uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
    return;
  }
-  uv_buf_t* wb = req->data;
-  assert(wb != NULL);
-  uv_write_t write_req;
-  uv_write(&write_req, req->handle, wb, 2, NULL);
-  uv_close((uv_handle_t*)req->handle, NULL);
+  uv_write(&cli->req, (uv_stream_t*)&cli->tcp, cli->buf, 2, clientSentCb);
 }

-int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
-  uint32_t ipv4 = taosGetIpv4FromFqdn(server);
-  if (ipv4 == 0xffffffff) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    uError("failed to get http server:%s ip since %s", server, terrstr());
-    return -1;
-  }
-
-  char ipv4Buf[128] = {0};
-  tinet_ntoa(ipv4Buf, ipv4);
-
-  struct sockaddr_in dest = {0};
-  uv_ip4_addr(ipv4Buf, port, &dest);
-
-  uv_tcp_t   socket_tcp = {0};
-  uv_loop_t* loop = uv_default_loop();
-  uv_tcp_init(loop, &socket_tcp);
-  uv_connect_t* connect = (uv_connect_t*)taosMemoryMalloc(sizeof(uv_connect_t));
-
-  if (flag == HTTP_GZIP) {
-    int32_t dstLen = taosCompressHttpRport(pCont, contLen);
-    if (dstLen > 0) {
-      contLen = dstLen;
-    } else {
-      flag = HTTP_FLAT;
-    }
-  }
-
-  char    header[1024] = {0};
-  int32_t headLen = taosBuildHttpHeader(server, contLen, header, sizeof(header), flag);
-
-  uv_buf_t wb[2];
-  wb[0] = uv_buf_init((char*)header, headLen);
-  wb[1] = uv_buf_init((char*)pCont, contLen);
-
-  connect->data = wb;
-  terrno = 0;
-  uv_tcp_connect(connect, &socket_tcp, (const struct sockaddr*)&dest, clientConnCb);
-  uv_run(loop, UV_RUN_DEFAULT);
-  uv_loop_close(loop);
-  taosMemoryFree(connect);
-  return terrno;
-}
-
-#else
-int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
-  int32_t     code = -1;
-  TdSocketPtr pSocket = NULL;
-
+static int32_t taosBuildDstAddr(const char* server, uint16_t port, struct sockaddr_in* dest) {
  uint32_t ip = taosGetIpv4FromFqdn(server);
  if (ip == 0xffffffff) {
    terrno = TAOS_SYSTEM_ERROR(errno);
-    uError("failed to get http server:%s ip since %s", server, terrstr());
-    goto SEND_OVER;
+    uError("http-report failed to get http server:%s ip since %s", server, terrstr());
+    return -1;
  }
-
-  pSocket = taosOpenTcpClientSocket(ip, port, 0);
-  if (pSocket == NULL) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    uError("failed to create http socket to %s:%u since %s", server, port, terrstr());
-    goto SEND_OVER;
+  char buf[128] = {0};
+  tinet_ntoa(buf, ip);
+  uv_ip4_addr(buf, port, dest);
+  return 0;
+}
+int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
+  struct sockaddr_in dest = {0};
+  if (taosBuildDstAddr(server, port, &dest) < 0) {
+    return -1;
  }
-
  if (flag == HTTP_GZIP) {
    int32_t dstLen = taosCompressHttpRport(pCont, contLen);
    if (dstLen > 0) {
@ -200,37 +176,38 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32
      flag = HTTP_FLAT;
    }
  }
+  terrno = 0;

-  char    header[1024] = {0};
+  char    header[2048] = {0};
  int32_t headLen = taosBuildHttpHeader(server, contLen, header, sizeof(header), flag);
-  if (taosWriteMsg(pSocket, header, headLen) < 0) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    uError("failed to send http header to %s:%u since %s", server, port, terrstr());
-    goto SEND_OVER;
+
+  uv_buf_t* wb = taosMemoryCalloc(2, sizeof(uv_buf_t));
+  wb[0] = uv_buf_init((char*)header, headLen);  // stack var
+  wb[1] = uv_buf_init((char*)pCont, contLen);   //  heap var
+
+  SHttpClient* cli = taosMemoryCalloc(1, sizeof(SHttpClient));
+  cli->conn.data = cli;
+  cli->tcp.data = cli;
+  cli->req.data = cli;
+  cli->buf = wb;
+  cli->addr = tstrdup(server);
+  cli->port = port;
+
+  uv_loop_t* loop = uv_default_loop();
+  uv_tcp_init(loop, &cli->tcp);
+  // set up timeout to avoid stuck;
+  int32_t fd = taosCreateSocketWithTimeout(5);
+  uv_tcp_open((uv_tcp_t*)&cli->tcp, fd);
+
+
+  int32_t ret = uv_tcp_connect(&cli->conn, &cli->tcp, (const struct sockaddr*)&dest, clientConnCb);
+  if (ret != 0) {
+    uError("http-report failed to connect to server, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr, cli->port);
+    destroyHttpClient(cli);
  }

-  if (taosWriteMsg(pSocket, (void*)pCont, contLen) < 0) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    uError("failed to send http content to %s:%u since %s", server, port, terrstr());
-    goto SEND_OVER;
-  }
-
-  // read something to avoid nginx error 499
-  if (taosWriteMsg(pSocket, header, 10) < 0) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    uError("failed to receive response from %s:%u since %s", server, port, terrstr());
-    goto SEND_OVER;
-  }
-
-  code = 0;
-
-SEND_OVER:
-  if (pSocket != NULL) {
-    taosCloseSocket(&pSocket);
-  }
-
-  return code;
+  uv_run(loop, UV_RUN_DEFAULT);
+  uv_loop_close(loop);
+  return terrno;
 }
-
 // clang-format on
-#endif
--- a/source/libs/transport/src/transCli.c
+++ b/source/libs/transport/src/transCli.c
@ -318,10 +318,17 @@ void cliHandleResp(SCliConn* conn) {
  }

  STransMsgHead* pHead = NULL;
-  transDumpFromBuffer(&conn->readBuf, (char**)&pHead);
+  if (transDumpFromBuffer(&conn->readBuf, (char**)&pHead) <= 0) {
+    tDebug("%s conn %p recv invalid packet ", CONN_GET_INST_LABEL(conn), conn);
+    return;
+  }
  pHead->code = htonl(pHead->code);
  pHead->msgLen = htonl(pHead->msgLen);

+  if (cliRecvReleaseReq(conn, pHead)) {
+    return;
+  }
+
  STransMsg transMsg = {0};
  transMsg.contLen = transContLenFromMsg(pHead->msgLen);
  transMsg.pCont = transContFromHead((char*)pHead);
@ -333,10 +340,6 @@ void cliHandleResp(SCliConn* conn) {

  SCliMsg*       pMsg = NULL;
  STransConnCtx* pCtx = NULL;
-  if (cliRecvReleaseReq(conn, pHead)) {
-    return;
-  }
-
  if (CONN_NO_PERSIST_BY_APP(conn)) {
    pMsg = transQueuePop(&conn->cliMsgs);

@ -598,8 +601,13 @@ static void cliRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
    pBuf->len += nread;
    while (transReadComplete(pBuf)) {
      tTrace("%s conn %p read complete", CONN_GET_INST_LABEL(conn), conn);
+      if (pBuf->invalid) {
+        cliHandleExcept(conn);
+        break;
+      } else {
        cliHandleResp(conn);
      }
+    }
    return;
  }

@ -759,6 +767,7 @@ void cliSend(SCliConn* pConn) {
  pHead->release = REQUEST_RELEASE_HANDLE(pCliMsg) ? 1 : 0;
  memcpy(pHead->user, pTransInst->user, strlen(pTransInst->user));
  pHead->traceId = pMsg->info.traceId;
+  pHead->magicNum = htonl(TRANS_MAGIC_NUM);

  uv_buf_t wb = uv_buf_init((char*)pHead, msgLen);

@ -1423,7 +1432,7 @@ int transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STran
  if (pThrd == NULL && valid == false) {
    transFreeMsg(pReq->pCont);
    transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
-    return -1;
+    return TSDB_CODE_RPC_BROKEN_LINK;
  }

  TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64());
@ -1468,7 +1477,7 @@ int transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransMs
  if (pThrd == NULL && valid == false) {
    transFreeMsg(pReq->pCont);
    transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
-    return -1;
+    return TSDB_CODE_RPC_BROKEN_LINK;
  }

  tsem_t* sem = taosMemoryCalloc(1, sizeof(tsem_t));
--- a/source/libs/transport/src/transComm.c
+++ b/source/libs/transport/src/transComm.c
@ -91,6 +91,7 @@ int transInitBuffer(SConnBuffer* buf) {
  buf->left = -1;
  buf->len = 0;
  buf->total = 0;
+  buf->invalid = 0;
  return 0;
 }
 int transDestroyBuffer(SConnBuffer* p) {
@ -108,19 +109,25 @@ int transClearBuffer(SConnBuffer* buf) {
  p->left = -1;
  p->len = 0;
  p->total = 0;
+  p->invalid = 0;
  return 0;
 }

 int transDumpFromBuffer(SConnBuffer* connBuf, char** buf) {
+  static const int HEADSIZE = sizeof(STransMsgHead);
+
  SConnBuffer* p = connBuf;
  if (p->left != 0) {
    return -1;
  }
  int total = connBuf->total;
+  if (total >= HEADSIZE && !p->invalid) {
    *buf = taosMemoryCalloc(1, total);
    memcpy(*buf, p->buf, total);
-
    transResetBuffer(connBuf);
+  } else {
+    total = -1;
+  }
  return total;
 }

@ -173,6 +180,7 @@ bool transReadComplete(SConnBuffer* connBuf) {
      memcpy((char*)&head, connBuf->buf, sizeof(head));
      int32_t msgLen = (int32_t)htonl(head.msgLen);
      p->total = msgLen;
+      p->invalid = TRANS_NOVALID_PACKET(htonl(head.magicNum));
    }
    if (p->total >= p->len) {
      p->left = p->total - p->len;
@ -180,7 +188,7 @@ bool transReadComplete(SConnBuffer* connBuf) {
      p->left = 0;
    }
  }
-  return p->left == 0 ? true : false;
+  return (p->left == 0 || p->invalid) ? true : false;
 }

 int transSetConnOption(uv_tcp_t* stream) {
--- a/source/libs/transport/src/transSvr.c
+++ b/source/libs/transport/src/transSvr.c
@ -183,17 +183,25 @@ static void uvHandleActivityTimeout(uv_timer_t* handle) {
  tDebug("%p timeout since no activity", conn);
 }

-static void uvHandleReq(SSvrConn* pConn) {
-  STransMsgHead* msg = NULL;
-  int            msgLen = 0;
+static bool uvHandleReq(SSvrConn* pConn) {
+  STrans* pTransInst = pConn->pTransInst;

-  msgLen = transDumpFromBuffer(&pConn->readBuf, (char**)&msg);
+  STransMsgHead* msg = NULL;
+  int            msgLen = transDumpFromBuffer(&pConn->readBuf, (char**)&msg);
+  if (msgLen <= 0) {
+    tError("%s conn %p read invalid packet", transLabel(pTransInst), pConn);
+    return false;
+  }

  STransMsgHead* pHead = (STransMsgHead*)msg;
  pHead->code = htonl(pHead->code);
  pHead->msgLen = htonl(pHead->msgLen);
  memcpy(pConn->user, pHead->user, strlen(pHead->user));

+  if (uvRecvReleaseReq(pConn, pHead)) {
+    return true;
+  }
+
  // TODO(dengyihao): time-consuming task throwed into BG Thread
  //  uv_work_t* wreq = taosMemoryMalloc(sizeof(uv_work_t));
  //  wreq->data = pConn;
@ -201,10 +209,6 @@ static void uvHandleReq(SSvrConn* pConn) {
  //  transRefSrvHandle(pConn);
  //  uv_queue_work(((SWorkThrd*)pConn->hostThrd)->loop, wreq, uvWorkDoTask, uvWorkAfterTask);

-  if (uvRecvReleaseReq(pConn, pHead)) {
-    return;
-  }
-
  STransMsg transMsg;
  memset(&transMsg, 0, sizeof(transMsg));
  transMsg.contLen = transContLenFromMsg(pHead->msgLen);
@ -220,7 +224,6 @@ static void uvHandleReq(SSvrConn* pConn) {
      tDebug("conn %p acquired by server app", pConn);
    }
  }
-  STrans*   pTransInst = pConn->pTransInst;
  STraceId* trace = &pHead->traceId;
  if (pConn->status == ConnNormal && pHead->noResp == 0) {
    transRefSrvHandle(pConn);
@ -258,21 +261,32 @@ static void uvHandleReq(SSvrConn* pConn) {
  transReleaseExHandle(transGetRefMgt(), pConn->refId);

  (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
+  return true;
 }

 void uvOnRecvCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) {
-  // opt
  SSvrConn* conn = cli->data;
-  SConnBuffer* pBuf = &conn->readBuf;
  STrans*   pTransInst = conn->pTransInst;
+
+  SConnBuffer* pBuf = &conn->readBuf;
  if (nread > 0) {
    pBuf->len += nread;
    tTrace("%s conn %p total read:%d, current read:%d", transLabel(pTransInst), conn, pBuf->len, (int)nread);
+    if (pBuf->len <= TRANS_PACKET_LIMIT) {
      while (transReadComplete(pBuf)) {
        tTrace("%s conn %p alread read complete packet", transLabel(pTransInst), conn);
-      uvHandleReq(conn);
+        if (true == pBuf->invalid || false == uvHandleReq(conn)) {
+          tError("%s conn %p read invalid packet", transLabel(pTransInst), conn);
+          destroyConn(conn, true);
+          return;
+        }
      }
      return;
+    } else {
+      tError("%s conn %p read invalid packet, exceed limit", transLabel(pTransInst), conn);
+      destroyConn(conn, true);
+      return;
+    }
  }
  if (nread == 0) {
    return;
@ -364,6 +378,7 @@ static void uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) {
  pHead->ahandle = (uint64_t)pMsg->info.ahandle;
  pHead->traceId = pMsg->info.traceId;
  pHead->hasEpSet = pMsg->info.hasEpSet;
+  pHead->magicNum = htonl(TRANS_MAGIC_NUM);

  if (pConn->status == ConnNormal) {
    pHead->msgType = (0 == pMsg->msgType ? pConn->inType + 1 : pMsg->msgType);
@ -859,6 +874,7 @@ static int reallocConnRef(SSvrConn* conn) {
 }
 static void uvDestroyConn(uv_handle_t* handle) {
  SSvrConn* conn = handle->data;
+
  if (conn == NULL) {
    return;
  }
@ -874,9 +890,8 @@ static void uvDestroyConn(uv_handle_t* handle) {
    SSvrMsg* msg = transQueueGet(&conn->srvMsgs, i);
    destroySmsg(msg);
  }
-
-  transReqQueueClear(&conn->wreqQueue);
  transQueueDestroy(&conn->srvMsgs);
+  transReqQueueClear(&conn->wreqQueue);

  QUEUE_REMOVE(&conn->queue);
  taosMemoryFree(conn->pTcp);
--- a/tests/script/tsim/stream/basic0.sim
+++ b/tests/script/tsim/stream/basic0.sim
@ -1,7 +1,7 @@
 system sh/stop_dnodes.sh
 system sh/deploy.sh -n dnode1 -i 1
-system sh/exec.sh -n dnode1 -s start
-sleep 50
+system sh/cfg.sh -n dnode1 -c debugflag -v 131
+system sh/exec.sh -n dnode1 -s start -v
 sql connect

 print =============== create database
@ -137,4 +137,17 @@ if $data13 != 789 then
  return -1
 endi

+_OVER:
 system sh/exec.sh -n dnode1 -s stop -x SIGINT
+print =============== check
+$null=
+
+system_content sh/checkValgrind.sh -n dnode1 
+print cmd return result ----> [ $system_content ]
+if $system_content > 0 then
+  return -1
+endi 
+
+if $system_content == $null then
+  return -1
+endi 
--- a/tests/system-test/1-insert/time_range_wise.py
+++ b/tests/system-test/1-insert/time_range_wise.py
@ -293,7 +293,7 @@ class TDTestCase:
        dbname =  tdSql.getData(0,0)
        tdSql.query("select * from information_schema.ins_databases")
        for index , value in enumerate(tdSql.cursor.description):
-            if value[0] == "retention":
+            if value[0] == "retentions":
                r_index = index
                break
        for row in tdSql.queryResult:
--- a/tests/system-test/2-query/interp.py
+++ b/tests/system-test/2-query/interp.py
@ -551,6 +551,29 @@ class TDTestCase:
        tdSql.checkData(0, 0, 15)
        tdSql.checkData(1, 0, 15)

+        tdLog.printNoPrefix("==========step9:test error cases")
+
+        tdSql.error(f"select interp(c0) from {dbname}.{tbname}")
+        tdSql.error(f"select interp(c0) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05')")
+        tdSql.error(f"select interp(c0) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d)")
+        tdSql.error(f"select interp(c0) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') fill(null)")
+        tdSql.error(f"select interp(c0) from {dbname}.{tbname} every(1s) fill(null)")
+        tdSql.error(f"select interp(c0) from {dbname}.{tbname} where ts >= '2020-02-10 00:00:05' and ts <= '2020-02-15 00:00:05' every(1s) fill(null)")
+
+        # input can only be numerical types
+        tdSql.error(f"select interp(ts) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
+        tdSql.error(f"select interp(c6) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
+        tdSql.error(f"select interp(c7) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
+        tdSql.error(f"select interp(c8) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
+
+        # input can only be columns
+        tdSql.error(f"select interp(1) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
+        tdSql.error(f"select interp(1.5) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
+        tdSql.error(f"select interp(true) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
+        tdSql.error(f"select interp(false) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
+        tdSql.error(f"select interp('abcd') from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
+        tdSql.error(f"select interp('中文字符') from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
+
    def stop(self):
        tdSql.close()
        tdLog.success(f"{__file__} successfully executed")
--- a/tools/CMakeLists.txt
+++ b/tools/CMakeLists.txt
@ -128,6 +128,7 @@ ELSE ()
                COMMAND cmake -E copy ./taosadapter.service ${CMAKE_BINARY_DIR}/test/cfg/
                COMMAND cmake -E copy taosadapter-debug ${CMAKE_BINARY_DIR}/build/bin
                )
+        unset(_upx_prefix)
    ELSEIF (TD_DARWIN)
        include(ExternalProject)
        ExternalProject_Add(taosadapter
@ -149,8 +150,42 @@ ELSE ()
                COMMAND cmake -E copy ./taosadapter.service ${CMAKE_BINARY_DIR}/test/cfg/
                COMMAND cmake -E copy taosadapter-debug ${CMAKE_BINARY_DIR}/build/bin
                )
+#        unset(_upx_prefix)
+    ELSEIF (TD_WINDOWS)
+        include(ExternalProject)
+        set(_upx_prefix "${CMAKE_BINARY_DIR}/.taos/externals/upx")
+        ExternalProject_Add(upx
+                PREFIX              "${_upx_prefix}"
+                URL                 https://github.com/upx/upx/releases/download/v3.96/upx-3.96-win32.zip
+                CONFIGURE_COMMAND   cmake -E true
+                BUILD_COMMAND       cmake -E true
+                INSTALL_COMMAND     cmake -E true
+                )
+
+        ExternalProject_Add(taosadapter
+                PREFIX        "taosadapter"
+                SOURCE_DIR    ${CMAKE_CURRENT_SOURCE_DIR}/taosadapter
+                BUILD_ALWAYS         off
+                DEPENDS             taos
+                BUILD_IN_SOURCE     1
+                CONFIGURE_COMMAND cmake -E echo "taosadapter no need cmake to config"
+                PATCH_COMMAND
+                COMMAND     git clean -f -d
+                BUILD_COMMAND
+                COMMAND set CGO_CFLAGS=-I${CMAKE_CURRENT_SOURCE_DIR}/../include/client
+                COMMAND set CGO_LDFLAGS=-L${CMAKE_BINARY_DIR}/build/lib
+                COMMAND go build -ldflags "-s -w -X github.com/taosdata/taosadapter/version.Version=${taos_version} -X github.com/taosdata/taosadapter/version.CommitID=${taosadapter_commit_sha1}"
+                COMMAND go build -o taosadapter-debug -ldflags "-X github.com/taosdata/taosadapter/version.Version=${taos_version} -X github.com/taosdata/taosadapter/version.CommitID=${taosadapter_commit_sha1}"
+                INSTALL_COMMAND
+                COMMAND ${_upx_prefix}/src/upx/upx taosadapter.exe
+                COMMAND cmake -E copy taosadapter.exe ${CMAKE_BINARY_DIR}/build/bin
+                COMMAND cmake -E make_directory ${CMAKE_BINARY_DIR}/test/cfg/
+                COMMAND cmake -E copy ./example/config/taosadapter.toml ${CMAKE_BINARY_DIR}/test/cfg/
+                COMMAND cmake -E copy ./taosadapter.service ${CMAKE_BINARY_DIR}/test/cfg/
+                COMMAND cmake -E copy taosadapter-debug ${CMAKE_BINARY_DIR}/build/bin
+                )
        unset(_upx_prefix)
    ELSE ()
-        MESSAGE("${Yellow} Windows system still use original embedded httpd ${ColourReset}")
+        MESSAGE("${Yellow} taosAdapter Not supported yet  ${ColourReset}")
    ENDIF ()
 ENDIF ()