From 43afdc5ad0564693671c87910841baa8024fe4cd Mon Sep 17 00:00:00 2001
From: afwerar <1296468573@qq.com>
Date: Tue, 14 Jun 2022 19:31:45 +0800
Subject: [PATCH 01/14] os: add taosd assert kill

---
 tests/system-test/7-tmq/basic5.py |  4 +++-
 tests/system-test/test-all.bat    | 10 +++++-----
 tests/system-test/test.py         | 15 +++++++++++----
 3 files changed, 19 insertions(+), 10 deletions(-)

diff --git a/tests/system-test/7-tmq/basic5.py b/tests/system-test/7-tmq/basic5.py
index 3d9efea938..a10eaf1fb5 100644
--- a/tests/system-test/7-tmq/basic5.py
+++ b/tests/system-test/7-tmq/basic5.py
@@ -134,7 +134,7 @@ class TDTestCase:
         parameterDict['cfg'] = cfgPath
         prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
         prepareEnvThread.start()
-        time.sleep(2)
+        prepareEnvThread.join()
         
         # wait stb ready
         while 1:
@@ -245,6 +245,7 @@ class TDTestCase:
 
         prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
         prepareEnvThread.start()
+        prepareEnvThread.join()
         
         # wait db ready
         while 1:
@@ -371,6 +372,7 @@ class TDTestCase:
 
         prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
         prepareEnvThread.start()
+        prepareEnvThread.join()
         
         # wait db ready
         while 1:
diff --git a/tests/system-test/test-all.bat b/tests/system-test/test-all.bat
index 076be6563a..a5a114f6eb 100644
--- a/tests/system-test/test-all.bat
+++ b/tests/system-test/test-all.bat
@@ -61,19 +61,19 @@ goto :eof
 set tt=%1
 set tt=%tt:.= %
 set tt=%tt::= %
-set index=1
+set /a index=1
 for %%a in (%tt%) do (
     if !index! EQU 1 (
-        set /a hh=%%a
+        set /a hh=%%a || echo 11 %%a
     )^
     else if  !index! EQU 2 (
-        set /a mm=%%a
+        set /a mm=%%a || echo 22 %%a
  
     )^
     else if  !index! EQU 3 (
-        set /a ss=%%a
+        set /a ss=%%a || echo 33 %%a
     )
    set /a index=index+1
 )
-set /a _timeTemp=(%hh%*60+%mm%)*60+%ss%  || echo hh:%hh% mm:%mm% ss:%ss%
+set /a _timeTemp=(%hh%*60+%mm%)*60+%ss%  || echo 44 hh:%hh% mm:%mm% ss:%ss%
 goto :eof
\ No newline at end of file
diff --git a/tests/system-test/test.py b/tests/system-test/test.py
index 47e0cefb52..aaa61d8127 100644
--- a/tests/system-test/test.py
+++ b/tests/system-test/test.py
@@ -21,6 +21,7 @@ import base64
 import json
 import platform
 import socket
+import threading
 from distutils.log import warn as printf
 from fabric2 import Connection
 sys.path.append("../pytest")
@@ -30,6 +31,13 @@ from util.cases import *
 
 import taos
 
+def checkRunTimeError():
+    import win32gui
+    while 1:
+        time.sleep(1)
+        hwnd = win32gui.FindWindow(None, "Microsoft Visual C++ Runtime Library")
+        if hwnd:
+            os.system("TASKKILL /F /IM taosd.exe")
 
 if __name__ == "__main__":
     
@@ -42,9 +50,6 @@ if __name__ == "__main__":
     logSql = True
     stop = 0
     restart = False
-    windows = 0
-    if platform.system().lower() == 'windows':
-        windows = 1
     updateCfgDict = {}
     execCmd = ""
     opts, args = getopt.gnu_getopt(sys.argv[1:], 'f:p:m:l:scghrd:k:e:', [
@@ -159,7 +164,9 @@ if __name__ == "__main__":
             host = masterIp
 
     tdLog.info("Procedures for tdengine deployed in %s" % (host))
-    if windows:
+    if platform.system().lower() == 'windows':
+        if (masterIp == ""):
+            threading.Thread(target=checkRunTimeError,daemon=True).start()
         tdCases.logSql(logSql)
         tdLog.info("Procedures for testing self-deployment")
         tdDnodes.init(deployPath, masterIp)

From c7f276462547783b1ef9843aaf7c38be15657628 Mon Sep 17 00:00:00 2001
From: afwerar <1296468573@qq.com>
Date: Tue, 14 Jun 2022 20:31:42 +0800
Subject: [PATCH 02/14] os: add taosd assert kill

---
 tests/system-test/test-all.bat | 9 +++++----
 tests/system-test/test.py      | 2 +-
 2 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/tests/system-test/test-all.bat b/tests/system-test/test-all.bat
index a5a114f6eb..0929b1fc6e 100644
--- a/tests/system-test/test-all.bat
+++ b/tests/system-test/test-all.bat
@@ -61,19 +61,20 @@ goto :eof
 set tt=%1
 set tt=%tt:.= %
 set tt=%tt::= %
+set tt=%tt: 0= %
 set /a index=1
 for %%a in (%tt%) do (
     if !index! EQU 1 (
-        set /a hh=%%a || echo 11 %%a
+        set /a hh=%%a
     )^
     else if  !index! EQU 2 (
-        set /a mm=%%a || echo 22 %%a
+        set /a mm=%%a
  
     )^
     else if  !index! EQU 3 (
-        set /a ss=%%a || echo 33 %%a
+        set /a ss=%%a
     )
    set /a index=index+1
 )
-set /a _timeTemp=(%hh%*60+%mm%)*60+%ss%  || echo 44 hh:%hh% mm:%mm% ss:%ss%
+set /a _timeTemp=(%hh%*60+%mm%)*60+%ss%
 goto :eof
\ No newline at end of file
diff --git a/tests/system-test/test.py b/tests/system-test/test.py
index aaa61d8127..0022a51329 100644
--- a/tests/system-test/test.py
+++ b/tests/system-test/test.py
@@ -165,7 +165,7 @@ if __name__ == "__main__":
 
     tdLog.info("Procedures for tdengine deployed in %s" % (host))
     if platform.system().lower() == 'windows':
-        if (masterIp == ""):
+        if (masterIp == "" and not fileName[0:3] == "udf"):
             threading.Thread(target=checkRunTimeError,daemon=True).start()
         tdCases.logSql(logSql)
         tdLog.info("Procedures for testing self-deployment")

From 435aca2eeb0eb5a9c9a4b1ff2265d32226a8e281 Mon Sep 17 00:00:00 2001
From: afwerar <1296468573@qq.com>
Date: Wed, 15 Jun 2022 09:04:06 +0800
Subject: [PATCH 03/14] os: add taosd assert kill

---
 tests/system-test/test.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/system-test/test.py b/tests/system-test/test.py
index 0022a51329..5a68f548da 100644
--- a/tests/system-test/test.py
+++ b/tests/system-test/test.py
@@ -165,7 +165,7 @@ if __name__ == "__main__":
 
     tdLog.info("Procedures for tdengine deployed in %s" % (host))
     if platform.system().lower() == 'windows':
-        if (masterIp == "" and not fileName[0:3] == "udf"):
+        if (masterIp == "" and not fileName[0:12] == "0-others\\udf"):
             threading.Thread(target=checkRunTimeError,daemon=True).start()
         tdCases.logSql(logSql)
         tdLog.info("Procedures for testing self-deployment")

From 6c7072286d4f82dd3e7353af02992a9a9df838f6 Mon Sep 17 00:00:00 2001
From: Minghao Li <castermode@gmail.com>
Date: Wed, 15 Jun 2022 11:06:06 +0800
Subject: [PATCH 04/14] refactor(sync): if eqmsg error, return

---
 source/libs/sync/src/syncMain.c | 30 ++++++++++++++++++++----------
 1 file changed, 20 insertions(+), 10 deletions(-)

diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c
index 9db9550681..920a0245fd 100644
--- a/source/libs/sync/src/syncMain.c
+++ b/source/libs/sync/src/syncMain.c
@@ -1743,16 +1743,26 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) {
     syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
     syncRpcMsgLog2((char*)"==syncNodeEqElectTimer==", &rpcMsg);
     if (pSyncNode->FpEqMsg != NULL) {
-      pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
+      int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
+      if (code != 0) {
+        sError("vgId:%d sync enqueue elect msg error, code:%d", pSyncNode->vgId, code);
+        rpcFreeCont(rpcMsg.pCont);
+        syncTimeoutDestroy(pSyncMsg);
+        return;
+      }
     } else {
-      sTrace("syncNodeEqElectTimer pSyncNode->FpEqMsg is NULL");
+      sTrace("syncNodeEqElectTimer FpEqMsg is NULL");
     }
     syncTimeoutDestroy(pSyncMsg);
 
     // reset timer ms
-    pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
-    taosTmrReset(syncNodeEqElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
-                 &pSyncNode->pElectTimer);
+    if (gSyncEnv != NULL) {
+      pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
+      taosTmrReset(syncNodeEqElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
+                   &pSyncNode->pElectTimer);
+    } else {
+      sError("sync env elect is already stop");
+    }
   } else {
     sTrace("==syncNodeEqElectTimer== electTimerLogicClock:%" PRIu64 ", electTimerLogicClockUser:%" PRIu64 "",
            pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
@@ -1774,19 +1784,19 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
       if (code != 0) {
         sError("vgId:%d sync enqueue timer msg error, code:%d", pSyncNode->vgId, code);
         rpcFreeCont(rpcMsg.pCont);
+        syncTimeoutDestroy(pSyncMsg);
         return;
       }
-
     } else {
-      sTrace("syncNodeEqHeartbeatTimer pSyncNode->FpEqMsg is NULL");
+      sError("syncNodeEqHeartbeatTimer FpEqMsg is NULL");
     }
     syncTimeoutDestroy(pSyncMsg);
 
     if (gSyncEnv != NULL) {
-    taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
-                 &pSyncNode->pHeartbeatTimer);
+      taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
+                   &pSyncNode->pHeartbeatTimer);
     } else {
-      sError("sync env is already stop");
+      sError("sync env heartbeat is already stop");
     }
   } else {
     sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64

From c9f6c2041091204289163212a170487d12ac41fa Mon Sep 17 00:00:00 2001
From: Minghao Li <castermode@gmail.com>
Date: Wed, 15 Jun 2022 11:42:04 +0800
Subject: [PATCH 05/14] refactor(sync): if eqmsg error, return

---
 source/libs/sync/src/syncMain.c | 57 +++++++++++++++++++++++----------
 1 file changed, 40 insertions(+), 17 deletions(-)

diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c
index 920a0245fd..7b1fd010ab 100644
--- a/source/libs/sync/src/syncMain.c
+++ b/source/libs/sync/src/syncMain.c
@@ -945,9 +945,13 @@ int32_t syncNodePingAll(SSyncNode* pSyncNode) {
 // timer control --------------
 int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
   int32_t ret = 0;
-  taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
-               &pSyncNode->pPingTimer);
-  atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
+  if (syncEnvIsStart()) {
+    taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
+                 &pSyncNode->pPingTimer);
+    atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
+  } else {
+    sError("sync env is stop, syncNodeStartPingTimer");
+  }
   return ret;
 }
 
@@ -961,10 +965,14 @@ int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode) {
 
 int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
   int32_t ret = 0;
-  pSyncNode->electTimerMS = ms;
-  taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
-               &pSyncNode->pElectTimer);
-  atomic_store_64(&pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
+  if (syncEnvIsStart()) {
+    pSyncNode->electTimerMS = ms;
+    taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
+                 &pSyncNode->pElectTimer);
+    atomic_store_64(&pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
+  } else {
+    sError("sync env is stop, syncNodeStartElectTimer");
+  }
   return ret;
 }
 
@@ -998,9 +1006,13 @@ int32_t syncNodeResetElectTimer(SSyncNode* pSyncNode) {
 
 int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
   int32_t ret = 0;
-  taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
-               &pSyncNode->pHeartbeatTimer);
-  atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
+  if (syncEnvIsStart()) {
+    taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
+                 &pSyncNode->pHeartbeatTimer);
+    atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
+  } else {
+    sError("sync env is stop, syncNodeStartHeartbeatTimer");
+  }
   return ret;
 }
 
@@ -1720,14 +1732,25 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) {
     syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
     syncRpcMsgLog2((char*)"==syncNodeEqPingTimer==", &rpcMsg);
     if (pSyncNode->FpEqMsg != NULL) {
-      pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
+      int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
+      if (code != 0) {
+        sError("vgId:%d sync enqueue ping msg error, code:%d", pSyncNode->vgId, code);
+        rpcFreeCont(rpcMsg.pCont);
+        syncTimeoutDestroy(pSyncMsg);
+        return;
+      }
     } else {
       sTrace("syncNodeEqPingTimer pSyncNode->FpEqMsg is NULL");
     }
     syncTimeoutDestroy(pSyncMsg);
 
-    taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
-                 &pSyncNode->pPingTimer);
+    if (syncEnvIsStart()) {
+      taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
+                   &pSyncNode->pPingTimer);
+    } else {
+      sError("sync env is stop, syncNodeEqPingTimer");
+    }
+
   } else {
     sTrace("==syncNodeEqPingTimer== pingTimerLogicClock:%" PRIu64 ", pingTimerLogicClockUser:%" PRIu64 "",
            pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
@@ -1756,12 +1779,12 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) {
     syncTimeoutDestroy(pSyncMsg);
 
     // reset timer ms
-    if (gSyncEnv != NULL) {
+    if (syncEnvIsStart()) {
       pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
       taosTmrReset(syncNodeEqElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                    &pSyncNode->pElectTimer);
     } else {
-      sError("sync env elect is already stop");
+      sError("sync env is stop, syncNodeEqElectTimer");
     }
   } else {
     sTrace("==syncNodeEqElectTimer== electTimerLogicClock:%" PRIu64 ", electTimerLogicClockUser:%" PRIu64 "",
@@ -1792,11 +1815,11 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
     }
     syncTimeoutDestroy(pSyncMsg);
 
-    if (gSyncEnv != NULL) {
+    if (syncEnvIsStart()) {
       taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                    &pSyncNode->pHeartbeatTimer);
     } else {
-      sError("sync env heartbeat is already stop");
+      sError("sync env is stop, syncNodeEqHeartbeatTimer");
     }
   } else {
     sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64

From 8a1d0812fe1fb85deedb34f38038ad981a509b6a Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Wed, 15 Jun 2022 13:22:36 +0800
Subject: [PATCH 06/14] fix last function crash on windows

---
 source/libs/function/src/builtinsimpl.c | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index c6ab9f859d..04d7396d12 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -2392,6 +2392,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
   SInputColumnInfoData* pInput = &pCtx->input;
   SColumnInfoData*      pInputCol = pInput->pData[0];
 
+  int32_t type  = pInputCol->info.type;
   int32_t bytes = pInputCol->info.bytes;
   pInfo->bytes  = bytes;
 
@@ -2428,6 +2429,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
       TSKEY cts = getRowPTs(pInput->pPTS, i);
 
       if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) > cts) {
+        if (IS_VAR_DATA_TYPE(type)) {
+          bytes = varDataTLen(data);
+          pInfo->bytes = bytes;
+        }
         memcpy(pInfo->buf, data, bytes);
         *(TSKEY*)(pInfo->buf + bytes) = cts;
         pInfo->hasResult = true;
@@ -2458,6 +2463,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
       TSKEY cts = getRowPTs(pInput->pPTS, i);
 
       if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) > cts) {
+        if (IS_VAR_DATA_TYPE(type)) {
+          bytes = varDataTLen(data);
+          pInfo->bytes = bytes;
+        }
         memcpy(pInfo->buf, data, bytes);
         *(TSKEY*)(pInfo->buf + bytes) = cts;
         pInfo->hasResult = true;
@@ -2481,6 +2490,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
   SInputColumnInfoData* pInput = &pCtx->input;
   SColumnInfoData*      pInputCol = pInput->pData[0];
 
+  int32_t type  = pInputCol->info.type;
   int32_t bytes = pInputCol->info.bytes;
   pInfo->bytes  = bytes;
 
@@ -2508,6 +2518,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
       char* data = colDataGetData(pInputCol, i);
       TSKEY cts = getRowPTs(pInput->pPTS, i);
       if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) < cts) {
+        if (IS_VAR_DATA_TYPE(type)) {
+          bytes = varDataTLen(data);
+          pInfo->bytes = bytes;
+        }
         memcpy(pInfo->buf, data, bytes);
         *(TSKEY*)(pInfo->buf + bytes) = cts;
         //        DO_UPDATE_TAG_COLUMNS(pCtx, ts);
@@ -2527,6 +2541,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
       char* data = colDataGetData(pInputCol, i);
       TSKEY cts = getRowPTs(pInput->pPTS, i);
       if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) < cts) {
+        if (IS_VAR_DATA_TYPE(type)) {
+          bytes = varDataTLen(data);
+          pInfo->bytes = bytes;
+        }
         memcpy(pInfo->buf, data, bytes);
         *(TSKEY*)(pInfo->buf + bytes) = cts;
         pInfo->hasResult = true;

From 045f0eb95121dfad6e9d1b830c31fe5ec3f84fdd Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Wed, 15 Jun 2022 14:11:06 +0800
Subject: [PATCH 07/14] fix jsons filter error

---
 tests/system-test/2-query/json_tag.py | 59 ++++++++++++++-------------
 1 file changed, 30 insertions(+), 29 deletions(-)

diff --git a/tests/system-test/2-query/json_tag.py b/tests/system-test/2-query/json_tag.py
index 1107e9693e..54b41e5a94 100644
--- a/tests/system-test/2-query/json_tag.py
+++ b/tests/system-test/2-query/json_tag.py
@@ -244,8 +244,8 @@ class TDTestCase:
         tdSql.query("select * from jsons1 where jtag->'tag2'!='beijing'")
         tdSql.checkRows(5)
 	#open 
-        #tdSql.query("select * from jsons1 where jtag->'tag2'=''")
-        #tdSql.checkRows(2)
+        tdSql.query("select * from jsons1 where jtag->'tag2'=''")
+        tdSql.checkRows(2)
         #
         # # where json value is int
         tdSql.query("select * from jsons1 where jtag->'tag1'=5")
@@ -254,10 +254,10 @@ class TDTestCase:
         tdSql.query("select * from jsons1 where jtag->'tag1'=10")
         tdSql.checkRows(0)
 	# open 
-        #tdSql.query("select * from jsons1 where jtag->'tag1'<54")
-        #tdSql.checkRows(3)
-        #tdSql.query("select * from jsons1 where jtag->'tag1'<=11")
-        #tdSql.checkRows(3)
+        tdSql.query("select * from jsons1 where jtag->'tag1'<54")
+        tdSql.checkRows(4)
+        tdSql.query("select * from jsons1 where jtag->'tag1'<=11")
+        tdSql.checkRows(4)
         tdSql.query("select * from jsons1 where jtag->'tag1'>4")
         tdSql.checkRows(2)
         tdSql.query("select * from jsons1 where jtag->'tag1'>=5")
@@ -271,30 +271,30 @@ class TDTestCase:
         tdSql.query("select * from jsons1 where jtag->'tag1'=1.232")
         tdSql.checkRows(1)
 	# open
-        #tdSql.query("select * from jsons1 where jtag->'tag1'<1.232")
-        #tdSql.checkRows(0)
-        #tdSql.query("select * from jsons1 where jtag->'tag1'<=1.232")
-        #tdSql.checkRows(1)
+        tdSql.query("select * from jsons1 where jtag->'tag1'<1.232")
+        tdSql.checkRows(1)
+        tdSql.query("select * from jsons1 where jtag->'tag1'<=1.232")
+        tdSql.checkRows(2)
         tdSql.query("select * from jsons1 where jtag->'tag1'>1.23")
         tdSql.checkRows(3)
         tdSql.query("select * from jsons1 where jtag->'tag1'>=1.232")
         tdSql.checkRows(3)
 	# open
-        #tdSql.query("select * from jsons1 where jtag->'tag1'!=1.232")
-        #tdSql.checkRows(2)
+        tdSql.query("select * from jsons1 where jtag->'tag1'!=1.232")
+        tdSql.checkRows(6)
         tdSql.query("select * from jsons1 where jtag->'tag1'!=3.232")
         tdSql.checkRows(7)
         #tdSql.error("select * from jsons1 where jtag->'tag1'/0=3")
         #tdSql.error("select * from jsons1 where jtag->'tag1'/5=1")
         #
         # # where json value is bool
-        #tdSql.query("select * from jsons1 where jtag->'tag1'=true")
+        tdSql.query("select * from jsons1 where jtag->'tag1'=true")
 	# open
-        #tdSql.checkRows(0)
+        tdSql.checkRows(0)
         #tdSql.query("select * from jsons1 where jtag->'tag1'=false")
         #tdSql.checkRows(1)
-        #tdSql.query("select * from jsons1 where jtag->'tag1'!=false")
-        #tdSql.checkRows(0)
+        tdSql.query("select * from jsons1 where jtag->'tag1'!=false")
+        tdSql.checkRows(3)
         #tdSql.error("select * from jsons1 where jtag->'tag1'>false")
         #
         # # where json value is null
@@ -304,17 +304,17 @@ class TDTestCase:
         #
         # # where json key is null
 	# open
-        #tdSql.query("select * from jsons1 where jtag->'tag_no_exist'=3")
-        #tdSql.checkRows(0)
+        tdSql.query("select * from jsons1 where jtag->'tag_no_exist'=3")
+        tdSql.checkRows(0)
         #
         # # where json value is not exist
-        #tdSql.query("select * from jsons1 where jtag->'tag1' is null")
-        #tdSql.checkData(0, 0, 'jsons1_9')
-        #tdSql.checkRows(1)
-        #tdSql.query("select * from jsons1 where jtag->'tag4' is null")
-        #tdSql.checkRows(9)
-        #tdSql.query("select * from jsons1 where jtag->'tag3' is not null")
-        #tdSql.checkRows(4)
+        tdSql.query("select * from jsons1 where jtag->'tag1' is null")
+        tdSql.checkData(0, 0, 'jsons1_9')
+        tdSql.checkRows(2)
+        tdSql.query("select * from jsons1 where jtag->'tag4' is null")
+        tdSql.checkRows(9)
+        tdSql.query("select * from jsons1 where jtag->'tag3' is not null")
+        tdSql.checkRows(3)
         #
         # # test contains
         tdSql.query("select * from jsons1 where jtag contains 'tag1'")
@@ -344,10 +344,10 @@ class TDTestCase:
         #
         #
         # # test with between and
-        #tdSql.query("select * from jsons1 where jtag->'tag1' between 1 and 30")
-        #tdSql.checkRows(3)
-        #tdSql.query("select * from jsons1 where jtag->'tag1' between 'femail' and 'beijing'")
-        #tdSql.checkRows(2)
+        tdSql.query("select * from jsons1 where jtag->'tag1' between 1 and 30")
+        tdSql.checkRows(3)
+        tdSql.query("select * from jsons1 where jtag->'tag1' between 'femail' and 'beijing'")
+        tdSql.checkRows(2)
         #
         # # test with tbname/normal column
         tdSql.query("select * from jsons1 where tbname = 'jsons1_1'")
@@ -362,6 +362,7 @@ class TDTestCase:
         #
         # # test where condition like
 	# open
+ 	# syntax error	
         #tdSql.query("select *,tbname from jsons1 where jtag->'tag2' like 'bei%'")
         #tdSql.checkRows(2)
         #tdSql.query("select *,tbname from jsons1 where jtag->'tag1' like 'fe%' and jtag->'tag2' is not null")

From 4fc7e0aae98b75c21fe915847e2a22269f4c38e1 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Wed, 15 Jun 2022 15:31:58 +0800
Subject: [PATCH 08/14] fix jsons filter error

---
 tests/system-test/2-query/json_tag.py | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/tests/system-test/2-query/json_tag.py b/tests/system-test/2-query/json_tag.py
index 54b41e5a94..b6890a8503 100644
--- a/tests/system-test/2-query/json_tag.py
+++ b/tests/system-test/2-query/json_tag.py
@@ -243,7 +243,6 @@ class TDTestCase:
         tdSql.checkRows(2)
         tdSql.query("select * from jsons1 where jtag->'tag2'!='beijing'")
         tdSql.checkRows(5)
-	#open 
         tdSql.query("select * from jsons1 where jtag->'tag2'=''")
         tdSql.checkRows(2)
         #
@@ -253,7 +252,6 @@ class TDTestCase:
         tdSql.checkData(0, 1, 2)
         tdSql.query("select * from jsons1 where jtag->'tag1'=10")
         tdSql.checkRows(0)
-	# open 
         tdSql.query("select * from jsons1 where jtag->'tag1'<54")
         tdSql.checkRows(4)
         tdSql.query("select * from jsons1 where jtag->'tag1'<=11")
@@ -270,7 +268,6 @@ class TDTestCase:
         # # where json value is double
         tdSql.query("select * from jsons1 where jtag->'tag1'=1.232")
         tdSql.checkRows(1)
-	# open
         tdSql.query("select * from jsons1 where jtag->'tag1'<1.232")
         tdSql.checkRows(1)
         tdSql.query("select * from jsons1 where jtag->'tag1'<=1.232")
@@ -279,7 +276,6 @@ class TDTestCase:
         tdSql.checkRows(3)
         tdSql.query("select * from jsons1 where jtag->'tag1'>=1.232")
         tdSql.checkRows(3)
-	# open
         tdSql.query("select * from jsons1 where jtag->'tag1'!=1.232")
         tdSql.checkRows(6)
         tdSql.query("select * from jsons1 where jtag->'tag1'!=3.232")
@@ -289,7 +285,6 @@ class TDTestCase:
         #
         # # where json value is bool
         tdSql.query("select * from jsons1 where jtag->'tag1'=true")
-	# open
         tdSql.checkRows(0)
         #tdSql.query("select * from jsons1 where jtag->'tag1'=false")
         #tdSql.checkRows(1)
@@ -303,7 +298,6 @@ class TDTestCase:
         #tdSql.checkRows(1)
         #
         # # where json key is null
-	# open
         tdSql.query("select * from jsons1 where jtag->'tag_no_exist'=3")
         tdSql.checkRows(0)
         #

From d8c2a68f27df8ec9f3c108c10682ca7fb6d0e13b Mon Sep 17 00:00:00 2001
From: 54liuyao <54liuyao@163.com>
Date: Wed, 15 Jun 2022 10:44:36 +0800
Subject: [PATCH 09/14] feat(stream): update data fo partition by

---
 source/libs/executor/inc/executorimpl.h       |   4 +
 source/libs/executor/src/scanoperator.c       | 154 +++++++++++++-----
 source/libs/executor/src/timewindowoperator.c |   4 +-
 source/libs/function/src/builtinsimpl.c       |   9 +-
 source/libs/planner/src/planOptimizer.c       |  34 ++--
 5 files changed, 145 insertions(+), 60 deletions(-)

diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h
index ab60acab53..c4d16c89e1 100644
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@@ -391,7 +391,9 @@ typedef struct SStreamBlockScanInfo {
   void*           streamBlockReader;// stream block reader handle
   SArray*         pColMatchInfo;    //
   SNode*          pCondition;
+  int32_t         tsArrayIndex;
   SArray*         tsArray;
+  uint64_t        groupId;
   SUpdateInfo*    pUpdateInfo;
 
   SExprInfo*      pPseudoExpr;
@@ -582,6 +584,7 @@ typedef struct SPartitionOperatorInfo {
   int32_t*       columnOffset;  // start position for each column data
   void*          pGroupIter;  // group iterator
   int32_t        pageIndex;   // page index of current group
+  SSDataBlock*   pUpdateRes;
 } SPartitionOperatorInfo;
 
 typedef struct SWindowRowsSup {
@@ -907,6 +910,7 @@ int32_t compareTimeWindow(const void* p1, const void* p2, const void* param);
 int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition,
                                        SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset,
                                        SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
+void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);
 
 #ifdef __cplusplus
 }
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index d30e4ef6db..e44e05224f 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -750,16 +750,103 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
   return true;
 }
 
+static void copyOneRow(SSDataBlock* dest, SSDataBlock* source, int32_t sourceRowId) {
+  for (int32_t j = 0; j < source->info.numOfCols; j++) {
+    SColumnInfoData* pDestCol = (SColumnInfoData*)taosArrayGet(dest->pDataBlock, j);
+    SColumnInfoData* pSourceCol = (SColumnInfoData*)taosArrayGet(source->pDataBlock, j);
+    if (colDataIsNull_s(pSourceCol, sourceRowId)) {
+      colDataAppendNULL(pDestCol, dest->info.rows);
+    } else {
+      colDataAppend(pDestCol, dest->info.rows, colDataGetData(pSourceCol, sourceRowId), false);
+    }
+  }
+  dest->info.rows++;
+}
+
+static uint64_t getGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t rowId) {
+  uint64_t* groupId = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &pBlock->info.uid, sizeof(int64_t));
+  if (groupId) {
+    return *groupId;
+  }
+  return 0;
+  /* Todo(liuyao) for partition by column
+  recordNewGroupKeys(pTableScanInfo->pGroupCols, pTableScanInfo->pGroupColVals, pBlock, rowId);
+  int32_t len = buildGroupKeys(pTableScanInfo->keyBuf, pTableScanInfo->pGroupColVals);
+  uint64_t resId = 0;
+  uint64_t* groupId = taosHashGet(pTableScanInfo->pGroupSet, pTableScanInfo->keyBuf, len);
+  if (groupId) {
+    return *groupId;
+  } else if (len != 0) {
+    resId = calcGroupId(pTableScanInfo->keyBuf, len);
+    taosHashPut(pTableScanInfo->pGroupSet, pTableScanInfo->keyBuf, len, &resId, sizeof(uint64_t));
+  }
+  return resId;
+  */
+}
+
 static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo) {
-  SSDataBlock* pResult = NULL;
-  pResult = doTableScan(pInfo->pOperatorDumy);
-  if (pResult == NULL) {
-    if (prepareDataScan(pInfo)) {
-      // scan next window data
-      pResult = doTableScan(pInfo->pOperatorDumy);
+  while (1) {
+    SSDataBlock* pResult = NULL;
+    pResult = doTableScan(pInfo->pOperatorDumy);
+    if (pResult == NULL) {
+      if (prepareDataScan(pInfo)) {
+        // scan next window data
+        pResult = doTableScan(pInfo->pOperatorDumy);
+      }
+    }
+    if (!pResult) {
+      return NULL;
+    }
+    
+    if (pResult->info.groupId == pInfo->groupId) {
+      return pResult;
+    }
+  }
+
+/* Todo(liuyao) for partition by column
+  SSDataBlock* pBlock = createOneDataBlock(pResult, true);
+  blockDataCleanup(pResult);
+  for (int32_t i = 0; i < pBlock->info.rows; i++) {
+    uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock, i);
+    if (id == pInfo->groupId) {
+      copyOneRow(pResult, pBlock, i);
     }
   }
   return pResult;
+*/
+}
+
+static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDataBlock* pUpdateBlock) {
+  blockDataCleanup(pUpdateBlock);
+  int32_t size = taosArrayGetSize(pInfo->tsArray);
+  if (pInfo->tsArrayIndex < size) {
+    SColumnInfoData* pCol = (SColumnInfoData*)taosArrayGet(pUpdateBlock->pDataBlock, pInfo->primaryTsIndex);
+    ASSERT(pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
+    blockDataEnsureCapacity(pUpdateBlock, size);
+    ASSERT(pBlock->info.numOfCols == pUpdateBlock->info.numOfCols);
+
+    int32_t rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, pInfo->tsArrayIndex);
+    pInfo->groupId = getGroupId(pInfo->pOperatorDumy, pBlock, rowId);
+    int32_t i = 0;
+    for ( ; i < size; i++) {
+      rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, i + pInfo->tsArrayIndex);
+      uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock, rowId);
+      if (pInfo->groupId != id) {
+        break;
+      }
+      copyOneRow(pUpdateBlock, pBlock, rowId);
+    }
+    pUpdateBlock->info.rows = i;
+    pInfo->tsArrayIndex += i;
+    pUpdateBlock->info.groupId = pInfo->groupId;
+    pUpdateBlock->info.type = STREAM_REPROCESS;
+    blockDataUpdateTsWindow(pUpdateBlock, 0);
+  }
+  // all rows have same group id
+  ASSERT(pInfo->tsArrayIndex >= size);
+  if (size > 0 && pInfo->tsArrayIndex == size) {
+    taosArrayClear(pInfo->tsArray);
+  }
 }
 
 static void getUpdateDataBlock(SStreamBlockScanInfo* pInfo, bool invertible, SSDataBlock* pBlock,
@@ -767,41 +854,21 @@ static void getUpdateDataBlock(SStreamBlockScanInfo* pInfo, bool invertible, SSD
   SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
   ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP);
   TSKEY* ts = (TSKEY*)pColDataInfo->pData;
-  for (int32_t i = 0; i < pBlock->info.rows; i++) {
-    if (updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, ts[i])) {
-      taosArrayPush(pInfo->tsArray, ts + i);
+  for (int32_t rowId = 0; rowId < pBlock->info.rows; rowId++) {
+    if (updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, ts[rowId])) {
+      taosArrayPush(pInfo->tsArray, &rowId);
     }
   }
   if (!pUpdateBlock) {
     taosArrayClear(pInfo->tsArray);
     return;
   }
-  int32_t size = taosArrayGetSize(pInfo->tsArray);
-  if (size > 0 && invertible) {
-    // Todo(liuyao) get from tsdb
-    //  SSDataBlock* p = createOneDataBlock(pBlock, true);
-    //  p->info.type = STREAM_INVERT;
-    //  taosArrayClear(pInfo->tsArray);
-    //  return p;
-    SColumnInfoData* pCol = (SColumnInfoData*)taosArrayGet(pUpdateBlock->pDataBlock, pInfo->primaryTsIndex);
-    ASSERT(pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
-    blockDataEnsureCapacity(pUpdateBlock, size);
-    for (int32_t i = 0; i < size; i++) {
-      TSKEY* pTs = (TSKEY*)taosArrayGet(pInfo->tsArray, i);
-      colDataAppend(pCol, i, (char*)pTs, false);
-    }
-    for (int32_t i = 0; i < pUpdateBlock->info.numOfCols; i++) {
-      if (i == pInfo->primaryTsIndex) {
-        continue;
-      }
-      SColumnInfoData* pCol = (SColumnInfoData*)taosArrayGet(pUpdateBlock->pDataBlock, i);
-      colDataAppendNNULL(pCol, 0, size);
-    }
-    pUpdateBlock->info.rows = size;
-    pUpdateBlock->info.type = STREAM_REPROCESS;
-    blockDataUpdateTsWindow(pUpdateBlock, 0);
-    taosArrayClear(pInfo->tsArray);
-  }
+  setUpdateData(pInfo, pBlock, pUpdateBlock);
+  // Todo(liuyao) get from tsdb
+  //  SSDataBlock* p = createOneDataBlock(pBlock, true);
+  //  p->info.type = STREAM_INVERT;
+  //  taosArrayClear(pInfo->tsArray);
+  //  return p;
 }
 
 static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
@@ -833,7 +900,6 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
       pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
       return pInfo->pRes;
     } else if (pInfo->scanMode == STREAM_SCAN_FROM_UPDATERES) {
-      blockDataCleanup(pInfo->pRes);
       pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
       if (!isStateWindow(pInfo)) {
         prepareDataScan(pInfo);
@@ -848,7 +914,15 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
       if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) {
         SSDataBlock* pSDB = doDataScan(pInfo);
         if (pSDB == NULL) {
-          pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
+          setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes);
+          if (pInfo->pUpdateRes->info.rows > 0) {
+            if (!isStateWindow(pInfo)) {
+              prepareDataScan(pInfo);
+            }
+            return pInfo->pUpdateRes;
+          } else {
+            pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
+          }
         } else {
           getUpdateDataBlock(pInfo, true, pSDB, NULL);
           return pSDB;
@@ -941,7 +1015,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     if (rows == 0) {
       pOperator->status = OP_EXEC_DONE;
     } else if (pInfo->pUpdateInfo) {
-      blockDataCleanup(pInfo->pUpdateRes);
+      pInfo->tsArrayIndex = 0;
       getUpdateDataBlock(pInfo, true, pInfo->pRes, pInfo->pUpdateRes);
       if (pInfo->pUpdateRes->info.rows > 0) {
         if (pInfo->pUpdateRes->info.type == STREAM_REPROCESS) {
@@ -1020,7 +1094,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan
     goto _error;
   }
 
-  pInfo->tsArray = taosArrayInit(4, sizeof(TSKEY));
+  pInfo->tsArray = taosArrayInit(4, sizeof(int32_t));
   if (pInfo->tsArray == NULL) {
     goto _error;
   }
@@ -1047,6 +1121,8 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan
   pInfo->pOperatorDumy = pTableScanDummy;
   pInfo->interval = pSTInfo->interval;
   pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1};
+  pInfo->groupId = 0;
+  
   pOperator->name = "StreamBlockScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
   pOperator->blocking = false;
diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c
index 0151474008..5a18649cab 100644
--- a/source/libs/executor/src/timewindowoperator.c
+++ b/source/libs/executor/src/timewindowoperator.c
@@ -1985,7 +1985,7 @@ static void clearUpdateDataBlock(SSDataBlock* pBlock) {
   blockDataCleanup(pBlock);
 }
 
-static void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex) {
+void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex) {
   ASSERT(pDest->info.capacity >= pSource->info.rows);
   clearUpdateDataBlock(pDest);
   SColumnInfoData* pDestCol = taosArrayGet(pDest->pDataBlock, 0);
@@ -1997,6 +1997,8 @@ static void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_
     colDataAppendNNULL(pCol, 0, pSource->info.rows);
   }
   pDest->info.rows = pSource->info.rows;
+  pDest->info.groupId = pSource->info.groupId;
+  pDest->info.type = pSource->info.type;
   blockDataUpdateTsWindow(pDest, 0);
 }
 
diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index c6ab9f859d..94208043e7 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -2345,14 +2345,7 @@ int32_t apercentileCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx)
   SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
   SAPercentileInfo*    pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
   ASSERT(pDBuf->algo == pSBuf->algo);
-  if (pDBuf->algo == APERCT_ALGO_TDIGEST) {
-    tdigestMerge(pDBuf->pTDigest, pSBuf->pTDigest);
-  } else {
-    SHistogramInfo* pTmp = tHistogramMerge(pDBuf->pHisto, pSBuf->pHisto, MAX_HISTOGRAM_BIN);
-    memcpy(pDBuf->pHisto, pTmp, sizeof(SHistogramInfo) + sizeof(SHistBin) * (MAX_HISTOGRAM_BIN + 1));
-    pDBuf->pHisto->elems = (SHistBin*)((char*)pDBuf->pHisto + sizeof(SHistogramInfo));
-    tHistogramDestroy(&pTmp);
-  }
+  apercentileTransferInfo(pSBuf, pDBuf);
   pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes);
   return TSDB_CODE_SUCCESS;
 }
diff --git a/source/libs/planner/src/planOptimizer.c b/source/libs/planner/src/planOptimizer.c
index 65d73b4cdb..41b80eaaa8 100644
--- a/source/libs/planner/src/planOptimizer.c
+++ b/source/libs/planner/src/planOptimizer.c
@@ -104,10 +104,14 @@ static bool osdMayBeOptimized(SLogicNode* pNode) {
     return false;
   }
   if (NULL == pNode->pParent || (QUERY_NODE_LOGIC_PLAN_WINDOW != nodeType(pNode->pParent) &&
-                                 QUERY_NODE_LOGIC_PLAN_AGG != nodeType(pNode->pParent))) {
+                                 QUERY_NODE_LOGIC_PLAN_AGG != nodeType(pNode->pParent) &&
+                                 QUERY_NODE_LOGIC_PLAN_PARTITION != nodeType(pNode->pParent))) {
     return false;
   }
-  if (QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pNode->pParent)) {
+  if (QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pNode->pParent) ||
+      (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode->pParent) &&
+          pNode->pParent->pParent &&
+          QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pNode->pParent->pParent)) ) {
     return true;
   }
   return !osdHaveNormalCol(((SAggLogicNode*)pNode->pParent)->pGroupKeys);
@@ -217,16 +221,22 @@ static int32_t osdGetDataRequired(SNodeList* pFuncs) {
 }
 
 static void setScanWindowInfo(SScanLogicNode* pScan) {
-  if (QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pScan->node.pParent)) {
-    pScan->interval = ((SWindowLogicNode*)pScan->node.pParent)->interval;
-    pScan->offset = ((SWindowLogicNode*)pScan->node.pParent)->offset;
-    pScan->sliding = ((SWindowLogicNode*)pScan->node.pParent)->sliding;
-    pScan->intervalUnit = ((SWindowLogicNode*)pScan->node.pParent)->intervalUnit;
-    pScan->slidingUnit = ((SWindowLogicNode*)pScan->node.pParent)->slidingUnit;
-    pScan->triggerType = ((SWindowLogicNode*)pScan->node.pParent)->triggerType;
-    pScan->watermark = ((SWindowLogicNode*)pScan->node.pParent)->watermark;
-    pScan->tsColId = ((SColumnNode*)((SWindowLogicNode*)pScan->node.pParent)->pTspk)->colId;
-    pScan->filesFactor = ((SWindowLogicNode*)pScan->node.pParent)->filesFactor;
+  SLogicNode* pParent = pScan->node.pParent;
+  if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pParent) &&
+          pParent->pParent &&
+          QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pParent->pParent)) {
+    pParent = pParent->pParent;
+  }
+  if (QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pParent)) {
+    pScan->interval = ((SWindowLogicNode*)pParent)->interval;
+    pScan->offset = ((SWindowLogicNode*)pParent)->offset;
+    pScan->sliding = ((SWindowLogicNode*)pParent)->sliding;
+    pScan->intervalUnit = ((SWindowLogicNode*)pParent)->intervalUnit;
+    pScan->slidingUnit = ((SWindowLogicNode*)pParent)->slidingUnit;
+    pScan->triggerType = ((SWindowLogicNode*)pParent)->triggerType;
+    pScan->watermark = ((SWindowLogicNode*)pParent)->watermark;
+    pScan->tsColId = ((SColumnNode*)((SWindowLogicNode*)pParent)->pTspk)->colId;
+    pScan->filesFactor = ((SWindowLogicNode*)pParent)->filesFactor;
   }
 }
 

From 5f73b165f59bb0e40bdd31a84dd258a5dac59c35 Mon Sep 17 00:00:00 2001
From: afwerar <1296468573@qq.com>
Date: Wed, 15 Jun 2022 15:59:56 +0800
Subject: [PATCH 10/14] os: add taosd assert kill

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

diff --git a/source/libs/catalog/src/ctgCache.c b/source/libs/catalog/src/ctgCache.c
index 62890b8326..277516686b 100644
--- a/source/libs/catalog/src/ctgCache.c
+++ b/source/libs/catalog/src/ctgCache.c
@@ -1899,7 +1899,7 @@ _return:
 
 
 void ctgUpdateThreadUnexpectedStopped(void) {
-  if (CTG_IS_LOCKED(&gCtgMgmt.lock) > 0) CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock);
+  if (!atomic_load_8((int8_t*)&gCtgMgmt.exit) && CTG_IS_LOCKED(&gCtgMgmt.lock) > 0) CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock);
 }
 
 void ctgCleanupCacheQueue(void) {

From 8eb08fa0d4c021f1a8b74e2096282641793475a8 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Wed, 15 Jun 2022 16:11:20 +0800
Subject: [PATCH 11/14] feat(stream): trigger mode add max delay

---
 examples/c/stream_demo.c                   |  3 +-
 include/libs/executor/executor.h           |  9 ++-
 include/libs/stream/tstream.h              | 32 ++++++++-
 source/client/src/tmq.c                    | 84 ++++++++++++++++++++++
 source/dnode/mnode/impl/inc/mndDef.h       |  4 +-
 source/dnode/mnode/impl/inc/mndStream.h    |  2 +-
 source/dnode/mnode/impl/src/mndDef.c       |  8 +--
 source/dnode/mnode/impl/src/mndScheduler.c |  3 +
 source/dnode/mnode/impl/src/mndSma.c       |  5 +-
 source/dnode/mnode/impl/src/mndStream.c    | 17 +++--
 source/dnode/vnode/src/tq/tq.c             | 21 +++---
 source/libs/executor/src/executor.c        | 20 ++++--
 source/libs/function/src/builtins.c        |  4 +-
 source/libs/stream/inc/streamInc.h         |  7 ++
 source/libs/stream/src/stream.c            | 68 +++++++++++++++++-
 source/libs/stream/src/streamExec.c        | 14 ++--
 source/libs/stream/src/streamTask.c        |  2 +
 source/util/src/tarray.c                   |  6 +-
 18 files changed, 261 insertions(+), 48 deletions(-)

diff --git a/examples/c/stream_demo.c b/examples/c/stream_demo.c
index 943fcbdb53..ab59fa5e47 100644
--- a/examples/c/stream_demo.c
+++ b/examples/c/stream_demo.c
@@ -82,7 +82,8 @@ int32_t create_stream() {
   /*const char* sql = "select sum(k) from tu1 interval(10m)";*/
   /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/
   pRes = taos_query(
-      pConn, "create stream stream1 trigger at_once into outstb as select _wstartts, sum(k) from st1 interval(10m)");
+      pConn,
+      "create stream stream1 trigger max_delay 10s into outstb as select _wstartts, sum(k) from st1 interval(10m)");
   if (taos_errno(pRes) != 0) {
     printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
     return -1;
diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h
index 288248422b..3f05c27453 100644
--- a/include/libs/executor/executor.h
+++ b/include/libs/executor/executor.h
@@ -38,8 +38,10 @@ typedef struct SReadHandle {
   SMsgCb* pMsgCb;
 } SReadHandle;
 
-#define STREAM_DATA_TYPE_SUBMIT_BLOCK 0x1
-#define STREAM_DATA_TYPE_SSDATA_BLOCK 0x2
+enum {
+  STREAM_DATA_TYPE_SUBMIT_BLOCK = 1,
+  STREAM_DATA_TYPE_SSDATA_BLOCK = 2,
+};
 
 typedef enum {
   OPTR_EXEC_MODEL_BATCH = 0x1,
@@ -102,7 +104,8 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
  * @param tversion
  * @return
  */
-int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion, int32_t* tversion);
+int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion,
+                                      int32_t* tversion);
 
 /**
  * The main task execution function, including query on both table and multiple tables,
diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
index 960794792b..2c9d66a828 100644
--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -58,6 +58,7 @@ enum {
 enum {
   STREAM_INPUT__DATA_SUBMIT = 1,
   STREAM_INPUT__DATA_BLOCK,
+  STREAM_INPUT__TRIGGER,
   STREAM_INPUT__CHECKPOINT,
 };
 
@@ -85,6 +86,11 @@ typedef struct {
   int8_t type;
 } SStreamCheckpoint;
 
+typedef struct {
+  int8_t       type;
+  SSDataBlock* pBlock;
+} SStreamTrigger;
+
 enum {
   STREAM_QUEUE__SUCESS = 1,
   STREAM_QUEUE__FAILED,
@@ -98,6 +104,9 @@ typedef struct {
   int8_t      status;
 } SStreamQueue;
 
+int32_t streamInit();
+void    streamCleanUp();
+
 SStreamQueue* streamQueueOpen();
 void          streamQueueClose(SStreamQueue* queue);
 
@@ -220,6 +229,11 @@ enum {
   TASK_INPUT_TYPE__DATA_BLOCK,
 };
 
+enum {
+  TASK_TRIGGER_STATUS__IN_ACTIVE = 1,
+  TASK_TRIGGER_STATUS__ACTIVE,
+};
+
 struct SStreamTask {
   int64_t streamId;
   int32_t taskId;
@@ -262,8 +276,16 @@ struct SStreamTask {
   SStreamQueue* inputQueue;
   SStreamQueue* outputQueue;
 
+  // trigger
+  int8_t  triggerStatus;
+  int64_t triggerParam;
+  void*   timer;
+
   // application storage
   // void* ahandle;
+
+  // msg handle
+  SMsgCb* pMsgCb;
 };
 
 SStreamTask* tNewSStreamTask(int64_t streamId);
@@ -292,6 +314,13 @@ static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem
     taosWriteQitem(pTask->inputQueue->queue, pItem);
   } else if (pItem->type == STREAM_INPUT__CHECKPOINT) {
     taosWriteQitem(pTask->inputQueue->queue, pItem);
+  } else if (pItem->type == STREAM_INPUT__TRIGGER) {
+    taosWriteQitem(pTask->inputQueue->queue, pItem);
+  }
+
+  if (pItem->type != STREAM_INPUT__TRIGGER && pItem->type != STREAM_INPUT__CHECKPOINT && pTask->triggerParam != 0 &&
+      pTask->triggerStatus == TASK_TRIGGER_STATUS__IN_ACTIVE) {
+    atomic_store_8(&pTask->triggerStatus, TASK_TRIGGER_STATUS__ACTIVE);
   }
 
   // TODO: back pressure
@@ -370,7 +399,8 @@ typedef struct {
 
 int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq);
 
-int32_t streamTriggerByWrite(SStreamTask* pTask, int32_t vgId, SMsgCb* pMsgCb);
+int32_t streamLaunchByWrite(SStreamTask* pTask, int32_t vgId, SMsgCb* pMsgCb);
+int32_t streamSetupTrigger(SStreamTask* pTask);
 
 int32_t streamTaskRun(SStreamTask* pTask);
 
diff --git a/source/client/src/tmq.c b/source/client/src/tmq.c
index 10c6c3623a..dd900e6045 100644
--- a/source/client/src/tmq.c
+++ b/source/client/src/tmq.c
@@ -992,6 +992,90 @@ CREATE_MSG_FAIL:
   return -1;
 }
 
+bool tmqUpdateEp2(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) {
+  bool set = false;
+
+  int32_t topicNumGet = taosArrayGetSize(pRsp->topics);
+  char    vgKey[TSDB_TOPIC_FNAME_LEN + 22];
+  tscDebug("consumer %ld update ep epoch %d to epoch %d, topic num: %d", tmq->consumerId, tmq->epoch, epoch,
+           topicNumGet);
+
+  SArray* newTopics = taosArrayInit(topicNumGet, sizeof(SMqClientTopic));
+  if (newTopics == NULL) {
+    return false;
+  }
+
+  SHashObj* pHash = taosHashInit(64, MurmurHash3_32, false, HASH_NO_LOCK);
+  if (pHash == NULL) {
+    taosArrayDestroy(newTopics);
+    return false;
+  }
+  int32_t topicNumCur = taosArrayGetSize(tmq->clientTopics);
+  for (int32_t i = 0; i < topicNumCur; i++) {
+    // find old topic
+    SMqClientTopic* pTopicCur = taosArrayGet(tmq->clientTopics, i);
+    if (pTopicCur->vgs) {
+      int32_t vgNumCur = taosArrayGetSize(pTopicCur->vgs);
+      tscDebug("consumer %ld new vg num: %d", tmq->consumerId, vgNumCur);
+      if (vgNumCur == 0) break;
+      for (int32_t j = 0; j < vgNumCur; j++) {
+        SMqClientVg* pVgCur = taosArrayGet(pTopicCur->vgs, j);
+        sprintf(vgKey, "%s:%d", pTopicCur->topicName, pVgCur->vgId);
+        tscDebug("consumer %ld epoch %d vg %d build %s", tmq->consumerId, epoch, pVgCur->vgId, vgKey);
+        taosHashPut(pHash, vgKey, strlen(vgKey), &pVgCur->currentOffset, sizeof(int64_t));
+      }
+      break;
+    }
+  }
+
+  for (int32_t i = 0; i < topicNumGet; i++) {
+    SMqClientTopic topic = {0};
+    SMqSubTopicEp* pTopicEp = taosArrayGet(pRsp->topics, i);
+    topic.schema = pTopicEp->schema;
+    taosHashClear(pHash);
+    topic.topicName = strdup(pTopicEp->topic);
+    tstrncpy(topic.db, pTopicEp->db, TSDB_DB_FNAME_LEN);
+
+    tscDebug("consumer %ld update topic: %s", tmq->consumerId, topic.topicName);
+
+    int32_t vgNumGet = taosArrayGetSize(pTopicEp->vgs);
+    topic.vgs = taosArrayInit(vgNumGet, sizeof(SMqClientVg));
+    for (int32_t j = 0; j < vgNumGet; j++) {
+      SMqSubVgEp* pVgEp = taosArrayGet(pTopicEp->vgs, j);
+      sprintf(vgKey, "%s:%d", topic.topicName, pVgEp->vgId);
+      int64_t* pOffset = taosHashGet(pHash, vgKey, strlen(vgKey));
+      int64_t  offset = tmq->resetOffsetCfg;
+      if (pOffset != NULL) {
+        offset = *pOffset;
+      }
+
+      tscDebug("consumer %ld(epoch %d) offset of vg %d updated to %ld", tmq->consumerId, epoch, pVgEp->vgId, offset);
+      SMqClientVg clientVg = {
+          .pollCnt = 0,
+          .currentOffset = offset,
+          .vgId = pVgEp->vgId,
+          .epSet = pVgEp->epSet,
+          .vgStatus = TMQ_VG_STATUS__IDLE,
+          .vgSkipCnt = 0,
+      };
+      taosArrayPush(topic.vgs, &clientVg);
+      set = true;
+    }
+    taosArrayPush(newTopics, &topic);
+  }
+  if (tmq->clientTopics) taosArrayDestroy(tmq->clientTopics);
+  taosHashCleanup(pHash);
+  tmq->clientTopics = newTopics;
+
+  if (taosArrayGetSize(tmq->clientTopics) == 0)
+    atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__NO_TOPIC);
+  else
+    atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY);
+
+  atomic_store_32(&tmq->epoch, epoch);
+  return set;
+}
+
 bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) {
   /*printf("call update ep %d\n", epoch);*/
   bool    set = false;
diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h
index 6dd5d1622c..2e543149c0 100644
--- a/source/dnode/mnode/impl/inc/mndDef.h
+++ b/source/dnode/mnode/impl/inc/mndDef.h
@@ -554,8 +554,8 @@ typedef struct {
   SVgObj         fixedSinkVg;
   int64_t        smaId;  // 0 for unused
   int8_t         trigger;
-  int32_t        triggerParam;
-  int64_t        waterMark;
+  int64_t        triggerParam;
+  int64_t        watermark;
   char*          sql;
   char*          physicalPlan;
   SArray*        tasks;  // SArray<SArray<SStreamTask>>
diff --git a/source/dnode/mnode/impl/inc/mndStream.h b/source/dnode/mnode/impl/inc/mndStream.h
index 15cd9fa043..b5d22cb7a5 100644
--- a/source/dnode/mnode/impl/inc/mndStream.h
+++ b/source/dnode/mnode/impl/inc/mndStream.h
@@ -31,7 +31,7 @@ void        mndReleaseStream(SMnode *pMnode, SStreamObj *pStream);
 SSdbRaw *mndStreamActionEncode(SStreamObj *pStream);
 SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw);
 
-int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, int8_t triggerType, int64_t watermark, STrans *pTrans);
+int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, STrans *pTrans);
 
 #ifdef __cplusplus
 }
diff --git a/source/dnode/mnode/impl/src/mndDef.c b/source/dnode/mnode/impl/src/mndDef.c
index 1e2c61ff89..6eac91bb43 100644
--- a/source/dnode/mnode/impl/src/mndDef.c
+++ b/source/dnode/mnode/impl/src/mndDef.c
@@ -33,8 +33,8 @@ int32_t tEncodeSStreamObj(SEncoder *pEncoder, const SStreamObj *pObj) {
   if (tEncodeI8(pEncoder, pObj->status) < 0) return -1;
   if (tEncodeI8(pEncoder, pObj->createdBy) < 0) return -1;
   if (tEncodeI8(pEncoder, pObj->trigger) < 0) return -1;
-  if (tEncodeI32(pEncoder, pObj->triggerParam) < 0) return -1;
-  if (tEncodeI64(pEncoder, pObj->waterMark) < 0) return -1;
+  if (tEncodeI64(pEncoder, pObj->triggerParam) < 0) return -1;
+  if (tEncodeI64(pEncoder, pObj->watermark) < 0) return -1;
   if (tEncodeI32(pEncoder, pObj->fixedSinkVgId) < 0) return -1;
   if (tEncodeI64(pEncoder, pObj->smaId) < 0) return -1;
   if (tEncodeCStr(pEncoder, pObj->sql) < 0) return -1;
@@ -85,8 +85,8 @@ int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj) {
   if (tDecodeI8(pDecoder, &pObj->status) < 0) return -1;
   if (tDecodeI8(pDecoder, &pObj->createdBy) < 0) return -1;
   if (tDecodeI8(pDecoder, &pObj->trigger) < 0) return -1;
-  if (tDecodeI32(pDecoder, &pObj->triggerParam) < 0) return -1;
-  if (tDecodeI64(pDecoder, &pObj->waterMark) < 0) return -1;
+  if (tDecodeI64(pDecoder, &pObj->triggerParam) < 0) return -1;
+  if (tDecodeI64(pDecoder, &pObj->watermark) < 0) return -1;
   if (tDecodeI32(pDecoder, &pObj->fixedSinkVgId) < 0) return -1;
   if (tDecodeI64(pDecoder, &pObj->smaId) < 0) return -1;
   if (tDecodeCStrAlloc(pDecoder, &pObj->sql) < 0) return -1;
diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c
index a3b7754a4e..5b745f3cea 100644
--- a/source/dnode/mnode/impl/src/mndScheduler.c
+++ b/source/dnode/mnode/impl/src/mndScheduler.c
@@ -387,6 +387,9 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
       // input
       pFinalTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
 
+      // trigger
+      pFinalTask->triggerParam = pStream->triggerParam;
+
       // dispatch
       if (mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pFinalTask) < 0) {
         qDestroyQueryPlan(pPlan);
diff --git a/source/dnode/mnode/impl/src/mndSma.c b/source/dnode/mnode/impl/src/mndSma.c
index a14eb78ffe..003afb4fa0 100644
--- a/source/dnode/mnode/impl/src/mndSma.c
+++ b/source/dnode/mnode/impl/src/mndSma.c
@@ -561,6 +561,8 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   streamObj.sql = pCreate->sql;
   streamObj.createdBy = STREAM_CREATED_BY__SMA;
   streamObj.smaId = smaObj.uid;
+  streamObj.watermark = 0;
+  streamObj.trigger = STREAM_TRIGGER_AT_ONCE;
 
   if (mndAllocSmaVgroup(pMnode, pDb, &streamObj.fixedSinkVg) != 0) {
     mError("sma:%s, failed to create since %s", smaObj.name, terrstr());
@@ -583,7 +585,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   if (mndSetUpdateSmaStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
   if (mndSetCreateSmaRedoActions(pMnode, pTrans, pDb, &smaObj) != 0) goto _OVER;
   if (mndSetCreateSmaVgroupRedoActions(pMnode, pTrans, pDb, &streamObj.fixedSinkVg, &smaObj) != 0) goto _OVER;
-  if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, STREAM_TRIGGER_AT_ONCE, 0, pTrans) != 0) goto _OVER;
+  if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, pTrans) != 0) goto _OVER;
   if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
 
   code = 0;
@@ -1012,7 +1014,6 @@ int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool
   rsp->suid = pStb->uid;
   rsp->version = pStb->smaVer;
   mndReleaseStb(pMnode, pStb);
-  
 
   while (1) {
     pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c
index 96d199fcb6..4a7da8f2d1 100644
--- a/source/dnode/mnode/impl/src/mndStream.c
+++ b/source/dnode/mnode/impl/src/mndStream.c
@@ -235,16 +235,15 @@ static int32_t mndStreamGetPlanString(const char *ast, int8_t triggerType, int64
   }
 
   if (TSDB_CODE_SUCCESS == code) {
-    code = nodesNodeToString((SNode*)pPlan, false, pStr, NULL);
+    code = nodesNodeToString((SNode *)pPlan, false, pStr, NULL);
   }
   nodesDestroyNode(pAst);
-  nodesDestroyNode((SNode*)pPlan);
+  nodesDestroyNode((SNode *)pPlan);
   terrno = code;
   return code;
 }
 
-int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, int8_t triggerType, int64_t watermark,
-                            STrans *pTrans) {
+int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, STrans *pTrans) {
   SNode *pAst = NULL;
 
   if (nodesStringToNode(ast, &pAst) < 0) {
@@ -258,7 +257,6 @@ int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast
   // free
   nodesDestroyNode(pAst);
 
-
 #if 0
   printf("|");
   for (int i = 0; i < pStream->outputSchema.nCols; i++) {
@@ -268,7 +266,7 @@ int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast
 
 #endif
 
-  if (TSDB_CODE_SUCCESS != mndStreamGetPlanString(ast, triggerType, watermark, &pStream->physicalPlan)) {
+  if (TSDB_CODE_SUCCESS != mndStreamGetPlanString(ast, pStream->trigger, pStream->watermark, &pStream->physicalPlan)) {
     mError("topic:%s, failed to get plan since %s", pStream->name, terrstr());
     return -1;
   }
@@ -391,7 +389,8 @@ static int32_t mndCreateStream(SMnode *pMnode, SRpcMsg *pReq, SCMCreateStreamReq
   streamObj.smaId = 0;
   /*streamObj.physicalPlan = "";*/
   streamObj.trigger = pCreate->triggerType;
-  streamObj.waterMark = pCreate->watermark;
+  streamObj.watermark = pCreate->watermark;
+  streamObj.triggerParam = pCreate->maxDelay;
 
   if (streamObj.targetSTbName[0]) {
     pDb = mndAcquireDbByStb(pMnode, streamObj.targetSTbName);
@@ -409,7 +408,7 @@ static int32_t mndCreateStream(SMnode *pMnode, SRpcMsg *pReq, SCMCreateStreamReq
   }
   mDebug("trans:%d, used to create stream:%s", pTrans->id, pCreate->name);
 
-  if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, pCreate->triggerType, pCreate->watermark, pTrans) != 0) {
+  if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, pTrans) != 0) {
     mError("trans:%d, failed to add stream since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
@@ -566,7 +565,7 @@ static int32_t mndRetrieveStream(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
     colDataAppend(pColInfo, numOfRows, (const char *)&pStream->targetSTbName, true);
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppend(pColInfo, numOfRows, (const char *)&pStream->waterMark, false);
+    colDataAppend(pColInfo, numOfRows, (const char *)&pStream->watermark, false);
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)&pStream->trigger, false);
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 9f34ae39c0..7d2dfeeba5 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -375,6 +375,8 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
 
   if (pTask->inputQueue == NULL || pTask->outputQueue == NULL) goto FAIL;
 
+  pTask->pMsgCb = &pTq->pVnode->msgCb;
+
   // exec
   if (pTask->execType != TASK_EXEC__NONE) {
     // expand runners
@@ -406,9 +408,12 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
         tdGetSTSChemaFromSSChema(&pTask->tbSink.pSchemaWrapper->pSchema, pTask->tbSink.pSchemaWrapper->nCols);
     ASSERT(pTask->tbSink.pTSchema);
   }
+
+  streamSetupTrigger(pTask);
+
   tqInfo("deploy stream task id %d child id %d on vg %d", pTask->taskId, pTask->childId, pTq->pVnode->config.vgId);
 
-  taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), pTask, sizeof(SStreamTask));
+  taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), &pTask, sizeof(void*));
 
   return 0;
 FAIL:
@@ -431,7 +436,7 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
   while (1) {
     pIter = taosHashIterate(pTq->pStreamTasks, pIter);
     if (pIter == NULL) break;
-    SStreamTask* pTask = (SStreamTask*)pIter;
+    SStreamTask* pTask = *(SStreamTask**)pIter;
     if (pTask->inputType != STREAM_INPUT__DATA_SUBMIT) continue;
 
     if (!failed) {
@@ -439,7 +444,7 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
         continue;
       }
 
-      if (streamTriggerByWrite(pTask, pTq->pVnode->config.vgId, &pTq->pVnode->msgCb) < 0) {
+      if (streamLaunchByWrite(pTask, pTq->pVnode->config.vgId, &pTq->pVnode->msgCb) < 0) {
         continue;
       }
     } else {
@@ -459,7 +464,7 @@ int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) {
   //
   SStreamTaskRunReq* pReq = pMsg->pCont;
   int32_t            taskId = pReq->taskId;
-  SStreamTask*       pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
+  SStreamTask*       pTask = *(SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
   streamTaskProcessRunReq(pTask, &pTq->pVnode->msgCb);
   return 0;
 }
@@ -473,7 +478,7 @@ int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg) {
   tDecoderInit(&decoder, msgBody, msgLen);
   tDecodeStreamDispatchReq(&decoder, &req);
   int32_t      taskId = req.taskId;
-  SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
+  SStreamTask* pTask = *(SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
   SRpcMsg      rsp = {
            .info = pMsg->info,
            .code = 0,
@@ -485,7 +490,7 @@ int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg) {
 int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg) {
   SStreamTaskRecoverReq* pReq = pMsg->pCont;
   int32_t                taskId = pReq->taskId;
-  SStreamTask*           pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
+  SStreamTask*           pTask = *(SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
   streamProcessRecoverReq(pTask, &pTq->pVnode->msgCb, pReq, pMsg);
   return 0;
 }
@@ -493,7 +498,7 @@ int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg) {
 int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) {
   SStreamDispatchRsp* pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
   int32_t             taskId = pRsp->taskId;
-  SStreamTask*        pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
+  SStreamTask*        pTask = *(SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
   streamProcessDispatchRsp(pTask, &pTq->pVnode->msgCb, pRsp);
   return 0;
 }
@@ -501,7 +506,7 @@ int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) {
 int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg) {
   SStreamTaskRecoverRsp* pRsp = pMsg->pCont;
   int32_t                taskId = pRsp->taskId;
-  SStreamTask*           pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
+  SStreamTask*           pTask = *(SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
   streamProcessRecoverRsp(pTask, pRsp);
   return 0;
 }
diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c
index fd62849e56..bebfc75f17 100644
--- a/source/libs/executor/src/executor.c
+++ b/source/libs/executor/src/executor.c
@@ -19,7 +19,8 @@
 #include "tdatablock.h"
 #include "vnode.h"
 
-static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, bool assignUid, char* id) {
+static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, bool assignUid,
+                                char* id) {
   ASSERT(pOperator != NULL);
   if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
     if (pOperator->numOfDownstream == 0) {
@@ -43,6 +44,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
     if (pInfo->blockType == 0) {
       pInfo->blockType = type;
     } else if (pInfo->blockType != type) {
+      ASSERT(0);
       return TSDB_CODE_QRY_APP_ERROR;
     }
 
@@ -51,7 +53,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
         qError("submit msg messed up when initing stream block, %s" PRIx64, id);
         return TSDB_CODE_QRY_APP_ERROR;
       }
-    } else {
+    } else if (type == STREAM_DATA_TYPE_SSDATA_BLOCK) {
       for (int32_t i = 0; i < numOfBlocks; ++i) {
         SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i];
 
@@ -62,6 +64,8 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
         taosArrayAddAll(p->pDataBlock, pDataBlock->pDataBlock);
         taosArrayPush(pInfo->pBlockLists, &p);
       }
+    } else {
+      ASSERT(0);
     }
 
     return TSDB_CODE_SUCCESS;
@@ -83,7 +87,8 @@ int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numO
 
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
 
-  int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void**)pBlocks, numOfBlocks, type, assignUid, GET_TASKID(pTaskInfo));
+  int32_t code =
+      doSetStreamBlock(pTaskInfo->pRoot, (void**)pBlocks, numOfBlocks, type, assignUid, GET_TASKID(pTaskInfo));
   if (code != TSDB_CODE_SUCCESS) {
     qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));
   } else {
@@ -162,7 +167,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
     pInfo = pInfo->pDownstream[0];
   }
 
-  int32_t code = 0;
+  int32_t               code = 0;
   SStreamBlockScanInfo* pScanInfo = pInfo->info;
   if (isAdd) {  // add new table id
     SArray* qa = filterQualifiedChildTables(pScanInfo, tableIdList);
@@ -178,9 +183,10 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
   return code;
 }
 
-int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion, int32_t* tversion) {
+int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion,
+                                      int32_t* tversion) {
   ASSERT(tinfo != NULL && dbName != NULL && tableName != NULL);
-  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) tinfo;
+  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
 
   *sversion = pTaskInfo->schemaVer.sversion;
   *tversion = pTaskInfo->schemaVer.tversion;
@@ -196,4 +202,4 @@ int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tab
   }
 
   return 0;
-}
\ No newline at end of file
+}
diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c
index 833f6db0c3..62a0c663b6 100644
--- a/source/libs/function/src/builtins.c
+++ b/source/libs/function/src/builtins.c
@@ -1682,7 +1682,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "top",
     .type = FUNCTION_TYPE_TOP,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
     .translateFunc = translateTopBot,
     .getEnvFunc   = getTopBotFuncEnv,
     .initFunc     = topBotFunctionSetup,
@@ -1717,7 +1717,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "bottom",
     .type = FUNCTION_TYPE_BOTTOM,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
     .translateFunc = translateTopBot,
     .getEnvFunc   = getTopBotFuncEnv,
     .initFunc     = topBotFunctionSetup,
diff --git a/source/libs/stream/inc/streamInc.h b/source/libs/stream/inc/streamInc.h
index 27746fff78..8aaf4953de 100644
--- a/source/libs/stream/inc/streamInc.h
+++ b/source/libs/stream/inc/streamInc.h
@@ -23,6 +23,13 @@
 extern "C" {
 #endif
 
+typedef struct {
+  int8_t inited;
+  void*  timer;
+} SStreamGlobalEnv;
+
+static SStreamGlobalEnv streamEnv;
+
 int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb);
 int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb);
 int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock* pData);
diff --git a/source/libs/stream/src/stream.c b/source/libs/stream/src/stream.c
index d3828d9ee4..6dcbfad957 100644
--- a/source/libs/stream/src/stream.c
+++ b/source/libs/stream/src/stream.c
@@ -14,8 +14,74 @@
  */
 
 #include "streamInc.h"
+#include "ttimer.h"
 
-int32_t streamTriggerByWrite(SStreamTask* pTask, int32_t vgId, SMsgCb* pMsgCb) {
+int32_t streamInit() {
+  int8_t old;
+  while (1) {
+    old = atomic_val_compare_exchange_8(&streamEnv.inited, 0, 2);
+    if (old != 2) break;
+  }
+
+  if (old == 0) {
+    streamEnv.timer = taosTmrInit(10000, 100, 10000, "STREAM");
+    if (streamEnv.timer == NULL) {
+      atomic_store_8(&streamEnv.inited, 0);
+      return -1;
+    }
+    atomic_store_8(&streamEnv.inited, 1);
+  }
+  return 0;
+}
+
+void streamCleanUp() {
+  int8_t old;
+  while (1) {
+    old = atomic_val_compare_exchange_8(&streamEnv.inited, 1, 2);
+    if (old != 2) break;
+  }
+
+  if (old == 1) {
+    taosTmrCleanUp(streamEnv.timer);
+    atomic_store_8(&streamEnv.inited, 0);
+  }
+}
+
+void streamTriggerByTimer(void* param, void* tmrId) {
+  SStreamTask* pTask = (void*)param;
+
+  if (atomic_load_8(&pTask->triggerStatus) == TASK_TRIGGER_STATUS__ACTIVE) {
+    SStreamTrigger* trigger = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM);
+    if (trigger == NULL) return;
+    trigger->type = STREAM_INPUT__TRIGGER;
+    trigger->pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
+    if (trigger->pBlock == NULL) {
+      taosFreeQitem(trigger);
+      return;
+    }
+    trigger->pBlock->info.type = STREAM_GET_ALL;
+
+    atomic_store_8(&pTask->triggerStatus, TASK_TRIGGER_STATUS__IN_ACTIVE);
+
+    streamTaskInput(pTask, (SStreamQueueItem*)trigger);
+    streamLaunchByWrite(pTask, pTask->nodeId, pTask->pMsgCb);
+  }
+
+  taosTmrReset(streamTriggerByTimer, (int32_t)pTask->triggerParam, pTask, streamEnv.timer, &pTask->timer);
+}
+
+int32_t streamSetupTrigger(SStreamTask* pTask) {
+  if (pTask->triggerParam != 0) {
+    if (streamInit() < 0) {
+      return -1;
+    }
+    pTask->timer = taosTmrStart(streamTriggerByTimer, (int32_t)pTask->triggerParam, pTask, streamEnv.timer);
+    pTask->triggerStatus = TASK_TRIGGER_STATUS__IN_ACTIVE;
+  }
+  return 0;
+}
+
+int32_t streamLaunchByWrite(SStreamTask* pTask, int32_t vgId, SMsgCb* pMsgCb) {
   int8_t execStatus = atomic_load_8(&pTask->status);
   if (execStatus == TASK_STATUS__IDLE || execStatus == TASK_STATUS__CLOSING) {
     SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq));
diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index fe1a857743..04428136ae 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -20,15 +20,17 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes)
   void* exec = pTask->exec.executor;
 
   // set input
-  if (pTask->inputType == STREAM_INPUT__DATA_SUBMIT) {
+  SStreamQueueItem* pItem = (SStreamQueueItem*)data;
+  if (pItem->type == STREAM_INPUT__TRIGGER) {
+    SStreamTrigger* pTrigger = (SStreamTrigger*)data;
+    qSetMultiStreamInput(exec, pTrigger->pBlock, 1, STREAM_DATA_TYPE_SSDATA_BLOCK, false);
+  } else if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {
     SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)data;
-    ASSERT(pSubmit->type == STREAM_INPUT__DATA_SUBMIT);
-
+    ASSERT(pTask->inputType == STREAM_INPUT__DATA_SUBMIT);
     qSetStreamInput(exec, pSubmit->data, STREAM_DATA_TYPE_SUBMIT_BLOCK, false);
-  } else if (pTask->inputType == STREAM_INPUT__DATA_BLOCK) {
+  } else if (pItem->type == STREAM_INPUT__DATA_BLOCK) {
     SStreamDataBlock* pBlock = (SStreamDataBlock*)data;
-    ASSERT(pBlock->type == STREAM_INPUT__DATA_BLOCK);
-
+    ASSERT(pTask->inputType == STREAM_INPUT__DATA_BLOCK);
     SArray* blocks = pBlock->blocks;
     qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_DATA_TYPE_SSDATA_BLOCK, false);
   }
diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c
index 5d8546bcb9..7a7d9d15ad 100644
--- a/source/libs/stream/src/streamTask.c
+++ b/source/libs/stream/src/streamTask.c
@@ -72,6 +72,7 @@ int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) {
     if (tSerializeSUseDbRspImp(pEncoder, &pTask->shuffleDispatcher.dbInfo) < 0) return -1;
     /*if (tEncodeI8(pEncoder, pTask->shuffleDispatcher.hashMethod) < 0) return -1;*/
   }
+  if (tEncodeI64(pEncoder, pTask->triggerParam) < 0) return -1;
 
   /*tEndEncode(pEncoder);*/
   return pEncoder->pos;
@@ -121,6 +122,7 @@ int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
     /*if (tDecodeI8(pDecoder, &pTask->shuffleDispatcher.hashMethod) < 0) return -1;*/
     if (tDeserializeSUseDbRspImp(pDecoder, &pTask->shuffleDispatcher.dbInfo) < 0) return -1;
   }
+  if (tDecodeI64(pDecoder, &pTask->triggerParam) < 0) return -1;
 
   /*tEndDecode(pDecoder);*/
   return 0;
diff --git a/source/util/src/tarray.c b/source/util/src/tarray.c
index 2357f760d1..ec90e5a9b9 100644
--- a/source/util/src/tarray.c
+++ b/source/util/src/tarray.c
@@ -174,7 +174,11 @@ void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp)
 }
 
 void* taosArrayAddAll(SArray* pArray, const SArray* pInput) {
-  return taosArrayAddBatch(pArray, pInput->pData, (int32_t)taosArrayGetSize(pInput));
+  if (pInput) {
+    return taosArrayAddBatch(pArray, pInput->pData, (int32_t)taosArrayGetSize(pInput));
+  } else {
+    return NULL;
+  }
 }
 
 void* taosArrayPop(SArray* pArray) {

From 4b18bd718eaa4767579aec660c902d2751632542 Mon Sep 17 00:00:00 2001
From: Minghao Li <castermode@gmail.com>
Date: Wed, 15 Jun 2022 16:14:17 +0800
Subject: [PATCH 12/14] refactor(sync): adjust errno

---
 include/libs/sync/sync.h                      | 10 +----
 include/util/taoserror.h                      |  2 +
 source/dnode/mnode/impl/src/mndMain.c         | 15 ++++---
 source/dnode/mnode/impl/src/mndSync.c         | 18 ++++----
 source/dnode/vnode/src/vnd/vnodeSvr.c         |  9 ++--
 source/dnode/vnode/src/vnd/vnodeSync.c        | 27 ++++--------
 source/libs/sync/src/syncMain.c               | 44 ++++++++++++-------
 .../test/syncConfigChangeSnapshotTest.cpp     |  2 +-
 .../libs/sync/test/syncConfigChangeTest.cpp   |  2 +-
 source/libs/sync/test/syncReplicateTest.cpp   |  2 +-
 source/libs/sync/test/syncTestTool.cpp        |  2 +-
 source/util/src/terror.c                      |  2 +
 12 files changed, 71 insertions(+), 64 deletions(-)

diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h
index a369b81d26..2d21b3531a 100644
--- a/include/libs/sync/sync.h
+++ b/include/libs/sync/sync.h
@@ -24,7 +24,7 @@ extern "C" {
 #include "tdef.h"
 #include "tmsgcb.h"
 
-#define SYNC_INDEX_BEGIN 0
+#define SYNC_INDEX_BEGIN   0
 #define SYNC_INDEX_INVALID -1
 
 typedef uint64_t SyncNodeId;
@@ -44,14 +44,6 @@ typedef enum {
   TAOS_SYNC_STATE_ERROR = 103,
 } ESyncState;
 
-typedef enum {
-  TAOS_SYNC_PROPOSE_SUCCESS = 0,
-  TAOS_SYNC_PROPOSE_NOT_LEADER = 1,
-  TAOS_SYNC_ONLY_ONE_REPLICA = 2,
-  TAOS_SYNC_NOT_IN_NEW_CONFIG = 3,
-  TAOS_SYNC_OTHER_ERROR = 100,
-} ESyncProposeCode;
-
 typedef enum {
   TAOS_SYNC_FSM_CB_SUCCESS = 0,
   TAOS_SYNC_FSM_CB_OTHER_ERROR = 1,
diff --git a/include/util/taoserror.h b/include/util/taoserror.h
index 30cbbbeb11..6fc84e023d 100644
--- a/include/util/taoserror.h
+++ b/include/util/taoserror.h
@@ -411,6 +411,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_SYN_INVALID_MSGTYPE           TAOS_DEF_ERROR_CODE(0, 0x090A)
 
 #define TSDB_CODE_SYN_NOT_LEADER                TAOS_DEF_ERROR_CODE(0, 0x0910)
+#define TSDB_CODE_SYN_ONE_REPLICA               TAOS_DEF_ERROR_CODE(0, 0x0911)
+#define TSDB_CODE_SYN_NOT_IN_NEW_CONFIG         TAOS_DEF_ERROR_CODE(0, 0x0912)
 #define TSDB_CODE_SYN_INTERNAL_ERROR            TAOS_DEF_ERROR_CODE(0, 0x09FF)
 
 // tq
diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c
index 59599ee134..00118eac1e 100644
--- a/source/dnode/mnode/impl/src/mndMain.c
+++ b/source/dnode/mnode/impl/src/mndMain.c
@@ -380,17 +380,19 @@ void mndStop(SMnode *pMnode) {
 int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
   SMnode    *pMnode = pMsg->info.node;
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
-  int32_t    code = TAOS_SYNC_OTHER_ERROR;
+  int32_t    code = 0;
 
   if (!syncEnvIsStart()) {
     mError("failed to process sync msg:%p type:%s since syncEnv stop", pMsg, TMSG_INFO(pMsg->msgType));
-    return TAOS_SYNC_OTHER_ERROR;
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+    return -1;
   }
 
   SSyncNode *pSyncNode = syncNodeAcquire(pMgmt->sync);
   if (pSyncNode == NULL) {
     mError("failed to process sync msg:%p type:%s since syncNode is null", pMsg, TMSG_INFO(pMsg->msgType));
-    return TAOS_SYNC_OTHER_ERROR;
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+    return -1;
   }
 
   char  logBuf[512] = {0};
@@ -451,7 +453,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
       tmsgSendRsp(&rsp);
     } else {
       mError("failed to process msg:%p since invalid type:%s", pMsg, TMSG_INFO(pMsg->msgType));
-      code = TAOS_SYNC_OTHER_ERROR;
+      code = -1;
     }
   } else {
     if (pMsg->msgType == TDMT_SYNC_TIMEOUT) {
@@ -492,10 +494,13 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
       tmsgSendRsp(&rsp);
     } else {
       mError("failed to process msg:%p since invalid type:%s", pMsg, TMSG_INFO(pMsg->msgType));
-      code = TAOS_SYNC_OTHER_ERROR;
+      code = -1;
     }
   }
 
+  if (code != 0) {
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+  }
   return code;
 }
 
diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c
index 9940037356..63708aef8a 100644
--- a/source/dnode/mnode/impl/src/mndSync.c
+++ b/source/dnode/mnode/impl/src/mndSync.c
@@ -234,9 +234,9 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
   int32_t    code = syncPropose(pMgmt->sync, &rsp, isWeak);
   if (code == 0) {
     tsem_wait(&pMgmt->syncSem);
-  } else if (code == TAOS_SYNC_PROPOSE_NOT_LEADER) {
+  } else if (code == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
     terrno = TSDB_CODE_APP_NOT_READY;
-  } else if (code == TAOS_SYNC_OTHER_ERROR) {
+  } else if (code == -1 && terrno == TSDB_CODE_SYN_INTERNAL_ERROR) {
     terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
   } else {
     terrno = TSDB_CODE_APP_ERROR;
@@ -257,13 +257,13 @@ void mndSyncStart(SMnode *pMnode) {
   syncStart(pMgmt->sync);
   mDebug("mnode sync started, id:%" PRId64 " standby:%d", pMgmt->sync, pMgmt->standby);
 
-/*
-  if (pMgmt->standby) {
-    syncStartStandBy(pMgmt->sync);
-  } else {
-    syncStart(pMgmt->sync);
-  }
-*/  
+  /*
+    if (pMgmt->standby) {
+      syncStartStandBy(pMgmt->sync);
+    } else {
+      syncStart(pMgmt->sync);
+    }
+  */
 }
 
 void mndSyncStop(SMnode *pMnode) {}
diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c
index cd64bc8a9c..34ff9ffa0f 100644
--- a/source/dnode/vnode/src/vnd/vnodeSvr.c
+++ b/source/dnode/vnode/src/vnd/vnodeSvr.c
@@ -296,7 +296,7 @@ void vnodeUpdateMetaRsp(SVnode *pVnode, STableMetaRsp *pMetaRsp) {
 }
 
 int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
-  int32_t ret = TAOS_SYNC_OTHER_ERROR;
+  int32_t ret = 0;
 
   if (syncEnvIsStart()) {
     SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync);
@@ -381,15 +381,18 @@ int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
       tmsgSendRsp(&rsp);
     } else {
       vError("==vnodeProcessSyncReq== error msg type:%d", pRpcMsg->msgType);
-      ret = TAOS_SYNC_OTHER_ERROR;
+      ret = -1;
     }
 
     syncNodeRelease(pSyncNode);
   } else {
     vError("==vnodeProcessSyncReq== error syncEnv stop");
-    ret = TAOS_SYNC_OTHER_ERROR;
+    ret = -1;
   }
 
+  if (ret != 0) {
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+  }
   return ret;
 }
 
diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c
index 17c2c186be..975dff9fb6 100644
--- a/source/dnode/vnode/src/vnd/vnodeSync.c
+++ b/source/dnode/vnode/src/vnd/vnodeSync.c
@@ -98,7 +98,8 @@ void vnodeProposeMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
 
     if (code == 0) {
       vnodeAccumBlockMsg(pVnode, pMsg->msgType);
-    } else if (code == TAOS_SYNC_PROPOSE_NOT_LEADER) {
+
+    } else if (code == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
       SEpSet newEpSet = {0};
       syncGetEpSet(pVnode->sync, &newEpSet);
       SEp *pEp = &newEpSet.eps[newEpSet.inUse];
@@ -247,29 +248,17 @@ static void vnodeSyncRollBackMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta
   syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg);
 }
 
-int32_t vnodeSnapshotStartRead(struct SSyncFSM *pFsm, void **ppReader) {
-  return 0;
-}
+int32_t vnodeSnapshotStartRead(struct SSyncFSM *pFsm, void **ppReader) { return 0; }
 
-int32_t vnodeSnapshotStopRead(struct SSyncFSM *pFsm, void *pReader) {
-  return 0;
-}
+int32_t vnodeSnapshotStopRead(struct SSyncFSM *pFsm, void *pReader) { return 0; }
 
-int32_t vnodeSnapshotDoRead(struct SSyncFSM *pFsm, void *pReader, void **ppBuf, int32_t *len) {
-  return 0;
-}
+int32_t vnodeSnapshotDoRead(struct SSyncFSM *pFsm, void *pReader, void **ppBuf, int32_t *len) { return 0; }
 
-int32_t vnodeSnapshotStartWrite(struct SSyncFSM *pFsm, void **ppWriter) {
-  return 0;
-}
+int32_t vnodeSnapshotStartWrite(struct SSyncFSM *pFsm, void **ppWriter) { return 0; }
 
-int32_t vnodeSnapshotStopWrite(struct SSyncFSM *pFsm, void *pWriter, bool isApply) {
-  return 0;
-}
+int32_t vnodeSnapshotStopWrite(struct SSyncFSM *pFsm, void *pWriter, bool isApply) { return 0; }
 
-int32_t vnodeSnapshotDoWrite(struct SSyncFSM *pFsm, void *pWriter, void *pBuf, int32_t len) {
-  return 0;
-}
+int32_t vnodeSnapshotDoWrite(struct SSyncFSM *pFsm, void *pWriter, void *pBuf, int32_t len) { return 0; }
 
 static SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode) {
   SSyncFSM *pFsm = taosMemoryCalloc(1, sizeof(SSyncFSM));
diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c
index 7b1fd010ab..953294e489 100644
--- a/source/libs/sync/src/syncMain.c
+++ b/source/libs/sync/src/syncMain.c
@@ -149,12 +149,14 @@ void syncStop(int64_t rid) {
 int32_t syncSetStandby(int64_t rid) {
   SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
-    return TAOS_SYNC_OTHER_ERROR;
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+    return -1;
   }
 
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
     taosReleaseRef(tsNodeRefId, pSyncNode->rid);
-    return TAOS_SYNC_OTHER_ERROR;
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+    return -1;
   }
 
   // state change
@@ -177,7 +179,8 @@ int32_t syncSetStandby(int64_t rid) {
 int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg) {
   SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
-    return TAOS_SYNC_OTHER_ERROR;
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+    return -1;
   }
   ASSERT(rid == pSyncNode->rid);
 
@@ -201,7 +204,8 @@ int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg
 
   if (!IamInNew) {
     taosReleaseRef(tsNodeRefId, pSyncNode->rid);
-    return TAOS_SYNC_NOT_IN_NEW_CONFIG;
+    terrno = TSDB_CODE_SYN_NOT_IN_NEW_CONFIG;
+    return -1;
   }
 
   char* newconfig = syncCfg2Str((SSyncCfg*)pNewCfg);
@@ -219,7 +223,8 @@ int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg
 int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg) {
   SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
-    return TAOS_SYNC_OTHER_ERROR;
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+    return -1;
   }
   ASSERT(rid == pSyncNode->rid);
 
@@ -246,7 +251,8 @@ int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg) {
   if (!IamInNew) {
     sError("sync reconfig error, not in new config");
     taosReleaseRef(tsNodeRefId, pSyncNode->rid);
-    return TAOS_SYNC_NOT_IN_NEW_CONFIG;
+    terrno = TSDB_CODE_SYN_NOT_IN_NEW_CONFIG;
+    return -1;
   }
 
   char* newconfig = syncCfg2Str((SSyncCfg*)pNewCfg);
@@ -272,13 +278,15 @@ int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg) {
 int32_t syncLeaderTransfer(int64_t rid) {
   SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
-    return TAOS_SYNC_OTHER_ERROR;
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+    return -1;
   }
   ASSERT(rid == pSyncNode->rid);
 
   if (pSyncNode->peersNum == 0) {
     taosReleaseRef(tsNodeRefId, pSyncNode->rid);
-    return TAOS_SYNC_OTHER_ERROR;
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+    return -1;
   }
 
   SNodeInfo newLeader = (pSyncNode->peersNodeInfo)[0];
@@ -291,7 +299,8 @@ int32_t syncLeaderTransfer(int64_t rid) {
 int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) {
   SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
-    return TAOS_SYNC_OTHER_ERROR;
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+    return -1;
   }
   ASSERT(rid == pSyncNode->rid);
   int32_t ret = 0;
@@ -299,7 +308,8 @@ int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) {
   if (pSyncNode->replicaNum == 1) {
     sError("only one replica, cannot drop leader");
     taosReleaseRef(tsNodeRefId, pSyncNode->rid);
-    return TAOS_SYNC_ONLY_ONE_REPLICA;
+    terrno = TSDB_CODE_SYN_ONE_REPLICA;
+    return -1;
   }
 
   SyncLeaderTransfer* pMsg = syncLeaderTransferBuild(pSyncNode->vgId);
@@ -538,11 +548,12 @@ void setHeartbeatTimerMS(int64_t rid, int32_t hbTimerMS) {
 }
 
 int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
-  int32_t ret = TAOS_SYNC_PROPOSE_SUCCESS;
+  int32_t ret = 0;
 
   SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
-    return TAOS_SYNC_OTHER_ERROR;
+    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
+    return -1;
   }
   assert(rid == pSyncNode->rid);
   sDebug("vgId:%d sync event propose msgType:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType));
@@ -553,7 +564,7 @@ int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
 }
 
 int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak) {
-  int32_t ret = TAOS_SYNC_PROPOSE_SUCCESS;
+  int32_t ret = 0;
   sDebug("vgId:%d sync event propose msgType:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType));
 
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
@@ -567,14 +578,17 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak)
     syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
 
     if (pSyncNode->FpEqMsg != NULL && (*pSyncNode->FpEqMsg)(pSyncNode->msgcb, &rpcMsg) == 0) {
-      ret = TAOS_SYNC_PROPOSE_SUCCESS;
+      ret = 0;
     } else {
+      ret = -1;
+      terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
       sError("syncPropose pSyncNode->FpEqMsg is NULL");
     }
     syncClientRequestDestroy(pSyncMsg);
   } else {
+    ret = -1;
+    terrno = TSDB_CODE_SYN_NOT_LEADER;
     sError("syncPropose not leader, %s", syncUtilState2String(pSyncNode->state));
-    ret = TAOS_SYNC_PROPOSE_NOT_LEADER;
   }
 
   return ret;
diff --git a/source/libs/sync/test/syncConfigChangeSnapshotTest.cpp b/source/libs/sync/test/syncConfigChangeSnapshotTest.cpp
index 10b54d0aa4..3ee2b458d6 100644
--- a/source/libs/sync/test/syncConfigChangeSnapshotTest.cpp
+++ b/source/libs/sync/test/syncConfigChangeSnapshotTest.cpp
@@ -338,7 +338,7 @@ int main(int argc, char** argv) {
     if (alreadySend < writeRecordNum) {
       SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex);
       int32_t  ret = syncPropose(rid, pRpcMsg, false);
-      if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) {
+      if (ret == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
         sTrace("%s value%d write not leader", s, alreadySend);
       } else {
         assert(ret == 0);
diff --git a/source/libs/sync/test/syncConfigChangeTest.cpp b/source/libs/sync/test/syncConfigChangeTest.cpp
index 1e64a8a6f7..b1dc53d804 100644
--- a/source/libs/sync/test/syncConfigChangeTest.cpp
+++ b/source/libs/sync/test/syncConfigChangeTest.cpp
@@ -251,7 +251,7 @@ int main(int argc, char** argv) {
     if (alreadySend < writeRecordNum) {
       SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex);
       int32_t  ret = syncPropose(rid, pRpcMsg, false);
-      if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) {
+      if (ret == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
         sTrace("%s value%d write not leader", s, alreadySend);
       } else {
         assert(ret == 0);
diff --git a/source/libs/sync/test/syncReplicateTest.cpp b/source/libs/sync/test/syncReplicateTest.cpp
index bf9e34fffb..d955e64f81 100644
--- a/source/libs/sync/test/syncReplicateTest.cpp
+++ b/source/libs/sync/test/syncReplicateTest.cpp
@@ -188,7 +188,7 @@ int main(int argc, char** argv) {
     if (alreadySend < writeRecordNum) {
       SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex);
       int32_t  ret = syncPropose(rid, pRpcMsg, false);
-      if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) {
+      if (ret == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
         sTrace("%s value%d write not leader", s, alreadySend);
       } else {
         assert(ret == 0);
diff --git a/source/libs/sync/test/syncTestTool.cpp b/source/libs/sync/test/syncTestTool.cpp
index ebcd7368cc..21049454f4 100644
--- a/source/libs/sync/test/syncTestTool.cpp
+++ b/source/libs/sync/test/syncTestTool.cpp
@@ -391,7 +391,7 @@ int main(int argc, char** argv) {
     if (alreadySend < writeRecordNum) {
       SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex);
       int32_t  ret = syncPropose(rid, pRpcMsg, false);
-      if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) {
+      if (ret == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
         sTrace("%s value%d write not leader, leaderTransferWait:%d", simpleStr, alreadySend, leaderTransferWait);
       } else {
         assert(ret == 0);
diff --git a/source/util/src/terror.c b/source/util/src/terror.c
index 8e6284d2cb..621f947b64 100644
--- a/source/util/src/terror.c
+++ b/source/util/src/terror.c
@@ -413,6 +413,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_MSGLEN,           "Invalid msg length")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_MSGTYPE,          "Invalid msg type")
 
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_LEADER,               "Sync not leader")
+TAOS_DEFINE_ERROR(TSDB_CODE_SYN_ONE_REPLICA,              "Sync one replica")
+TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_IN_NEW_CONFIG,        "Sync not in new config")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INTERNAL_ERROR,           "Sync internal error")
 
 // wal

From 87bcbe003ee84d5d3dd38fe685bbc966a009f46b Mon Sep 17 00:00:00 2001
From: 54liuyao <54liuyao@163.com>
Date: Wed, 15 Jun 2022 16:40:45 +0800
Subject: [PATCH 13/14] feat(stream): stream state&session support partition by

---
 source/libs/executor/inc/executorimpl.h       |   8 +-
 source/libs/executor/src/executorimpl.c       |   7 +-
 source/libs/executor/src/scanoperator.c       |   2 +-
 source/libs/executor/src/timewindowoperator.c | 178 ++++++++++--------
 4 files changed, 108 insertions(+), 87 deletions(-)

diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h
index c4d16c89e1..18d35ecc56 100644
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@@ -365,7 +365,9 @@ typedef struct SCatchSupporter {
 } SCatchSupporter;
 
 typedef struct SStreamAggSupporter {
-  SArray*        pResultRows;
+  SHashObj*      pResultRows;
+  SArray*        pCurWins;
+  int32_t        valueSize;
   int32_t        keySize;
   char*          pKeyBuf;              // window key buffer
   SDiskbasedBuf* pResultBuf;           // query result buffer based on blocked-wised disk file
@@ -899,9 +901,9 @@ int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimary
     __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
 int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
 int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey,
-    SqlFunctionCtx* pCtx, int32_t numOfOutput, size_t size);
+    SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t size);
 SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, int32_t interBufSize);
-SResultWindowInfo* getSessionTimeWindow(SArray* pWinInfos, TSKEY ts, int64_t gap, int32_t* pIndex);
+SResultWindowInfo* getSessionTimeWindow(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, int64_t gap, int32_t* pIndex);
 int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pTs, int32_t rows,
     int32_t start, int64_t gap, SHashObj* pStDeleted);
 bool functionNeedToExecute(SqlFunctionCtx* pCtx);
diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index 99a9dcb6af..53477eb56f 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -4814,6 +4814,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
         .calTrigger = pIntervalPhyNode->window.triggerType,
         .maxTs = INT64_MIN,
     };
+    ASSERT(as.calTrigger != STREAM_TRIGGER_MAX_DELAY);
 
     int32_t tsSlotId = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
     bool    isStream = (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type);
@@ -5498,14 +5499,16 @@ int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SExplainExecInfo
 }
 
 int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput,
-                               size_t size) {
+                               int32_t size) {
   pSup->resultRowSize = getResultRowSize(pCtx, numOfOutput);
   pSup->keySize = sizeof(int64_t) + sizeof(TSKEY);
   pSup->pKeyBuf = taosMemoryCalloc(1, pSup->keySize);
-  pSup->pResultRows = taosArrayInit(1024, size);
+  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
+  pSup->pResultRows = taosHashInit(1024, hashFn, false, HASH_NO_LOCK);
   if (pSup->pKeyBuf == NULL || pSup->pResultRows == NULL) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
+  pSup->valueSize = size;
 
   pSup->pScanWindow = taosArrayInit(4, sizeof(STimeWindow));
 
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index e44e05224f..571e9ae120 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -716,7 +716,7 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
       int64_t              gap = pInfo->sessionSup.gap;
       int32_t              winIndex = 0;
       SResultWindowInfo*   pCurWin =
-          getSessionTimeWindow(pAggSup->pResultRows, tsCols[pInfo->updateResIndex], gap, &winIndex);
+          getSessionTimeWindow(pAggSup, tsCols[pInfo->updateResIndex], pSDB->info.groupId, gap, &winIndex);
       win = pCurWin->win;
       pInfo->updateResIndex +=
           updateSessionWindowInfo(pCurWin, tsCols, pSDB->info.rows, pInfo->updateResIndex, gap, NULL);
diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c
index 5a18649cab..738c5b2b26 100644
--- a/source/libs/executor/src/timewindowoperator.c
+++ b/source/libs/executor/src/timewindowoperator.c
@@ -1320,6 +1320,15 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
       break;
     }
 
+    if (pBlock->info.type == STREAM_REPROCESS) {
+      doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval, 0, pOperator->numOfExprs, pBlock, NULL);
+      qDebug("%s clear existed time window results for updates checked", GET_TASKID(pTaskInfo));
+      continue;
+    } else if (pBlock->info.type == STREAM_GET_ALL) {
+      getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated);
+      continue;
+    }
+
     // The timewindow that overlaps the timestamps of the input pBlock need to be recalculated and return to the
     // caller. Note that all the time window are not close till now.
     // the pDataBlock are always the same one, no need to call this again
@@ -1328,16 +1337,6 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
       setInverFunction(pInfo->binfo.pCtx, pOperator->numOfExprs, pBlock->info.type);
     }
 
-    if (pBlock->info.type == STREAM_REPROCESS) {
-      doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval, 0, pOperator->numOfExprs, pBlock, NULL);
-      qDebug("%s clear existed time window results for updates checked", GET_TASKID(pTaskInfo));
-      continue;
-    } else if (pBlock->info.type == STREAM_GET_ALL &&
-        pInfo->twAggSup.calTrigger == STREAM_TRIGGER_MAX_DELAY) {
-      getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated);
-      continue;
-    }
-
     pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
     hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, MAIN_SCAN, pUpdated);
   }
@@ -2038,7 +2037,6 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
       break;
     }
 
-    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
     if (pBlock->info.type == STREAM_REPROCESS) {
       SArray* pUpWins = taosArrayInit(8, sizeof(STimeWindow));
       doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval, pInfo->primaryTsIndex, pOperator->numOfExprs,
@@ -2058,12 +2056,12 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
       copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex);
       taosArrayDestroy(pUpWins);
       break;
-    } else if (pBlock->info.type == STREAM_GET_ALL && isFinalInterval(pInfo) &&
-        pInfo->twAggSup.calTrigger == STREAM_TRIGGER_MAX_DELAY) {
+    } else if (pBlock->info.type == STREAM_GET_ALL && isFinalInterval(pInfo)) {
       getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated);
       continue;
     }
 
+    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
     if (isFinalInterval(pInfo)) {
       int32_t chIndex = getChildIndex(pBlock);
       int32_t size = taosArrayGetSize(pInfo->pChildren);
@@ -2125,6 +2123,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
       .calTrigger = pIntervalPhyNode->window.triggerType,
       .maxTs = INT64_MIN,
   };
+  ASSERT(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY);
   pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
   size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
   initResultSizeInfo(pOperator, 4096);
@@ -2190,8 +2189,13 @@ _error:
 }
 
 void destroyStreamAggSupporter(SStreamAggSupporter* pSup) {
-  taosArrayDestroy(pSup->pResultRows);
   taosMemoryFreeClear(pSup->pKeyBuf);
+  void **pIte = NULL;
+  while ((pIte = taosHashIterate(pSup->pResultRows, pIte)) != NULL) {
+    SArray *pWins = (SArray *) (*pIte);
+    taosArrayDestroy(pWins);
+  }
+  taosHashCleanup(pSup->pResultRows);
   destroyDiskbasedBuf(pSup->pResultBuf);
 }
 
@@ -2333,7 +2337,22 @@ static SResultWindowInfo* addNewSessionWindow(SArray* pWinInfos, TSKEY ts) {
   return taosArrayPush(pWinInfos, &win);
 }
 
-SResultWindowInfo* getSessionTimeWindow(SArray* pWinInfos, TSKEY ts, int64_t gap, int32_t* pIndex) {
+SArray* getWinInfos(SStreamAggSupporter* pAggSup, uint64_t groupId) {
+  void** ite = taosHashGet(pAggSup->pResultRows, &groupId, sizeof(uint64_t));
+  SArray* pWinInfos = NULL;
+  if (ite == NULL) {
+    pWinInfos = taosArrayInit(1024, pAggSup->valueSize);
+    taosHashPut(pAggSup->pResultRows, &groupId, sizeof(uint64_t), &pWinInfos, sizeof(void *));
+  } else {
+    pWinInfos = *ite;
+  }
+  return pWinInfos;
+}
+
+SResultWindowInfo* getSessionTimeWindow(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, int64_t gap, int32_t* pIndex) {
+  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
+  pAggSup->pCurWins = pWinInfos;
+
   int32_t size = taosArrayGetSize(pWinInfos);
   if (size == 0) {
     *pIndex = 0;
@@ -2389,7 +2408,7 @@ static int32_t setWindowOutputBuf(SResultWindowInfo* pWinInfo, SResultRow** pRes
                                   SStreamAggSupporter* pAggSup, SExecTaskInfo* pTaskInfo) {
   assert(pWinInfo->win.skey <= pWinInfo->win.ekey);
   // too many time window in query
-  int32_t size = taosArrayGetSize(pAggSup->pResultRows);
+  int32_t size = taosArrayGetSize(pAggSup->pCurWins);
   if (size > MAX_INTERVAL_TIME_WINDOW) {
     longjmp(pTaskInfo->env, TSDB_CODE_QRY_TOO_MANY_TIMEWINDOW);
   }
@@ -2449,25 +2468,6 @@ static int32_t doOneStateWindowAgg(SStreamStateAggOperatorInfo* pInfo, SSDataBlo
                             pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pTaskInfo);
 }
 
-int32_t copyWinInfoToDataBlock(SSDataBlock* pBlock, SStreamAggSupporter* pAggSup, int32_t start, int32_t num,
-                               int32_t numOfExprs, SOptrBasicInfo* pBinfo) {
-  for (int32_t i = start; i < num; i += 1) {
-    SResultWindowInfo* pWinInfo = taosArrayGet(pAggSup->pResultRows, start);
-    SFilePage*         bufPage = getBufPage(pAggSup->pResultBuf, pWinInfo->pos.pageId);
-    SResultRow*        pRow = (SResultRow*)((char*)bufPage + pWinInfo->pos.offset);
-    for (int32_t j = 0; j < numOfExprs; ++j) {
-      SResultRowEntryInfo* pResultInfo = getResultCell(pRow, j, pBinfo->rowCellInfoOffset);
-      SColumnInfoData*     pColInfoData = taosArrayGet(pBlock->pDataBlock, j);
-      char*                in = GET_ROWCELL_INTERBUF(pBinfo->pCtx[j].resultInfo);
-      colDataAppend(pColInfoData, pBlock->info.rows, in, pResultInfo->isNullRes);
-    }
-    pBlock->info.rows += pRow->numOfRows;
-    releaseBufPage(pAggSup->pResultBuf, bufPage);
-  }
-  blockDataUpdateTsWindow(pBlock, -1);
-  return TSDB_CODE_SUCCESS;
-}
-
 int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap) {
   SResultWindowInfo* pCurWin = taosArrayGet(pWinInfos, startIndex);
   int32_t            size = taosArrayGetSize(pWinInfos);
@@ -2484,15 +2484,15 @@ int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap)
 
 void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex, int32_t num, int32_t groupId,
                        int32_t numOfOutput, SExecTaskInfo* pTaskInfo, SHashObj* pStUpdated, SHashObj* pStDeleted) {
-  SResultWindowInfo* pCurWin = taosArrayGet(pInfo->streamAggSup.pResultRows, startIndex);
+  SResultWindowInfo* pCurWin = taosArrayGet(pInfo->streamAggSup.pCurWins, startIndex);
   SResultRow*        pCurResult = NULL;
   setWindowOutputBuf(pCurWin, &pCurResult, pInfo->binfo.pCtx, groupId, numOfOutput, pInfo->binfo.rowCellInfoOffset,
                      &pInfo->streamAggSup, pTaskInfo);
   num += startIndex + 1;
-  ASSERT(num <= taosArrayGetSize(pInfo->streamAggSup.pResultRows));
+  ASSERT(num <= taosArrayGetSize(pInfo->streamAggSup.pCurWins));
   // Just look for the window behind StartIndex
   for (int32_t i = startIndex + 1; i < num; i++) {
-    SResultWindowInfo* pWinInfo = taosArrayGet(pInfo->streamAggSup.pResultRows, i);
+    SResultWindowInfo* pWinInfo = taosArrayGet(pInfo->streamAggSup.pCurWins, i);
     SResultRow*        pWinResult = NULL;
     setWindowOutputBuf(pWinInfo, &pWinResult, pInfo->pDummyCtx, groupId, numOfOutput, pInfo->binfo.rowCellInfoOffset,
                        &pInfo->streamAggSup, pTaskInfo);
@@ -2503,7 +2503,7 @@ void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex,
       taosHashPut(pStDeleted, &pWinInfo->pos, sizeof(SResultRowPosition), &pWinInfo->win.skey, sizeof(TSKEY));
       pWinInfo->isOutput = false;
     }
-    taosArrayRemove(pInfo->streamAggSup.pResultRows, i);
+    taosArrayRemove(pInfo->streamAggSup.pCurWins, i);
   }
 }
 
@@ -2533,7 +2533,7 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
   SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
   for (int32_t i = 0; i < pSDataBlock->info.rows;) {
     int32_t            winIndex = 0;
-    SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup->pResultRows, tsCols[i], gap, &winIndex);
+    SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, tsCols[i], pSDataBlock->info.groupId, gap, &winIndex);
     winRows = updateSessionWindowInfo(pCurWin, tsCols, pSDataBlock->info.rows, i, pInfo->gap, pStDeleted);
     code = doOneWindowAgg(pInfo, pSDataBlock, pCurWin, &pResult, i, winRows, numOfOutput, pTaskInfo);
     if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
@@ -2543,7 +2543,7 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
     // doWindowBorderInterpolation(pOperatorInfo, pSDataBlock, pInfo->binfo.pCtx, pResult, &nextWin, startPos,
     // forwardRows,
     //                             pInfo->order, false);
-    int32_t winNum = getNumCompactWindow(pAggSup->pResultRows, winIndex, gap);
+    int32_t winNum = getNumCompactWindow(pAggSup->pCurWins, winIndex, gap);
     if (winNum > 0) {
       compactTimeWindow(pInfo, winIndex, winNum, groupId, numOfOutput, pTaskInfo, pStUpdated, pStDeleted);
     }
@@ -2566,7 +2566,7 @@ static void doClearSessionWindows(SStreamAggSupporter* pAggSup, SOptrBasicInfo*
   int32_t          step = 0;
   for (int32_t i = 0; i < pBlock->info.rows; i += step) {
     int32_t            winIndex = 0;
-    SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup->pResultRows, tsCols[i], gap, &winIndex);
+    SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, tsCols[i], pBlock->info.groupId, gap, &winIndex);
     step = updateSessionWindowInfo(pCurWin, tsCols, pBlock->info.rows, i, gap, NULL);
     ASSERT(isInWindow(pCurWin, tsCols[i], gap));
     doClearWindowImpl(&pCurWin->pos, pAggSup->pResultBuf, pBinfo, numOfOutput);
@@ -2627,7 +2627,7 @@ static void rebuildTimeWindow(SStreamSessionAggOperatorInfo* pInfo, SArray* pWin
     for (int32_t j = 0; j < numOfChildren; j++) {
       SOperatorInfo*                 pChild = taosArrayGetP(pInfo->pChildren, j);
       SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
-      SArray*                        pChWins = pChInfo->streamAggSup.pResultRows;
+      SArray*                        pChWins = getWinInfos(&pChInfo->streamAggSup, groupId);
       int32_t                        chWinSize = taosArrayGetSize(pChWins);
       int32_t index = binarySearch(pChWins, chWinSize, pParentWin->win.skey, TSDB_ORDER_DESC, getSessionWindowEndkey);
       for (int32_t k = index; k > 0 && k < chWinSize; k++) {
@@ -2651,36 +2651,44 @@ typedef SResultWindowInfo* (*__get_win_info_)(void*);
 SResultWindowInfo* getSessionWinInfo(void* pData) { return (SResultWindowInfo*)pData; }
 SResultWindowInfo* getStateWinInfo(void* pData) { return &((SStateWindowInfo*)pData)->winInfo; }
 
-int32_t closeSessionWindow(SArray* pWins, STimeWindowAggSupp* pTwSup, SArray* pClosed,
+int32_t closeSessionWindow(SHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SArray* pClosed,
     __get_win_info_ fn) {
   // Todo(liuyao) save window to tdb
-  int32_t size = taosArrayGetSize(pWins);
-  for (int32_t i = 0; i < size; i++) {
-    void*              pWin = taosArrayGet(pWins, i);
-    SResultWindowInfo* pSeWin = fn(pWin);
-    if (pSeWin->win.ekey < pTwSup->maxTs - pTwSup->waterMark) {
-      if (!pSeWin->isClosed) {
-        pSeWin->isClosed = true;
-        if (pTwSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
-          int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, 0, pClosed);
-          pSeWin->isOutput = true;
+  void **pIte = NULL;
+  while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
+    SArray *pWins = (SArray *) (*pIte);
+    int32_t size = taosArrayGetSize(pWins);
+    for (int32_t i = 0; i < size; i++) {
+      void*              pWin = taosArrayGet(pWins, i);
+      SResultWindowInfo* pSeWin = fn(pWin);
+      if (pSeWin->win.ekey < pTwSup->maxTs - pTwSup->waterMark) {
+        if (!pSeWin->isClosed) {
+          pSeWin->isClosed = true;
+          if (pTwSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
+            int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, 0, pClosed);
+            pSeWin->isOutput = true;
+          }
         }
+        continue;
       }
-      continue;
+      break;
     }
-    break;
   }
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t getAllSessionWindow(SArray* pWins, SArray* pClosed, __get_win_info_ fn) {
-  int32_t size = taosArrayGetSize(pWins);
-  for (int32_t i = 0; i < size; i++) {
-    void*              pWin = taosArrayGet(pWins, i);
-    SResultWindowInfo* pSeWin = fn(pWin);
-    if (!pSeWin->isClosed) {
-      int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, 0, pClosed);
-      pSeWin->isOutput = true;
+int32_t getAllSessionWindow(SHashObj* pHashMap, SArray* pClosed, __get_win_info_ fn) {
+  void **pIte = NULL;
+  while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
+    SArray *pWins = (SArray *) (*pIte);
+    int32_t size = taosArrayGetSize(pWins);
+    for (int32_t i = 0; i < size; i++) {
+      void*              pWin = taosArrayGet(pWins, i);
+      SResultWindowInfo* pSeWin = fn(pWin);
+      if (!pSeWin->isClosed) {
+        int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, 0, pClosed);
+        pSeWin->isOutput = true;
+      }
     }
   }
   return TSDB_CODE_SUCCESS;
@@ -2714,8 +2722,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
     if (pBlock == NULL) {
       break;
     }
-    // the pDataBlock are always the same one, no need to call this again
-    setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
+
     if (pBlock->info.type == STREAM_REPROCESS) {
       SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
       doClearSessionWindows(&pInfo->streamAggSup, &pInfo->binfo, pBlock, 0, pOperator->numOfExprs, pInfo->gap, pWins);
@@ -2729,12 +2736,13 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
       }
       taosArrayDestroy(pWins);
       continue;
-    } else if (pBlock->info.type == STREAM_GET_ALL &&
-        pInfo->twAggSup.calTrigger == STREAM_TRIGGER_MAX_DELAY) {
+    } else if (pBlock->info.type == STREAM_GET_ALL) {
       getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getSessionWinInfo);
       continue;
     }
 
+    // the pDataBlock are always the same one, no need to call this again
+    setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
     if (isFinalSession(pInfo)) {
       int32_t         childIndex = 0;  // Todo(liuyao) get child id from SSDataBlock
       SOptrBasicInfo* pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
@@ -2873,7 +2881,9 @@ bool isEqualStateKey(SStateWindowInfo* pWin, char* pKeyData) {
   return pKeyData && compareVal(pKeyData, &pWin->stateKey);
 }
 
-SStateWindowInfo* getStateWindowByTs(SArray* pWinInfos, TSKEY ts, int32_t* pIndex) {
+SStateWindowInfo* getStateWindowByTs(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, int32_t* pIndex) {
+  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
+  pAggSup->pCurWins = pWinInfos;
   int32_t           size = taosArrayGetSize(pWinInfos);
   int32_t           index = binarySearch(pWinInfos, size, ts, TSDB_ORDER_DESC, getStateWinTsKey);
   SStateWindowInfo* pWin = NULL;
@@ -2896,7 +2906,10 @@ SStateWindowInfo* getStateWindowByTs(SArray* pWinInfos, TSKEY ts, int32_t* pInde
   return NULL;
 }
 
-SStateWindowInfo* getStateWindow(SArray* pWinInfos, TSKEY ts, char* pKeyData, SColumn* pCol, int32_t* pIndex) {
+SStateWindowInfo* getStateWindow(SStreamAggSupporter* pAggSup, TSKEY ts,
+    uint64_t groupId, char* pKeyData, SColumn* pCol, int32_t* pIndex) {
+  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
+  pAggSup->pCurWins = pWinInfos;
   int32_t size = taosArrayGetSize(pWinInfos);
   if (size == 0) {
     *pIndex = 0;
@@ -2987,16 +3000,16 @@ static void doClearStateWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBloc
   for (int32_t i = 0; i < pBlock->info.rows; i += step) {
     char*             pKeyData = colDataGetData(pKeyColInfo, i);
     int32_t           winIndex = 0;
-    SStateWindowInfo* pCurWin = getStateWindowByTs(pAggSup->pResultRows, tsCol[i], &winIndex);
+    SStateWindowInfo* pCurWin = getStateWindowByTs(pAggSup, tsCol[i], pBlock->info.groupId, &winIndex);
     if (!pCurWin) {
       continue;
     }
-    step = updateStateWindowInfo(pAggSup->pResultRows, winIndex, tsCol, pKeyColInfo, pBlock->info.rows, i, &allEqual,
+    step = updateStateWindowInfo(pAggSup->pCurWins, winIndex, tsCol, pKeyColInfo, pBlock->info.rows, i, &allEqual,
                                  pSeDeleted);
     ASSERT(isTsInWindow(pCurWin, tsCol[i]) || isEqualStateKey(pCurWin, pKeyData));
     taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win);
     taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition));
-    deleteWindow(pAggSup->pResultRows, winIndex);
+    deleteWindow(pAggSup->pCurWins, winIndex);
   }
 }
 
@@ -3026,13 +3039,15 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
     char*             pKeyData = colDataGetData(pKeyColInfo, i);
     int32_t           winIndex = 0;
     bool              allEqual = true;
-    SStateWindowInfo* pCurWin = getStateWindow(pAggSup->pResultRows, tsCols[i], pKeyData, &pInfo->stateCol, &winIndex);
-    winRows = updateStateWindowInfo(pAggSup->pResultRows, winIndex, tsCols, pKeyColInfo, pSDataBlock->info.rows, i,
-                                    &allEqual, pInfo->pSeDeleted);
+    SStateWindowInfo* pCurWin = 
+        getStateWindow(pAggSup, tsCols[i], pSDataBlock->info.groupId, pKeyData,
+            &pInfo->stateCol, &winIndex);
+    winRows = updateStateWindowInfo(pAggSup->pCurWins, winIndex, tsCols, pKeyColInfo,
+        pSDataBlock->info.rows, i, &allEqual, pInfo->pSeDeleted);
     if (!allEqual) {
       taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win);
       taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition));
-      deleteWindow(pAggSup->pResultRows, winIndex);
+      deleteWindow(pAggSup->pCurWins, winIndex);
       continue;
     }
     code = doOneStateWindowAgg(pInfo, pSDataBlock, &pCurWin->winInfo, &pResult, i, winRows, numOfOutput, pTaskInfo);
@@ -3079,17 +3094,18 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
     if (pBlock == NULL) {
       break;
     }
-    // the pDataBlock are always the same one, no need to call this again
-    setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
+
     if (pBlock->info.type == STREAM_REPROCESS) {
       doClearStateWindows(&pInfo->streamAggSup, pBlock, pInfo->primaryTsIndex, &pInfo->stateCol, pInfo->stateCol.slotId,
                           pSeUpdated, pInfo->pSeDeleted);
       continue;
-    } else if (pBlock->info.type == STREAM_GET_ALL &&
-        pInfo->twAggSup.calTrigger == STREAM_TRIGGER_MAX_DELAY) {
+    } else if (pBlock->info.type == STREAM_GET_ALL) {
       getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getStateWinInfo);
       continue;
     }
+
+    // the pDataBlock are always the same one, no need to call this again
+    setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
     doStreamStateAggImpl(pOperator, pBlock, pSeUpdated, pInfo->pSeDeleted);
     pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
   }

From d7a6c4976c413b019744139d5639508e50a41db7 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Wed, 15 Jun 2022 17:27:01 +0800
Subject: [PATCH 14/14] fix(stream): max delay should not be seen by operator
 (#13864)

---
 source/dnode/mnode/impl/src/mndStream.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c
index 4a7da8f2d1..f447a93392 100644
--- a/source/dnode/mnode/impl/src/mndStream.c
+++ b/source/dnode/mnode/impl/src/mndStream.c
@@ -228,7 +228,7 @@ static int32_t mndStreamGetPlanString(const char *ast, int8_t triggerType, int64
         .pAstRoot = pAst,
         .topicQuery = false,
         .streamQuery = true,
-        .triggerType = triggerType,
+        .triggerType = triggerType == STREAM_TRIGGER_MAX_DELAY ? STREAM_TRIGGER_WINDOW_CLOSE : triggerType,
         .watermark = watermark,
     };
     code = qCreateQueryPlan(&cxt, &pPlan, NULL);