Merge branch '3.0' into feature/TD-11274-3.0

.gitignore

@@ -118,3 +118,4 @@ contrib/*
 !contrib/test
 sql
 debug*/
+.env
\ No newline at end of file

cmake/taosws_CMakeLists.txt.in

@@ -2,7 +2,7 @@
 # taosws-rs
 ExternalProject_Add(taosws-rs
         GIT_REPOSITORY https://github.com/taosdata/taosws-rs.git
-        GIT_TAG 648cc62
+        GIT_TAG 29424d5
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosws-rs"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

docs/zh/07-develop/09-udf.md

@@ -124,52 +124,49 @@ gcc -g -O0 -fPIC -shared add_one.c -o add_one.so
 
 用户可以通过 SQL 指令在系统中加载客户端所在主机上的 UDF 函数库（不能通过 RESTful 接口或 HTTP 管理界面来进行这一过程）。一旦创建成功，则当前 TDengine 集群的所有用户都可以在 SQL 指令中使用这些函数。UDF 存储在系统的 MNode 节点上，因此即使重启 TDengine 系统，已经创建的 UDF 也仍然可用。
 
-在创建 UDF 时，需要区分标量函数和聚合函数。如果创建时声明了错误的函数类别，则可能导致通过 SQL 指令调用函数时出错。此外， UDF 支持输入与输出类型不一致，用户需要保证输入数据类型与 UDF 程序匹配，UDF 输出数据类型与 OUTPUTTYPE 匹配。
+在创建 UDF 时，需要区分标量函数和聚合函数。如果创建时声明了错误的函数类别，则可能导致通过 SQL 指令调用函数时出错。此外，用户需要保证输入数据类型与 UDF 程序匹配，UDF 输出数据类型与 OUTPUTTYPE 匹配。
 
 - 创建标量函数
 ```sql
-CREATE FUNCTION ids(X) AS ids(Y) OUTPUTTYPE typename(Z) [ BUFSIZE B ];
+CREATE FUNCTION function_name AS library_path OUTPUTTYPE output_type;
 ```
 
-  - ids(X)：标量函数未来在 SQL 指令中被调用时的函数名，必须与函数实现中 udfNormalFunc 的实际名称一致；
-  - ids(Y)：包含 UDF 函数实现的动态链接库的库文件绝对路径（指的是库文件在当前客户端所在主机上的保存路径，通常是指向一个 .so 文件），这个路径需要用英文单引号或英文双引号括起来；
-  - typename(Z)：此函数计算结果的数据类型，与上文中 udfNormalFunc 的 itype 参数不同，这里不是使用数字表示法，而是直接写类型名称即可；
-  - B：中间计算结果的缓冲区大小，单位是字节，最小 0，最大 512，如果不使用可以不设置。
+  - function_name：标量函数未来在 SQL 中被调用时的函数名，必须与函数实现中 udf 的实际名称一致；
+  - library_path：包含 UDF 函数实现的动态链接库的库文件绝对路径（指的是库文件在当前客户端所在主机上的保存路径，通常是指向一个 .so 文件），这个路径需要用英文单引号或英文双引号括起来；
+  - output_type：此函数计算结果的数据类型名称；
 
-  例如，如下语句可以把 add_one.so 创建为系统中可用的 UDF：
+  例如，如下语句可以把 libbitand.so 创建为系统中可用的 UDF：
 
   ```sql
-  CREATE FUNCTION add_one AS "/home/taos/udf_example/add_one.so" OUTPUTTYPE INT;
+  CREATE FUNCTION bit_and AS "/home/taos/udf_example/libbitand.so" OUTPUTTYPE INT;
   ```
 
 - 创建聚合函数：
 ```sql
-CREATE AGGREGATE FUNCTION ids(X) AS ids(Y) OUTPUTTYPE typename(Z) [ BUFSIZE B ];
+CREATE AGGREGATE FUNCTION function_name AS library_path OUTPUTTYPE output_type [ BUFSIZE buffer_size ];
 ```
 
-  - ids(X)：聚合函数未来在 SQL 指令中被调用时的函数名，必须与函数实现中 udfNormalFunc 的实际名称一致；
-  - ids(Y)：包含 UDF 函数实现的动态链接库的库文件绝对路径（指的是库文件在当前客户端所在主机上的保存路径，通常是指向一个 .so 文件），这个路径需要用英文单引号或英文双引号括起来；
-  - typename(Z)：此函数计算结果的数据类型，与上文中 udfNormalFunc 的 itype 参数不同，这里不是使用数字表示法，而是直接写类型名称即可；
-  - B：中间计算结果的缓冲区大小，单位是字节，最小 0，最大 512，如果不使用可以不设置。
+  - function_name：聚合函数未来在 SQL 中被调用时的函数名，必须与函数实现中 udfNormalFunc 的实际名称一致；
+  - library_path：包含 UDF 函数实现的动态链接库的库文件绝对路径（指的是库文件在当前客户端所在主机上的保存路径，通常是指向一个 .so 文件），这个路径需要用英文单引号或英文双引号括起来；
+  - output_type：此函数计算结果的数据类型，与上文中 udfNormalFunc 的 itype 参数不同，这里不是使用数字表示法，而是直接写类型名称即可；
+  - buffer_size：中间计算结果的缓冲区大小，单位是字节。如果不使用可以不设置。
 
-  关于中间计算结果的使用，可以参考示例程序[demo.c](https://github.com/taosdata/TDengine/blob/develop/tests/script/sh/demo.c)
-
-  例如，如下语句可以把 demo.so 创建为系统中可用的 UDF：
+  例如，如下语句可以把 libsqrsum.so 创建为系统中可用的 UDF：
 
   ```sql
-  CREATE AGGREGATE FUNCTION demo AS "/home/taos/udf_example/demo.so" OUTPUTTYPE DOUBLE bufsize 14;
+  CREATE AGGREGATE FUNCTION sqr_sum AS "/home/taos/udf_example/libsqrsum.so" OUTPUTTYPE DOUBLE bufsize 8;
   ```
 
 ### 管理 UDF
 
 - 删除指定名称的用户定义函数：
 ```
-DROP FUNCTION ids(X);
+DROP FUNCTION function_name;
 ```
 
-- ids(X)：此参数的含义与 CREATE 指令中的 ids(X) 参数一致，也即要删除的函数的名字，例如 
+- function_name：此参数的含义与 CREATE 指令中的 function_name 参数一致，也即要删除的函数的名字，例如 
 ```sql
-DROP FUNCTION add_one;
+DROP FUNCTION bit_and;
 ```
 - 显示系统中当前可用的所有 UDF：
 ```sql
@@ -180,53 +177,32 @@ SHOW FUNCTIONS;
 
 在 SQL 指令中，可以直接以在系统中创建 UDF 时赋予的函数名来调用用户定义函数。例如：
 ```sql
-SELECT X(c) FROM table/stable;
+SELECT X(c1,c2) FROM table/stable;
 ```
 
-表示对名为 c 的数据列调用名为 X 的用户定义函数。SQL 指令中用户定义函数可以配合 WHERE 等查询特性来使用。
-
-## UDF 的一些使用限制
-
-在当前版本下，使用 UDF 存在如下这些限制：
+表示对名为 c1, c2 的数据列调用名为 X 的用户定义函数。SQL 指令中用户定义函数可以配合 WHERE 等查询特性来使用。
 
-1. 在创建和调用 UDF 时，服务端和客户端都只支持 Linux 操作系统；
-2. UDF 不能与系统内建的 SQL 函数混合使用，暂不支持在一条 SQL 语句中使用多个不同名的 UDF ；
-3. UDF 只支持以单个数据列作为输入；
-4. UDF 只要创建成功，就会被持久化存储到 MNode 节点中；
-5. 无法通过 RESTful 接口来创建 UDF；
-6. UDF 在 SQL 中定义的函数名，必须与 .so 库文件实现中的接口函数名前缀保持一致，也即必须是 udfNormalFunc 的名称，而且不可与 TDengine 中已有的内建 SQL 函数重名。
 
 ## 示例代码
 
-### 标量函数示例 [add_one](https://github.com/taosdata/TDengine/blob/develop/tests/script/sh/add_one.c)
-
-<details>
-<summary>add_one.c</summary>
-
-```c
-{{#include tests/script/sh/add_one.c}}
-```
-
-</details>
-
-### 向量函数示例 [abs_max](https://github.com/taosdata/TDengine/blob/develop/tests/script/sh/abs_max.c)
+### 标量函数示例 [bit_and](https://github.com/taosdata/TDengine/blob/develop/tests/script/sh/bit_and.c)
 
 <details>
-<summary>abs_max.c</summary>
+<summary>bit_and.c</summary>
 
 ```c
-{{#include tests/script/sh/abs_max.c}}
+{{#include tests/script/sh/bit_and.c}}
 ```
 
 </details>
 
-### 使用中间计算结果示例 [demo](https://github.com/taosdata/TDengine/blob/develop/tests/script/sh/demo.c)
+### 聚合函数示例 [sqr_sum](https://github.com/taosdata/TDengine/blob/develop/tests/script/sh/sqr_sum.c)
 
 <details>
-<summary>demo.c</summary>
+<summary>sqr_sum.c</summary>
 
 ```c
-{{#include tests/script/sh/demo.c}}
+{{#include tests/script/sh/sqr_sum.c}}
 ```
 
 </details>

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

@@ -8,21 +8,30 @@ title: 用户自定义函数
 ## 创建函数
 
 ```sql
-CREATE [AGGREGATE] FUNCTION func_name AS library_path OUTPUTTYPE type_name [BUFSIZE value]
+CREATE [AGGREGATE] FUNCTION func_name AS library_path OUTPUTTYPE type_name [BUFSIZE buffer_size]
 ```
 
 语法说明：
 
 AGGREGATE：标识此函数是标量函数还是聚集函数。
-func_name：函数名，必须与函数实现中udfNormalFunc的实际名称一致。
+func_name：函数名，必须与函数实现中 udf 的实际名称一致。
 library_path：包含UDF函数实现的动态链接库的绝对路径，是在客户端侧主机上的绝对路径。
-OUTPUTTYPE：标识此函数的返回类型。
-BUFSIZE：中间结果的缓冲区大小，单位是字节。不设置则默认为0。最大不可超过512字节。
+type_name：标识此函数的返回类型。
+buffer_size：中间结果的缓冲区大小，单位是字节。不设置则默认为0。
 
 关于如何开发自定义函数，请参考 [UDF使用说明](../../develop/udf)。
 
 ## 删除自定义函数
 
+```
+DROP FUNCTION function_name;
+```
+
+- function_name：此参数的含义与 CREATE 指令中的 function_name 参数一致，也即要删除的函数的名字，例如 
+
+
+## 显示 UDF
+
 ```sql
-DROP FUNCTION func_name
-```
+SHOW FUNCTION;
+```
\ No newline at end of file

docs/zh/14-reference/05-taosbenchmark.md

@@ -233,11 +233,28 @@ taosBenchmark -A INT,DOUBLE,NCHAR,BINARY\(16\)
 
 - **drop** : 插入前是否删除数据库，默认为 true。
 
+#### 流式计算相关配置参数
+
+创建流式计算的相关参数在 json 配置文件中的 `stream` 中配置，具体参数如下。
+
+- **stream_name** : 流式计算的名称，必填项。
+
+- **stream_stb** : 流式计算对应的超级表名称，必填项。
+
+- **stream_sql** : 流式计算的sql语句，必填项。
+
+- **trigger_mode** : 流式计算的触发模式，可选项。
+
+- **watermark** : 流式计算的水印，可选项。
+
+- **drop** : 是否创建流式计算，可选项为 "yes" 或者 "no", 为 "no" 时不创建。
+
 #### 超级表相关配置参数
 
-创建超级表时的相关参数在 json 配置文件中的 `super_tables` 中配置，具体参数如下表。
+创建超级表时的相关参数在 json 配置文件中的 `super_tables` 中配置，具体参数如下。
 
 - **name**: 超级表名，必须配置，没有默认值。
+
 - **child_table_exists** : 子表是否已经存在，默认值为 "no"，可选值为 "yes" 或 "no"。
 
 - **child_table_count** : 子表的数量，默认值为 10。
@@ -288,6 +305,22 @@ taosBenchmark -A INT,DOUBLE,NCHAR,BINARY\(16\)
 
 - **tags_file** : 仅当 insert_mode 为 taosc, rest 的模式下生效。 最终的 tag 的数值与 childtable_count 有关，如果 csv 文件内的 tag 数据行小于给定的子表数量，那么会循环读取 csv 文件数据直到生成 childtable_count 指定的子表数量；否则则只会读取 childtable_count 行 tag 数据。也即最终生成的子表数量为二者取小。
 
+#### tsma配置参数
+
+指定tsma的配置参数在 `super_tables` 中的 `tsmas` 中，具体参数如下。
+
+- **name** : 指定 tsma 的名字，必选项。
+
+- **function** : 指定 tsma 的函数，必选项。
+
+- **interval** : 指定 tsma 的时间间隔，必选项。
+
+- **sliding** : 指定 tsma 的窗口时间位移，必选项。
+
+- **custom** : 指定 tsma 的创建语句结尾追加的自定义配置，可选项。
+
+- **start_when_inserted** : 指定当插入多少行时创建 tsma，可选项，默认为 0。
+
 #### 标签列与数据列配置参数
 
 指定超级表标签列与数据列的配置参数分别在 `super_tables` 中的 `columns` 和 `tag` 中。

docs/zh/28-releases.md

+---
+sidebar_label: 发布历史
+title: 发布历史
+---
+
+import Release from "/components/Release";
+
+
+<Release versionPrefix="3.0" />

include/common/tcommon.h

@@ -107,6 +107,7 @@ typedef struct SDataBlockInfo {
   int32_t     childId;  // used for stream, do not serialize
   EStreamType type;     // used for stream, do not serialize
   STimeWindow calWin;   // used for stream, do not serialize
+  TSKEY       watermark;// used for stream
 } SDataBlockInfo;
 
 typedef struct SSDataBlock {

include/common/tmsg.h

@@ -1369,6 +1369,7 @@ typedef struct {
   int64_t skey;
   int64_t ekey;
   int64_t version;  // for stream
+  TSKEY   watermark;// for stream
   char    data[];
 } SRetrieveTableRsp;
 
@@ -3054,6 +3055,7 @@ int32_t tEncodeDeleteRes(SEncoder* pCoder, const SDeleteRes* pRes);
 int32_t tDecodeDeleteRes(SDecoder* pCoder, SDeleteRes* pRes);
 
 typedef struct {
+  int32_t msgIdx;
   int32_t msgType;
   int32_t msgLen;
   void*   msg;
@@ -3067,6 +3069,7 @@ typedef struct {
 
 typedef struct {
   int32_t reqType;
+  int32_t msgIdx;
   int32_t msgLen;
   int32_t rspCode;
   void*   msg;

include/libs/catalog/catalog.h

@@ -58,12 +58,17 @@ typedef struct SDbInfo {
   int64_t dbId;
 } SDbInfo;
 
+typedef struct STablesReq {
+  char    dbFName[TSDB_DB_FNAME_LEN];
+  SArray* pTables;
+} STablesReq;
+
 typedef struct SCatalogReq {
   SArray* pDbVgroup;      // element is db full name
   SArray* pDbCfg;         // element is db full name
   SArray* pDbInfo;        // element is db full name
-  SArray* pTableMeta;     // element is SNAME
-  SArray* pTableHash;     // element is SNAME
+  SArray* pTableMeta;     // element is STablesReq
+  SArray* pTableHash;     // element is STablesReq
   SArray* pUdf;           // element is udf name
   SArray* pIndex;         // element is index name
   SArray* pUser;          // element is SUserAuthInfo

source/client/src/clientImpl.c

@@ -1308,8 +1308,8 @@ int32_t doProcessMsgFromServer(void* param) {
   char      tbuf[40] = {0};
   TRACE_TO_STR(trace, tbuf);
 
-  tscDebug("processMsgFromServer handle %p, message: %s, code: %s, gtid: %s", pMsg->info.handle,
-           TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code), tbuf);
+  tscDebug("processMsgFromServer handle %p, message: %s, size:%d, code: %s, gtid: %s", pMsg->info.handle,
+           TMSG_INFO(pMsg->msgType), pMsg->contLen, tstrerror(pMsg->code), tbuf);
 
   if (pSendInfo->requestObjRefId != 0) {
     SRequestObj* pRequest = (SRequestObj*)taosAcquireRef(clientReqRefPool, pSendInfo->requestObjRefId);
@@ -1922,7 +1922,7 @@ _OVER:
   return code;
 }
 
-int32_t appendTbToReq(SArray* pList, int32_t pos1, int32_t len1, int32_t pos2, int32_t len2, const char* str,
+int32_t appendTbToReq(SHashObj* pHash, int32_t pos1, int32_t len1, int32_t pos2, int32_t len2, const char* str,
                       int32_t acctId, char* db) {
   SName name;
 
@@ -1957,20 +1957,33 @@ int32_t appendTbToReq(SArray* pList, int32_t pos1, int32_t len1, int32_t pos2, i
     return -1;
   }
 
-  taosArrayPush(pList, &name);
+  char dbFName[TSDB_DB_FNAME_LEN];
+  sprintf(dbFName, "%d.%.*s", acctId, dbLen, dbName);
+
+  STablesReq* pDb = taosHashGet(pHash, dbFName, strlen(dbFName));
+  if (pDb) {
+    taosArrayPush(pDb->pTables, &name);
+  } else {
+    STablesReq db;
+    db.pTables = taosArrayInit(20, sizeof(SName));
+    strcpy(db.dbFName, dbFName);
+    taosArrayPush(db.pTables, &name);
+    taosHashPut(pHash, dbFName, strlen(dbFName), &db, sizeof(db));
+  }
 
   return TSDB_CODE_SUCCESS;
 }
 
 int32_t transferTableNameList(const char* tbList, int32_t acctId, char* dbName, SArray** pReq) {
-  *pReq = taosArrayInit(10, sizeof(SName));
-  if (NULL == *pReq) {
+  SHashObj* pHash = taosHashInit(3, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
+  if (NULL == pHash) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return terrno;
   }
 
   bool    inEscape = false;
   int32_t code = 0;
+  void *pIter = NULL;
 
   int32_t vIdx = 0;
   int32_t vPos[2];
@@ -1985,7 +1998,7 @@ int32_t transferTableNameList(const char* tbList, int32_t acctId, char* dbName,
         vLen[vIdx] = i - vPos[vIdx];
       }
 
-      code = appendTbToReq(*pReq, vPos[0], vLen[0], vPos[1], vLen[1], tbList, acctId, dbName);
+      code = appendTbToReq(pHash, vPos[0], vLen[0], vPos[1], vLen[1], tbList, acctId, dbName);
       if (code) {
         goto _return;
       }
@@ -2035,7 +2048,7 @@ int32_t transferTableNameList(const char* tbList, int32_t acctId, char* dbName,
         vLen[vIdx] = i - vPos[vIdx];
       }
 
-      code = appendTbToReq(*pReq, vPos[0], vLen[0], vPos[1], vLen[1], tbList, acctId, dbName);
+      code = appendTbToReq(pHash, vPos[0], vLen[0], vPos[1], vLen[1], tbList, acctId, dbName);
       if (code) {
         goto _return;
       }
@@ -2067,14 +2080,31 @@ int32_t transferTableNameList(const char* tbList, int32_t acctId, char* dbName,
     goto _return;
   }
 
+  int32_t dbNum = taosHashGetSize(pHash);
+  *pReq = taosArrayInit(dbNum, sizeof(STablesReq));
+  pIter = taosHashIterate(pHash, NULL);
+  while (pIter) {
+    STablesReq* pDb = (STablesReq*)pIter;
+    taosArrayPush(*pReq, pDb);
+    pIter = taosHashIterate(pHash, pIter);
+  }
+
+  taosHashCleanup(pHash);
+
   return TSDB_CODE_SUCCESS;
 
 _return:
 
   terrno = TSDB_CODE_TSC_INVALID_OPERATION;
 
-  taosArrayDestroy(*pReq);
-  *pReq = NULL;
+  pIter = taosHashIterate(pHash, NULL);
+  while (pIter) {
+    STablesReq* pDb = (STablesReq*)pIter;
+    taosArrayDestroy(pDb->pTables);
+    pIter = taosHashIterate(pHash, pIter);
+  }
+
+  taosHashCleanup(pHash);
 
   return terrno;
 }

source/common/src/systable.c

@@ -308,9 +308,9 @@ static const SSysDbTableSchema offsetSchema[] = {
 };
 
 static const SSysDbTableSchema querySchema[] = {
-    {.name = "query_id", .bytes = TSDB_QUERY_ID_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
-    {.name = "req_id", .bytes = 8, .type = TSDB_DATA_TYPE_UBIGINT},
-    {.name = "connId", .bytes = 4, .type = TSDB_DATA_TYPE_UINT},
+    {.name = "kill_id", .bytes = TSDB_QUERY_ID_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "query_id", .bytes = 8, .type = TSDB_DATA_TYPE_UBIGINT},
+    {.name = "conn_id", .bytes = 4, .type = TSDB_DATA_TYPE_UINT},
     {.name = "app", .bytes = TSDB_APP_NAME_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
     {.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "user", .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},

source/common/src/tdatablock.c

@@ -1272,8 +1272,7 @@ int32_t assignOneDataBlock(SSDataBlock* dst, const SSDataBlock* src) {
     colDataAssign(pDst, pSrc, src->info.rows, &src->info);
   }
 
-  dst->info.rows = src->info.rows;
-  dst->info.capacity = src->info.rows;
+  dst->info = src->info;
   return 0;
 }
 

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

@@ -84,6 +84,9 @@ int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
   }
   
   for (int32_t i = 0; i < msgNum; ++i) {
+    req.msgIdx = ntohl(*(int32_t*)((char*)pMsg->pCont + offset));
+    offset += sizeof(req.msgIdx);
+
     req.msgType = ntohl(*(int32_t*)((char*)pMsg->pCont + offset));
     offset += sizeof(req.msgType);
 
@@ -111,6 +114,7 @@ int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
     } else {
       rsp.rspCode = 0;
     }
+    rsp.msgIdx = req.msgIdx;
     rsp.reqType = reqMsg.msgType;
     rsp.msgLen = reqMsg.info.rspLen;
     rsp.msg = reqMsg.info.rsp;
@@ -136,6 +140,8 @@ int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
     
     *(int32_t*)((char*)pRsp + offset) = htonl(p->reqType);
     offset += sizeof(p->reqType);
+    *(int32_t*)((char*)pRsp + offset) = htonl(p->msgIdx);
+    offset += sizeof(p->msgIdx);
     *(int32_t*)((char*)pRsp + offset) = htonl(p->msgLen);
     offset += sizeof(p->msgLen);
     *(int32_t*)((char*)pRsp + offset) = htonl(p->rspCode);

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

@@ -223,6 +223,8 @@ int vnodeCommit(SVnode *pVnode) {
   vnodeBufPoolUnRef(pVnode->inUse);
   pVnode->inUse = NULL;
 
+  pVnode->state.commitTerm = pVnode->state.applyTerm;
+
   // save info
   info.config = pVnode->config;
   info.state.committed = pVnode->state.applied;

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

@@ -273,6 +273,9 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
   }
 
   for (int32_t i = 0; i < msgNum; ++i) {
+    req.msgIdx = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
+    offset += sizeof(req.msgIdx);
+
     req.msgType = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
     offset += sizeof(req.msgType);
 
@@ -301,6 +304,7 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
         break;
     }
 
+    rsp.msgIdx = req.msgIdx;
     rsp.reqType = reqMsg.msgType;
     rsp.msgLen = reqMsg.contLen;
     rsp.rspCode = reqMsg.code;
@@ -327,6 +331,8 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
 
     *(int32_t *)((char *)pRsp + offset) = htonl(p->reqType);
     offset += sizeof(p->reqType);
+    *(int32_t *)((char *)pRsp + offset) = htonl(p->msgIdx);
+    offset += sizeof(p->msgIdx);
     *(int32_t *)((char *)pRsp + offset) = htonl(p->msgLen);
     offset += sizeof(p->msgLen);
     *(int32_t *)((char *)pRsp + offset) = htonl(p->rspCode);

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

@@ -708,8 +708,8 @@ int32_t vnodeSyncOpen(SVnode *pVnode, char *path) {
   }
 
   setPingTimerMS(pVnode->sync, 5000);
-  setElectTimerMS(pVnode->sync, 2800);
-  setHeartbeatTimerMS(pVnode->sync, 900);
+  setElectTimerMS(pVnode->sync, 4000);
+  setHeartbeatTimerMS(pVnode->sync, 700);
   return 0;
 }
 

source/libs/catalog/inc/catalogInt.h

@@ -32,6 +32,7 @@ extern "C" {
 #define CTG_DEFAULT_RENT_SLOT_SIZE 10
 #define CTG_DEFAULT_MAX_RETRY_TIMES 3
 #define CTG_DEFAULT_BATCH_NUM 64
+#define CTG_DEFAULT_FETCH_NUM 8
 
 #define CTG_RENT_SLOT_SECOND 1.5
 
@@ -80,6 +81,8 @@ typedef enum {
   CTG_TASK_GET_UDF,
   CTG_TASK_GET_USER,
   CTG_TASK_GET_SVR_VER,
+  CTG_TASK_GET_TB_META_BATCH,
+  CTG_TASK_GET_TB_HASH_BATCH,
 } CTG_TASK_TYPE;
 
 typedef enum {
@@ -110,6 +113,23 @@ typedef struct SCtgTbMetaCtx {
   int32_t flag;
 } SCtgTbMetaCtx;
 
+typedef struct SCtgFetch {
+  int32_t dbIdx;
+  int32_t tbIdx;
+  int32_t fetchIdx;
+  int32_t resIdx;
+  int32_t flag;
+  SCtgTbCacheInfo tbInfo;
+  int32_t vgId;
+} SCtgFetch;
+
+typedef struct SCtgTbMetasCtx {
+  int32_t fetchNum;
+  SArray* pNames;
+  SArray* pResList;
+  SArray* pFetchs;
+} SCtgTbMetasCtx;
+
 typedef struct SCtgTbIndexCtx {
   SName* pName;
 } SCtgTbIndexCtx;
@@ -137,6 +157,14 @@ typedef struct SCtgTbHashCtx {
   SName* pName;
 } SCtgTbHashCtx;
 
+typedef struct SCtgTbHashsCtx {
+  int32_t fetchNum;
+  SArray* pNames;
+  SArray* pResList;
+  SArray* pFetchs;
+} SCtgTbHashsCtx;
+
+
 typedef struct SCtgIndexCtx {
   char indexFName[TSDB_INDEX_FNAME_LEN];
 } SCtgIndexCtx;
@@ -211,6 +239,7 @@ typedef struct SCtgBatch {
   SRequestConnInfo conn;
   char    dbFName[TSDB_DB_FNAME_LEN];
   SArray* pTaskIds;
+  SArray* pMsgIdxs;
 } SCtgBatch;
 
 typedef struct SCtgJob {
@@ -218,6 +247,7 @@ typedef struct SCtgJob {
   int32_t           batchId;
   SHashObj*         pBatchs;
   SArray*           pTasks;
+  int32_t           subTaskNum;
   int32_t           taskDone;
   SMetaData         jobRes;
   int32_t           jobResCode;
@@ -258,6 +288,7 @@ typedef struct SCtgTaskCallbackParam {
   SArray*                 taskId;
   int32_t                 reqType;
   int32_t                 batchId;
+  SArray*                 msgIdx;
 } SCtgTaskCallbackParam;
 
 
@@ -276,6 +307,7 @@ typedef struct SCtgTask {
   int32_t         taskId;
   SCtgJob*        pJob;
   void*           taskCtx;
+  SArray*         msgCtxs;
   SCtgMsgCtx      msgCtx;
   int32_t         code;
   void*           res;
@@ -286,9 +318,14 @@ typedef struct SCtgTask {
   SHashObj*       pBatchs;
 } SCtgTask;
 
+typedef struct SCtgTaskReq {
+  SCtgTask* pTask;
+  int32_t   msgIdx;
+} SCtgTaskReq;
+
 typedef int32_t (*ctgInitTaskFp)(SCtgJob*, int32_t, void*);
 typedef int32_t (*ctgLanchTaskFp)(SCtgTask*);
-typedef int32_t (*ctgHandleTaskMsgRspFp)(SCtgTask*, int32_t, const SDataBuf *, int32_t);
+typedef int32_t (*ctgHandleTaskMsgRspFp)(SCtgTaskReq*, int32_t, const SDataBuf *, int32_t);
 typedef int32_t (*ctgDumpTaskResFp)(SCtgTask*);
 typedef int32_t (*ctgCloneTaskResFp)(SCtgTask*, void**);
 typedef int32_t (*ctgCompTaskFp)(SCtgTask*, void*, bool*);
@@ -487,6 +524,8 @@ typedef struct SCtgOperation {
 #define CTG_FLAG_MAKE_STB(_isStb) (((_isStb) == 1) ? CTG_FLAG_STB : ((_isStb) == 0 ? CTG_FLAG_NOT_STB : CTG_FLAG_UNKNOWN_STB))
 #define CTG_FLAG_MATCH_STB(_flag, tbType) (CTG_FLAG_IS_UNKNOWN_STB(_flag) || (CTG_FLAG_IS_STB(_flag) && (tbType) == TSDB_SUPER_TABLE) || (CTG_FLAG_IS_NOT_STB(_flag) && (tbType) != TSDB_SUPER_TABLE))
 
+#define CTG_GET_TASK_MSGCTX(_task, _id) (((CTG_TASK_GET_TB_META_BATCH == (_task)->type) || (CTG_TASK_GET_TB_HASH_BATCH == (_task)->type)) ? taosArrayGet((_task)->msgCtxs, (_id)) : &(_task)->msgCtx)
+
 #define CTG_META_SIZE(pMeta) (sizeof(STableMeta) + ((pMeta)->tableInfo.numOfTags + (pMeta)->tableInfo.numOfColumns) * sizeof(SSchema))
 
 #define CTG_TABLE_NOT_EXIST(code) (code == CTG_ERR_CODE_TABLE_NOT_EXIST) 
@@ -586,6 +625,7 @@ int32_t ctgdShowCacheInfo(void);
 
 int32_t ctgRemoveTbMetaFromCache(SCatalog* pCtg, SName* pTableName, bool syncReq);
 int32_t ctgGetTbMetaFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta);
+int32_t ctgGetTbMetasFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMetasCtx* ctx, int32_t dbIdx, int32_t *fetchIdx, int32_t baseResIdx, SArray* pList);
 
 int32_t ctgOpUpdateVgroup(SCtgCacheOperation *action);
 int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *action);
@@ -631,7 +671,7 @@ int32_t ctgGetTbHashVgroupFromCache(SCatalog *pCtg, const SName *pTableName, SVg
 
 
 int32_t ctgProcessRspMsg(void* out, int32_t reqType, char* msg, int32_t msgSize, int32_t rspCode, char* target);
-int32_t ctgGetDBVgInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SBuildUseDBInput *input, SUseDbOutput *out, SCtgTask* pTask);
+int32_t ctgGetDBVgInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SBuildUseDBInput *input, SUseDbOutput *out, SCtgTaskReq* tReq);
 int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SArray *out, SCtgTask* pTask);
 int32_t ctgGetDnodeListFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SArray **out, SCtgTask* pTask);
 int32_t ctgGetDBCfgFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *dbFName, SDbCfgInfo *out, SCtgTask* pTask);
@@ -639,9 +679,9 @@ int32_t ctgGetIndexInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const
 int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SName *name, STableIndex* out, SCtgTask* pTask);
 int32_t ctgGetUdfInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *funcName, SFuncInfo *out, SCtgTask* pTask);
 int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *user, SGetUserAuthRsp *out, SCtgTask* pTask);
-int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo *pConn, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTask* pTask);
-int32_t ctgGetTbMetaFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableMetaOutput* out, SCtgTask* pTask);
-int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* out, SCtgTask* pTask);
+int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo *pConn, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTaskReq* tReq);
+int32_t ctgGetTbMetaFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableMetaOutput* out, SCtgTaskReq* tReq);
+int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* out, SCtgTaskReq* tReq);
 int32_t ctgGetTableCfgFromVnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, SVgroupInfo *vgroupInfo, STableCfg **out, SCtgTask* pTask);
 int32_t ctgGetTableCfgFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableCfg **out, SCtgTask* pTask);
 int32_t ctgGetSvrVerFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, char **out, SCtgTask* pTask);
@@ -664,6 +704,7 @@ void    ctgFreeJob(void* job);
 void    ctgFreeHandleImpl(SCatalog* pCtg);
 void    ctgFreeVgInfo(SDBVgInfo *vgInfo);
 int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName *pTableName, SVgroupInfo *pVgroup);
+int32_t ctgGetVgInfosFromHashValue(SCatalog *pCtg, SCtgTaskReq* tReq, SDBVgInfo *dbInfo, SCtgTbHashsCtx *pCtx, char* dbFName, SArray* pNames, bool update);
 void    ctgResetTbMetaTask(SCtgTask* pTask);
 void    ctgFreeDbCache(SCtgDBCache *dbCache);
 int32_t ctgStbVersionSortCompare(const void* key1, const void* key2);
@@ -672,8 +713,11 @@ int32_t ctgStbVersionSearchCompare(const void* key1, const void* key2);
 int32_t ctgDbVgVersionSearchCompare(const void* key1, const void* key2);
 void    ctgFreeSTableMetaOutput(STableMetaOutput* pOutput);
 int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* target);
+int32_t ctgAddMsgCtx(SArray* pCtxs, int32_t reqType, void* out, char* target);
 char *  ctgTaskTypeStr(CTG_TASK_TYPE type);
 int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, char* dbFName, int32_t vgId);
+int32_t ctgGetTablesReqNum(SArray *pList);
+int32_t ctgAddFetch(SArray** pFetchs, int32_t dbIdx, int32_t tbIdx, int32_t *fetchIdx, int32_t resIdx, int32_t flag);
 int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes);
 void    ctgFreeSTableIndex(void *info);
 void    ctgClearSubTaskRes(SCtgSubRes *pRes);
@@ -682,6 +726,7 @@ void    ctgClearHandle(SCatalog* pCtg);
 void    ctgFreeTbCacheImpl(SCtgTbCache *pCache);
 int32_t ctgRemoveTbMeta(SCatalog* pCtg, SName* pTableName);
 int32_t ctgGetTbHashVgroup(SCatalog *pCtg, SRequestConnInfo *pConn, const SName *pTableName, SVgroupInfo *pVgroup);
+SName*  ctgGetFetchName(SArray* pNames, SCtgFetch* pFetch);
 
 
 extern SCatalogMgmt gCtgMgmt;

source/libs/catalog/src/ctgCache.c

@@ -242,7 +242,6 @@ int32_t ctgAcquireTbMetaFromCache(SCatalog* pCtg, char *dbFName, char* tbName, S
     goto _return;
   }
   
-  int32_t sz = 0;
   pCache = taosHashAcquire(dbCache->tbCache, tbName, strlen(tbName));
   if (NULL == pCache) {
     ctgDebug("tb %s not in cache, dbFName:%s", tbName, dbFName);
@@ -282,7 +281,6 @@ int32_t ctgAcquireStbMetaFromCache(SCatalog* pCtg, char *dbFName, uint64_t suid,
     goto _return;
   }
   
-  int32_t sz = 0;
   char* stName = taosHashAcquire(dbCache->stbCache, &suid, sizeof(suid));
   if (NULL == stName) {
     ctgDebug("stb 0x%" PRIx64 " not in cache, dbFName:%s", suid, dbFName);
@@ -2152,6 +2150,254 @@ int32_t ctgGetTbMetaFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMet
   return TSDB_CODE_SUCCESS;
 }
 
+#if 0
+int32_t ctgGetTbMetaBFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMetasCtx* ctx, SArray** pResList) {
+  int32_t tbNum = taosArrayGetSize(ctx->pNames);
+  SName* fName = taosArrayGet(ctx->pNames, 0);
+  int32_t fIdx = 0;
+  
+  for (int32_t i = 0; i < tbNum; ++i) {
+    SName* pName = taosArrayGet(ctx->pNames, i);
+    SCtgTbMetaCtx nctx = {0};
+    nctx.flag = CTG_FLAG_UNKNOWN_STB;
+    nctx.pName = pName;
+    
+    if (IS_SYS_DBNAME(pName->dbname)) {
+      CTG_FLAG_SET_SYS_DB(nctx.flag);
+    }
+
+    STableMeta *pTableMeta = NULL;
+    CTG_ERR_RET(ctgReadTbMetaFromCache(pCtg, &nctx, &pTableMeta));
+    SMetaRes res = {0};
+    
+    if (pTableMeta) {
+      if (CTG_FLAG_MATCH_STB(nctx.flag, pTableMeta->tableType) &&
+          ((!CTG_FLAG_IS_FORCE_UPDATE(nctx.flag)) || (CTG_FLAG_IS_SYS_DB(nctx.flag)))) {
+        res.pRes = pTableMeta;
+      } else {
+        taosMemoryFreeClear(pTableMeta);
+      }
+    }
+
+    if (NULL == res.pRes) {
+      if (NULL == ctx->pFetchs) {
+        ctx->pFetchs = taosArrayInit(tbNum, sizeof(SCtgFetch));
+      }
+      
+      if (CTG_FLAG_IS_UNKNOWN_STB(nctx.flag)) {
+        CTG_FLAG_SET_STB(nctx.flag, nctx.tbInfo.tbType);
+      }
+
+      SCtgFetch fetch = {0};
+      fetch.tbIdx = i;
+      fetch.fetchIdx = fIdx++;
+      fetch.flag = nctx.flag;
+
+      taosArrayPush(ctx->pFetchs, &fetch);
+    }
+    
+    taosArrayPush(ctx->pResList, &res);
+  }
+
+  if (NULL == ctx->pFetchs) {
+    TSWAP(*pResList, ctx->pResList);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+#endif
+
+int32_t ctgGetTbMetasFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMetasCtx* ctx, int32_t dbIdx, int32_t *fetchIdx, int32_t baseResIdx, SArray* pList) {
+  int32_t tbNum = taosArrayGetSize(pList);
+  SName* pName = taosArrayGet(pList, 0);
+  char dbFName[TSDB_DB_FNAME_LEN] = {0};
+  int32_t flag = CTG_FLAG_UNKNOWN_STB;
+  uint64_t lastSuid = 0;
+  STableMeta* lastTableMeta = NULL;
+  
+  if (IS_SYS_DBNAME(pName->dbname)) {
+    CTG_FLAG_SET_SYS_DB(flag);
+    strcpy(dbFName, pName->dbname);
+  } else {
+    tNameGetFullDbName(pName, dbFName);
+  }
+
+  SCtgDBCache *dbCache = NULL;
+  SCtgTbCache* pCache = NULL;
+  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
+  
+  if (NULL == dbCache) {
+    ctgDebug("db %s not in cache", dbFName);
+    for (int32_t i = 0; i < tbNum; ++i) {
+      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
+      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
+    }
+
+    return TSDB_CODE_SUCCESS;
+  }
+
+  for (int32_t i = 0; i < tbNum; ++i) {
+    SName* pName = taosArrayGet(pList, i);
+
+    pCache = taosHashAcquire(dbCache->tbCache, pName->tname, strlen(pName->tname));
+    if (NULL == pCache) {
+      ctgDebug("tb %s not in cache, dbFName:%s", pName->tname, dbFName);
+      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
+      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
+      
+      continue;
+    }
+
+    CTG_LOCK(CTG_READ, &pCache->metaLock);
+    if (NULL == pCache->pMeta) {
+      ctgDebug("tb %s meta not in cache, dbFName:%s", pName->tname, dbFName);
+      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
+      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
+      
+      continue;
+    }
+
+    STableMeta* tbMeta = pCache->pMeta;
+
+    SCtgTbMetaCtx nctx = {0};
+    nctx.flag = flag;
+    nctx.tbInfo.inCache = true;
+    nctx.tbInfo.dbId = dbCache->dbId;
+    nctx.tbInfo.suid = tbMeta->suid;
+    nctx.tbInfo.tbType = tbMeta->tableType;
+
+    SMetaRes res = {0};
+    STableMeta* pTableMeta = NULL;
+    if (tbMeta->tableType != TSDB_CHILD_TABLE) {
+      int32_t metaSize = CTG_META_SIZE(tbMeta);
+      pTableMeta = taosMemoryCalloc(1, metaSize);
+      if (NULL == pTableMeta) {
+        ctgReleaseTbMetaToCache(pCtg, dbCache, pCache);
+        CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+      }
+    
+      memcpy(pTableMeta, tbMeta, metaSize);
+      
+      CTG_UNLOCK(CTG_READ, &pCache->metaLock);
+      taosHashRelease(dbCache->tbCache, pCache); 
+      
+      ctgDebug("Got tb %s meta from cache, type:%d, dbFName:%s", pName->tname, tbMeta->tableType, dbFName);
+      
+      res.pRes = pTableMeta;
+      taosArrayPush(ctx->pResList, &res);
+
+      continue;
+    }
+    
+    // PROCESS FOR CHILD TABLE
+
+    if (lastSuid && tbMeta->suid == lastSuid && lastTableMeta) {
+      cloneTableMeta(lastTableMeta, &pTableMeta);
+      memcpy(pTableMeta, tbMeta, sizeof(SCTableMeta));
+
+      CTG_UNLOCK(CTG_READ, &pCache->metaLock);
+      taosHashRelease(dbCache->tbCache, pCache); 
+      
+      ctgDebug("Got tb %s meta from cache, type:%d, dbFName:%s", pName->tname, tbMeta->tableType, dbFName);
+      
+      res.pRes = pTableMeta;
+      taosArrayPush(ctx->pResList, &res);
+      
+      continue;
+    }
+    
+    int32_t metaSize = sizeof(SCTableMeta);
+    pTableMeta = taosMemoryCalloc(1, metaSize);
+    if (NULL == pTableMeta) {
+      ctgReleaseTbMetaToCache(pCtg, dbCache, pCache);    
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+    
+    memcpy(pTableMeta, tbMeta, metaSize);
+    
+    CTG_UNLOCK(CTG_READ, &pCache->metaLock);
+    taosHashRelease(dbCache->tbCache, pCache); 
+    
+    ctgDebug("Got ctb %s meta from cache, will continue to get its stb meta, type:%d, dbFName:%s", 
+             pName->tname, nctx.tbInfo.tbType, dbFName);
+    
+    char* stName = taosHashAcquire(dbCache->stbCache, &pTableMeta->suid, sizeof(pTableMeta->suid));
+    if (NULL == stName) {
+      ctgDebug("stb 0x%" PRIx64 " not in cache, dbFName:%s", pTableMeta->suid, dbFName);
+      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
+      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
+
+      taosMemoryFreeClear(pTableMeta);
+      continue;
+    }
+
+    pCache = taosHashAcquire(dbCache->tbCache, stName, strlen(stName));
+    if (NULL == pCache) {
+      ctgDebug("stb 0x%" PRIx64 " name %s not in cache, dbFName:%s", pTableMeta->suid, stName, dbFName);
+      taosHashRelease(dbCache->stbCache, stName);
+      
+      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
+      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
+
+      taosMemoryFreeClear(pTableMeta);      
+      continue;
+    }
+
+    taosHashRelease(dbCache->stbCache, stName);
+
+    CTG_LOCK(CTG_READ, &pCache->metaLock);
+    if (NULL == pCache->pMeta) {
+      ctgDebug("stb 0x%" PRIx64 " meta not in cache, dbFName:%s", pTableMeta->suid, dbFName);
+      CTG_UNLOCK(CTG_READ, &pCache->metaLock);
+      taosHashRelease(dbCache->tbCache, pCache); 
+      
+      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
+      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
+
+      taosMemoryFreeClear(pTableMeta);
+
+      continue;
+    }
+    
+    STableMeta* stbMeta = pCache->pMeta;
+    if (stbMeta->suid != nctx.tbInfo.suid) {    
+      CTG_UNLOCK(CTG_READ, &pCache->metaLock);
+      taosHashRelease(dbCache->tbCache, pCache); 
+      
+      ctgError("stb suid 0x%" PRIx64 " in stbCache mis-match, expected suid 0x%"PRIx64 , stbMeta->suid, nctx.tbInfo.suid);
+      
+      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
+      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
+
+      taosMemoryFreeClear(pTableMeta);
+
+      continue;
+    }
+    
+    metaSize = CTG_META_SIZE(stbMeta);
+    pTableMeta = taosMemoryRealloc(pTableMeta, metaSize);
+    if (NULL == pTableMeta) {    
+      ctgReleaseTbMetaToCache(pCtg, dbCache, pCache);
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+    
+    memcpy(&pTableMeta->sversion, &stbMeta->sversion, metaSize - sizeof(SCTableMeta));
+    
+    CTG_UNLOCK(CTG_READ, &pCache->metaLock);
+    taosHashRelease(dbCache->tbCache, pCache); 
+    
+    res.pRes = pTableMeta;
+    taosArrayPush(ctx->pResList, &res);
+
+    lastSuid = pTableMeta->suid;
+    lastTableMeta = pTableMeta;
+  }
+
+  ctgReleaseDBCache(pCtg, dbCache);
+  
+  return TSDB_CODE_SUCCESS;
+}
+
 
 int32_t ctgRemoveTbMetaFromCache(SCatalog* pCtg, SName* pTableName, bool syncReq) {
   int32_t       code = 0;

source/libs/catalog/src/ctgUtil.c

@@ -26,6 +26,7 @@ void ctgFreeMsgSendParam(void* param) {
 
   SCtgTaskCallbackParam* pParam = (SCtgTaskCallbackParam*)param;
   taosArrayDestroy(pParam->taskId);
+  taosArrayDestroy(pParam->msgIdx);
 
   taosMemoryFree(param);
 }
@@ -88,6 +89,10 @@ char *ctgTaskTypeStr(CTG_TASK_TYPE type) {
       return "[get user]";
     case CTG_TASK_GET_SVR_VER:
       return "[get svr ver]";
+    case CTG_TASK_GET_TB_META_BATCH:
+      return "[bget table meta]";
+    case CTG_TASK_GET_TB_HASH_BATCH:
+      return "[bget table hash]";
     default:
       return "unknown";
   }
@@ -460,6 +465,25 @@ void ctgResetTbMetaTask(SCtgTask* pTask) {
   taosMemoryFreeClear(pTask->res);
 }
 
+void ctgFreeBatchMeta(void* meta) {
+  if (NULL == meta) {
+    return;
+  }
+  
+  SMetaRes* pRes = (SMetaRes*)meta;
+  taosMemoryFreeClear(pRes->pRes);
+}
+
+void ctgFreeBatchHash(void* hash) {
+  if (NULL == hash) {
+    return;
+  }
+  
+  SMetaRes* pRes = (SMetaRes*)hash;
+  taosMemoryFreeClear(pRes->pRes);
+}
+
+
 void ctgFreeTaskRes(CTG_TASK_TYPE type, void **pRes) {
   switch (type) {
     case CTG_TASK_GET_QNODE:
@@ -500,6 +524,24 @@ void ctgFreeTaskRes(CTG_TASK_TYPE type, void **pRes) {
       taosMemoryFreeClear(*pRes);
       break;
     }
+    case CTG_TASK_GET_TB_META_BATCH: {
+      SArray* pArray = (SArray*)*pRes;
+      int32_t num = taosArrayGetSize(pArray);
+      for (int32_t i = 0; i < num; ++i) {
+        ctgFreeBatchMeta(taosArrayGet(pArray, i));
+      }
+      *pRes = NULL; // no need to free it
+      break;
+    }
+    case CTG_TASK_GET_TB_HASH_BATCH: {
+      SArray* pArray = (SArray*)*pRes;
+      int32_t num = taosArrayGetSize(pArray);
+      for (int32_t i = 0; i < num; ++i) {
+        ctgFreeBatchHash(taosArrayGet(pArray, i));
+      }
+      *pRes = NULL; // no need to free it
+      break;
+    }    
     default:
       qError("invalid task type %d", type);
       break;
@@ -554,6 +596,16 @@ void ctgFreeSubTaskRes(CTG_TASK_TYPE type, void **pRes) {
       taosMemoryFreeClear(*pRes);
       break;
     }
+    case CTG_TASK_GET_TB_META_BATCH: {
+      taosArrayDestroyEx(*pRes, ctgFreeBatchMeta);
+      *pRes = NULL;
+      break;
+    }
+    case CTG_TASK_GET_TB_HASH_BATCH: {
+      taosArrayDestroyEx(*pRes, ctgFreeBatchHash);
+      *pRes = NULL;
+      break;
+    }
     default:
       qError("invalid task type %d", type);
       break;
@@ -583,12 +635,38 @@ void ctgFreeTaskCtx(SCtgTask* pTask) {
       taosMemoryFreeClear(pTask->taskCtx);
       break;
     }
+    case CTG_TASK_GET_TB_META_BATCH: {
+      SCtgTbMetasCtx* taskCtx = (SCtgTbMetasCtx*)pTask->taskCtx;
+      taosArrayDestroyEx(taskCtx->pResList, ctgFreeBatchMeta);
+      taosArrayDestroy(taskCtx->pFetchs);
+      // NO NEED TO FREE pNames
+
+      taosArrayDestroyEx(pTask->msgCtxs, (FDelete)ctgFreeMsgCtx);
+      
+      if (pTask->msgCtx.lastOut) {
+        ctgFreeSTableMetaOutput((STableMetaOutput*)pTask->msgCtx.lastOut);
+        pTask->msgCtx.lastOut = NULL;
+      }
+      taosMemoryFreeClear(pTask->taskCtx);
+      break;
+    }
     case CTG_TASK_GET_TB_HASH: {
       SCtgTbHashCtx* taskCtx = (SCtgTbHashCtx*)pTask->taskCtx;
       taosMemoryFreeClear(taskCtx->pName);
       taosMemoryFreeClear(pTask->taskCtx);      
       break;
     }
+    case CTG_TASK_GET_TB_HASH_BATCH: {
+      SCtgTbHashsCtx* taskCtx = (SCtgTbHashsCtx*)pTask->taskCtx;
+      taosArrayDestroyEx(taskCtx->pResList, ctgFreeBatchHash);
+      taosArrayDestroy(taskCtx->pFetchs);
+      // NO NEED TO FREE pNames
+
+      taosArrayDestroyEx(pTask->msgCtxs, (FDelete)ctgFreeMsgCtx);
+      
+      taosMemoryFreeClear(pTask->taskCtx);
+      break;
+    }    
     case CTG_TASK_GET_TB_INDEX: {
       SCtgTbIndexCtx* taskCtx = (SCtgTbIndexCtx*)pTask->taskCtx;
       taosMemoryFreeClear(taskCtx->pName);
@@ -679,6 +757,23 @@ int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* targ
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgAddMsgCtx(SArray* pCtxs, int32_t reqType, void* out, char* target) {
+  SCtgMsgCtx ctx = {0};
+  
+  ctx.reqType = reqType;
+  ctx.out = out;
+  if (target) {
+    ctx.target = strdup(target);
+    if (NULL == ctx.target) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+  }
+
+  taosArrayPush(pCtxs, &ctx);
+
+  return TSDB_CODE_SUCCESS;
+}
+
 
 int32_t ctgGetHashFunction(int8_t hashMethod, tableNameHashFp *fp) {
   switch (hashMethod) {
@@ -780,6 +875,104 @@ int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName
   CTG_RET(code);
 }
 
+int32_t ctgGetVgInfosFromHashValue(SCatalog *pCtg, SCtgTaskReq* tReq, SDBVgInfo *dbInfo, SCtgTbHashsCtx *pCtx, char* dbFName, SArray* pNames, bool update) {
+  int32_t code = 0;
+  SCtgTask* pTask = tReq->pTask;
+  SMetaRes res = {0};
+  int32_t vgNum = taosHashGetSize(dbInfo->vgHash);
+  if (vgNum <= 0) {
+    ctgError("db vgroup cache invalid, db:%s, vgroup number:%d", dbFName, vgNum);
+    CTG_ERR_RET(TSDB_CODE_TSC_DB_NOT_SELECTED);
+  }
+
+  tableNameHashFp fp = NULL;
+  SVgroupInfo *vgInfo = NULL;
+
+  CTG_ERR_RET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
+
+  int32_t tbNum = taosArrayGetSize(pNames);
+
+  if (1 == vgNum) {
+    void *pIter = taosHashIterate(dbInfo->vgHash, NULL);
+    for (int32_t i = 0; i < tbNum; ++i) {
+      vgInfo = taosMemoryMalloc(sizeof(SVgroupInfo));
+      if (NULL == vgInfo) {
+        CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+      }
+
+      *vgInfo = *(SVgroupInfo*)pIter;
+
+      ctgDebug("Got tb hash vgroup, vgId:%d, epNum %d, current %s port %d", vgInfo->vgId, vgInfo->epSet.numOfEps,
+        vgInfo->epSet.eps[vgInfo->epSet.inUse].fqdn, vgInfo->epSet.eps[vgInfo->epSet.inUse].port);
+
+      if (update) {
+        SCtgFetch* pFetch = taosArrayGet(pCtx->pFetchs, tReq->msgIdx);
+        SMetaRes *pRes = taosArrayGet(pCtx->pResList, pFetch->resIdx + i);
+        pRes->pRes = vgInfo;
+      } else {
+        res.pRes = vgInfo;
+        taosArrayPush(pCtx->pResList, &res);
+      }
+    }
+
+    return TSDB_CODE_SUCCESS;
+  }
+
+  char tbFullName[TSDB_TABLE_FNAME_LEN];
+  sprintf(tbFullName, "%s.", dbFName);
+  int32_t offset = strlen(tbFullName);
+  SName* pName = NULL;
+  int32_t tbNameLen = 0;
+  
+  for (int32_t i = 0; i < tbNum; ++i) {
+    pName = taosArrayGet(pNames, i);
+
+    tbNameLen = offset + strlen(pName->tname);
+    strcpy(tbFullName + offset, pName->tname);
+
+    uint32_t hashValue = (*fp)(tbFullName, (uint32_t)tbNameLen);
+
+    void *pIter = taosHashIterate(dbInfo->vgHash, NULL);
+    while (pIter) {
+      vgInfo = pIter;
+      if (hashValue >= vgInfo->hashBegin && hashValue <= vgInfo->hashEnd) {
+        taosHashCancelIterate(dbInfo->vgHash, pIter);
+        break;
+      }
+      
+      pIter = taosHashIterate(dbInfo->vgHash, pIter);
+      vgInfo = NULL;
+    }
+
+    if (NULL == vgInfo) {
+      ctgError("no hash range found for hash value [%u], db:%s, numOfVgId:%d", hashValue, dbFName, taosHashGetSize(dbInfo->vgHash));
+      CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
+    }
+
+    SVgroupInfo* pNewVg = taosMemoryMalloc(sizeof(SVgroupInfo));
+    if (NULL == pNewVg) {
+      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+
+    *pNewVg = *vgInfo;
+
+    ctgDebug("Got tb %s hash vgroup, vgId:%d, epNum %d, current %s port %d", tbFullName, vgInfo->vgId, vgInfo->epSet.numOfEps,
+             vgInfo->epSet.eps[vgInfo->epSet.inUse].fqdn, vgInfo->epSet.eps[vgInfo->epSet.inUse].port);
+
+    if (update) {
+      SCtgFetch* pFetch = taosArrayGet(pCtx->pFetchs, tReq->msgIdx);
+      SMetaRes *pRes = taosArrayGet(pCtx->pResList, pFetch->resIdx + i);
+      pRes->pRes = pNewVg;
+    } else {
+      res.pRes = pNewVg;
+      taosArrayPush(pCtx->pResList, &res);
+    }    
+  }
+
+  CTG_RET(code);
+}
+
+
 int32_t ctgStbVersionSearchCompare(const void* key1, const void* key2) {
   if (*(uint64_t *)key1 < ((SSTableVersion*)key2)->suid) {
     return -1;
@@ -921,4 +1114,41 @@ int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, cha
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgGetTablesReqNum(SArray *pList) {
+  if (NULL == pList) {
+    return 0;
+  }
+
+  int32_t total = 0;
+  int32_t n = taosArrayGetSize(pList);
+  for (int32_t i = 0; i < n; ++i) {
+    STablesReq *pReq = taosArrayGet(pList, i);
+    total += taosArrayGetSize(pReq->pTables);
+  }
+
+  return total;
+}
+
+int32_t ctgAddFetch(SArray** pFetchs, int32_t dbIdx, int32_t tbIdx, int32_t *fetchIdx, int32_t resIdx, int32_t flag) {
+  if (NULL == (*pFetchs)) {
+    *pFetchs = taosArrayInit(CTG_DEFAULT_FETCH_NUM, sizeof(SCtgFetch));
+  }
+  
+  SCtgFetch fetch = {0};
+  fetch.dbIdx = dbIdx;
+  fetch.tbIdx = tbIdx;
+  fetch.fetchIdx = (*fetchIdx)++;
+  fetch.resIdx = resIdx;
+  fetch.flag = flag;
+  
+  taosArrayPush(*pFetchs, &fetch);
+
+  return TSDB_CODE_SUCCESS;
+}
+
+SName* ctgGetFetchName(SArray* pNames, SCtgFetch* pFetch) {
+  STablesReq* pReq = (STablesReq*)taosArrayGet(pNames, pFetch->dbIdx);
+  return (SName*)taosArrayGet(pReq->pTables, pFetch->tbIdx);
+}
+
 

source/libs/executor/inc/executorimpl.h

@@ -737,6 +737,9 @@ typedef struct STimeSliceOperatorInfo {
   SInterval      interval;
   int64_t        current;
   SArray*        pPrevRow;      // SArray<SGroupValue>
+  SArray*        pNextRow;      // SArray<SGroupValue>
+  bool           isPrevRowSet;
+  bool           isNextRowSet;
   int32_t        fillType;      // fill type
   SColumn        tsCol;         // primary timestamp column
   SExprSupp      scalarSup;     // scalar calculation

source/libs/executor/src/executil.c

@@ -191,6 +191,7 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
   pBlock->info.blockId = pNode->dataBlockId;
   pBlock->info.type = STREAM_INVALID;
   pBlock->info.calWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
+  pBlock->info.watermark = INT64_MIN;
 
   for (int32_t i = 0; i < numOfCols; ++i) {
     SSlotDescNode*  pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);

source/libs/executor/src/executorimpl.c

@@ -4180,7 +4180,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
     int32_t children = 0;
     pOptr = createStreamFinalSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
   } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION == type) {
-    int32_t children = 1;
+    int32_t children = pHandle->numOfVgroups;
     pOptr = createStreamFinalSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
   } else if (QUERY_NODE_PHYSICAL_PLAN_PARTITION == type) {
     pOptr = createPartitionOperatorInfo(ops[0], (SPartitionPhysiNode*)pPhyNode, pTaskInfo);

source/libs/executor/src/timewindowoperator.c

@@ -1752,30 +1752,11 @@ void increaseTs(SqlFunctionCtx* pCtx) {
   }
 }
 
-SSDataBlock* createDeleteBlock() {
-  SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
-  pBlock->info.hasVarCol = false;
-  pBlock->info.groupId = 0;
-  pBlock->info.rows = 0;
-  pBlock->info.type = STREAM_DELETE_RESULT;
-  pBlock->info.rowSize = sizeof(TSKEY) + sizeof(uint64_t);
-
-  pBlock->pDataBlock = taosArrayInit(2, sizeof(SColumnInfoData));
-  SColumnInfoData infoData = {0};
-  infoData.info.type = TSDB_DATA_TYPE_TIMESTAMP;
-  infoData.info.bytes = sizeof(TSKEY);
-  // window start ts
-  taosArrayPush(pBlock->pDataBlock, &infoData);
-
-  infoData.info.type = TSDB_DATA_TYPE_UBIGINT;
-  infoData.info.bytes = sizeof(uint64_t);
-  taosArrayPush(pBlock->pDataBlock, &infoData);
-
-  return pBlock;
-}
-
 void initIntervalDownStream(SOperatorInfo* downstream, uint8_t type, SAggSupporter* pSup) {
-  ASSERT(downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN);
+  if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
+    // Todo(liuyao) support partition by column
+    return;
+  }
   SStreamScanInfo* pScanInfo = downstream->info;
   pScanInfo->sessionSup.parentType = type;
   pScanInfo->sessionSup.pIntervalAggSup = pSup;
@@ -2084,10 +2065,30 @@ static void doKeepPrevRows(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlock
       memcpy(pkey->pData, val, pkey->bytes);
     }
   }
+
+  pSliceInfo->isPrevRowSet = true;
+}
+
+static void doKeepNextRows(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlock* pBlock, int32_t rowIndex) {
+  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
+
+    // null data should not be kept since it can not be used to perform interpolation
+    if (!colDataIsNull_s(pColInfoData, i)) {
+      SGroupKeys* pkey = taosArrayGet(pSliceInfo->pNextRow, i);
+
+      pkey->isNull = false;
+      char* val = colDataGetData(pColInfoData, rowIndex);
+      memcpy(pkey->pData, val, pkey->bytes);
+    }
+  }
+
+  pSliceInfo->isNextRowSet = true;
 }
 
 static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp* pExprSup, SSDataBlock* pBlock,
-                                   int32_t rowIndex, SSDataBlock* pResBlock) {
+                                   SSDataBlock* pResBlock) {
   int32_t rows = pResBlock->info.rows;
 
   // todo set the correct primary timestamp column
@@ -2163,6 +2164,10 @@ static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
         break;
       }
       case TSDB_FILL_PREV: {
+        if (!pSliceInfo->isPrevRowSet) {
+          break;
+        }
+
         SGroupKeys* pkey = taosArrayGet(pSliceInfo->pPrevRow, srcSlot);
         colDataAppend(pDst, rows, pkey->pData, false);
         pResBlock->info.rows += 1;
@@ -2170,8 +2175,12 @@ static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
       }
 
       case TSDB_FILL_NEXT: {
-        char* p = colDataGetData(pSrc, rowIndex);
-        colDataAppend(pDst, rows, p, colDataIsNull_s(pSrc, rowIndex));
+        if (!pSliceInfo->isNextRowSet) {
+          break;
+        }
+
+        SGroupKeys* pkey = taosArrayGet(pSliceInfo->pNextRow, srcSlot);
+        colDataAppend(pDst, rows, pkey->pData, false);
         pResBlock->info.rows += 1;
         break;
       }
@@ -2205,6 +2214,35 @@ static int32_t initPrevRowsKeeper(STimeSliceOperatorInfo* pInfo, SSDataBlock* pB
     taosArrayPush(pInfo->pPrevRow, &key);
   }
 
+  pInfo->isPrevRowSet = false;
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t initNextRowsKeeper(STimeSliceOperatorInfo* pInfo, SSDataBlock* pBlock) {
+  if (pInfo->pNextRow != NULL) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  pInfo->pNextRow = taosArrayInit(4, sizeof(SGroupKeys));
+  if (pInfo->pNextRow == NULL) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);
+
+    SGroupKeys key = {0};
+    key.bytes = pColInfo->info.bytes;
+    key.type = pColInfo->info.type;
+    key.isNull = false;
+    key.pData = taosMemoryCalloc(1, pColInfo->info.bytes);
+    taosArrayPush(pInfo->pNextRow, &key);
+  }
+
+  pInfo->isNextRowSet = false;
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -2234,14 +2272,19 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
 
   blockDataCleanup(pResBlock);
 
-  int32_t numOfRows = 0;
   while (1) {
     SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
     if (pBlock == NULL) {
       break;
     }
 
-    int32_t code = initPrevRowsKeeper(pSliceInfo, pBlock);
+    int32_t code;
+    code = initPrevRowsKeeper(pSliceInfo, pBlock);
+    if (code != TSDB_CODE_SUCCESS) {
+      longjmp(pTaskInfo->env, code);
+    }
+
+    code = initNextRowsKeeper(pSliceInfo, pBlock);
     if (code != TSDB_CODE_SUCCESS) {
       longjmp(pTaskInfo->env, code);
     }
@@ -2263,7 +2306,7 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
           SColumnInfoData* pDst = taosArrayGet(pResBlock->pDataBlock, dstSlot);
 
           char* v = colDataGetData(pSrc, i);
-          colDataAppend(pDst, numOfRows, v, false);
+          colDataAppend(pDst, pResBlock->info.rows, v, false);
         }
 
         pResBlock->info.rows += 1;
@@ -2280,11 +2323,16 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
           break;
         }
       } else if (ts < pSliceInfo->current) {
+        // in case interpolation window starts and ends between two datapoints, fill(prev) need to interpolate
+        doKeepPrevRows(pSliceInfo, pBlock, i);
+
         if (i < pBlock->info.rows - 1) {
+          // in case interpolation window starts and ends between two datapoints, fill(next) need to interpolate
+          doKeepNextRows(pSliceInfo, pBlock, i + 1);
           int64_t nextTs = *(int64_t*)colDataGetData(pTsCol, i + 1);
           if (nextTs > pSliceInfo->current) {
             while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
-              genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, i, pResBlock);
+              genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, pResBlock);
               pSliceInfo->current =
                   taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
               if (pResBlock->info.rows >= pResBlock->info.capacity) {
@@ -2303,8 +2351,11 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
           doKeepPrevRows(pSliceInfo, pBlock, i);
         }
       } else {  // ts > pSliceInfo->current
+        // in case interpolation window starts and ends between two datapoints, fill(next) need to interpolate
+        doKeepNextRows(pSliceInfo, pBlock, i);
+
         while (pSliceInfo->current < ts && pSliceInfo->current <= pSliceInfo->win.ekey) {
-          genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, i, pResBlock);
+          genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, pResBlock);
           pSliceInfo->current =
               taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
           if (pResBlock->info.rows >= pResBlock->info.capacity) {
@@ -2312,12 +2363,48 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
           }
         }
 
+        // add current row if timestamp match
+        if (ts == pSliceInfo->current && pSliceInfo->current <= pSliceInfo->win.ekey) {
+          for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {
+            SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[j];
+            int32_t    dstSlot = pExprInfo->base.resSchema.slotId;
+            int32_t    srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
+
+            SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, srcSlot);
+            SColumnInfoData* pDst = taosArrayGet(pResBlock->pDataBlock, dstSlot);
+
+            char* v = colDataGetData(pSrc, i);
+            colDataAppend(pDst, pResBlock->info.rows, v, false);
+          }
+
+          pResBlock->info.rows += 1;
+          doKeepPrevRows(pSliceInfo, pBlock, i);
+
+          pSliceInfo->current =
+              taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+
+          if (pResBlock->info.rows >= pResBlock->info.capacity) {
+            break;
+          }
+        }
+
         if (pSliceInfo->current > pSliceInfo->win.ekey) {
           doSetOperatorCompleted(pOperator);
           break;
         }
       }
     }
+
+    // check if need to interpolate after ts range
+    // except for fill(next)
+    while (pSliceInfo->current <= pSliceInfo->win.ekey && pSliceInfo->fillType != TSDB_FILL_NEXT) {
+      genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, pResBlock);
+      pSliceInfo->current =
+          taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+      if (pResBlock->info.rows >= pResBlock->info.capacity) {
+        break;
+      }
+    }
   }
 
   // restore the value
@@ -2836,13 +2923,6 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
       // process the rest of the data
       return pInfo->pUpdateRes;
     }
-    // doBuildPullDataBlock(pInfo->pPullWins, &pInfo->pullIndex, pInfo->pPullDataRes);
-    // if (pInfo->pPullDataRes->info.rows != 0) {
-    //   // process the rest of the data
-    //   ASSERT(IS_FINAL_OP(pInfo));
-    //   printDataBlock(pInfo->pPullDataRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
-    //   return pInfo->pPullDataRes;
-    // }
     doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
     if (pInfo->pDelRes->info.rows != 0) {
       // process the rest of the data
@@ -2862,6 +2942,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
     }
     printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "interval final recv" : "interval semi recv");
     maxTs = TMAX(maxTs, pBlock->info.window.ekey);
+    maxTs = TMAX(maxTs, pBlock->info.watermark);
 
     if (pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_PULL_DATA ||
         pBlock->info.type == STREAM_INVALID) {
@@ -2950,6 +3031,8 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
     closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, pInfo->pPullDataMap,
                         pUpdated, pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
     closeChildIntervalWindow(pInfo->pChildren, pInfo->twAggSup.maxTs);
+  } else {
+    pInfo->binfo.pRes->info.watermark = pInfo->twAggSup.maxTs;
   }
 
   finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset);
@@ -2984,7 +3067,6 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
     printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
     return pInfo->pDelRes;
   }
-  // ASSERT(false);
   return NULL;
 }
 
@@ -2996,6 +3078,7 @@ SSDataBlock* createSpecialDataBlock(EStreamType type) {
   pBlock->info.type = type;
   pBlock->info.rowSize =
       sizeof(TSKEY) + sizeof(TSKEY) + sizeof(uint64_t) + sizeof(uint64_t) + sizeof(TSKEY) + sizeof(TSKEY);
+  pBlock->info.watermark = INT64_MIN;
 
   pBlock->pDataBlock = taosArrayInit(6, sizeof(SColumnInfoData));
   SColumnInfoData infoData = {0};
@@ -3185,7 +3268,6 @@ void destroyStreamSessionAggOperatorInfo(void* param, int32_t numOfOutput) {
       SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
       destroyStreamSessionAggOperatorInfo(pChInfo, numOfOutput);
       taosMemoryFreeClear(pChild);
-      taosMemoryFreeClear(pChInfo);
     }
   }
   colDataDestroy(&pInfo->twAggSup.timeWindowData);
@@ -3950,6 +4032,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
       doStreamSessionAggImpl(pChildOp, pBlock, NULL, NULL, true);
     }
     maxTs = TMAX(maxTs, pBlock->info.window.ekey);
+    maxTs = TMAX(maxTs, pBlock->info.watermark);
   }
 
   pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
@@ -4073,6 +4156,7 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
   }
 
   pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
+  pBInfo->pRes->info.watermark = pInfo->twAggSup.maxTs;
   // restore the value
   pOperator->status = OP_RES_TO_RETURN;
   // semi operator

source/libs/parser/inc/parUtil.h

@@ -39,6 +39,11 @@ typedef struct SMsgBuf {
   char*   buf;
 } SMsgBuf;
 
+typedef struct SParseTablesMetaReq {
+  char      dbFName[TSDB_DB_FNAME_LEN];
+  SHashObj* pTables;
+} SParseTablesMetaReq;
+
 typedef struct SParseMetaCache {
   SHashObj* pTableMeta;    // key is tbFName, element is STableMeta*
   SHashObj* pDbVgroup;     // key is dbFName, element is SArray<SVgroupInfo>*
@@ -95,7 +100,7 @@ int32_t getUdfInfoFromCache(SParseMetaCache* pMetaCache, const char* pFunc, SFun
 int32_t getTableIndexFromCache(SParseMetaCache* pMetaCache, const SName* pName, SArray** pIndexes);
 int32_t getTableCfgFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableCfg** pOutput);
 int32_t getDnodeListFromCache(SParseMetaCache* pMetaCache, SArray** pDnodes);
-void    destoryParseMetaCache(SParseMetaCache* pMetaCache);
+void    destoryParseMetaCache(SParseMetaCache* pMetaCache, bool request);
 
 #ifdef __cplusplus
 }

source/libs/parser/src/parTranslater.c

@@ -1253,20 +1253,21 @@ static int32_t translateRepeatScanFunc(STranslateContext* pCxt, SFunctionNode* p
   if (!fmIsRepeatScanFunc(pFunc->funcId)) {
     return TSDB_CODE_SUCCESS;
   }
-  if (isSelectStmt(pCxt->pCurrStmt)) {
-    // select percentile() without from clause is also valid
-    if (NULL == ((SSelectStmt*)pCxt->pCurrStmt)->pFromTable) {
-      return TSDB_CODE_SUCCESS;
-    }
-    SNode* pTable = ((SSelectStmt*)pCxt->pCurrStmt)->pFromTable;
-    if (QUERY_NODE_REAL_TABLE == nodeType(pTable) &&
-        (TSDB_CHILD_TABLE == ((SRealTableNode*)pTable)->pMeta->tableType ||
-         TSDB_NORMAL_TABLE == ((SRealTableNode*)pTable)->pMeta->tableType)) {
-      return TSDB_CODE_SUCCESS;
-    }
+  if (!isSelectStmt(pCxt->pCurrStmt)) {
+    return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_ONLY_SUPPORT_SINGLE_TABLE,
+                                   "%s is only supported in single table query", pFunc->functionName);
   }
-  return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_ONLY_SUPPORT_SINGLE_TABLE,
-                                 "%s is only supported in single table query", pFunc->functionName);
+  SSelectStmt* pSelect = (SSelectStmt*)pCxt->pCurrStmt;
+  SNode*       pTable = pSelect->pFromTable;
+  // select percentile() without from clause is also valid
+  if ((NULL != pTable && (QUERY_NODE_REAL_TABLE != nodeType(pTable) ||
+                          (TSDB_CHILD_TABLE != ((SRealTableNode*)pTable)->pMeta->tableType &&
+                           TSDB_NORMAL_TABLE != ((SRealTableNode*)pTable)->pMeta->tableType))) ||
+      NULL != pSelect->pPartitionByList) {
+    return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_ONLY_SUPPORT_SINGLE_TABLE,
+                                   "%s is only supported in single table query", pFunc->functionName);
+  }
+  return TSDB_CODE_SUCCESS;
 }
 
 static bool isStar(SNode* pNode) {
@@ -2509,9 +2510,31 @@ static EDealRes checkStateExpr(SNode* pNode, void* pContext) {
   return DEAL_RES_CONTINUE;
 }
 
-static int32_t translateStateWindow(STranslateContext* pCxt, SStateWindowNode* pState) {
+static bool isPartitionByTbname(SNodeList* pPartitionByList) {
+  if (1 != LIST_LENGTH(pPartitionByList)) {
+    return false;
+  }
+  SNode* pPartKey = nodesListGetNode(pPartitionByList, 0);
+  return QUERY_NODE_FUNCTION != nodeType(pPartKey) || FUNCTION_TYPE_TBNAME != ((SFunctionNode*)pPartKey)->funcType;
+}
+
+static int32_t checkStateWindowForStream(STranslateContext* pCxt, SSelectStmt* pSelect) {
+  if (!pCxt->createStream) {
+    return TSDB_CODE_SUCCESS;
+  }
+  if (TSDB_SUPER_TABLE == ((SRealTableNode*)pSelect->pFromTable)->pMeta->tableType &&
+      !isPartitionByTbname(pSelect->pPartitionByList)) {
+    return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "Unsupported stream query");
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateStateWindow(STranslateContext* pCxt, SSelectStmt* pSelect) {
+  SStateWindowNode* pState = (SStateWindowNode*)pSelect->pWindow;
   nodesWalkExprPostOrder(pState->pExpr, checkStateExpr, pCxt);
-  // todo check for "function not support for state_window"
+  if (TSDB_CODE_SUCCESS == pCxt->errCode) {
+    pCxt->errCode = checkStateWindowForStream(pCxt, pSelect);
+  }
   return pCxt->errCode;
 }
 
@@ -2522,14 +2545,13 @@ static int32_t translateSessionWindow(STranslateContext* pCxt, SSessionWindowNod
   if (PRIMARYKEY_TIMESTAMP_COL_ID != pSession->pCol->colId) {
     return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INTER_SESSION_COL);
   }
-  // todo check for "function not support for session"
   return TSDB_CODE_SUCCESS;
 }
 
 static int32_t translateSpecificWindow(STranslateContext* pCxt, SSelectStmt* pSelect) {
   switch (nodeType(pSelect->pWindow)) {
     case QUERY_NODE_STATE_WINDOW:
-      return translateStateWindow(pCxt, (SStateWindowNode*)pSelect->pWindow);
+      return translateStateWindow(pCxt, pSelect);
     case QUERY_NODE_SESSION_WINDOW:
       return translateSessionWindow(pCxt, (SSessionWindowNode*)pSelect->pWindow);
     case QUERY_NODE_INTERVAL_WINDOW:
@@ -4708,7 +4730,7 @@ static int32_t checkCreateStream(STranslateContext* pCxt, SCreateStreamStmt* pSt
     }
   }
 
-  return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY);
+  return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "Unsupported stream query");
 }
 
 static void getSourceDatabase(SNode* pStmt, int32_t acctId, char* pDbFName) {

source/libs/parser/src/parUtil.c

@@ -474,6 +474,24 @@ static int32_t buildDbReq(SHashObj* pDbsHash, SArray** pDbs) {
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t buildTableReqFromDb(SHashObj* pDbsHash, SArray** pDbs) {
+  if (NULL != pDbsHash) {
+    *pDbs = taosArrayInit(taosHashGetSize(pDbsHash), sizeof(STablesReq));
+    if (NULL == *pDbs) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+    SParseTablesMetaReq* p = taosHashIterate(pDbsHash, NULL);
+    while (NULL != p) {
+      STablesReq req = {0};
+      strcpy(req.dbFName, p->dbFName);
+      buildTableReq(p->pTables, &req.pTables);
+      taosArrayPush(*pDbs, &req);
+      p = taosHashIterate(pDbsHash, p);
+    }
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t buildUserAuthReq(SHashObj* pUserAuthHash, SArray** pUserAuth) {
   if (NULL != pUserAuthHash) {
     *pUserAuth = taosArrayInit(taosHashGetSize(pUserAuthHash), sizeof(SUserAuthInfo));
@@ -513,12 +531,12 @@ static int32_t buildUdfReq(SHashObj* pUdfHash, SArray** pUdf) {
 }
 
 int32_t buildCatalogReq(const SParseMetaCache* pMetaCache, SCatalogReq* pCatalogReq) {
-  int32_t code = buildTableReq(pMetaCache->pTableMeta, &pCatalogReq->pTableMeta);
+  int32_t code = buildTableReqFromDb(pMetaCache->pTableMeta, &pCatalogReq->pTableMeta);
   if (TSDB_CODE_SUCCESS == code) {
     code = buildDbReq(pMetaCache->pDbVgroup, &pCatalogReq->pDbVgroup);
   }
   if (TSDB_CODE_SUCCESS == code) {
-    code = buildTableReq(pMetaCache->pTableVgroup, &pCatalogReq->pTableHash);
+    code = buildTableReqFromDb(pMetaCache->pTableVgroup, &pCatalogReq->pTableHash);
   }
   if (TSDB_CODE_SUCCESS == code) {
     code = buildDbReq(pMetaCache->pDbCfg, &pCatalogReq->pDbCfg);
@@ -587,6 +605,24 @@ static int32_t putDbDataToCache(const SArray* pDbReq, const SArray* pDbData, SHa
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t putDbTableDataToCache(const SArray* pDbReq, const SArray* pTableData, SHashObj** pTable) {
+  int32_t ndbs = taosArrayGetSize(pDbReq);
+  int32_t tableNo = 0;
+  for (int32_t i = 0; i < ndbs; ++i) {
+    STablesReq* pReq = taosArrayGet(pDbReq, i);
+    int32_t     ntables = taosArrayGetSize(pReq->pTables);
+    for (int32_t j = 0; j < ntables; ++j) {
+      char fullName[TSDB_TABLE_FNAME_LEN];
+      tNameExtractFullName(taosArrayGet(pReq->pTables, j), fullName);
+      if (TSDB_CODE_SUCCESS != putMetaDataToHash(fullName, strlen(fullName), pTableData, tableNo, pTable)) {
+        return TSDB_CODE_OUT_OF_MEMORY;
+      }
+      ++tableNo;
+    }
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t putUserAuthToCache(const SArray* pUserAuthReq, const SArray* pUserAuthData, SHashObj** pUserAuth) {
   int32_t nvgs = taosArrayGetSize(pUserAuthReq);
   for (int32_t i = 0; i < nvgs; ++i) {
@@ -612,12 +648,12 @@ static int32_t putUdfToCache(const SArray* pUdfReq, const SArray* pUdfData, SHas
 }
 
 int32_t putMetaDataToCache(const SCatalogReq* pCatalogReq, const SMetaData* pMetaData, SParseMetaCache* pMetaCache) {
-  int32_t code = putTableDataToCache(pCatalogReq->pTableMeta, pMetaData->pTableMeta, &pMetaCache->pTableMeta);
+  int32_t code = putDbTableDataToCache(pCatalogReq->pTableMeta, pMetaData->pTableMeta, &pMetaCache->pTableMeta);
   if (TSDB_CODE_SUCCESS == code) {
     code = putDbDataToCache(pCatalogReq->pDbVgroup, pMetaData->pDbVgroup, &pMetaCache->pDbVgroup);
   }
   if (TSDB_CODE_SUCCESS == code) {
-    code = putTableDataToCache(pCatalogReq->pTableHash, pMetaData->pTableHash, &pMetaCache->pTableVgroup);
+    code = putDbTableDataToCache(pCatalogReq->pTableHash, pMetaData->pTableHash, &pMetaCache->pTableVgroup);
   }
   if (TSDB_CODE_SUCCESS == code) {
     code = putDbDataToCache(pCatalogReq->pDbCfg, pMetaData->pDbCfg, &pMetaCache->pDbCfg);
@@ -657,14 +693,38 @@ static int32_t reserveTableReqInCache(int32_t acctId, const char* pDb, const cha
   return reserveTableReqInCacheImpl(fullName, len, pTables);
 }
 
+static int32_t reserveTableReqInDbCacheImpl(int32_t acctId, const char* pDb, const char* pTable, SHashObj* pDbs) {
+  SParseTablesMetaReq req = {0};
+  int32_t             len = snprintf(req.dbFName, sizeof(req.dbFName), "%d.%s", acctId, pDb);
+  int32_t             code = reserveTableReqInCache(acctId, pDb, pTable, &req.pTables);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = taosHashPut(pDbs, req.dbFName, len, &req, sizeof(SParseTablesMetaReq));
+  }
+  return code;
+}
+
+static int32_t reserveTableReqInDbCache(int32_t acctId, const char* pDb, const char* pTable, SHashObj** pDbs) {
+  if (NULL == *pDbs) {
+    *pDbs = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
+    if (NULL == *pDbs) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+  }
+  char                 fullName[TSDB_DB_FNAME_LEN];
+  int32_t              len = snprintf(fullName, sizeof(fullName), "%d.%s", acctId, pDb);
+  SParseTablesMetaReq* pReq = taosHashGet(*pDbs, fullName, len);
+  if (NULL == pReq) {
+    return reserveTableReqInDbCacheImpl(acctId, pDb, pTable, *pDbs);
+  }
+  return reserveTableReqInCache(acctId, pDb, pTable, &pReq->pTables);
+}
+
 int32_t reserveTableMetaInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache) {
-  return reserveTableReqInCache(acctId, pDb, pTable, &pMetaCache->pTableMeta);
+  return reserveTableReqInDbCache(acctId, pDb, pTable, &pMetaCache->pTableMeta);
 }
 
 int32_t reserveTableMetaInCacheExt(const SName* pName, SParseMetaCache* pMetaCache) {
-  char fullName[TSDB_TABLE_FNAME_LEN];
-  tNameExtractFullName(pName, fullName);
-  return reserveTableReqInCacheImpl(fullName, strlen(fullName), &pMetaCache->pTableMeta);
+  return reserveTableReqInDbCache(pName->acctId, pName->dbname, pName->tname, &pMetaCache->pTableMeta);
 }
 
 int32_t getTableMetaFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableMeta** pMeta) {
@@ -711,13 +771,11 @@ int32_t getDbVgInfoFromCache(SParseMetaCache* pMetaCache, const char* pDbFName,
 }
 
 int32_t reserveTableVgroupInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache) {
-  return reserveTableReqInCache(acctId, pDb, pTable, &pMetaCache->pTableVgroup);
+  return reserveTableReqInDbCache(acctId, pDb, pTable, &pMetaCache->pTableVgroup);
 }
 
 int32_t reserveTableVgroupInCacheExt(const SName* pName, SParseMetaCache* pMetaCache) {
-  char fullName[TSDB_TABLE_FNAME_LEN];
-  tNameExtractFullName(pName, fullName);
-  return reserveTableReqInCacheImpl(fullName, strlen(fullName), &pMetaCache->pTableVgroup);
+  return reserveTableReqInDbCache(pName->acctId, pName->dbname, pName->tname, &pMetaCache->pTableVgroup);
 }
 
 int32_t getTableVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName, SVgroupInfo* pVgroup) {
@@ -919,10 +977,24 @@ int32_t getDnodeListFromCache(SParseMetaCache* pMetaCache, SArray** pDnodes) {
   return TSDB_CODE_SUCCESS;
 }
 
-void destoryParseMetaCache(SParseMetaCache* pMetaCache) {
-  taosHashCleanup(pMetaCache->pTableMeta);
+void destoryParseTablesMetaReqHash(SHashObj* pHash) {
+  SParseTablesMetaReq* p = taosHashIterate(pHash, NULL);
+  while (NULL != p) {
+    taosHashCleanup(p->pTables);
+    p = taosHashIterate(pHash, p);
+  }
+  taosHashCleanup(pHash);
+}
+
+void destoryParseMetaCache(SParseMetaCache* pMetaCache, bool request) {
+  if (request) {
+    destoryParseTablesMetaReqHash(pMetaCache->pTableMeta);
+    destoryParseTablesMetaReqHash(pMetaCache->pTableVgroup);
+  } else {
+    taosHashCleanup(pMetaCache->pTableMeta);
+    taosHashCleanup(pMetaCache->pTableVgroup);
+  }
   taosHashCleanup(pMetaCache->pDbVgroup);
-  taosHashCleanup(pMetaCache->pTableVgroup);
   taosHashCleanup(pMetaCache->pDbCfg);
   taosHashCleanup(pMetaCache->pDbInfo);
   taosHashCleanup(pMetaCache->pUserAuth);

source/libs/parser/src/parser.c

@@ -85,13 +85,13 @@ static int32_t setValueByBindParam(SValueNode* pVal, TAOS_MULTI_BIND* pParam) {
   if (IS_VAR_DATA_TYPE(pVal->node.resType.type)) {
     taosMemoryFreeClear(pVal->datum.p);
   }
-  
+
   if (pParam->is_null && 1 == *(pParam->is_null)) {
     pVal->node.resType.type = TSDB_DATA_TYPE_NULL;
     pVal->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_NULL].bytes;
     return TSDB_CODE_SUCCESS;
   }
-  
+
   int32_t inputSize = (NULL != pParam->length ? *(pParam->length) : tDataTypes[pParam->buffer_type].bytes);
   pVal->node.resType.type = pParam->buffer_type;
   pVal->node.resType.bytes = inputSize;
@@ -187,7 +187,7 @@ int32_t qParseSqlSyntax(SParseContext* pCxt, SQuery** pQuery, struct SCatalogReq
   if (TSDB_CODE_SUCCESS == code) {
     code = buildCatalogReq(&metaCache, pCatalogReq);
   }
-  destoryParseMetaCache(&metaCache);
+  destoryParseMetaCache(&metaCache, true);
   terrno = code;
   return code;
 }
@@ -203,7 +203,7 @@ int32_t qAnalyseSqlSemantic(SParseContext* pCxt, const struct SCatalogReq* pCata
       code = analyseSemantic(pCxt, pQuery, &metaCache);
     }
   }
-  destoryParseMetaCache(&metaCache);
+  destoryParseMetaCache(&metaCache, false);
   terrno = code;
   return code;
 }

source/libs/parser/test/mockCatalogService.cpp

@@ -472,12 +472,16 @@ class MockCatalogServiceImpl {
 
   int32_t getAllTableMeta(SArray* pTableMetaReq, SArray** pTableMetaData) const {
     if (NULL != pTableMetaReq) {
-      int32_t ntables = taosArrayGetSize(pTableMetaReq);
-      *pTableMetaData = taosArrayInit(ntables, sizeof(SMetaRes));
-      for (int32_t i = 0; i < ntables; ++i) {
-        SMetaRes res = {0};
-        res.code = catalogGetTableMeta((const SName*)taosArrayGet(pTableMetaReq, i), (STableMeta**)&res.pRes);
-        taosArrayPush(*pTableMetaData, &res);
+      int32_t ndbs = taosArrayGetSize(pTableMetaReq);
+      *pTableMetaData = taosArrayInit(ndbs, sizeof(SMetaRes));
+      for (int32_t i = 0; i < ndbs; ++i) {
+        STablesReq* pReq = (STablesReq*)taosArrayGet(pTableMetaReq, i);
+        int32_t     ntables = taosArrayGetSize(pReq->pTables);
+        for (int32_t j = 0; j < ntables; ++j) {
+          SMetaRes res = {0};
+          res.code = catalogGetTableMeta((const SName*)taosArrayGet(pReq->pTables, j), (STableMeta**)&res.pRes);
+          taosArrayPush(*pTableMetaData, &res);
+        }
       }
     }
     return TSDB_CODE_SUCCESS;
@@ -485,13 +489,17 @@ class MockCatalogServiceImpl {
 
   int32_t getAllTableVgroup(SArray* pTableVgroupReq, SArray** pTableVgroupData) const {
     if (NULL != pTableVgroupReq) {
-      int32_t ntables = taosArrayGetSize(pTableVgroupReq);
-      *pTableVgroupData = taosArrayInit(ntables, sizeof(SMetaRes));
-      for (int32_t i = 0; i < ntables; ++i) {
-        SMetaRes res = {0};
-        res.pRes = taosMemoryCalloc(1, sizeof(SVgroupInfo));
-        res.code = catalogGetTableHashVgroup((const SName*)taosArrayGet(pTableVgroupReq, i), (SVgroupInfo*)res.pRes);
-        taosArrayPush(*pTableVgroupData, &res);
+      int32_t ndbs = taosArrayGetSize(pTableVgroupReq);
+      *pTableVgroupData = taosArrayInit(ndbs, sizeof(SMetaRes));
+      for (int32_t i = 0; i < ndbs; ++i) {
+        STablesReq* pReq = (STablesReq*)taosArrayGet(pTableVgroupReq, i);
+        int32_t     ntables = taosArrayGetSize(pReq->pTables);
+        for (int32_t j = 0; j < ntables; ++j) {
+          SMetaRes res = {0};
+          res.pRes = taosMemoryCalloc(1, sizeof(SVgroupInfo));
+          res.code = catalogGetTableHashVgroup((const SName*)taosArrayGet(pReq->pTables, j), (SVgroupInfo*)res.pRes);
+          taosArrayPush(*pTableVgroupData, &res);
+        }
       }
     }
     return TSDB_CODE_SUCCESS;
@@ -677,12 +685,17 @@ int32_t MockCatalogService::catalogGetAllMeta(const SCatalogReq* pCatalogReq, SM
   return impl_->catalogGetAllMeta(pCatalogReq, pMetaData);
 }
 
+void MockCatalogService::destoryTablesReq(void* p) {
+  STablesReq* pRes = (STablesReq*)p;
+  taosArrayDestroy(pRes->pTables);
+}
+
 void MockCatalogService::destoryCatalogReq(SCatalogReq* pReq) {
   taosArrayDestroy(pReq->pDbVgroup);
   taosArrayDestroy(pReq->pDbCfg);
   taosArrayDestroy(pReq->pDbInfo);
-  taosArrayDestroy(pReq->pTableMeta);
-  taosArrayDestroy(pReq->pTableHash);
+  taosArrayDestroyEx(pReq->pTableMeta, destoryTablesReq);
+  taosArrayDestroyEx(pReq->pTableHash, destoryTablesReq);
   taosArrayDestroy(pReq->pUdf);
   taosArrayDestroy(pReq->pIndex);
   taosArrayDestroy(pReq->pUser);

source/libs/parser/test/mockCatalogService.h

@@ -50,6 +50,7 @@ struct MockTableMeta {
 class MockCatalogServiceImpl;
 class MockCatalogService {
  public:
+  static void destoryTablesReq(void* p);
   static void destoryCatalogReq(SCatalogReq* pReq);
   static void destoryMetaRes(void* p);
   static void destoryMetaArrayRes(void* p);

source/libs/parser/test/parInsertTest.cpp

@@ -13,6 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include <functional>
+
 #include <gtest/gtest.h>
 
 #include "mockCatalogService.h"
@@ -20,6 +22,7 @@
 #include "parInt.h"
 
 using namespace std;
+using namespace std::placeholders;
 using namespace testing;
 
 namespace {
@@ -63,7 +66,9 @@ class InsertTest : public Test {
 
   int32_t runAsync() {
     cxt_.async = true;
-    unique_ptr<SParseMetaCache, void (*)(SParseMetaCache*)> metaCache(new SParseMetaCache(), _destoryParseMetaCache);
+    bool                                                           request = true;
+    unique_ptr<SParseMetaCache, function<void(SParseMetaCache*)> > metaCache(
+        new SParseMetaCache(), std::bind(_destoryParseMetaCache, _1, cref(request)));
     code_ = parseInsertSyntax(&cxt_, &res_, metaCache.get());
     if (code_ != TSDB_CODE_SUCCESS) {
       cout << "parseInsertSyntax code:" << toString(code_) << ", msg:" << errMagBuf_ << endl;
@@ -81,6 +86,8 @@ class InsertTest : public Test {
     unique_ptr<SMetaData, void (*)(SMetaData*)> metaData(new SMetaData(), MockCatalogService::destoryMetaData);
     g_mockCatalogService->catalogGetAllMeta(catalogReq.get(), metaData.get());
 
+    metaCache.reset(new SParseMetaCache());
+    request = false;
     code_ = putMetaDataToCache(catalogReq.get(), metaData.get(), metaCache.get());
     if (code_ != TSDB_CODE_SUCCESS) {
       cout << "putMetaDataToCache code:" << toString(code_) << ", msg:" << errMagBuf_ << endl;
@@ -144,8 +151,8 @@ class InsertTest : public Test {
   static const int max_err_len = 1024;
   static const int max_sql_len = 1024 * 1024;
 
-  static void _destoryParseMetaCache(SParseMetaCache* pMetaCache) {
-    destoryParseMetaCache(pMetaCache);
+  static void _destoryParseMetaCache(SParseMetaCache* pMetaCache, bool request) {
+    destoryParseMetaCache(pMetaCache, request);
     delete pMetaCache;
   }
 

source/libs/parser/test/parTestUtil.cpp

@@ -24,6 +24,7 @@
 #include "parInt.h"
 
 using namespace std;
+using namespace std::placeholders;
 using namespace testing;
 
 namespace ParserTest {
@@ -118,8 +119,8 @@ class ParserTestBaseImpl {
     TEST_INTERFACE_ASYNC_API
   };
 
-  static void _destoryParseMetaCache(SParseMetaCache* pMetaCache) {
-    destoryParseMetaCache(pMetaCache);
+  static void _destoryParseMetaCache(SParseMetaCache* pMetaCache, bool request) {
+    destoryParseMetaCache(pMetaCache, request);
     delete pMetaCache;
   }
 
@@ -340,7 +341,9 @@ class ParserTestBaseImpl {
       doParse(&cxt, query.get());
       SQuery* pQuery = *(query.get());
 
-      unique_ptr<SParseMetaCache, void (*)(SParseMetaCache*)> metaCache(new SParseMetaCache(), _destoryParseMetaCache);
+      bool                                                           request = true;
+      unique_ptr<SParseMetaCache, function<void(SParseMetaCache*)> > metaCache(
+          new SParseMetaCache(), bind(_destoryParseMetaCache, _1, cref(request)));
       doCollectMetaKey(&cxt, pQuery, metaCache.get());
 
       unique_ptr<SCatalogReq, void (*)(SCatalogReq*)> catalogReq(new SCatalogReq(),
@@ -353,6 +356,8 @@ class ParserTestBaseImpl {
           unique_ptr<SMetaData, void (*)(SMetaData*)> metaData(new SMetaData(), MockCatalogService::destoryMetaData);
           doGetAllMeta(catalogReq.get(), metaData.get());
 
+          metaCache.reset(new SParseMetaCache());
+          request = false;
           doPutMetaDataToCache(catalogReq.get(), metaData.get(), metaCache.get());
 
           doAuthenticate(&cxt, pQuery, metaCache.get());

source/libs/planner/src/planner.c

@@ -85,7 +85,7 @@ static int32_t setSubplanExecutionNode(SPhysiNode* pNode, int32_t groupId, SDown
 }
 
 int32_t qSetSubplanExecutionNode(SSubplan* subplan, int32_t groupId, SDownstreamSourceNode* pSource) {
-  planDebug("QID:0x%" PRIx64 " set subplan execution node, groupId:%d", subplan->id.groupId, groupId);
+  planDebug("QID:0x%" PRIx64 " set subplan execution node, groupId:%d", subplan->id.queryId, groupId);
   return setSubplanExecutionNode(subplan->pNode, groupId, pSource);
 }
 
@@ -104,7 +104,10 @@ static void clearSubplanExecutionNode(SPhysiNode* pNode) {
   FOREACH(pChild, pNode->pChildren) { clearSubplanExecutionNode((SPhysiNode*)pChild); }
 }
 
-void qClearSubplanExecutionNode(SSubplan* pSubplan) { clearSubplanExecutionNode(pSubplan->pNode); }
+void qClearSubplanExecutionNode(SSubplan* pSubplan) {
+  planDebug("QID:0x%" PRIx64 " clear subplan execution node, groupId:%d", pSubplan->id.queryId, pSubplan->id.groupId);
+  clearSubplanExecutionNode(pSubplan->pNode);
+}
 
 int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {
   if (SUBPLAN_TYPE_MODIFY == pSubplan->subplanType && NULL == pSubplan->pNode) {

source/libs/qcom/src/queryUtil.c

@@ -462,3 +462,5 @@ int32_t cloneDbVgInfo(SDBVgInfo* pSrc, SDBVgInfo** pDst) {
 
   return TSDB_CODE_SUCCESS;
 }
+
+

source/libs/qworker/src/qwMsg.c

@@ -387,10 +387,13 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int
   
   char * sql = strndup(msg->msg, msg->sqlLen);
   QW_SCH_TASK_DLOG("processQuery start, node:%p, type:%s, handle:%p, SQL:%s", node, TMSG_INFO(pMsg->msgType), pMsg->info.handle, sql);
-  QW_ERR_RET(qwProcessQuery(QW_FPARAMS(), &qwMsg, sql));
-  QW_SCH_TASK_DLOG("processQuery end, node:%p", node);
+  QW_ERR_JRET(qwProcessQuery(QW_FPARAMS(), &qwMsg, sql));
 
-  return TSDB_CODE_SUCCESS;
+_return:
+
+  QW_SCH_TASK_DLOG("processQuery end, node:%p, code:%d", node, code);
+
+  return code;
 }
 
 int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int64_t ts) {

source/libs/scheduler/src/schStatus.c

@@ -64,7 +64,7 @@ _return:
 int32_t schHandleOpBeginEvent(int64_t jobId, SSchJob** job, SCH_OP_TYPE type, SSchedulerReq* pReq) {
   SSchJob *pJob = schAcquireJob(jobId);
   if (NULL == pJob) {
-    qError("Acquire sch job failed, may be dropped, jobId:0x%" PRIx64, jobId);
+    qWarn("Acquire sch job failed, may be dropped, jobId:0x%" PRIx64, jobId);
     SCH_ERR_RET(TSDB_CODE_SCH_STATUS_ERROR);
   }
 

source/libs/scheduler/src/schTask.c

@@ -284,7 +284,6 @@ int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask) {
 
   for (int32_t i = 0; i < parentNum; ++i) {
     SSchTask *parent = *(SSchTask **)taosArrayGet(pTask->parents, i);
-    int32_t   readyNum = atomic_add_fetch_32(&parent->childReady, 1);
 
     SCH_LOCK(SCH_WRITE, &parent->planLock);
     SDownstreamSourceNode source = {
@@ -298,6 +297,8 @@ int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask) {
     qSetSubplanExecutionNode(parent->plan, pTask->plan->id.groupId, &source);
     SCH_UNLOCK(SCH_WRITE, &parent->planLock);
 
+    int32_t readyNum = atomic_add_fetch_32(&parent->childReady, 1);
+
     if (SCH_TASK_READY_FOR_LAUNCH(readyNum, parent)) {
       SCH_TASK_DLOG("all %d children task done, start to launch parent task 0x%" PRIx64, readyNum, parent->taskId);
       SCH_ERR_RET(schLaunchTask(pJob, parent));
@@ -536,6 +537,7 @@ int32_t schMoveTaskToExecList(SSchJob *pJob, SSchTask *pTask, bool *moved) {
 int32_t schTaskCheckSetRetry(SSchJob *pJob, SSchTask *pTask, int32_t errCode, bool *needRetry) {
   if (TSDB_CODE_SCH_TIMEOUT_ERROR == errCode) {
     pTask->maxExecTimes++;
+    pTask->maxRetryTimes++;
     if (pTask->timeoutUsec < SCH_MAX_TASK_TIMEOUT_USEC) {
       pTask->timeoutUsec *= 2;
       if (pTask->timeoutUsec > SCH_MAX_TASK_TIMEOUT_USEC) {

source/libs/scheduler/src/scheduler.c

@@ -150,7 +150,7 @@ void schedulerFreeJob(int64_t* jobId, int32_t errCode) {
 
   SSchJob *pJob = schAcquireJob(*jobId);
   if (NULL == pJob) {
-    qError("Acquire sch job failed, may be dropped, jobId:0x%" PRIx64, *jobId);
+    qWarn("Acquire sch job failed, may be dropped, jobId:0x%" PRIx64, *jobId);
     return;
   }
 

source/libs/stream/src/streamData.c

@@ -35,6 +35,7 @@ int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock
     pDataBlock->info.window.skey = be64toh(pRetrieve->skey);
     pDataBlock->info.window.ekey = be64toh(pRetrieve->ekey);
     pDataBlock->info.version = be64toh(pRetrieve->version);
+    pDataBlock->info.watermark = be64toh(pRetrieve->watermark);
 
     pDataBlock->info.type = pRetrieve->streamBlockType;
     pDataBlock->info.childId = pReq->upstreamChildId;

source/libs/stream/src/streamDispatch.c

@@ -184,6 +184,7 @@ static int32_t streamAddBlockToDispatchMsg(const SSDataBlock* pBlock, SStreamDis
   pRetrieve->skey = htobe64(pBlock->info.window.skey);
   pRetrieve->ekey = htobe64(pBlock->info.window.ekey);
   pRetrieve->version = htobe64(pBlock->info.version);
+  pRetrieve->watermark = htobe64(pBlock->info.watermark);
 
   int32_t numOfCols = (int32_t)taosArrayGetSize(pBlock->pDataBlock);
   pRetrieve->numOfCols = htonl(numOfCols);

source/libs/sync/src/syncMain.c

@@ -1270,6 +1270,8 @@ int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode) {
   atomic_add_fetch_64(&pSyncNode->electTimerLogicClockUser, 1);
   taosTmrStop(pSyncNode->pElectTimer);
   pSyncNode->pElectTimer = NULL;
+
+  sTrace("vgId:%d, sync %s stop elect timer", pSyncNode->vgId, syncUtilState2String(pSyncNode->state));
   return ret;
 }
 
@@ -1343,7 +1345,8 @@ int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) {
   atomic_add_fetch_64(&pSyncNode->heartbeatTimerLogicClockUser, 1);
   taosTmrStop(pSyncNode->pHeartbeatTimer);
   pSyncNode->pHeartbeatTimer = NULL;
-  sTrace("vgId:%d, stop heartbeat timer", pSyncNode->vgId);
+
+  sTrace("vgId:%d, sync %s stop heartbeat timer", pSyncNode->vgId, syncUtilState2String(pSyncNode->state));
 
   return ret;
 }
@@ -2965,7 +2968,7 @@ bool syncNodeCanChange(SSyncNode* pSyncNode) {
   return true;
 }
 
-inline void syncLogSendRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, const char* s) {
+void syncLogSendRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, const char* s) {
   char     host[64];
   uint16_t port;
   syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
@@ -2976,7 +2979,7 @@ inline void syncLogSendRequestVote(SSyncNode* pSyncNode, const SyncRequestVote*
   syncNodeEventLog(pSyncNode, logBuf);
 }
 
-inline void syncLogRecvRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, const char* s) {
+void syncLogRecvRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, const char* s) {
   char     logBuf[256];
   char     host[64];
   uint16_t port;
@@ -2987,7 +2990,7 @@ inline void syncLogRecvRequestVote(SSyncNode* pSyncNode, const SyncRequestVote*
   syncNodeEventLog(pSyncNode, logBuf);
 }
 
-inline void syncLogSendRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteReply* pMsg, const char* s) {
+void syncLogSendRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteReply* pMsg, const char* s) {
   char     host[64];
   uint16_t port;
   syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
@@ -2997,7 +3000,7 @@ inline void syncLogSendRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestV
   syncNodeEventLog(pSyncNode, logBuf);
 }
 
-inline void syncLogRecvRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteReply* pMsg, const char* s) {
+void syncLogRecvRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteReply* pMsg, const char* s) {
   char     host[64];
   uint16_t port;
   syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
@@ -3007,7 +3010,7 @@ inline void syncLogRecvRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestV
   syncNodeEventLog(pSyncNode, logBuf);
 }
 
-inline void syncLogSendAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntries* pMsg, const char* s) {
+void syncLogSendAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntries* pMsg, const char* s) {
   char     host[64];
   uint16_t port;
   syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
@@ -3022,7 +3025,7 @@ inline void syncLogSendAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntri
   syncNodeEventLog(pSyncNode, logBuf);
 }
 
-inline void syncLogRecvAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntries* pMsg, const char* s) {
+void syncLogRecvAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntries* pMsg, const char* s) {
   char     host[64];
   uint16_t port;
   syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
@@ -3037,7 +3040,7 @@ inline void syncLogRecvAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntri
   syncNodeEventLog(pSyncNode, logBuf);
 }
 
-inline void syncLogSendAppendEntriesBatch(SSyncNode* pSyncNode, const SyncAppendEntriesBatch* pMsg, const char* s) {
+void syncLogSendAppendEntriesBatch(SSyncNode* pSyncNode, const SyncAppendEntriesBatch* pMsg, const char* s) {
   char     host[64];
   uint16_t port;
   syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
@@ -3050,7 +3053,7 @@ inline void syncLogSendAppendEntriesBatch(SSyncNode* pSyncNode, const SyncAppend
   syncNodeEventLog(pSyncNode, logBuf);
 }
 
-inline void syncLogRecvAppendEntriesBatch(SSyncNode* pSyncNode, const SyncAppendEntriesBatch* pMsg, const char* s) {
+void syncLogRecvAppendEntriesBatch(SSyncNode* pSyncNode, const SyncAppendEntriesBatch* pMsg, const char* s) {
   char     host[64];
   uint16_t port;
   syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
@@ -3063,7 +3066,7 @@ inline void syncLogRecvAppendEntriesBatch(SSyncNode* pSyncNode, const SyncAppend
   syncNodeEventLog(pSyncNode, logBuf);
 }
 
-inline void syncLogSendAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s) {
+ void syncLogSendAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s) {
   char     host[64];
   uint16_t port;
   syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
@@ -3075,7 +3078,7 @@ inline void syncLogSendAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppend
   syncNodeEventLog(pSyncNode, logBuf);
 }
 
-inline void syncLogRecvAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s) {
+void syncLogRecvAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s) {
   char     host[64];
   uint16_t port;
   syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);

source/libs/sync/src/syncRequestVote.c

@@ -57,11 +57,14 @@ int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {
 
   // maybe update term
   if (pMsg->term > ths->pRaftStore->currentTerm) {
+    syncNodeUpdateTerm(ths, pMsg->term);
+#if 0  
     if (logOK) {
       syncNodeUpdateTerm(ths, pMsg->term);
     } else {
       syncNodeUpdateTermWithoutStepDown(ths, pMsg->term);
     }
+#endif
   }
   ASSERT(pMsg->term <= ths->pRaftStore->currentTerm);
 
@@ -167,11 +170,14 @@ int32_t syncNodeOnRequestVoteSnapshotCb(SSyncNode* ths, SyncRequestVote* pMsg) {
 
   // maybe update term
   if (pMsg->term > ths->pRaftStore->currentTerm) {
+    syncNodeUpdateTerm(ths, pMsg->term);
+#if 0
     if (logOK) {
       syncNodeUpdateTerm(ths, pMsg->term);
     } else {
       syncNodeUpdateTermWithoutStepDown(ths, pMsg->term);
     }
+#endif
   }
   ASSERT(pMsg->term <= ths->pRaftStore->currentTerm);
 

source/libs/sync/test/sh/a.sh

@@ -89,7 +89,7 @@ for file in `ls ${logpath}/log.dnode*.vgId*.commit`;do
 	line=`cat ${file} | tail -n1`
 	echo $line | awk '{print $5, $0}' >> ${tmpfile}
 done
-cat ${tmpfile} | sort -k1 > ${logpath}/log.commits
+cat ${tmpfile} | sort -k1 | awk 'BEGIN{vgid=$1}{if($1==vgid){print $0}else{print ""; print $0; vgid=$1;}}END{}' > ${logpath}/log.commits
 
 exit 0
 

source/libs/sync/test/sh/insert.tpl.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 8,
-    "thread_count_create_tbl": 8,
+    "create_table_thread_count": 8,
     "result_file": "./tpl_insert_result_tpl",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/parallel_test/run_case.sh

@@ -49,11 +49,13 @@ if [ $ent -eq 0 ]; then
     export PATH=$PATH:/home/TDengine/debug/build/bin
     export LD_LIBRARY_PATH=/home/TDengine/debug/build/lib
     ln -s /home/TDengine/debug/build/lib/libtaos.so /usr/lib/libtaos.so 2>/dev/null
+    ln -s /home/TDengine/debug/build/lib/libtaos.so /usr/lib/libtaos.so.1 2>/dev/null
     CONTAINER_TESTDIR=/home/TDengine
 else
     export PATH=$PATH:/home/TDinternal/debug/build/bin
     export LD_LIBRARY_PATH=/home/TDinternal/debug/build/lib
     ln -s /home/TDinternal/debug/build/lib/libtaos.so /usr/lib/libtaos.so 2>/dev/null
+    ln -s /home/TDinternal/debug/build/lib/libtaos.so /usr/lib/libtaos.so.1 2>/dev/null
     CONTAINER_TESTDIR=/home/TDinternal/community
 fi
 mkdir -p /var/lib/taos/subscribe

tests/parallel_test/run_container.sh

@@ -100,6 +100,7 @@ docker run \
     -v "$TMP_DIR/thread_volume/$thread_no/sim:${SIM_DIR}" \
     -v ${TMP_DIR}/thread_volume/$thread_no/coredump:$coredump_dir \
     -v $WORKDIR/taos-connector-python/taos:/usr/local/lib/python3.8/site-packages/taos:ro \
+    -v $WORKDIR/taos-connector-python/taosrest:/usr/local/lib/python3.8/site-packages/taosrest:ro \
     --rm --ulimit core=-1 taos_test:v1.0 $CONTAINER_TESTDIR/tests/parallel_test/run_case.sh -d "$exec_dir" -c "$cmd" $extra_param
 ret=$?
 exit $ret

tests/pytest/cluster/TD-3693/insert1Data.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/cluster/TD-3693/insert2Data.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/dockerCluster/insert.json

@@ -6,7 +6,7 @@
 	"user": "root",
 	"password": "taosdata",
 	"thread_count": 4,
-	"thread_count_create_tbl": 1,
+	"create_table_thread_count": 1,
 	"result_file": "./insert_res.txt",
         "confirm_parameter_prompt": "no",
 	"databases": [{

tests/pytest/manualTest/TD-5114/insertDataDb3Replica2.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/perfbenchmark/bug3433.py

@@ -185,7 +185,7 @@ class TDTestCase:
             "user": "root",
             "password": "taosdata",
             "thread_count": 4,
-            "thread_count_create_tbl": 4,
+            "create_table_thread_count": 4,
             "result_file": "/tmp/insert_res.txt",
             "confirm_parameter_prompt": "no",
             "insert_interval": 0,

tests/pytest/perfbenchmark/joinPerformance.py

@@ -168,7 +168,7 @@ class JoinPerf:
             "user": self.user,
             "password": self.password,
             "thread_count": cpu_count(),
-            "thread_count_create_tbl": cpu_count(),
+            "create_table_thread_count": cpu_count(),
             "result_file": "/tmp/insert_res.txt",
             "confirm_parameter_prompt": "no",
             "insert_interval": 0,

tests/pytest/perfbenchmark/taosdemoInsert.py

@@ -172,7 +172,7 @@ class Taosdemo:
             "user": self.user,
             "password": self.password,
             "thread_count": cpu_count(),
-            "thread_count_create_tbl": cpu_count(),
+            "create_table_thread_count": cpu_count(),
             "result_file": "/tmp/insert_res.txt",
             "confirm_parameter_prompt": "no",
             "insert_interval": 0,

tests/pytest/query/nestedQuery/insertData.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file":"./insert_res.txt",
     "confirm_parameter_prompt": "no",   
     "insert_interval": 0,

tests/pytest/query/query1970YearsAf.py

@@ -133,7 +133,7 @@ class TDTestCase:
             "user": "root",
             "password": "taosdata",
             "thread_count": 4,
-            "thread_count_create_tbl": 4,
+            "create_table_thread_count": 4,
             "result_file": "/tmp/insert_res.txt",
             "confirm_parameter_prompt": "no",
             "insert_interval": 0,

tests/pytest/tools/insert-interlace.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 5000,

tests/pytest/tools/insert-tblimit-tboffset-createdb.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/insert-tblimit-tboffset-insertrec.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/insert-tblimit-tboffset.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/insert-tblimit-tboffset0.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/insert-tblimit1-tboffset.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/insert.json

@@ -7,7 +7,7 @@
     "password": "taosdata",
     "thread_count": 2,
     "num_of_records_per_req": 10,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "databases": [{
         "dbinfo": {
             "name": "db01",

tests/pytest/tools/taosdemoAllTest/NanoTestCase/taosdemoInsertMSDB.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 10,
-    "thread_count_create_tbl": 10,
+    "create_table_thread_count": 10,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/NanoTestCase/taosdemoInsertNanoDB.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 10,
-    "thread_count_create_tbl": 10,
+    "create_table_thread_count": 10,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/NanoTestCase/taosdemoInsertUSDB.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 10,
-    "thread_count_create_tbl": 10,
+    "create_table_thread_count": 10,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/NanoTestCase/taosdemoTestNanoDatabase.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 10,
-    "thread_count_create_tbl": 10,
+    "create_table_thread_count": 10,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/NanoTestCase/taosdemoTestNanoDatabaseInsertForSub.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 10,
-    "thread_count_create_tbl": 10,
+    "create_table_thread_count": 10,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/NanoTestCase/taosdemoTestNanoDatabaseNow.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 10,
-    "thread_count_create_tbl": 10,
+    "create_table_thread_count": 10,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/NanoTestCase/taosdemoTestNanoDatabasecsv.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 10,
-    "thread_count_create_tbl": 10,
+    "create_table_thread_count": 10,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/TD-3453/query-interrupt.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/TD-4985/query-limit-offset.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 10,
-    "thread_count_create_tbl": 10,
+    "create_table_thread_count": 10,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/TD-5213/insertSigcolumnsNum4096.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 10,
-    "thread_count_create_tbl": 10,
+    "create_table_thread_count": 10,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-1s1tnt1r.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-1s1tntmr.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-disorder.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file":"./insert_res.txt",
     "confirm_parameter_prompt": "no", 
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-drop-exist-auto-N00.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,                   
-    "thread_count_create_tbl": 4,        
+    "create_table_thread_count": 4,        
     "result_file": "./insert_res.txt",   
     "confirm_parameter_prompt": "no",    
     "insert_interval": 0,         

tests/pytest/tools/taosdemoAllTest/insert-drop-exist-auto-Y00.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,                   
-    "thread_count_create_tbl": 4,        
+    "create_table_thread_count": 4,        
     "result_file": "./insert_res.txt",   
     "confirm_parameter_prompt": "no",    
     "insert_interval": 0,         

tests/pytest/tools/taosdemoAllTest/insert-illegal.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-interlace-row.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-interval-speed.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 100,

tests/pytest/tools/taosdemoAllTest/insert-newdb.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-newtable.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-nodbnodrop.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-offset.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-renewdb.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-sample.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file":"./insert_res.txt",
     "confirm_parameter_prompt": "no",   
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insert-timestep.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file":"./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertBinaryLenLarge16374AllcolLar49151.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertChildTab0.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertChildTabLess0.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertColumnsAndTagNum4096.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertColumnsAndTagNumLarge4096.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertColumnsNum0.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertInterlaceRowsLarge1M.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertMaxNumPerReq.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertNumOfrecordPerReq0.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertNumOfrecordPerReqless0.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertRestful.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,

tests/pytest/tools/taosdemoAllTest/insertSigcolumnsNum4096.json

@@ -6,7 +6,7 @@
     "user": "root",
     "password": "taosdata",
     "thread_count": 4,
-    "thread_count_create_tbl": 4,
+    "create_table_thread_count": 4,
     "result_file": "./insert_res.txt",
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,