From ff8908dedebfffc72cb798a7f4a46de496d52bdd Mon Sep 17 00:00:00 2001
From: 54liuyao <54liuyao>
Date: Thu, 18 Jan 2024 18:30:21 +0800
Subject: [PATCH] ignore invalid state

---
 source/libs/stream/src/streamSessionState.c | 4 ++++
 source/libs/stream/src/streamState.c        | 1 +
 tests/script/tsim/stream/partitionby1.sim   | 2 ++
 3 files changed, 7 insertions(+)

diff --git a/source/libs/stream/src/streamSessionState.c b/source/libs/stream/src/streamSessionState.c
index 2cb77c60dc..1f991d309f 100644
--- a/source/libs/stream/src/streamSessionState.c
+++ b/source/libs/stream/src/streamSessionState.c
@@ -90,6 +90,7 @@ SRowBuffPos* createSessionWinBuff(SStreamFileState* pFileState, SSessionKey* pKe
   SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);
   memcpy(pNewPos->pKey, pKey, sizeof(SSessionKey));
   pNewPos->needFree = true;
+  pNewPos->beFlushed = true;
   memcpy(pNewPos->pRowBuff, p, *pVLen);
   taosMemoryFree(p);
   return pNewPos;
@@ -217,6 +218,7 @@ int32_t getSessionFlushedBuff(SStreamFileState* pFileState, SSessionKey* pKey, v
   SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);
   memcpy(pNewPos->pKey, pKey, sizeof(SSessionKey));
   pNewPos->needFree = true;
+  pNewPos->beFlushed = true;
   void*   pBuff = NULL;
   int32_t code = streamStateSessionGet_rocksdb(getStateFileStore(pFileState), pKey, &pBuff, pVLen);
   if (code != TSDB_CODE_SUCCESS) {
@@ -307,6 +309,7 @@ int32_t allocSessioncWinBuffByNextPosition(SStreamFileState* pFileState, SStream
     }
     pNewPos = getNewRowPosForWrite(pFileState);
     pNewPos->needFree = true;
+    pNewPos->beFlushed = true;
   }
 
 _end:
@@ -482,6 +485,7 @@ int32_t sessionWinStateGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, void
       SRowBuffPos* pNewPos = getNewRowPosForWrite(pCur->pStreamFileState);
       memcpy(pNewPos->pKey, pKey, sizeof(SSessionKey));
       pNewPos->needFree = true;
+      pNewPos->beFlushed = true;
       memcpy(pNewPos->pRowBuff, pData, *pVLen);
       (*pVal) = pNewPos;
     }
diff --git a/source/libs/stream/src/streamState.c b/source/libs/stream/src/streamState.c
index 19b7359981..ea20f0e2b1 100644
--- a/source/libs/stream/src/streamState.c
+++ b/source/libs/stream/src/streamState.c
@@ -698,6 +698,7 @@ int32_t streamStateSessionPut(SStreamState* pState, const SSessionKey* key, void
       stDebug("===stream===save skey:%" PRId64 ", ekey:%" PRId64 ", groupId:%" PRIu64 ".code:%d", key->win.skey,
               key->win.ekey, key->groupId, code);
     } else {
+      pos->beFlushed = false;
       code = putSessionWinResultBuff(pState->pFileState, value);
     }
   }
diff --git a/tests/script/tsim/stream/partitionby1.sim b/tests/script/tsim/stream/partitionby1.sim
index d92aecb3a6..24c588d410 100644
--- a/tests/script/tsim/stream/partitionby1.sim
+++ b/tests/script/tsim/stream/partitionby1.sim
@@ -13,6 +13,8 @@ sql create table ts3 using st tags(3,2,2);
 sql create table ts4 using st tags(4,2,2);
 sql create stream stream_t1 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0  into streamtST1 as select  _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b)  c4, min(c) c5 from st partition by tbname interval(10s);
 
+sleep 1000
+
 sql insert into ts1 values(1648791213001,1,12,3,1.0);
 sql insert into ts2 values(1648791213001,1,12,3,1.0);