diff --git a/source/libs/executor/src/streamfilloperator.c b/source/libs/executor/src/streamfilloperator.c
index c992bd15b7..9039a26f0b 100644
--- a/source/libs/executor/src/streamfilloperator.c
+++ b/source/libs/executor/src/streamfilloperator.c
@@ -987,7 +987,14 @@ _end:
 }
 
 void resetStreamFillSup(SStreamFillSupporter* pFillSup) {
-  tSimpleHashClear(pFillSup->pResMap);
+  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
+  SSHashObj* pNewMap = tSimpleHashInit(16, hashFn);
+  if (pNewMap != NULL) {
+    tSimpleHashCleanup(pFillSup->pResMap);
+    pFillSup->pResMap = pNewMap;
+  } else {
+    tSimpleHashClear(pFillSup->pResMap);
+  }
   pFillSup->hasDelete = false;
 }
 void resetStreamFillInfo(SStreamFillOperatorInfo* pInfo) {
@@ -1406,6 +1413,7 @@ static int32_t doStreamForceFillNext(SOperatorInfo* pOperator, SSDataBlock** ppR
     }
 
     pInfo->stateStore.streamStateClearExpiredState(pInfo->pState);
+    resetStreamFillInfo(pInfo);
     setStreamOperatorCompleted(pOperator);
     (*ppRes) = NULL;
     goto _end;
@@ -1477,6 +1485,7 @@ static int32_t doStreamForceFillNext(SOperatorInfo* pOperator, SSDataBlock** ppR
 
   if ((*ppRes) == NULL) {
     pInfo->stateStore.streamStateClearExpiredState(pInfo->pState);
+    resetStreamFillInfo(pInfo);
     setStreamOperatorCompleted(pOperator);
   }
 
diff --git a/source/libs/stream/src/tstreamFileState.c b/source/libs/stream/src/tstreamFileState.c
index dcabadb8bd..592523e70b 100644
--- a/source/libs/stream/src/tstreamFileState.c
+++ b/source/libs/stream/src/tstreamFileState.c
@@ -1228,6 +1228,8 @@ void setFillInfo(SStreamFileState* pFileState) {
 }
 
 void clearExpiredState(SStreamFileState* pFileState) {
+  int32_t    code = TSDB_CODE_SUCCESS;
+  int32_t    lino = 0;
   SSHashObj* pSearchBuff = pFileState->searchBuff;
   void*      pIte = NULL;
   int32_t    iter = 0;
@@ -1246,6 +1248,13 @@ void clearExpiredState(SStreamFileState* pFileState) {
     }
     taosArrayRemoveBatch(pWinStates, 0, size - 1, NULL);
   }
+  code = clearRowBuff(pFileState);
+  QUERY_CHECK_CODE(code, lino, _end);
+
+_end:
+  if (code != TSDB_CODE_SUCCESS) {
+    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
+  }
 }
 
 int32_t getStateSearchRowBuff(SStreamFileState* pFileState, const SWinKey* pKey, void** pVal, int32_t* pVLen,