From c368db30c36b6dccecec954505fd634baefecdd5 Mon Sep 17 00:00:00 2001
From: liuyao <54liuyao@163.com>
Date: Wed, 26 Jul 2023 17:38:00 +0800
Subject: [PATCH] stream fill operator mem leak

---
 source/libs/executor/src/filloperator.c       | 18 +++++++++++++-----
 source/libs/stream/src/streamBackendRocksdb.c |  2 ++
 2 files changed, 15 insertions(+), 5 deletions(-)

diff --git a/source/libs/executor/src/filloperator.c b/source/libs/executor/src/filloperator.c
index 7798ded61b..0ddf74437c 100644
--- a/source/libs/executor/src/filloperator.c
+++ b/source/libs/executor/src/filloperator.c
@@ -502,9 +502,13 @@ void* destroyStreamFillSupporter(SStreamFillSupporter* pFillSup) {
   pFillSup->pAllColInfo = destroyFillColumnInfo(pFillSup->pAllColInfo, pFillSup->numOfFillCols, pFillSup->numOfAllCols);
   tSimpleHashCleanup(pFillSup->pResMap);
   pFillSup->pResMap = NULL;
-  releaseOutputBuf(NULL, NULL, (SResultRow*)pFillSup->cur.pRowVal, &pFillSup->pAPI->stateStore);   //?????
-  pFillSup->cur.pRowVal = NULL;
   cleanupExprSupp(&pFillSup->notFillExprSup);
+  if (pFillSup->cur.pRowVal != pFillSup->prev.pRowVal && pFillSup->cur.pRowVal != pFillSup->next.pRowVal) {
+    taosMemoryFree(pFillSup->cur.pRowVal);
+  }
+  taosMemoryFree(pFillSup->prev.pRowVal);
+  taosMemoryFree(pFillSup->next.pRowVal);
+  taosMemoryFree(pFillSup->nextNext.pRowVal);
 
   taosMemoryFree(pFillSup);
   return NULL;
@@ -546,13 +550,17 @@ static void destroyStreamFillOperatorInfo(void* param) {
 
 static void resetFillWindow(SResultRowData* pRowData) {
   pRowData->key = INT64_MIN;
-  pRowData->pRowVal = NULL;
+  taosMemoryFreeClear(pRowData->pRowVal);
 }
 
 void resetPrevAndNextWindow(SStreamFillSupporter* pFillSup, void* pState, SStorageAPI* pAPI) {
+  if (pFillSup->cur.pRowVal != pFillSup->prev.pRowVal && pFillSup->cur.pRowVal != pFillSup->next.pRowVal) {
+    resetFillWindow(&pFillSup->cur);
+  } else {
+    pFillSup->cur.key = INT64_MIN;
+    pFillSup->cur.pRowVal = NULL;
+  }
   resetFillWindow(&pFillSup->prev);
-  releaseOutputBuf(NULL, NULL, (SResultRow*)pFillSup->cur.pRowVal, &pAPI->stateStore);   //???
-  resetFillWindow(&pFillSup->cur);
   resetFillWindow(&pFillSup->next);
   resetFillWindow(&pFillSup->nextNext);
 }
diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c
index ffefdb0003..6d6d6e1848 100644
--- a/source/libs/stream/src/streamBackendRocksdb.c
+++ b/source/libs/stream/src/streamBackendRocksdb.c
@@ -1263,6 +1263,8 @@ int32_t streamStateGetGroupKVByCur_rocksdb(SStreamStateCur* pCur, SWinKey* pKey,
     if (pKey->groupId == groupId) {
       return 0;
     }
+    taosMemoryFree((void*)*pVal);
+    *pVal = NULL;
   }
   return -1;
 }
-- 
GitLab