From 7c383446428ee2443df5c09929bad94cb9b4d231 Mon Sep 17 00:00:00 2001
From: liuyao <54liuyao@163.com>
Date: Fri, 3 Nov 2023 17:53:36 +0800
Subject: [PATCH 1/2] adj builtin function

---
 source/libs/function/src/builtins.c | 40 ++++++++++++++---------------
 1 file changed, 20 insertions(+), 20 deletions(-)

diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c
index fdbc0b4038..6464243d86 100644
--- a/source/libs/function/src/builtins.c
+++ b/source/libs/function/src/builtins.c
@@ -3312,26 +3312,6 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .sprocessFunc = castFunction,
     .finalizeFunc = NULL
   },
-  {
-    .name = "to_timestamp",
-    .type = FUNCTION_TYPE_TO_TIMESTAMP,
-    .classification = FUNC_MGT_SCALAR_FUNC | FUNC_MGT_DATETIME_FUNC,
-    .translateFunc = translateToTimestamp,
-    .getEnvFunc = NULL,
-    .initFunc = NULL,
-    .sprocessFunc = toTimestampFunction,
-    .finalizeFunc = NULL
-  },
-  {
-    .name = "to_char",
-    .type = FUNCTION_TYPE_TO_CHAR,
-    .classification = FUNC_MGT_SCALAR_FUNC,
-    .translateFunc = translateToChar,
-    .getEnvFunc = NULL,
-    .initFunc = NULL,
-    .sprocessFunc = toCharFunction,
-    .finalizeFunc = NULL
-  },
   {
     .name = "to_iso8601",
     .type = FUNCTION_TYPE_TO_ISO8601,
@@ -3709,6 +3689,26 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .sprocessFunc = qVgIdFunction,
     .finalizeFunc = NULL
   },
+  {
+    .name = "to_timestamp",
+    .type = FUNCTION_TYPE_TO_TIMESTAMP,
+    .classification = FUNC_MGT_SCALAR_FUNC | FUNC_MGT_DATETIME_FUNC,
+    .translateFunc = translateToTimestamp,
+    .getEnvFunc = NULL,
+    .initFunc = NULL,
+    .sprocessFunc = toTimestampFunction,
+    .finalizeFunc = NULL
+  },
+  {
+    .name = "to_char",
+    .type = FUNCTION_TYPE_TO_CHAR,
+    .classification = FUNC_MGT_SCALAR_FUNC,
+    .translateFunc = translateToChar,
+    .getEnvFunc = NULL,
+    .initFunc = NULL,
+    .sprocessFunc = toCharFunction,
+    .finalizeFunc = NULL
+  },
   
 };
 // clang-format on

From 4ce81f25da9d6c85079d05138d3830f2521cbb59 Mon Sep 17 00:00:00 2001
From: liuyao <54liuyao@163.com>
Date: Wed, 1 Nov 2023 09:48:53 +0800
Subject: [PATCH 2/2] session window max delay

---
 source/libs/executor/inc/executorInt.h        |  2 ++
 .../executor/src/streamtimewindowoperator.c   | 25 +++++++++++++++++++
 2 files changed, 27 insertions(+)

diff --git a/source/libs/executor/inc/executorInt.h b/source/libs/executor/inc/executorInt.h
index 288919d709..52acdfaaa8 100644
--- a/source/libs/executor/inc/executorInt.h
+++ b/source/libs/executor/inc/executorInt.h
@@ -580,6 +580,7 @@ typedef struct SStreamSessionAggOperatorInfo {
   bool                reCkBlock;
   SSDataBlock*        pCheckpointRes;
   bool                clearState;
+  bool                recvGetAll;
 } SStreamSessionAggOperatorInfo;
 
 typedef struct SStreamStateAggOperatorInfo {
@@ -603,6 +604,7 @@ typedef struct SStreamStateAggOperatorInfo {
   SArray*             historyWins;
   bool                reCkBlock;
   SSDataBlock*        pCheckpointRes;
+  bool                recvGetAll;
 } SStreamStateAggOperatorInfo;
 
 typedef struct SStreamPartitionOperatorInfo {
diff --git a/source/libs/executor/src/streamtimewindowoperator.c b/source/libs/executor/src/streamtimewindowoperator.c
index 8bfa8e1a5d..ab933a87ac 100644
--- a/source/libs/executor/src/streamtimewindowoperator.c
+++ b/source/libs/executor/src/streamtimewindowoperator.c
@@ -2532,6 +2532,15 @@ void doStreamSessionSaveCheckpoint(SOperatorInfo* pOperator) {
   taosMemoryFree(buf);
 }
 
+static void resetUnCloseSessionWinInfo(SSHashObj* winMap) {
+  void*   pIte = NULL;
+  int32_t iter = 0;
+  while ((pIte = tSimpleHashIterate(winMap, pIte, &iter)) != NULL) {
+    SResultWindowInfo* pResInfo = pIte;
+    pResInfo->pStatePos->beUsed = true;
+  }
+}
+
 static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
   SExprSupp*                     pSup = &pOperator->exprSupp;
   SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
@@ -2546,6 +2555,12 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
     if (opRes) {
       return opRes;
     }
+
+    if (pInfo->recvGetAll) {
+      pInfo->recvGetAll = false;
+      resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);
+    }
+
     setOperatorCompleted(pOperator);
     return NULL;
   }
@@ -2583,6 +2598,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
       taosArrayDestroy(pWins);
       continue;
     } else if (pBlock->info.type == STREAM_GET_ALL) {
+      pInfo->recvGetAll = true;
       getAllSessionWindow(pAggSup->pResultRows, pInfo->pStUpdated);
       continue;
     } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
@@ -2838,6 +2854,8 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
 
   pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT);
   pInfo->clearState = false;
+  pInfo->recvGetAll = false;
+
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
     // for stream
   void*   buff = NULL;
@@ -3454,6 +3472,11 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
       return resBlock;
     }
 
+    if (pInfo->recvGetAll) {
+      pInfo->recvGetAll = false;
+      resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);
+    }
+
     setOperatorCompleted(pOperator);
     return NULL;
   }
@@ -3482,6 +3505,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
       taosArrayDestroy(pWins);
       continue;
     } else if (pBlock->info.type == STREAM_GET_ALL) {
+      pInfo->recvGetAll = true;
       getAllSessionWindow(pInfo->streamAggSup.pResultRows, pInfo->pSeUpdated);
       continue;
     } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
@@ -3712,6 +3736,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
   }
 
   pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT);
+  pInfo->recvGetAll = false;
 
   // for stream
   void*   buff = NULL;