From 9e5684479e6a15169d3d5b1c0dd5372760e62b54 Mon Sep 17 00:00:00 2001
From: kailixu <klxu@taosdata.com>
Date: Thu, 27 Jul 2023 14:19:49 +0800
Subject: [PATCH 01/20] fix: taos print version

---
 tools/shell/src/shellUtil.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tools/shell/src/shellUtil.c b/tools/shell/src/shellUtil.c
index e15b49efcc..93451c85a9 100644
--- a/tools/shell/src/shellUtil.c
+++ b/tools/shell/src/shellUtil.c
@@ -68,7 +68,7 @@ int32_t shellCheckIntSize() {
   return 0;
 }
 
-void shellPrintVersion() { printf("version: %s\r\n", version); }
+void shellPrintVersion() { printf("%s\r\n", shell.info.programVersion); }
 
 void shellGenerateAuth() {
   char secretEncrypt[TSDB_PASSWORD_LEN + 1] = {0};

From 37fa70c487414d1e8232c9dc1c0d9a3659795c30 Mon Sep 17 00:00:00 2001
From: Shuduo Sang <sangshuduo@gmail.com>
Date: Thu, 27 Jul 2023 18:42:41 +0800
Subject: [PATCH 02/20] docs: fix sum() in en/25-application/_03-immigrate.md

---
 docs/en/25-application/_03-immigrate.md | 2 +-
 docs/zh/25-application/03-immigrate.md  | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/en/25-application/_03-immigrate.md b/docs/en/25-application/_03-immigrate.md
index f780423532..457a40614e 100644
--- a/docs/en/25-application/_03-immigrate.md
+++ b/docs/en/25-application/_03-immigrate.md
@@ -338,7 +338,7 @@ Remark:
 Equivalent function: sum
 
 ```sql
-Select max(value) from (select first(val) value from table_name interval(10s) fill(linear)) interval(10s)
+Select sum(value) from (select first(val) value from table_name interval(10s) fill(linear)) interval(10s)
 ```
 
 Note: This function has no interpolation requirements, so it can be directly calculated.
diff --git a/docs/zh/25-application/03-immigrate.md b/docs/zh/25-application/03-immigrate.md
index 75788c0cc7..389a2b2c5a 100644
--- a/docs/zh/25-application/03-immigrate.md
+++ b/docs/zh/25-application/03-immigrate.md
@@ -371,7 +371,7 @@ Select min(val) from table_name
 等效函数：sum
 
 ```sql
-Select max(value) from (select first(val) value from table_name interval(10s) fill(linear)) interval(10s)
+Select sum(value) from (select first(val) value from table_name interval(10s) fill(linear)) interval(10s)
 ```
 
 备注：该函数无插值需求，因此可用直接计算。

From b00af27de59ddc7dcd118964e579168411d1d2d2 Mon Sep 17 00:00:00 2001
From: meeki007 <5952964+meeki007@users.noreply.github.com>
Date: Thu, 27 Jul 2023 21:12:54 -0400
Subject: [PATCH 03/20] Update 02-rest-api.mdx - note about URL Encoding
 (#22221)

Explanation on proper URL Encoding for parameters.
BAD: ?tz=Etc/GMT+10
GOOD: ?tz=Etc%2FGMT%2B10
---
 docs/en/14-reference/02-rest-api/02-rest-api.mdx | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/docs/en/14-reference/02-rest-api/02-rest-api.mdx b/docs/en/14-reference/02-rest-api/02-rest-api.mdx
index ea010f42db..4da987213c 100644
--- a/docs/en/14-reference/02-rest-api/02-rest-api.mdx
+++ b/docs/en/14-reference/02-rest-api/02-rest-api.mdx
@@ -79,6 +79,12 @@ Parameter Description:
 - tz: Optional parameter that specifies the timezone of the returned time, following the IANA Time Zone rules, e.g. `America/New_York`.
 - req_id: Optional parameter that specifies the request id for tracing.
 
+:::note
+
+URL Encoding. Make sure that parameters are properly encoded. For example, when specifying a timezone you must properly encode special characters. ?tz=Etc/GMT+10 will not work because the <+> plus symbol is recognized as a space in the url. It's best practice to encode all special characters in a parameter. Instead use ?tz=Etc%2FGMT%2B10 for the parameter.
+
+:::
+
 For example, `http://h1.taos.com:6041/rest/sql/test` is a URL to `h1.taos.com:6041` and sets the default database name to `test`.
 
 TDengine supports both Basic authentication and custom authentication mechanisms, and subsequent versions will provide a standard secure digital signature mechanism for authentication.

From f70709be3c0a382fbc5da225ff20265c08811a76 Mon Sep 17 00:00:00 2001
From: kailixu <klxu@taosdata.com>
Date: Fri, 28 Jul 2023 14:54:03 +0800
Subject: [PATCH 04/20] fix: run sim cases on windows

---
 tests/script/sh/stop_dnodes.bat | 4 +++-
 utils/tsim/CMakeLists.txt       | 2 +-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/tests/script/sh/stop_dnodes.bat b/tests/script/sh/stop_dnodes.bat
index ab7af2ca92..65aee26ed4 100644
--- a/tests/script/sh/stop_dnodes.bat
+++ b/tests/script/sh/stop_dnodes.bat
@@ -3,4 +3,6 @@
 rem echo taskkill /F /IM taosd.exe
 
 wmic process where "name='taosd.exe'" call terminate > NUL 2>&1
-taskkill /F /IM taosd.exe > NUL 2>&1
\ No newline at end of file
+taskkill /F /IM taosd.exe > NUL 2>&1
+
+rem echo taskkill /F /IM taosd.exe finished
\ No newline at end of file
diff --git a/utils/tsim/CMakeLists.txt b/utils/tsim/CMakeLists.txt
index 81737809d9..c2cf7ac3c5 100644
--- a/utils/tsim/CMakeLists.txt
+++ b/utils/tsim/CMakeLists.txt
@@ -2,7 +2,7 @@ aux_source_directory(src TSIM_SRC)
 add_executable(tsim ${TSIM_SRC})
 target_link_libraries(
     tsim
-    PUBLIC taos
+    PUBLIC taos_static
     PUBLIC util
     PUBLIC common
     PUBLIC os

From 8999bb927fa9cdd7b0f110863e8fc2aa9fb48283 Mon Sep 17 00:00:00 2001
From: Shuduo Sang <sangshuduo@gmail.com>
Date: Fri, 28 Jul 2023 15:20:45 +0800
Subject: [PATCH 05/20] docs: add unit in wal_retention_period description

---
 docs/en/12-taos-sql/22-meta.md | 2 +-
 docs/zh/12-taos-sql/22-meta.md | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/en/12-taos-sql/22-meta.md b/docs/en/12-taos-sql/22-meta.md
index ae7b81864e..194c75b19c 100644
--- a/docs/en/12-taos-sql/22-meta.md
+++ b/docs/en/12-taos-sql/22-meta.md
@@ -98,7 +98,7 @@ Provides information about user-created databases. Similar to SHOW DATABASES.
 | 21  |      cachesize       | INT           | Memory per vnode used for caching the newest data. It should be noted that `cachesize` is a TDengine keyword and needs to be escaped with ` when used as a column name.                                                           |
 | 22  |      wal_level       | INT           | WAL level. It should be noted that `wal_level` is a TDengine keyword and needs to be escaped with ` when used as a column name.                                                                                                   |
 | 23  |   wal_fsync_period   | INT           | Interval at which WAL is written to disk. It should be noted that `wal_fsync_period` is a TDengine keyword and needs to be escaped with ` when used as a column name.                                                             |
-| 24  | wal_retention_period | INT           | WAL retention period. It should be noted that `wal_retention_period` is a TDengine keyword and needs to be escaped with ` when used as a column name.                                                                             |
+| 24  | wal_retention_period | INT           | WAL retention period, in seconds. It should be noted that `wal_retention_period` is a TDengine keyword and needs to be escaped with ` when used as a column name.                                                                             |
 | 25  |  wal_retention_size  | INT           | Maximum WAL size. It should be noted that `wal_retention_size` is a TDengine keyword and needs to be escaped with ` when used as a column name.                                                                                   |
 | 26  |   wal_roll_period    | INT           | WAL rotation period. It should be noted that `wal_roll_period` is a TDengine keyword and needs to be escaped with ` when used as a column name.                                                                                   |
 | 27  |   wal_segment_size   | BIGINT        | WAL file size. It should be noted that `wal_segment_size` is a TDengine keyword and needs to be escaped with ` when used as a column name.                                                                                        |
diff --git a/docs/zh/12-taos-sql/22-meta.md b/docs/zh/12-taos-sql/22-meta.md
index 5df79204d0..62356b76d1 100644
--- a/docs/zh/12-taos-sql/22-meta.md
+++ b/docs/zh/12-taos-sql/22-meta.md
@@ -98,7 +98,7 @@ TDengine 内置了一个名为 `INFORMATION_SCHEMA` 的数据库，提供对数
 | 21  |      cachesize       | INT          | 表示每个 vnode 中用于缓存子表最近数据的内存大小。需要注意，`cachesize` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。                      |
 | 22  |      wal_level       | INT          | WAL 级别。需要注意，`wal_level` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。                                                             |
 | 23  |   wal_fsync_period   | INT          | 数据落盘周期。需要注意，`wal_fsync_period` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。                                                  |
-| 24  | wal_retention_period | INT          | WAL 的保存时长。需要注意，`wal_retention_period` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。                                            |
+| 24  | wal_retention_period | INT          | WAL 的保存时长，单位为秒。需要注意，`wal_retention_period` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。                                            |
 | 25  |  wal_retention_size  | INT          | WAL 的保存上限。需要注意，`wal_retention_size` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。                                              |
 | 26  |   wal_roll_period    | INT          | wal 文件切换时长。需要注意，`wal_roll_period` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。                                               |
 | 27  |   wal_segment_size   | BIGINT       | wal 单个文件大小。需要注意，`wal_segment_size` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。                                              |

From b38e10034a439cef9973440e6e7a855e3e387fff Mon Sep 17 00:00:00 2001
From: Adam Ji <jrgwyz@163.com>
Date: Fri, 28 Jul 2023 15:45:49 +0800
Subject: [PATCH 06/20] docs: update connector rust version history

---
 docs/en/14-reference/03-connector/06-rust.mdx | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/docs/en/14-reference/03-connector/06-rust.mdx b/docs/en/14-reference/03-connector/06-rust.mdx
index 56f5e20cb4..a98683d43c 100644
--- a/docs/en/14-reference/03-connector/06-rust.mdx
+++ b/docs/en/14-reference/03-connector/06-rust.mdx
@@ -31,7 +31,8 @@ Websocket connections are supported on all platforms that can run Go.
 
 | connector-rust version |  TDengine version |                   major features                    |
 | :----------------: | :--------------: | :--------------------------------------------------: |
-|       v0.8.12       |      3.0.5.0 or later    | TMQ: Get consuming progress and seek offset to consume. |
+|       v0.9.2       |      3.0.7.0 or later    | STMT: Get tag_fields and col_fields under ws. |
+|       v0.8.12       |      3.0.5.0    | TMQ: Get consuming progress and seek offset to consume. |
 |       v0.8.0       |      3.0.4.0     | Support schemaless insert. |
 |       v0.7.6       |      3.0.3.0     | Support req_id in query. |
 |       v0.6.0       |      3.0.0.0     | Base features. |

From 46421bc32165b3a86ffcbb88c8c586c4e0f08d6e Mon Sep 17 00:00:00 2001
From: Adam Ji <jrgwyz@163.com>
Date: Fri, 28 Jul 2023 15:46:29 +0800
Subject: [PATCH 07/20] docs: update connector rust version history

---
 docs/zh/08-connector/26-rust.mdx | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/docs/zh/08-connector/26-rust.mdx b/docs/zh/08-connector/26-rust.mdx
index 79a6badfea..3e51aa72bb 100644
--- a/docs/zh/08-connector/26-rust.mdx
+++ b/docs/zh/08-connector/26-rust.mdx
@@ -30,7 +30,8 @@ Websocket 连接支持所有能运行 Rust 的平台。
 
 |   Rust 连接器版本   |   TDengine 版本   |                        主要功能                       |
 | :----------------: | :--------------: | :--------------------------------------------------: |
-|       v0.8.12       |      3.0.5.0 or later    | 消息订阅：获取消费进度及按照指定进度开始消费。 |
+|       v0.9.2       |      3.0.7.0 or later    | STMT：ws 下获取 tag_fields、col_fields。 |
+|       v0.8.12       |      3.0.5.0    | 消息订阅：获取消费进度及按照指定进度开始消费。 |
 |       v0.8.0       |      3.0.4.0     | 支持无模式写入。 |
 |       v0.7.6       |      3.0.3.0     | 支持在请求中使用 req_id。 |
 |       v0.6.0       |      3.0.0.0     | 基础功能。 |

From 5db8990f53e73cc699dfd53ca71ec33174f5fe3f Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Fri, 28 Jul 2023 18:06:27 +0800
Subject: [PATCH 08/20] test: disable roll upgrade case.

---
 tests/parallel_test/cases.task | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task
index bb45ffdf72..1ec5102d9b 100644
--- a/tests/parallel_test/cases.task
+++ b/tests/parallel_test/cases.task
@@ -450,7 +450,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 6-cluster/5dnode3mnodeRestartDnodeInsertDataAsync.py -N 6 -M 3
 #,,y,system-test,./pytest.sh python3 ./test.py -f 6-cluster/5dnode3mnodeRestartDnodeInsertDataAsync.py -N 6 -M 3 -n 3
 ,,n,system-test,python3 ./test.py -f 6-cluster/manually-test/6dnode3mnodeInsertLessDataAlterRep3to1to3.py -N 6 -M 3
-,,n,system-test,python ./test.py  -f 6-cluster/5dnode3mnodeRoll.py  -N 3  -C 1
+#,,n,system-test,python ./test.py  -f 6-cluster/5dnode3mnodeRoll.py  -N 3  -C 1
 ,,y,system-test,./pytest.sh python3 ./test.py -f 6-cluster/5dnode3mnodeAdd1Ddnoe.py -N 7 -M 3 -C 6
 ,,y,system-test,./pytest.sh python3 ./test.py -f 6-cluster/5dnode3mnodeAdd1Ddnoe.py -N 7 -M 3 -C 6 -n 3
 #,,y,system-test,./pytest.sh python3 ./test.py -f 6-cluster/5dnode3mnodeDrop.py -N 5

From 2deeed87ae00d24e095a3b81230e7f9ccf314275 Mon Sep 17 00:00:00 2001
From: kailixu <klxu@taosdata.com>
Date: Sat, 29 Jul 2023 13:18:17 +0800
Subject: [PATCH 09/20] fix: community for mac os

---
 source/dnode/mgmt/node_mgmt/src/dmEnv.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/dnode/mgmt/node_mgmt/src/dmEnv.c b/source/dnode/mgmt/node_mgmt/src/dmEnv.c
index 848e123448..3f9c5bbeaf 100644
--- a/source/dnode/mgmt/node_mgmt/src/dmEnv.c
+++ b/source/dnode/mgmt/node_mgmt/src/dmEnv.c
@@ -42,7 +42,7 @@
 
 static SDnode      globalDnode = {0};
 static const char *dmOS[10] = {"Ubuntu",  "CentOS Linux", "Red Hat", "Debian GNU", "CoreOS",
-                               "FreeBSD", "openSUSE",     "SLES",    "Fedora",     "MacOS"};
+                               "FreeBSD", "openSUSE",     "SLES",    "Fedora",     "macOS"};
 
 SDnode *dmInstance() { return &globalDnode; }
 

From ff39aadb3483d8c647c22a421b431a2430c0ffc3 Mon Sep 17 00:00:00 2001
From: kailixu <klxu@taosdata.com>
Date: Mon, 31 Jul 2023 08:42:17 +0800
Subject: [PATCH 10/20] fix: mac OS for community

---
 source/dnode/mgmt/node_mgmt/src/dmEnv.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/dnode/mgmt/node_mgmt/src/dmEnv.c b/source/dnode/mgmt/node_mgmt/src/dmEnv.c
index 848e123448..3f9c5bbeaf 100644
--- a/source/dnode/mgmt/node_mgmt/src/dmEnv.c
+++ b/source/dnode/mgmt/node_mgmt/src/dmEnv.c
@@ -42,7 +42,7 @@
 
 static SDnode      globalDnode = {0};
 static const char *dmOS[10] = {"Ubuntu",  "CentOS Linux", "Red Hat", "Debian GNU", "CoreOS",
-                               "FreeBSD", "openSUSE",     "SLES",    "Fedora",     "MacOS"};
+                               "FreeBSD", "openSUSE",     "SLES",    "Fedora",     "macOS"};
 
 SDnode *dmInstance() { return &globalDnode; }
 

From 196ae283aa2fa5afadbc9a1439718fb9bb23d9d6 Mon Sep 17 00:00:00 2001
From: wangjiaming0909 <604227650@qq.com>
Date: Mon, 31 Jul 2023 14:03:07 +0800
Subject: [PATCH 11/20] force stt_trigger to be 1

---
 include/util/tdef.h | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/include/util/tdef.h b/include/util/tdef.h
index 3ee35b3b6b..26d2ef060c 100644
--- a/include/util/tdef.h
+++ b/include/util/tdef.h
@@ -369,8 +369,13 @@ typedef enum ELogicConditionType {
 #define TSDB_DB_SCHEMALESS_OFF          0
 #define TSDB_DEFAULT_DB_SCHEMALESS      TSDB_DB_SCHEMALESS_OFF
 #define TSDB_MIN_STT_TRIGGER            1
-#define TSDB_MAX_STT_TRIGGER            16
-#define TSDB_DEFAULT_SST_TRIGGER        2
+#ifdef TD_ENTERPRISE
+#define TSDB_MAX_STT_TRIGGER     16
+#define TSDB_DEFAULT_SST_TRIGGER 2
+#else
+#define TSDB_MAX_STT_TRIGGER     1
+#define TSDB_DEFAULT_SST_TRIGGER 1
+#endif
 #define TSDB_MIN_HASH_PREFIX            (2 - TSDB_TABLE_NAME_LEN)
 #define TSDB_MAX_HASH_PREFIX            (TSDB_TABLE_NAME_LEN - 2)
 #define TSDB_DEFAULT_HASH_PREFIX        0

From f359a0dca3d763245f72f5308a03c984144a3406 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 31 Jul 2023 14:22:46 +0800
Subject: [PATCH 12/20] fix(stream): set the correct end key of delete block.

---
 source/libs/executor/src/scanoperator.c | 47 ++++++++++++++++---------
 1 file changed, 30 insertions(+), 17 deletions(-)

diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index 7434db61db..555ca938c0 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -1590,38 +1590,51 @@ static void doBlockDataWindowFilter(SSDataBlock* pBlock, int32_t tsIndex, STimeW
 }
 
 // re-build the delete block, ONLY according to the split timestamp
-static void rebuildDeleteBlockData(SSDataBlock* pBlock, int64_t skey, const char* id) {
-  if (skey == INT64_MIN) {
-    return;
-  }
-
+static void rebuildDeleteBlockData(SSDataBlock* pBlock, STimeWindow* pWindow, const char* id) {
   int32_t numOfRows = pBlock->info.rows;
-
-  bool* p = taosMemoryCalloc(numOfRows, sizeof(bool));
-  bool  hasUnqualified = false;
+  bool*   p = taosMemoryCalloc(numOfRows, sizeof(bool));
+  bool    hasUnqualified = false;
+  int64_t skey = pWindow->skey;
+  int64_t ekey = pWindow->ekey;
 
   SColumnInfoData* pSrcStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
   uint64_t*        tsStartCol = (uint64_t*)pSrcStartCol->pData;
   SColumnInfoData* pSrcEndCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
   uint64_t*        tsEndCol = (uint64_t*)pSrcEndCol->pData;
 
-  for (int32_t i = 0; i < numOfRows; i++) {
-    if (tsStartCol[i] < skey) {
-      tsStartCol[i] = skey;
-    }
+  if (pWindow->skey != INT64_MIN) {
+    for (int32_t i = 0; i < numOfRows; i++) {
+      if (tsStartCol[i] < skey) {
+        tsStartCol[i] = skey;
+      }
 
-    if (tsEndCol[i] >= skey) {
-      p[i] = true;
-    } else { // this row should be removed, since it is not in this query time window, which is [skey, INT64_MAX]
-      hasUnqualified = true;
+      if (tsEndCol[i] >= skey) {
+        p[i] = true;
+      } else {  // this row should be removed, since it is not in this query time window, which is [skey, INT64_MAX]
+        hasUnqualified = true;
+      }
+    }
+  } else if (pWindow->ekey != INT64_MAX) {
+    for(int32_t i = 0; i < numOfRows; ++i) {
+      if (tsEndCol[i] > ekey) {
+        tsEndCol[i] = ekey;
+      }
+
+      if (tsStartCol[i] <= ekey) {
+        p[i] = true;
+      } else {
+        hasUnqualified = true;
+      }
     }
   }
 
   if (hasUnqualified) {
     trimDataBlock(pBlock, pBlock->info.rows, p);
+    qDebug("%s re-build delete datablock, start key revised to:%"PRId64", rows:%"PRId64, id, skey, pBlock->info.rows);
+  } else {
+    qDebug("%s not update the delete block", id);
   }
 
-  qDebug("%s re-build delete datablock, start key revised to:%"PRId64", rows:%"PRId64, id, skey, pBlock->info.rows);
   taosMemoryFree(p);
 }
 

From fe21943c9c27024a98ee624a8f735abc02b1ed12 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 31 Jul 2023 14:23:46 +0800
Subject: [PATCH 13/20] fix(stream): set the correct end key of delete block.

---
 source/libs/executor/src/scanoperator.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index 555ca938c0..738809f588 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -2043,7 +2043,7 @@ FETCH_NEXT_BLOCK:
         }
 
         setBlockGroupIdByUid(pInfo, pDelBlock);
-        rebuildDeleteBlockData(pDelBlock, pStreamInfo->fillHistoryWindow.skey, id);
+        rebuildDeleteBlockData(pDelBlock, &pStreamInfo->fillHistoryWindow, id);
         printDataBlock(pDelBlock, "stream scan delete recv filtered");
         if (pDelBlock->info.rows == 0) {
           if (pInfo->tqReader) {

From 079d7ff69ec525c69be84f30c4295662843cb547 Mon Sep 17 00:00:00 2001
From: Minglei Jin <mljin@taosdata.com>
Date: Mon, 31 Jul 2023 15:21:47 +0800
Subject: [PATCH 14/20] fix(tsdb/read2): reset stt reader when suspended

---
 source/dnode/vnode/src/tsdb/tsdbRead2.c | 131 +++++++++++-------------
 1 file changed, 62 insertions(+), 69 deletions(-)

diff --git a/source/dnode/vnode/src/tsdb/tsdbRead2.c b/source/dnode/vnode/src/tsdb/tsdbRead2.c
index d5d8ba130c..e1756333c5 100644
--- a/source/dnode/vnode/src/tsdb/tsdbRead2.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRead2.c
@@ -1729,45 +1729,41 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader
 
     // row in last file block
     TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
-    int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
+    int64_t ts = getCurrentKeyInLastBlock(pLastBlockReader);
+
     if (ASCENDING_TRAVERSE(pReader->info.order)) {
-      if (key < tsLast) {
+      if (key < ts) {  // imem, mem are all empty, file blocks (data blocks and last block) exist
         return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader);
-      } else if (key > tsLast) {
-        return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false);
-      }
-    } else {
-      if (key > tsLast) {
-        return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader);
-      } else if (key < tsLast) {
+      } else if (key == ts) {
+        SRow*   pTSRow = NULL;
+        int32_t code = tsdbRowMergerAdd(pMerger, &fRow, pReader->info.pSchema);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
+
+        doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader);
+
+        TSDBROW* pRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
+        tsdbRowMergerAdd(pMerger, pRow1, NULL);
+
+        doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, ts, pMerger, &pReader->info.verRange, pReader->idStr);
+
+        code = tsdbRowMergerGetRow(pMerger, &pTSRow);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
+
+        code = doAppendRowFromTSRow(pReader->resBlockInfo.pResBlock, pReader, pTSRow, pBlockScanInfo);
+
+        taosMemoryFree(pTSRow);
+        tsdbRowMergerClear(pMerger);
+        return code;
+      } else {  // key > ts
         return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false);
       }
+    } else {  // desc order
+      return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, pBlockData, true);
     }
-    // the following for key == tsLast
-    SRow*   pTSRow = NULL;
-    int32_t code = tsdbRowMergerAdd(pMerger, &fRow, pReader->info.pSchema);
-    if (code != TSDB_CODE_SUCCESS) {
-      return code;
-    }
-
-    doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader);
-
-    TSDBROW* pRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
-    tsdbRowMergerAdd(pMerger, pRow1, NULL);
-
-    doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, pMerger, &pReader->info.verRange, pReader->idStr);
-
-    code = tsdbRowMergerGetRow(pMerger, &pTSRow);
-    if (code != TSDB_CODE_SUCCESS) {
-      return code;
-    }
-
-    code = doAppendRowFromTSRow(pReader->resBlockInfo.pResBlock, pReader, pTSRow, pBlockScanInfo);
-
-    taosMemoryFree(pTSRow);
-    tsdbRowMergerClear(pMerger);
-    return code;
-
   } else {  // only last block exists
     return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false);
   }
@@ -2194,8 +2190,7 @@ static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, STableBlockScanI
   SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
 
   TSDBROW *pRow = NULL, *piRow = NULL;
-  int64_t key = (pBlockData->nRow > 0 && (!pDumpInfo->allDumped)) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] :
-                                                                   (ASCENDING_TRAVERSE(pReader->info.order) ? INT64_MAX : INT64_MIN);
+  int64_t key = (pBlockData->nRow > 0 && (!pDumpInfo->allDumped)) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] : INT64_MIN;
   if (pBlockScanInfo->iter.hasVal) {
     pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader);
   }
@@ -2569,18 +2564,9 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
 
     // load the last data block of current table
     STableBlockScanInfo* pScanInfo = *(STableBlockScanInfo**)pStatus->pTableIter;
-    if (pScanInfo == NULL) {
-      tsdbError("table Iter is null, invalid pScanInfo, try next table %s", pReader->idStr);
-      bool hasNexTable = moveToNextTable(pUidList, pStatus);
-      if (!hasNexTable) {
-        return TSDB_CODE_SUCCESS;
-      }
-
-      continue;
-    }
-
     if (pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pScanInfo->uid, sizeof(pScanInfo->uid))) {
       // reset the index in last block when handing a new file
+      //      doCleanupTableScanInfo(pScanInfo);
       bool hasNexTable = moveToNextTable(pUidList, pStatus);
       if (!hasNexTable) {
         return TSDB_CODE_SUCCESS;
@@ -2589,6 +2575,9 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
       continue;
     }
 
+    // reset the index in last block when handing a new file
+    //    doCleanupTableScanInfo(pScanInfo);
+
     bool hasDataInLastFile = initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
     if (!hasDataInLastFile) {
       bool hasNexTable = moveToNextTable(pUidList, pStatus);
@@ -2678,32 +2667,16 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
         (ASCENDING_TRAVERSE(pReader->info.order)) ? pBlockInfo->record.firstKey : pBlockInfo->record.lastKey;
     code = buildDataBlockFromBuf(pReader, pScanInfo, endKey);
   } else {
-    bool bHasDataInLastBlock = hasDataInLastBlock(pLastBlockReader);
-    int64_t tsLast = bHasDataInLastBlock ? getCurrentKeyInLastBlock(pLastBlockReader) : INT64_MIN;
-    if (!bHasDataInLastBlock || ((ASCENDING_TRAVERSE(pReader->info.order) && pBlockInfo->record.lastKey < tsLast) ||
-                                 (!ASCENDING_TRAVERSE(pReader->info.order) && pBlockInfo->record.firstKey > tsLast))) {
-      // whole block is required, return it directly
-      SDataBlockInfo* pInfo = &pReader->resBlockInfo.pResBlock->info;
-      pInfo->rows = pBlockInfo->record.numRow;
-      pInfo->id.uid = pScanInfo->uid;
-      pInfo->dataLoad = 0;
-      pInfo->window = (STimeWindow){.skey = pBlockInfo->record.firstKey, .ekey = pBlockInfo->record.lastKey};
-      setComposedBlockFlag(pReader, false);
-      setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlockInfo->record.lastKey, pReader->info.order);
+    if (hasDataInLastBlock(pLastBlockReader) && !ASCENDING_TRAVERSE(pReader->info.order)) {
+      // only return the rows in last block
+      int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
+      ASSERT(tsLast >= pBlockInfo->record.lastKey);
 
-      // update the last key for the corresponding table
-      pScanInfo->lastKey = ASCENDING_TRAVERSE(pReader->info.order) ? pInfo->window.ekey : pInfo->window.skey;
-      tsdbDebug("%p uid:%" PRIu64
-                " clean file block retrieved from file, global index:%d, "
-                "table index:%d, rows:%d, brange:%" PRId64 "-%" PRId64 ", %s",
-                pReader, pScanInfo->uid, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlockInfo->record.numRow,
-                pBlockInfo->record.firstKey, pBlockInfo->record.lastKey, pReader->idStr);
-    } else {
       SBlockData* pBData = &pReader->status.fileBlockData;
       tBlockDataReset(pBData);
 
       SSDataBlock* pResBlock = pReader->resBlockInfo.pResBlock;
-      tsdbDebug("load data in last block firstly %s", pReader->idStr);
+      tsdbDebug("load data in last block firstly, due to desc scan data, %s", pReader->idStr);
 
       int64_t st = taosGetTimestampUs();
 
@@ -2734,8 +2707,23 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
                   pReader, pResBlock->info.id.uid, pResBlock->info.window.skey, pResBlock->info.window.ekey,
                   pResBlock->info.rows, el, pReader->idStr);
       }
-    }
+    } else {  // whole block is required, return it directly
+      SDataBlockInfo* pInfo = &pReader->resBlockInfo.pResBlock->info;
+      pInfo->rows = pBlockInfo->record.numRow;
+      pInfo->id.uid = pScanInfo->uid;
+      pInfo->dataLoad = 0;
+      pInfo->window = (STimeWindow){.skey = pBlockInfo->record.firstKey, .ekey = pBlockInfo->record.lastKey};
+      setComposedBlockFlag(pReader, false);
+      setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlockInfo->record.lastKey, pReader->info.order);
 
+      // update the last key for the corresponding table
+      pScanInfo->lastKey = ASCENDING_TRAVERSE(pReader->info.order) ? pInfo->window.ekey : pInfo->window.skey;
+      tsdbDebug("%p uid:%" PRIu64
+                " clean file block retrieved from file, global index:%d, "
+                "table index:%d, rows:%d, brange:%" PRId64 "-%" PRId64 ", %s",
+                pReader, pScanInfo->uid, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlockInfo->record.numRow,
+                pBlockInfo->record.firstKey, pBlockInfo->record.lastKey, pReader->idStr);
+    }
   }
 
   return (pReader->code != TSDB_CODE_SUCCESS) ? pReader->code : code;
@@ -4109,6 +4097,11 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
 
     tsdbDataFileReaderClose(&pReader->pFileReader);
 
+    int64_t loadBlocks = 0;
+    double  elapse = 0;
+    pReader->status.pLDataIterArray = destroySttBlockReader(pReader->status.pLDataIterArray, &loadBlocks, &elapse);
+    pReader->status.pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
+
     // resetDataBlockScanInfo excluding lastKey
     STableBlockScanInfo** p = NULL;
     int32_t               iter = 0;
@@ -4179,7 +4172,7 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
     }
   }
 
-  tsdbUntakeReadSnap(pReader, pReader->pReadSnap, false);
+  tsdbUntakeReadSnap2(pReader, pReader->pReadSnap, false);
   pReader->pReadSnap = NULL;
   pReader->flag = READER_STATUS_SUSPEND;
 

From 9981cecbed4b02f0e64d34d8c42efaf657740d95 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 31 Jul 2023 15:24:35 +0800
Subject: [PATCH 15/20] refactor: do some internal refactor.

---
 include/libs/executor/executor.h       |   6 +-
 include/libs/stream/tstream.h          |   2 -
 source/dnode/vnode/src/tq/tq.c         |   2 +-
 source/libs/executor/inc/querytask.h   |   4 +-
 source/libs/executor/src/executor.c    |  37 ++-----
 source/libs/stream/src/streamExec.c    | 132 +++++++------------------
 source/libs/stream/src/streamRecover.c |  12 +--
 7 files changed, 46 insertions(+), 149 deletions(-)

diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h
index 3bef15f3a7..f90c38f341 100644
--- a/include/libs/executor/executor.h
+++ b/include/libs/executor/executor.h
@@ -221,13 +221,9 @@ int32_t qStreamSourceScanParamForHistoryScanStep2(qTaskInfo_t tinfo, SVersionRan
 int32_t qStreamRecoverFinish(qTaskInfo_t tinfo);
 int32_t qRestoreStreamOperatorOption(qTaskInfo_t tinfo);
 bool    qStreamRecoverScanFinished(qTaskInfo_t tinfo);
-bool    qStreamRecoverScanStep1Finished(qTaskInfo_t tinfo);
-bool    qStreamRecoverScanStep2Finished(qTaskInfo_t tinfo);
-int32_t qStreamRecoverSetAllStepFinished(qTaskInfo_t tinfo);
+int32_t qStreamInfoResetTimewindowFilter(qTaskInfo_t tinfo);
 void    resetTaskInfo(qTaskInfo_t tinfo);
 
-void qResetStreamInfoTimeWindow(qTaskInfo_t tinfo);
-
 int32_t qStreamOperatorReleaseState(qTaskInfo_t tInfo);
 int32_t qStreamOperatorReloadState(qTaskInfo_t tInfo);
 
diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
index 066f83fbcb..9d3a42f502 100644
--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -607,8 +607,6 @@ int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask);
 int32_t streamStartRecoverTask(SStreamTask* pTask, int8_t igUntreated);
 bool    streamHistoryTaskSetVerRangeStep2(SStreamTask* pTask, int64_t latestVer);
 
-bool    streamTaskRecoverScanStep1Finished(SStreamTask* pTask);
-bool    streamTaskRecoverScanStep2Finished(SStreamTask* pTask);
 int32_t streamTaskRecoverSetAllStepFinished(SStreamTask* pTask);
 
 // common
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index db9ef7805f..bd4a35d39c 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -1296,7 +1296,7 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
           "s-task:%s scan-history in stream time window completed, no related fill-history task, reset the time "
           "window:%" PRId64 " - %" PRId64,
           id, pWindow->skey, pWindow->ekey);
-      qResetStreamInfoTimeWindow(pTask->exec.pExecutor);
+      qStreamInfoResetTimewindowFilter(pTask->exec.pExecutor);
     } else {
       // when related fill-history task exists, update the fill-history time window only when the
       // state transfer is completed.
diff --git a/source/libs/executor/inc/querytask.h b/source/libs/executor/inc/querytask.h
index cdf37bcc6b..7241b015a0 100644
--- a/source/libs/executor/inc/querytask.h
+++ b/source/libs/executor/inc/querytask.h
@@ -62,8 +62,8 @@ typedef struct {
   SSchemaWrapper*      schema;
   char                 tbName[TSDB_TABLE_NAME_LEN];  // this is the current scan table: todo refactor
   int8_t               recoverStep;
-  bool                 recoverStep1Finished;
-  bool                 recoverStep2Finished;
+//  bool                 recoverStep1Finished;
+//  bool                 recoverStep2Finished;
   int8_t               recoverScanFinished;
   SQueryTableDataCond  tableCond;
   SVersionRange        fillHistoryVer;
diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c
index 5a99c1ea9a..e4ddf9ca6c 100644
--- a/source/libs/executor/src/executor.c
+++ b/source/libs/executor/src/executor.c
@@ -116,17 +116,6 @@ void resetTaskInfo(qTaskInfo_t tinfo) {
   clearStreamBlock(pTaskInfo->pRoot);
 }
 
-void qResetStreamInfoTimeWindow(qTaskInfo_t tinfo) {
-  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) tinfo;
-  if (pTaskInfo == NULL) {
-    return;
-  }
-
-  qDebug("%s set stream fill-history window:%" PRId64"-%"PRId64, GET_TASKID(pTaskInfo), INT64_MIN, INT64_MAX);
-  pTaskInfo->streamInfo.fillHistoryWindow.skey = INT64_MIN;
-  pTaskInfo->streamInfo.fillHistoryWindow.ekey = INT64_MAX;
-}
-
 static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, const char* id) {
   if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
     if (pOperator->numOfDownstream == 0) {
@@ -341,7 +330,7 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers, int32_t v
     return NULL;
   }
 
-  qResetStreamInfoTimeWindow(pTaskInfo);
+  qStreamInfoResetTimewindowFilter(pTaskInfo);
   return pTaskInfo;
 }
 
@@ -891,8 +880,6 @@ int32_t qStreamSourceScanParamForHistoryScanStep1(qTaskInfo_t tinfo, SVersionRan
   pStreamInfo->fillHistoryVer = *pVerRange;
   pStreamInfo->fillHistoryWindow = *pWindow;
   pStreamInfo->recoverStep = STREAM_RECOVER_STEP__PREPARE1;
-  pStreamInfo->recoverStep1Finished = false;
-  pStreamInfo->recoverStep2Finished = false;
 
   qDebug("%s step 1. set param for stream scanner for scan-history data, verRange:%" PRId64 " - %" PRId64 ", window:%" PRId64
          " - %" PRId64,
@@ -910,8 +897,6 @@ int32_t qStreamSourceScanParamForHistoryScanStep2(qTaskInfo_t tinfo, SVersionRan
   pStreamInfo->fillHistoryVer = *pVerRange;
   pStreamInfo->fillHistoryWindow = *pWindow;
   pStreamInfo->recoverStep = STREAM_RECOVER_STEP__PREPARE2;
-  pStreamInfo->recoverStep1Finished = true;
-  pStreamInfo->recoverStep2Finished = false;
 
   qDebug("%s step 2. set param for stream scanner for scan-history data, verRange:%" PRId64 " - %" PRId64
          ", window:%" PRId64 " - %" PRId64,
@@ -1050,23 +1035,15 @@ bool qStreamRecoverScanFinished(qTaskInfo_t tinfo) {
   return pTaskInfo->streamInfo.recoverScanFinished;
 }
 
-bool qStreamRecoverScanStep1Finished(qTaskInfo_t tinfo) {
+int32_t qStreamInfoResetTimewindowFilter(qTaskInfo_t tinfo) {
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
-  return pTaskInfo->streamInfo.recoverStep1Finished;
-}
+  STimeWindow* pWindow = &pTaskInfo->streamInfo.fillHistoryWindow;
 
-bool qStreamRecoverScanStep2Finished(qTaskInfo_t tinfo) {
-  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
-  return pTaskInfo->streamInfo.recoverStep2Finished;
-}
+  qDebug("%s set remove scan-history filter window:%" PRId64 "-%" PRId64 ", new window:%" PRId64 "-%" PRId64,
+         GET_TASKID(pTaskInfo), pWindow->skey, pWindow->ekey, INT64_MIN, INT64_MAX);
 
-int32_t qStreamRecoverSetAllStepFinished(qTaskInfo_t tinfo) {
-  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
-  pTaskInfo->streamInfo.recoverStep1Finished = true;
-  pTaskInfo->streamInfo.recoverStep2Finished = true;
-
-  // reset the time window
-  pTaskInfo->streamInfo.fillHistoryWindow.skey = INT64_MIN;
+  pWindow->skey = INT64_MIN;
+  pWindow->ekey = INT64_MAX;
   return 0;
 }
 
diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index 4ef7d6084d..763a2fdcc6 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -163,15 +163,14 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, i
 }
 
 int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
-  int32_t code = 0;
-
   ASSERT(pTask->info.taskLevel == TASK_LEVEL__SOURCE);
-  void* exec = pTask->exec.pExecutor;
+  int32_t code = TSDB_CODE_SUCCESS;
+  void*   exec = pTask->exec.pExecutor;
+  bool    finished = false;
 
   qSetStreamOpOpen(exec);
-  bool finished = false;
 
-  while (1) {
+  while (!finished) {
     if (streamTaskShouldPause(&pTask->status)) {
       double el = (taosGetTimestampMs() - pTask->tsInfo.step1Start) / 1000.0;
       qDebug("s-task:%s paused from the scan-history task, elapsed time:%.2fsec", pTask->id.idStr, el);
@@ -184,44 +183,30 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
       return -1;
     }
 
-    int32_t batchCnt = 0;
+    int32_t numOfBlocks = 0;
     while (1) {
       if (streamTaskShouldStop(&pTask->status)) {
         taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
         return 0;
       }
 
+      if (streamTaskShouldPause(&pTask->status)) {
+        break;
+      }
+
       SSDataBlock* output = NULL;
       uint64_t     ts = 0;
       if (qExecTask(exec, &output, &ts) < 0) {
         continue;
       }
 
-      if (output == NULL) {
-        if (qStreamRecoverScanFinished(exec)) {
-          finished = true;
-        } else {
-          qSetStreamOpOpen(exec);
-          if (streamTaskShouldPause(&pTask->status)) {
-            SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, 0);
-            if (qRes == NULL) {
-              taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
-              terrno = TSDB_CODE_OUT_OF_MEMORY;
-              return -1;
-            }
-
-            qRes->type = STREAM_INPUT__DATA_BLOCK;
-            qRes->blocks = pRes;
-            code = streamTaskOutputResultBlock(pTask, qRes);
-            if (code == TSDB_CODE_UTIL_QUEUE_OUT_OF_MEMORY) {
-              taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
-              taosFreeQitem(qRes);
-              return code;
-            }
-            return 0;
-          }
-        }
+      if (output == NULL && qStreamRecoverScanFinished(exec)) {
+        finished = true;
         break;
+      } else {
+        if (output == NULL) {
+          ASSERT(0);
+        }
       }
 
       SSDataBlock block = {0};
@@ -229,86 +214,37 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
       block.info.childId = pTask->info.selfChildId;
       taosArrayPush(pRes, &block);
 
-      batchCnt++;
-
-      qDebug("s-task:%s scan exec numOfBlocks:%d, limit:%d", pTask->id.idStr, batchCnt, batchSz);
-      if (batchCnt >= batchSz) {
+      numOfBlocks++;
+      qDebug("s-task:%s scan exec numOfBlocks:%d, limit:%d", pTask->id.idStr, numOfBlocks, batchSz);
+      if (numOfBlocks >= batchSz) {
         break;
       }
     }
 
-    if (taosArrayGetSize(pRes) == 0) {
-      taosArrayDestroy(pRes);
-
-      if (finished) {
-        qDebug("s-task:%s finish recover exec task ", pTask->id.idStr);
-        break;
-      } else {
-        qDebug("s-task:%s continue recover exec task ", pTask->id.idStr);
-        continue;
+    if (taosArrayGetSize(pRes) > 0) {
+      SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, 0);
+      if (qRes == NULL) {
+        taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        return -1;
       }
-    }
 
-    SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, 0);
-    if (qRes == NULL) {
-      taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      return -1;
-    }
+      qRes->type = STREAM_INPUT__DATA_BLOCK;
+      qRes->blocks = pRes;
 
-    qRes->type = STREAM_INPUT__DATA_BLOCK;
-    qRes->blocks = pRes;
-    code = streamTaskOutputResultBlock(pTask, qRes);
-    if (code == TSDB_CODE_UTIL_QUEUE_OUT_OF_MEMORY) {
-      taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
-      taosFreeQitem(qRes);
-      return code;
-    }
-
-    if (finished) {
-      break;
-    }
-  }
-  return 0;
-}
-
-#if 0
-int32_t streamBatchExec(SStreamTask* pTask, int32_t batchLimit) {
-  // fetch all queue item, merge according to batchLimit
-  int32_t numOfItems = taosReadAllQitems(pTask->inputQueue1, pTask->inputQall);
-  if (numOfItems == 0) {
-    qDebug("task: %d, stream task exec over, queue empty", pTask->id.taskId);
-    return 0;
-  }
-  SStreamQueueItem* pMerged = NULL;
-  SStreamQueueItem* pItem = NULL;
-  taosGetQitem(pTask->inputQall, (void**)&pItem);
-  if (pItem == NULL) {
-    if (pMerged != NULL) {
-      // process merged item
+      code = streamTaskOutputResultBlock(pTask, qRes);
+      if (code == TSDB_CODE_UTIL_QUEUE_OUT_OF_MEMORY) {
+        taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
+        taosFreeQitem(qRes);
+        return code;
+      }
     } else {
-      return 0;
+      taosArrayDestroy(pRes);
     }
   }
 
-  // if drop
-  if (pItem->type == STREAM_INPUT__DESTROY) {
-    // set status drop
-    return -1;
-  }
-
-  if (pTask->info.taskLevel == TASK_LEVEL__SINK) {
-    ASSERT(((SStreamQueueItem*)pItem)->type == STREAM_INPUT__DATA_BLOCK);
-    streamTaskOutputResultBlock(pTask, (SStreamDataBlock*)pItem);
-  }
-
-  // exec impl
-
-  // output
-  // try dispatch
   return 0;
 }
-#endif
 
 int32_t updateCheckPointInfo(SStreamTask* pTask) {
   int64_t ckId = 0;
@@ -404,7 +340,7 @@ static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
 
   // expand the query time window for stream scanner
   pTimeWindow->skey = INT64_MIN;
-  qResetStreamInfoTimeWindow(pStreamTask->exec.pExecutor);
+  qStreamInfoResetTimewindowFilter(pStreamTask->exec.pExecutor);
 
   // transfer the ownership of executor state
   streamTaskReleaseState(pTask);
diff --git a/source/libs/stream/src/streamRecover.c b/source/libs/stream/src/streamRecover.c
index df45ff2759..4fa6150412 100644
--- a/source/libs/stream/src/streamRecover.c
+++ b/source/libs/stream/src/streamRecover.c
@@ -647,19 +647,9 @@ int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask) {
   return 0;
 }
 
-bool streamTaskRecoverScanStep1Finished(SStreamTask* pTask) {
-  void* exec = pTask->exec.pExecutor;
-  return qStreamRecoverScanStep1Finished(exec);
-}
-
-bool streamTaskRecoverScanStep2Finished(SStreamTask* pTask) {
-  void* exec = pTask->exec.pExecutor;
-  return qStreamRecoverScanStep2Finished(exec);
-}
-
 int32_t streamTaskRecoverSetAllStepFinished(SStreamTask* pTask) {
   void* exec = pTask->exec.pExecutor;
-  return qStreamRecoverSetAllStepFinished(exec);
+  return qStreamInfoResetTimewindowFilter(exec);
 }
 
 bool streamHistoryTaskSetVerRangeStep2(SStreamTask* pTask, int64_t latestVer) {

From 7619bbcf22ae75fc634a07bd1a9da3622869f786 Mon Sep 17 00:00:00 2001
From: "chao.feng" <cfeng@taosdata.com>
Date: Mon, 31 Jul 2023 16:58:19 +0800
Subject: [PATCH 16/20] Update the doc content of taos-sql/22-meta.md to add
 ins_user_privilege by charles

---
 docs/en/12-taos-sql/22-meta.md | 10 ++++++++++
 docs/zh/12-taos-sql/22-meta.md | 10 ++++++++++
 2 files changed, 20 insertions(+)

diff --git a/docs/en/12-taos-sql/22-meta.md b/docs/en/12-taos-sql/22-meta.md
index 2d7c517dca..37304633e7 100644
--- a/docs/en/12-taos-sql/22-meta.md
+++ b/docs/en/12-taos-sql/22-meta.md
@@ -297,3 +297,13 @@ Provides dnode configuration information.
 | 7   | target_table | BINARY(192)   | Target table                                                                                                                                                                                   |
 | 8   |  watermark   | BIGINT        | Watermark (see stream processing documentation). It should be noted that `watermark` is a TDengine keyword and needs to be escaped with ` when used as a column name.                          |
 | 9   |   trigger    | INT           | Method of triggering the result push (see stream processing documentation). It should be noted that `trigger` is a TDengine keyword and needs to be escaped with ` when used as a column name. |
+
+## INS_USER_PRIVILEGES
+
+| #   |   **Column**   | **Data Type** | **Description**                         |**                                                                               |
+| --- | :----------: | ------------ | -------------------------------------------|
+| 1   | user_name    | VARCHAR(24)       | Username                              |
+| 2   | privilege    | VARCHAR(10)       | Privilege description                 |
+| 3   | db_name      | VARCHAR(65)       | Database name                         |
+| 4   | table_name   | VARCHAR(193)      | Table name                            |
+| 5   | condition    | VARCHAR(49152)    | The privilege filter for child tables |
diff --git a/docs/zh/12-taos-sql/22-meta.md b/docs/zh/12-taos-sql/22-meta.md
index a5396d8607..35794ec269 100644
--- a/docs/zh/12-taos-sql/22-meta.md
+++ b/docs/zh/12-taos-sql/22-meta.md
@@ -298,3 +298,13 @@ TDengine 内置了一个名为 `INFORMATION_SCHEMA` 的数据库，提供对数
 | 7   | target_table | BINARY(192)  | 流计算写入的目标表                                                                                                   |
 | 8   |  watermark   | BIGINT       | watermark，详见 SQL 手册流式计算。需要注意，`watermark` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。      |
 | 9   |   trigger    | INT          | 计算结果推送模式，详见 SQL 手册流式计算。需要注意，`trigger` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。 |
+
+## INS_USER_PRIVILEGES
+
+| #   |   **列名**    | **数据类型** | **说明**                                                                                                             |
+| --- | :----------: | ------------ | -------------------------------------------------------------------------------------------------------------------- |
+| 1   | user_name    | VARCHAR(24)       | 用户名
+| 2   | privilege    | VARCHAR(10)       | 权限描述
+| 3   | db_name      | VARCHAR(65)       | 数据库名称
+| 4   | table_name   | VARCHAR(193)      | 表名称
+| 5   | condition    | VARCHAR(49152)    | 子表权限过滤条件

From 32b59af4b18a19ae749042f55d4be91c9769c6d7 Mon Sep 17 00:00:00 2001
From: liuyao <54liuyao@163.com>
Date: Mon, 31 Jul 2023 17:04:41 +0800
Subject: [PATCH 17/20] check update data

---
 include/libs/executor/storageapi.h        | 2 ++
 include/libs/stream/tstreamUpdate.h       | 2 ++
 source/dnode/snode/src/snodeInitApi.c     | 2 ++
 source/dnode/vnode/src/vnd/vnodeInitApi.c | 2 ++
 source/libs/executor/src/scanoperator.c   | 4 +++-
 source/libs/stream/src/streamUpdate.c     | 4 ++--
 6 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/include/libs/executor/storageapi.h b/include/libs/executor/storageapi.h
index 2fbd7851e8..773f373a2d 100644
--- a/include/libs/executor/storageapi.h
+++ b/include/libs/executor/storageapi.h
@@ -368,6 +368,8 @@ typedef struct SStateStore {
   bool (*updateInfoIsUpdated)(SUpdateInfo* pInfo, uint64_t tableId, TSKEY ts);
   bool (*updateInfoIsTableInserted)(SUpdateInfo* pInfo, int64_t tbUid);
   void (*updateInfoDestroy)(SUpdateInfo* pInfo);
+  void (*windowSBfDelete)(SUpdateInfo *pInfo, uint64_t count);
+  void (*windowSBfAdd)(SUpdateInfo *pInfo, uint64_t count);
 
   SUpdateInfo* (*updateInfoInitP)(SInterval* pInterval, int64_t watermark);
   void (*updateInfoAddCloseWindowSBF)(SUpdateInfo* pInfo);
diff --git a/include/libs/stream/tstreamUpdate.h b/include/libs/stream/tstreamUpdate.h
index bd5a3be8de..7bb1d027c9 100644
--- a/include/libs/stream/tstreamUpdate.h
+++ b/include/libs/stream/tstreamUpdate.h
@@ -53,6 +53,8 @@ void         updateInfoAddCloseWindowSBF(SUpdateInfo *pInfo);
 void         updateInfoDestoryColseWinSBF(SUpdateInfo *pInfo);
 int32_t      updateInfoSerialize(void *buf, int32_t bufLen, const SUpdateInfo *pInfo);
 int32_t      updateInfoDeserialize(void *buf, int32_t bufLen, SUpdateInfo *pInfo);
+void         windowSBfDelete(SUpdateInfo *pInfo, uint64_t count);
+void         windowSBfAdd(SUpdateInfo *pInfo, uint64_t count);
 
 #ifdef __cplusplus
 }
diff --git a/source/dnode/snode/src/snodeInitApi.c b/source/dnode/snode/src/snodeInitApi.c
index c046505630..e737e3fa37 100644
--- a/source/dnode/snode/src/snodeInitApi.c
+++ b/source/dnode/snode/src/snodeInitApi.c
@@ -78,6 +78,8 @@ void initStateStoreAPI(SStateStore* pStore) {
   pStore->updateInfoIsUpdated = updateInfoIsUpdated;
   pStore->updateInfoIsTableInserted = updateInfoIsTableInserted;
   pStore->updateInfoDestroy = updateInfoDestroy;
+  pStore->windowSBfDelete = windowSBfDelete;
+  pStore->windowSBfAdd = windowSBfAdd;
 
   pStore->updateInfoInitP = updateInfoInitP;
   pStore->updateInfoAddCloseWindowSBF = updateInfoAddCloseWindowSBF;
diff --git a/source/dnode/vnode/src/vnd/vnodeInitApi.c b/source/dnode/vnode/src/vnd/vnodeInitApi.c
index 3dfaa28c09..5c8d563d73 100644
--- a/source/dnode/vnode/src/vnd/vnodeInitApi.c
+++ b/source/dnode/vnode/src/vnd/vnodeInitApi.c
@@ -180,6 +180,8 @@ void initStateStoreAPI(SStateStore* pStore) {
   pStore->updateInfoIsUpdated = updateInfoIsUpdated;
   pStore->updateInfoIsTableInserted = updateInfoIsTableInserted;
   pStore->updateInfoDestroy = updateInfoDestroy;
+  pStore->windowSBfDelete = windowSBfDelete;
+  pStore->windowSBfAdd = windowSBfAdd;
 
   pStore->updateInfoInitP = updateInfoInitP;
   pStore->updateInfoAddCloseWindowSBF = updateInfoAddCloseWindowSBF;
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index 7434db61db..6331f4bd35 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -2411,7 +2411,9 @@ void streamScanReloadState(SOperatorInfo* pOperator) {
       pInfo->stateStore.updateInfoDestroy(pInfo->pUpdateInfo);
       pInfo->pUpdateInfo = pUpInfo;
     } else {
-      pInfo->pUpdateInfo->minTS = TMAX(pInfo->pUpdateInfo->minTS, pUpInfo->minTS);
+      pInfo->stateStore.windowSBfDelete(pInfo->pUpdateInfo, 1);
+      pInfo->stateStore.windowSBfAdd(pInfo->pUpdateInfo, 1);
+      ASSERT(pInfo->pUpdateInfo->minTS > pUpInfo->minTS);
       pInfo->pUpdateInfo->maxDataVersion = TMAX(pInfo->pUpdateInfo->maxDataVersion, pUpInfo->maxDataVersion);
       SHashObj* curMap = pInfo->pUpdateInfo->pMap;
       void *pIte = taosHashIterate(curMap, NULL);
diff --git a/source/libs/stream/src/streamUpdate.c b/source/libs/stream/src/streamUpdate.c
index 85be120dbd..7a8de91d77 100644
--- a/source/libs/stream/src/streamUpdate.c
+++ b/source/libs/stream/src/streamUpdate.c
@@ -33,7 +33,7 @@
 
 static int64_t adjustExpEntries(int64_t entries) { return TMIN(DEFAULT_EXPECTED_ENTRIES, entries); }
 
-static void windowSBfAdd(SUpdateInfo *pInfo, uint64_t count) {
+void windowSBfAdd(SUpdateInfo *pInfo, uint64_t count) {
   if (pInfo->numSBFs < count) {
     count = pInfo->numSBFs;
   }
@@ -49,7 +49,7 @@ static void clearItemHelper(void *p) {
   tScalableBfDestroy(*pBf);
 }
 
-static void windowSBfDelete(SUpdateInfo *pInfo, uint64_t count) {
+void windowSBfDelete(SUpdateInfo *pInfo, uint64_t count) {
   if (count < pInfo->numSBFs) {
     for (uint64_t i = 0; i < count; ++i) {
       SScalableBf *pTsSBFs = taosArrayGetP(pInfo->pTsSBFs, 0);

From d45596bb33daf19347d2fdc46215994abd860433 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 31 Jul 2023 18:07:37 +0800
Subject: [PATCH 18/20] fix(stream): kill task when pause the stream execution.

---
 include/libs/stream/tstream.h          |  6 +--
 source/dnode/vnode/src/tq/tq.c         | 14 +++----
 source/libs/stream/inc/streamInt.h     |  1 +
 source/libs/stream/src/streamExec.c    |  2 +-
 source/libs/stream/src/streamRecover.c | 51 +++++++++++++++-----------
 5 files changed, 37 insertions(+), 37 deletions(-)

diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
index 9d3a42f502..db0509d81d 100644
--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -604,13 +604,10 @@ int32_t streamSendCheckRsp(const SStreamMeta* pMeta, const SStreamTaskCheckReq*
 int32_t streamProcessCheckRsp(SStreamTask* pTask, const SStreamTaskCheckRsp* pRsp);
 int32_t streamLaunchFillHistoryTask(SStreamTask* pTask);
 int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask);
-int32_t streamStartRecoverTask(SStreamTask* pTask, int8_t igUntreated);
+int32_t streamStartScanHistoryAsync(SStreamTask* pTask, int8_t igUntreated);
 bool    streamHistoryTaskSetVerRangeStep2(SStreamTask* pTask, int64_t latestVer);
 
-int32_t streamTaskRecoverSetAllStepFinished(SStreamTask* pTask);
-
 // common
-int32_t     streamSetParamForScanHistory(SStreamTask* pTask);
 int32_t     streamRestoreParam(SStreamTask* pTask);
 int32_t     streamSetStatusNormal(SStreamTask* pTask);
 const char* streamGetTaskStatusStr(int32_t status);
@@ -624,7 +621,6 @@ void        streamTaskEnablePause(SStreamTask* pTask);
 // source level
 int32_t streamSetParamForStreamScannerStep1(SStreamTask* pTask, SVersionRange* pVerRange, STimeWindow* pWindow);
 int32_t streamSetParamForStreamScannerStep2(SStreamTask* pTask, SVersionRange* pVerRange, STimeWindow* pWindow);
-int32_t streamBuildSourceRecover1Req(SStreamTask* pTask, SStreamScanHistoryReq* pReq, int8_t igUntreated);
 int32_t streamSourceScanHistoryData(SStreamTask* pTask);
 int32_t streamDispatchScanHistoryFinishMsg(SStreamTask* pTask);
 
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index bd4a35d39c..809de13bcf 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -1567,9 +1567,8 @@ int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
   SStreamMeta* pMeta = pTq->pStreamMeta;
   SStreamTask* pTask = streamMetaAcquireTask(pMeta, pReq->taskId);
   if (pTask == NULL) {
-    tqError("vgId:%d failed to acquire task:0x%x, it may have been dropped already", pMeta->vgId,
+    tqError("vgId:%d process pause req, failed to acquire task:0x%x, it may have been dropped already", pMeta->vgId,
             pReq->taskId);
-
     // since task is in [STOP|DROPPING] state, it is safe to assume the pause is active
     return TSDB_CODE_SUCCESS;
   }
@@ -1581,9 +1580,8 @@ int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
   if (pTask->historyTaskId.taskId != 0) {
     pHistoryTask = streamMetaAcquireTask(pMeta, pTask->historyTaskId.taskId);
     if (pHistoryTask == NULL) {
-      tqError("vgId:%d failed to acquire fill-history task:0x%x, it may have been dropped already. Pause success",
+      tqError("vgId:%d process pause req, failed to acquire fill-history task:0x%x, it may have been dropped already",
               pMeta->vgId, pTask->historyTaskId.taskId);
-
       streamMetaReleaseTask(pMeta, pTask);
 
       // since task is in [STOP|DROPPING] state, it is safe to assume the pause is active
@@ -1591,14 +1589,12 @@ int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
     }
 
     tqDebug("s-task:%s fill-history task handle paused along with related stream task", pHistoryTask->id.idStr);
-    streamTaskPause(pHistoryTask);
-  }
 
-  streamMetaReleaseTask(pMeta, pTask);
-  if (pHistoryTask != NULL) {
+    streamTaskPause(pHistoryTask);
     streamMetaReleaseTask(pMeta, pHistoryTask);
   }
 
+  streamMetaReleaseTask(pMeta, pTask);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -1627,7 +1623,7 @@ int32_t tqProcessTaskResumeImpl(STQ* pTq, SStreamTask* pTask, int64_t sversion,
     }
 
     if (level == TASK_LEVEL__SOURCE && pTask->info.fillHistory && pTask->status.taskStatus == TASK_STATUS__SCAN_HISTORY) {
-      streamStartRecoverTask(pTask, igUntreated);
+      streamStartScanHistoryAsync(pTask, igUntreated);
     } else if (level == TASK_LEVEL__SOURCE && (taosQueueItemSize(pTask->inputQueue->queue) == 0)) {
       tqStartStreamTasks(pTq);
     } else {
diff --git a/source/libs/stream/inc/streamInt.h b/source/libs/stream/inc/streamInt.h
index add893c8c7..32d6dc65d9 100644
--- a/source/libs/stream/inc/streamInt.h
+++ b/source/libs/stream/inc/streamInt.h
@@ -62,6 +62,7 @@ SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem*
 
 int32_t streamAddEndScanHistoryMsg(SStreamTask* pTask, SRpcHandleInfo* pRpcInfo, SStreamScanHistoryFinishReq* pReq);
 int32_t streamNotifyUpstreamContinue(SStreamTask* pTask);
+int32_t streamTaskFillHistoryFinished(SStreamTask* pTask);
 
 extern int32_t streamBackendId;
 extern int32_t streamBackendCfWrapperId;
diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index 763a2fdcc6..7aa82ed5f6 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -561,7 +561,7 @@ int32_t streamTryExec(SStreamTask* pTask) {
     if (taosQueueEmpty(pTask->inputQueue->queue)) {
       // fill-history WAL scan has completed
       if (pTask->info.taskLevel == TASK_LEVEL__SOURCE && pTask->status.transferState == true) {
-        streamTaskRecoverSetAllStepFinished(pTask);
+        streamTaskFillHistoryFinished(pTask);
         streamTaskEndScanWAL(pTask);
       } else {
         atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
diff --git a/source/libs/stream/src/streamRecover.c b/source/libs/stream/src/streamRecover.c
index 4fa6150412..bd2d67e14a 100644
--- a/source/libs/stream/src/streamRecover.c
+++ b/source/libs/stream/src/streamRecover.c
@@ -17,23 +17,30 @@
 #include "ttimer.h"
 #include "wal.h"
 
-static void launchFillHistoryTask(SStreamTask* pTask);
-static void streamTaskSetRangeStreamCalc(SStreamTask* pTask);
+typedef struct SStreamTaskRetryInfo {
+  SStreamMeta* pMeta;
+  int32_t taskId;
+} SStreamTaskRetryInfo;
 
-static void streamTaskSetForReady(SStreamTask* pTask, int32_t numOfReqs) {
+static int32_t streamSetParamForScanHistory(SStreamTask* pTask);
+static void    launchFillHistoryTask(SStreamTask* pTask);
+static void    streamTaskSetRangeStreamCalc(SStreamTask* pTask);
+static int32_t initScanHistoryReq(SStreamTask* pTask, SStreamScanHistoryReq* pReq, int8_t igUntreated);
+
+static void streamTaskSetReady(SStreamTask* pTask, int32_t numOfReqs) {
   ASSERT(pTask->status.downstreamReady == 0);
   pTask->status.downstreamReady = 1;
-  int64_t el = (taosGetTimestampMs() - pTask->tsInfo.init);
 
+  int64_t el = (taosGetTimestampMs() - pTask->tsInfo.init);
   qDebug("s-task:%s all %d downstream ready, init completed, elapsed time:%dms, task status:%s",
          pTask->id.idStr, numOfReqs, (int32_t) el, streamGetTaskStatusStr(pTask->status.taskStatus));
 }
 
-int32_t streamStartRecoverTask(SStreamTask* pTask, int8_t igUntreated) {
+int32_t streamStartScanHistoryAsync(SStreamTask* pTask, int8_t igUntreated) {
   SStreamScanHistoryReq req;
-  streamBuildSourceRecover1Req(pTask, &req, igUntreated);
-  int32_t len = sizeof(SStreamScanHistoryReq);
+  initScanHistoryReq(pTask, &req, igUntreated);
 
+  int32_t len = sizeof(SStreamScanHistoryReq);
   void* serializedReq = rpcMallocCont(len);
   if (serializedReq == NULL) {
     return -1;
@@ -65,9 +72,9 @@ static int32_t doLaunchScanHistoryTask(SStreamTask* pTask) {
   if (pTask->info.fillHistory) {
     streamSetParamForScanHistory(pTask);
   }
-  streamSetParamForStreamScannerStep1(pTask, pRange, &pTask->dataRange.window);
 
-  int32_t code = streamStartRecoverTask(pTask, 0);
+  streamSetParamForStreamScannerStep1(pTask, pRange, &pTask->dataRange.window);
+  int32_t code = streamStartScanHistoryAsync(pTask, 0);
   return code;
 }
 
@@ -142,7 +149,7 @@ int32_t streamTaskDoCheckDownstreamTasks(SStreamTask* pTask) {
   } else {
     qDebug("s-task:%s (vgId:%d) set downstream ready, since no downstream", pTask->id.idStr, pTask->info.nodeId);
 
-    streamTaskSetForReady(pTask, 0);
+    streamTaskSetReady(pTask, 0);
     streamTaskSetRangeStreamCalc(pTask);
     streamTaskLaunchScanHistory(pTask);
 
@@ -188,7 +195,7 @@ int32_t streamTaskCheckStatus(SStreamTask* pTask) {
 }
 
 static void doProcessDownstreamReadyRsp(SStreamTask* pTask, int32_t numOfReqs) {
-  streamTaskSetForReady(pTask, numOfReqs);
+  streamTaskSetReady(pTask, numOfReqs);
   const char* id = pTask->id.idStr;
 
   int8_t      status = pTask->status.taskStatus;
@@ -319,7 +326,7 @@ int32_t streamSetParamForStreamScannerStep2(SStreamTask* pTask, SVersionRange *p
   return qStreamSourceScanParamForHistoryScanStep2(pTask->exec.pExecutor, pVerRange, pWindow);
 }
 
-int32_t streamBuildSourceRecover1Req(SStreamTask* pTask, SStreamScanHistoryReq* pReq, int8_t igUntreated) {
+int32_t initScanHistoryReq(SStreamTask* pTask, SStreamScanHistoryReq* pReq, int8_t igUntreated) {
   pReq->msgHead.vgId = pTask->info.nodeId;
   pReq->streamId = pTask->id.streamId;
   pReq->taskId = pTask->id.taskId;
@@ -524,11 +531,6 @@ static void doCheckDownstreamStatus(SStreamTask* pTask, SStreamTask* pHTask) {
   streamTaskDoCheckDownstreamTasks(pHTask);
 }
 
-typedef struct SStreamTaskRetryInfo {
-  SStreamMeta* pMeta;
-  int32_t taskId;
-} SStreamTaskRetryInfo;
-
 static void tryLaunchHistoryTask(void* param, void* tmrId) {
   SStreamTaskRetryInfo* pInfo = param;
   SStreamMeta*          pMeta = pInfo->pMeta;
@@ -638,7 +640,7 @@ int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask) {
     }
   }
 
-  // dispatch recover finish req to all related downstream task
+  // dispatch scan-history finish req to all related downstream task
   code = streamDispatchScanHistoryFinishMsg(pTask);
   if (code < 0) {
     return -1;
@@ -647,7 +649,7 @@ int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask) {
   return 0;
 }
 
-int32_t streamTaskRecoverSetAllStepFinished(SStreamTask* pTask) {
+int32_t streamTaskFillHistoryFinished(SStreamTask* pTask) {
   void* exec = pTask->exec.pExecutor;
   return qStreamInfoResetTimewindowFilter(exec);
 }
@@ -659,7 +661,7 @@ bool streamHistoryTaskSetVerRangeStep2(SStreamTask* pTask, int64_t latestVer) {
   int64_t nextStartVer = pRange->maxVer + 1;
   if (nextStartVer > latestVer - 1) {
     // no input data yet. no need to execute the secondardy scan while stream task halt
-    streamTaskRecoverSetAllStepFinished(pTask);
+    streamTaskFillHistoryFinished(pTask);
     qDebug(
         "s-task:%s no need to perform secondary scan-history data(step 2), since no data ingest during step1 scan, "
         "related stream task currentVer:%" PRId64,
@@ -674,7 +676,6 @@ bool streamHistoryTaskSetVerRangeStep2(SStreamTask* pTask, int64_t latestVer) {
   }
 }
 
-
 int32_t tEncodeStreamTaskCheckReq(SEncoder* pEncoder, const SStreamTaskCheckReq* pReq) {
   if (tStartEncode(pEncoder) < 0) return -1;
   if (tEncodeI64(pEncoder, pReq->reqId) < 0) return -1;
@@ -847,7 +848,7 @@ void streamTaskPause(SStreamTask* pTask) {
     taosMsleep(100);
   }
 
-  // todo: use the lock of the task.
+  // todo: use the task lock, stead of meta lock
   taosWLockLatch(&pMeta->lock);
 
   status = pTask->status.taskStatus;
@@ -861,6 +862,12 @@ void streamTaskPause(SStreamTask* pTask) {
   atomic_store_8(&pTask->status.taskStatus, TASK_STATUS__PAUSE);
   taosWUnLockLatch(&pMeta->lock);
 
+  // in case of fill-history task, stop the tsdb file scan operation.
+  if (pTask->info.fillHistory == 1) {
+    void* pExecutor = pTask->exec.pExecutor;
+    qKillTask(pExecutor, TSDB_CODE_SUCCESS);
+  }
+
   int64_t el = taosGetTimestampMs() - st;
   qDebug("vgId:%d s-task:%s set pause flag, prev:%s, elapsed time:%dms", pMeta->vgId, pTask->id.idStr,
          streamGetTaskStatusStr(pTask->status.keepTaskStatus), (int32_t)el);

From d72580dad768517cdf8435ed19253dcdb79875d3 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 31 Jul 2023 19:12:23 +0800
Subject: [PATCH 19/20] fix(stream): transfer the state for agg tasks.

---
 source/libs/stream/src/streamExec.c | 58 ++++++++++++++++++++---------
 1 file changed, 40 insertions(+), 18 deletions(-)

diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index 7aa82ed5f6..e4dc28b76b 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -292,12 +292,12 @@ static void waitForTaskIdle(SStreamTask* pTask, SStreamTask* pStreamTask) {
   }
 }
 
-static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
+static int32_t streamDoTransferStateToStreamTask(SStreamTask* pTask) {
   SStreamMeta* pMeta = pTask->pMeta;
 
   SStreamTask* pStreamTask = streamMetaAcquireTask(pMeta, pTask->streamTaskId.taskId);
   if (pStreamTask == NULL) {
-    // todo: destroy this task here
+    // todo: destroy the fill-history task here
     qError("s-task:%s failed to find related stream task:0x%x, it may have been destroyed or closed", pTask->id.idStr,
            pTask->streamTaskId.taskId);
     return TSDB_CODE_STREAM_TASK_NOT_EXIST;
@@ -338,34 +338,36 @@ static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
     qDebug("s-task:%s no need to update time window for non-source task", pStreamTask->id.idStr);
   }
 
-  // expand the query time window for stream scanner
+  // 1. expand the query time window for stream task of WAL scanner
   pTimeWindow->skey = INT64_MIN;
   qStreamInfoResetTimewindowFilter(pStreamTask->exec.pExecutor);
 
-  // transfer the ownership of executor state
+  // 2. transfer the ownership of executor state
   streamTaskReleaseState(pTask);
   streamTaskReloadState(pStreamTask);
 
-  // clear the link between fill-history task and stream task info
+  // 3. clear the link between fill-history task and stream task info
   pStreamTask->historyTaskId.taskId = 0;
+
+  // 4. resume the state of stream task, after this function, the stream task will run immidately. But it can not be
+  // pause, since the pause allowed attribute is not set yet.
   streamTaskResumeFromHalt(pStreamTask);
 
   qDebug("s-task:%s fill-history task set status to be dropping, save the state into disk", pTask->id.idStr);
   int32_t taskId = pTask->id.taskId;
 
-  // free it and remove it from disk meta-store
+  // 5. free it and remove fill-history task from disk meta-store
   streamMetaUnregisterTask(pMeta, taskId);
 
-  // save to disk
+  // 6. save to disk
   taosWLockLatch(&pMeta->lock);
-
   streamMetaSaveTask(pMeta, pStreamTask);
   if (streamMetaCommit(pMeta) < 0) {
     // persist to disk
   }
   taosWUnLockLatch(&pMeta->lock);
 
-  // pause allowed
+  // 7. pause allowed.
   streamTaskEnablePause(pStreamTask);
 
   streamSchedExec(pStreamTask);
@@ -373,6 +375,25 @@ static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
+  int32_t code = TSDB_CODE_SUCCESS;
+  if (!pTask->status.transferState) {
+    return code;
+  }
+
+  if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {
+    streamTaskFillHistoryFinished(pTask);
+    streamTaskEndScanWAL(pTask);
+  } else { // do transfer task operator states.
+    code = streamDoTransferStateToStreamTask(pTask);
+    if (code != TSDB_CODE_SUCCESS) {  // todo handle this
+      return code;
+    }
+  }
+
+  return code;
+}
+
 static int32_t extractMsgFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInput, int32_t* numOfBlocks,
                                     const char* id) {
   int32_t retryTimes = 0;
@@ -526,17 +547,16 @@ int32_t streamTaskEndScanWAL(SStreamTask* pTask) {
   double      el = (taosGetTimestampMs() - pTask->tsInfo.step2Start) / 1000.0;
   qDebug("s-task:%s scan-history from WAL stage(step 2) ended, elapsed time:%.2fs", id, el);
 
-  // 3. notify downstream tasks to transfer executor state after handle all history blocks.
-  pTask->status.transferState = true;
-
+  // 1. notify all downstream tasks to transfer executor state after handle all history blocks.
   int32_t code = streamDispatchTransferStateMsg(pTask);
   if (code != TSDB_CODE_SUCCESS) {
     // todo handle error
   }
 
-  // the last execution of fill-history task, in order to transfer task operator states.
-  code = streamTransferStateToStreamTask(pTask);
-  if (code != TSDB_CODE_SUCCESS) {  // todo handle this
+  // 2. do transfer stream task operator states.
+  pTask->status.transferState = true;
+  code = streamDoTransferStateToStreamTask(pTask);
+  if (code != TSDB_CODE_SUCCESS) { // todo handle error
     return code;
   }
 
@@ -560,9 +580,11 @@ int32_t streamTryExec(SStreamTask* pTask) {
     // todo the task should be commit here
     if (taosQueueEmpty(pTask->inputQueue->queue)) {
       // fill-history WAL scan has completed
-      if (pTask->info.taskLevel == TASK_LEVEL__SOURCE && pTask->status.transferState == true) {
-        streamTaskFillHistoryFinished(pTask);
-        streamTaskEndScanWAL(pTask);
+      if (pTask->status.transferState) {
+        code = streamTransferStateToStreamTask(pTask);
+        if (code != TSDB_CODE_SUCCESS) {
+          return code;
+        }
       } else {
         atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
         qDebug("s-task:%s exec completed, status:%s, sched-status:%d", id, streamGetTaskStatusStr(pTask->status.taskStatus),

From 8c1c17e31e01521432413d3cd6a73e857bb34c24 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 31 Jul 2023 19:23:53 +0800
Subject: [PATCH 20/20] fix(stream): ignore the sink task transfer state.

---
 source/libs/stream/src/streamExec.c | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index e4dc28b76b..d1dff0f2e7 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -381,10 +381,11 @@ static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
     return code;
   }
 
-  if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {
+  int32_t level = pTask->info.taskLevel;
+  if (level == TASK_LEVEL__SOURCE) {
     streamTaskFillHistoryFinished(pTask);
     streamTaskEndScanWAL(pTask);
-  } else { // do transfer task operator states.
+  } else if (level == TASK_LEVEL__AGG) { // do transfer task operator states.
     code = streamDoTransferStateToStreamTask(pTask);
     if (code != TSDB_CODE_SUCCESS) {  // todo handle this
       return code;