feat(stream): distribute interval disc buff

include/libs/stream/streamState.h

@@ -33,6 +33,7 @@ typedef struct {
   TTB*         pFuncStateDb;
   TTB*         pFillStateDb;  // todo refactor
   TXN          txn;
+  int32_t      number;
 } SStreamState;
 
 SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath);
@@ -42,7 +43,8 @@ int32_t       streamStateCommit(SStreamState* pState);
 int32_t       streamStateAbort(SStreamState* pState);
 
 typedef struct {
-  TBC* pCur;
+  TBC*    pCur;
+  int64_t number;
 } SStreamStateCur;
 
 int32_t streamStateFuncPut(SStreamState* pState, const STupleKey* key, const void* value, int32_t vLen);
@@ -52,6 +54,8 @@ int32_t streamStateFuncDel(SStreamState* pState, const STupleKey* key);
 int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
 int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
 int32_t streamStateDel(SStreamState* pState, const SWinKey* key);
+int32_t streamStateClear(SStreamState* pState);
+void    streamStateSetNumber(SStreamState* pState, int32_t number);
 
 int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
 int32_t streamStateFillGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
@@ -63,6 +67,7 @@ void    streamFreeVal(void* val);
 
 SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key);
 SStreamStateCur* streamStateGetAndCheckCur(SStreamState* pState, SWinKey* key);
+SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key);
 SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey* key);
 SStreamStateCur* streamStateFillSeekKeyPrev(SStreamState* pState, const SWinKey* key);
 void             streamStateFreeCur(SStreamStateCur* pCur);
@@ -70,6 +75,7 @@ void             streamStateFreeCur(SStreamStateCur* pCur);
 int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);
 int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);
 
+int32_t streamStateGetFirst(SStreamState* pState, SWinKey* key);
 int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur);
 int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur);
 

source/libs/executor/inc/executil.h

@@ -15,7 +15,6 @@
 #ifndef TDENGINE_QUERYUTIL_H
 #define TDENGINE_QUERYUTIL_H
 
-#include "vnode.h"
 #include "function.h"
 #include "nodes.h"
 #include "plannodes.h"
@@ -23,6 +22,7 @@
 #include "tcommon.h"
 #include "tpagedbuf.h"
 #include "tsimplehash.h"
+#include "vnode.h"
 
 #define T_LONG_JMP(_obj, _c) \
   do {                       \
@@ -93,7 +93,7 @@ void   resetResultRow(SResultRow* pResultRow, size_t entrySize);
 struct SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset);
 
 static FORCE_INLINE SResultRow* getResultRowByPos(SDiskbasedBuf* pBuf, SResultRowPosition* pos, bool forUpdate) {
-  SFilePage*  bufPage = (SFilePage*)getBufPage(pBuf, pos->pageId);
+  SFilePage* bufPage = (SFilePage*)getBufPage(pBuf, pos->pageId);
   if (forUpdate) {
     setBufPageDirty(bufPage, true);
   }
@@ -101,11 +101,6 @@ static FORCE_INLINE SResultRow* getResultRowByPos(SDiskbasedBuf* pBuf, SResultRo
   return pRow;
 }
 
-static FORCE_INLINE void setResultBufPageDirty(SDiskbasedBuf* pBuf, SResultRowPosition* pos) {
-  void* pPage = getBufPage(pBuf, pos->pageId);
-  setBufPageDirty(pPage, true);
-}
-
 void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SSHashObj* pHashmap, int32_t order);
 void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo);
 
@@ -117,17 +112,18 @@ int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo);
 SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode);
 
 EDealRes doTranslateTagExpr(SNode** pNode, void* pContext);
-int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond, STableListInfo* pListInfo);
-int32_t getGroupIdFromTagsVal(void* pMeta, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId);
-int32_t getColInfoResultForGroupby(void* metaHandle, SNodeList* group, STableListInfo* pTableListInfo);
-size_t  getTableTagsBufLen(const SNodeList* pGroups);
-
-SArray*  createSortInfo(SNodeList* pNodeList);
-SArray*  extractPartitionColInfo(SNodeList* pNodeList);
-SArray*  extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,
-                             int32_t type);
-
-void createExprFromTargetNode(SExprInfo* pExp, STargetNode* pTargetNode);
+int32_t  getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond,
+                      STableListInfo* pListInfo);
+int32_t  getGroupIdFromTagsVal(void* pMeta, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId);
+int32_t  getColInfoResultForGroupby(void* metaHandle, SNodeList* group, STableListInfo* pTableListInfo);
+size_t   getTableTagsBufLen(const SNodeList* pGroups);
+
+SArray* createSortInfo(SNodeList* pNodeList);
+SArray* extractPartitionColInfo(SNodeList* pNodeList);
+SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,
+                            int32_t type);
+
+void       createExprFromTargetNode(SExprInfo* pExp, STargetNode* pTargetNode);
 SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t* numOfExprs);
 
 SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowEntryInfoOffset);

source/libs/executor/inc/executorimpl.h

@@ -577,13 +577,7 @@ typedef struct SIntervalAggOperatorInfo {
   int32_t            inputOrder;         // input data ts order
   EOPTR_EXEC_MODEL   execModel;          // operator execution model [batch model|stream model]
   STimeWindowAggSupp twAggSup;
-  bool               invertible;
   SArray*            pPrevValues;  //  SArray<SGroupKeys> used to keep the previous not null value for interpolation.
-  bool               ignoreExpiredData;
-  SArray*            pRecycledPages;
-  SArray*            pDelWins;  // SWinRes
-  int32_t            delIndex;
-  SSDataBlock*       pDelRes;
   SNode*             pCondition;
 } SIntervalAggOperatorInfo;
 
@@ -609,38 +603,21 @@ typedef struct SStreamIntervalOperatorInfo {
   STimeWindowAggSupp twAggSup;
   bool               invertible;
   bool               ignoreExpiredData;
-  SArray*            pRecycledPages;
   SArray*            pDelWins;           // SWinRes
   int32_t            delIndex;
   SSDataBlock*       pDelRes;
-  bool               isFinal;
-} SStreamIntervalOperatorInfo;
-
-typedef struct SStreamFinalIntervalOperatorInfo {
-  // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
-  SOptrBasicInfo     binfo;           // basic info
-  SAggSupporter      aggSup;          // aggregate supporter
-  SExprSupp          scalarSupp;      // supporter for perform scalar function
-  SGroupResInfo      groupResInfo;    // multiple results build supporter
-  SInterval          interval;        // interval info
-  int32_t            primaryTsIndex;  // primary time stamp slot id from result of downstream operator.
-  int32_t            order;           // current SSDataBlock scan order
-  STimeWindowAggSupp twAggSup;
-  SArray*            pChildren;
   SSDataBlock*       pUpdateRes;
   bool               returnUpdate;
-  SPhysiNode*        pPhyNode;  // create new child
-  bool               isFinal;
+  SPhysiNode*        pPhyNode;           // create new child
   SHashObj*          pPullDataMap;
-  SArray*            pPullWins;  // SPullWindowInfo
+  SArray*            pPullWins;          // SPullWindowInfo
   int32_t            pullIndex;
   SSDataBlock*       pPullDataRes;
-  bool               ignoreExpiredData;
-  SArray*            pRecycledPages;
-  SArray*            pDelWins;  // SWinRes
-  int32_t            delIndex;
-  SSDataBlock*       pDelRes;
-} SStreamFinalIntervalOperatorInfo;
+  bool               isFinal;
+  SArray*            pChildren;
+  SStreamState*      pState;
+  SWinKey            delKey;
+} SStreamIntervalOperatorInfo;
 
 typedef struct SAggOperatorInfo {
   // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
@@ -1086,7 +1063,7 @@ bool               functionNeedToExecute(SqlFunctionCtx* pCtx);
 bool               isOverdue(TSKEY ts, STimeWindowAggSupp* pSup);
 bool               isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup);
 bool               isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
-bool               isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SOperatorInfo* pOperator, STimeWindowAggSupp* pTwSup);
+bool               isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup);
 void               appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid, uint64_t* pGp);
 void               printDataBlock(SSDataBlock* pBlock, const char* flag);
 uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
@@ -1108,13 +1085,12 @@ void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsCol
 bool    groupbyTbname(SNodeList* pGroupList);
 int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, SNodeList* groupKey);
 void*   destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput);
-int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup,
+int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
                                    SGroupResInfo* pGroupResInfo);
-int32_t setOutputBuf(STimeWindow* win, SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx,
-                           int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup,
-                           SExecTaskInfo* pTaskInfo);
-int32_t releaseOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult);
-int32_t saveOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult, int32_t resSize);
+int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx,
+                     int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup);
+int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult);
+int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize);
 void    getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order);
 
 #ifdef __cplusplus

source/libs/executor/src/executorimpl.c

@@ -4183,9 +4183,8 @@ int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlF
   return code;
 }
 
-int32_t setOutputBuf(STimeWindow* win, SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx,
-                     int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup,
-                     SExecTaskInfo* pTaskInfo) {
+int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId,
+                     SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup) {
   SWinKey key = {
       .ts = win->skey,
       .groupId = tableGroupId,
@@ -4194,7 +4193,7 @@ int32_t setOutputBuf(STimeWindow* win, SResultRow** pResult, int64_t tableGroupI
   int32_t size = pAggSup->resultRowSize;
 
   tSimpleHashPut(pAggSup->pResultRowHashTable, &key, sizeof(SWinKey), NULL, 0);
-  if (streamStateAddIfNotExist(pTaskInfo->streamInfo.pState, &key, (void**)&value, &size) < 0) {
+  if (streamStateAddIfNotExist(pState, &key, (void**)&value, &size) < 0) {
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
   *pResult = (SResultRow*)value;
@@ -4205,18 +4204,17 @@ int32_t setOutputBuf(STimeWindow* win, SResultRow** pResult, int64_t tableGroupI
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t releaseOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult) {
-  streamStateReleaseBuf(pTaskInfo->streamInfo.pState, pKey, pResult);
-  /*taosMemoryFree((*(void**)pResult));*/
+int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult) {
+  streamStateReleaseBuf(pState, pKey, pResult);
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t saveOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult, int32_t resSize) {
-  streamStatePut(pTaskInfo->streamInfo.pState, pKey, pResult, resSize);
+int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize) {
+  streamStatePut(pState, pKey, pResult, resSize);
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup,
+int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
                                    SGroupResInfo* pGroupResInfo) {
   SExprInfo*      pExprInfo = pSup->pExprInfo;
   int32_t         numOfExprs = pSup->numOfExprs;
@@ -4233,14 +4231,14 @@ int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock
             .ts = *(TSKEY*)pPos->key,
             .groupId = pPos->groupId,
     };
-    int32_t code = streamStateGet(pTaskInfo->streamInfo.pState, &key, &pVal, &size);
+    int32_t code = streamStateGet(pState, &key, &pVal, &size);
     ASSERT(code == 0);
     SResultRow* pRow = (SResultRow*)pVal;
     doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowEntryOffset);
     // no results, continue to check the next one
     if (pRow->numOfRows == 0) {
       pGroupResInfo->index += 1;
-      releaseOutputBuf(pTaskInfo, &key, pRow);
+      releaseOutputBuf(pState, &key, pRow);
       continue;
     }
 
@@ -4249,14 +4247,14 @@ int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock
     } else {
       // current value belongs to different group, it can't be packed into one datablock
       if (pBlock->info.groupId != pPos->groupId) {
-        releaseOutputBuf(pTaskInfo, &key, pRow);
+        releaseOutputBuf(pState, &key, pRow);
         break;
       }
     }
 
     if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
       ASSERT(pBlock->info.rows > 0);
-      releaseOutputBuf(pTaskInfo, &key, pRow);
+      releaseOutputBuf(pState, &key, pRow);
       break;
     }
 
@@ -4286,7 +4284,7 @@ int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock
     }
 
     pBlock->info.rows += pRow->numOfRows;
-    releaseOutputBuf(pTaskInfo, &key, pRow);
+    releaseOutputBuf(pState, &key, pRow);
   }
   blockDataUpdateTsWindow(pBlock, 0);
   return TSDB_CODE_SUCCESS;

source/libs/executor/src/scanoperator.c

@@ -1357,7 +1357,8 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
     // must check update info first.
     bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, tsCol[rowId]);
     bool closedWin = isClosed && isSignleIntervalWindow(pInfo) &&
-                     isDeletedStreamWindow(&win, pBlock->info.groupId, pInfo->pTableScanOp, &pInfo->twAggSup);
+                     isDeletedStreamWindow(&win, pBlock->info.groupId,
+                                           pInfo->pTableScanOp->pTaskInfo->streamInfo.pState, &pInfo->twAggSup);
     if ((update || closedWin) && out) {
       qDebug("stream update check not pass, update %d, closedWin %d", update, closedWin);
       uint64_t gpId = 0;
@@ -2135,6 +2136,9 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
 
     pInfo->pUpdateInfo = NULL;
     pInfo->pTableScanOp = pTableScanOp;
+    if (pInfo->pTableScanOp->pTaskInfo->streamInfo.pState) {
+      streamStateSetNumber(pInfo->pTableScanOp->pTaskInfo->streamInfo.pState, -1);
+    }
 
     pInfo->readHandle = *pHandle;
     pInfo->tableUid = pScanPhyNode->uid;

source/libs/executor/src/tfill.c

@@ -1620,9 +1620,7 @@ SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFi
     goto _error;
   }
 
-  SInterval* pInterval = QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL == downstream->operatorType
-                             ? &((SStreamFinalIntervalOperatorInfo*)downstream->info)->interval
-                             : &((SStreamIntervalOperatorInfo*)downstream->info)->interval;
+  SInterval* pInterval = &((SStreamIntervalOperatorInfo*)downstream->info)->interval;
   int32_t    numOfFillCols = 0;
   SExprInfo* pFillExprInfo = createExprInfo(pPhyFillNode->pFillExprs, NULL, &numOfFillCols);
   pInfo->pFillSup = initStreamFillSup(pPhyFillNode, pInterval, pFillExprInfo, numOfFillCols);

source/libs/stream/src/streamState.c

@@ -18,6 +18,37 @@
 #include "tcommon.h"
 #include "ttimer.h"
 
+// todo refactor
+typedef struct SStateKey {
+  SWinKey key;
+  int64_t opNum;
+} SStateKey;
+
+static inline int SStateKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
+  SStateKey* pWin1 = (SStateKey*)pKey1;
+  SStateKey* pWin2 = (SStateKey*)pKey2;
+
+  if (pWin1->opNum > pWin2->opNum) {
+    return 1;
+  } else if (pWin1->opNum < pWin2->opNum) {
+    return -1;
+  }
+
+  if (pWin1->key.ts > pWin2->key.ts) {
+    return 1;
+  } else if (pWin1->key.ts < pWin2->key.ts) {
+    return -1;
+  }
+
+  if (pWin1->key.groupId > pWin2->key.groupId) {
+    return 1;
+  } else if (pWin1->key.groupId < pWin2->key.groupId) {
+    return -1;
+  }
+
+  return 0;
+}
+
 SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath) {
   SStreamState* pState = taosMemoryCalloc(1, sizeof(SStreamState));
   if (pState == NULL) {
@@ -36,7 +67,7 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath) {
   }
 
   // open state storage backend
-  if (tdbTbOpen("state.db", sizeof(SWinKey), -1, SWinKeyCmpr, pState->db, &pState->pStateDb) < 0) {
+  if (tdbTbOpen("state.db", sizeof(SStateKey), -1, SStateKeyCmpr, pState->db, &pState->pStateDb) < 0) {
     goto _err;
   }
 
@@ -130,8 +161,10 @@ int32_t streamStateFuncDel(SStreamState* pState, const STupleKey* key) {
   return tdbTbDelete(pState->pFuncStateDb, key, sizeof(STupleKey), &pState->txn);
 }
 
+// todo refactor
 int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) {
-  return tdbTbUpsert(pState->pStateDb, key, sizeof(SWinKey), value, vLen, &pState->txn);
+  SStateKey sKey = {.key = *key, .opNum = pState->number};
+  return tdbTbUpsert(pState->pStateDb, &sKey, sizeof(SStateKey), value, vLen, &pState->txn);
 }
 
 // todo refactor
@@ -139,8 +172,10 @@ int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void*
   return tdbTbUpsert(pState->pFillStateDb, key, sizeof(SWinKey), value, vLen, &pState->txn);
 }
 
+// todo refactor
 int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) {
-  return tdbTbGet(pState->pStateDb, key, sizeof(SWinKey), pVal, pVLen);
+  SStateKey sKey = {.key = *key, .opNum = pState->number};
+  return tdbTbGet(pState->pStateDb, &sKey, sizeof(SStateKey), pVal, pVLen);
 }
 
 // todo refactor
@@ -148,10 +183,30 @@ int32_t streamStateFillGet(SStreamState* pState, const SWinKey* key, void** pVal
   return tdbTbGet(pState->pFillStateDb, key, sizeof(SWinKey), pVal, pVLen);
 }
 
+// todo refactor
 int32_t streamStateDel(SStreamState* pState, const SWinKey* key) {
-  return tdbTbDelete(pState->pStateDb, key, sizeof(SWinKey), &pState->txn);
+  SStateKey sKey = {.key = *key, .opNum = pState->number};
+  return tdbTbDelete(pState->pStateDb, &sKey, sizeof(SStateKey), &pState->txn);
 }
 
+int32_t streamStateClear(SStreamState* pState) {
+  SWinKey key = {.ts = 0, .groupId = 0};
+  streamStatePut(pState, &key, NULL, 0);
+  while (1) {
+    SStreamStateCur* pCur = streamStateSeekKeyNext(pState, &key);
+    SWinKey          delKey = {0};
+    int32_t          code = streamStateGetKVByCur(pCur, &delKey, NULL, 0);
+    if (code == 0) {
+      streamStateDel(pState, &delKey);
+    } else {
+      break;
+    }
+  }
+  return 0;
+}
+
+void streamStateSetNumber(SStreamState* pState, int32_t number) { pState->number = number; }
+
 // todo refactor
 int32_t streamStateFillDel(SStreamState* pState, const SWinKey* key) {
   return tdbTbDelete(pState->pFillStateDb, key, sizeof(SWinKey), &pState->txn);
@@ -179,12 +234,14 @@ SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key) {
   if (pCur == NULL) return NULL;
   tdbTbcOpen(pState->pStateDb, &pCur->pCur, NULL);
 
-  int32_t c;
+  int32_t   c;
+  SStateKey sKey = {.key = *key, .opNum = pState->number};
   tdbTbcMoveTo(pCur->pCur, key, sizeof(SWinKey), &c);
   if (c != 0) {
     taosMemoryFree(pCur);
     return NULL;
   }
+  pCur->number = pState->number;
   return pCur;
 }
 
@@ -214,6 +271,25 @@ SStreamStateCur* streamStateGetAndCheckCur(SStreamState* pState, SWinKey* key) {
 }
 
 int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
+  if (!pCur) {
+    return -1;
+  }
+  const SStateKey* pKTmp = NULL;
+  int32_t          kLen;
+  if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, pVal, pVLen) < 0) {
+    return -1;
+  }
+  if (pKTmp->opNum != pCur->number) {
+    return -1;
+  }
+  *pKey = pKTmp->key;
+  return 0;
+}
+
+int32_t streamStateFillGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
+  if (!pCur) {
+    return -1;
+  }
   const SWinKey* pKTmp = NULL;
   int32_t        kLen;
   if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, pVal, pVLen) < 0) {
@@ -225,7 +301,7 @@ int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void**
 
 int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
   uint64_t groupId = pKey->groupId;
-  int32_t  code = streamStateGetKVByCur(pCur, pKey, pVal, pVLen);
+  int32_t  code = streamStateFillGetKVByCur(pCur, pKey, pVal, pVLen);
   if (code == 0) {
     if (pKey->groupId == groupId) {
       return 0;
@@ -234,6 +310,16 @@ int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const v
   return -1;
 }
 
+int32_t streamStateGetFirst(SStreamState* pState, SWinKey* key) {
+  // todo refactor
+  SWinKey tmp = {.ts = 0, .groupId = 0};
+  streamStatePut(pState, &tmp, NULL, 0);
+  SStreamStateCur* pCur = streamStateSeekKeyNext(pState, &tmp);
+  int32_t          code = streamStateGetKVByCur(pCur, key, NULL, 0);
+  streamStateDel(pState, &tmp);
+  return code;
+}
+
 int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur) {
   //
   return tdbTbcMoveToFirst(pCur->pCur);
@@ -244,6 +330,34 @@ int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur) {
   return tdbTbcMoveToLast(pCur->pCur);
 }
 
+SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key) {
+  SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
+  if (pCur == NULL) {
+    return NULL;
+  }
+  pCur->number = pState->number;
+  if (tdbTbcOpen(pState->pStateDb, &pCur->pCur, NULL) < 0) {
+    taosMemoryFree(pCur);
+    return NULL;
+  }
+
+  SStateKey sKey = {.key = *key, .opNum = pState->number};
+  int32_t   c;
+  if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateKey), &c) < 0) {
+    tdbTbcClose(pCur->pCur);
+    taosMemoryFree(pCur);
+    return NULL;
+  }
+  if (c > 0) return pCur;
+
+  if (tdbTbcMoveToNext(pCur->pCur) < 0) {
+    taosMemoryFree(pCur);
+    return NULL;
+  }
+
+  return pCur;
+}
+
 SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey* key) {
   SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
   if (pCur == NULL) {
@@ -303,9 +417,15 @@ int32_t streamStateCurNext(SStreamState* pState, SStreamStateCur* pCur) {
 
 int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur) {
   //
+  if (!pCur) {
+    return -1;
+  }
   return tdbTbcMoveToPrev(pCur->pCur);
 }
 void streamStateFreeCur(SStreamStateCur* pCur) {
+  if (!pCur) {
+    return;
+  }
   tdbTbcClose(pCur->pCur);
   taosMemoryFree(pCur);
 }

tests/script/tsim/stream/distributeInterval0.sim

@@ -33,7 +33,8 @@ if $data(2)[4] != ready then
 endi
 
 print ===== step2
-
+sql drop stream if exists stream_t1;
+sql drop database if exists test;
 sql create database test  vgroups 4;
 sql use test;
 sql create stable st(ts timestamp, a int, b int , c int, d double) tags(ta int,tb int,tc int);