From 59912fedf116a28330d601a6d576f185fe9be4a0 Mon Sep 17 00:00:00 2001
From: Cary Xu <klxu@taosdata.com>
Date: Sun, 4 Sep 2022 20:14:21 +0800
Subject: [PATCH] fix: stream input data block memory leak

---
 source/dnode/vnode/src/sma/smaEnv.c            |  3 ++-
 source/dnode/vnode/src/tsdb/tsdbReaderWriter.c |  2 +-
 source/libs/executor/src/executor.c            | 10 ++++------
 3 files changed, 7 insertions(+), 8 deletions(-)

diff --git a/source/dnode/vnode/src/sma/smaEnv.c b/source/dnode/vnode/src/sma/smaEnv.c
index 013481ab3d..2595a629ba 100644
--- a/source/dnode/vnode/src/sma/smaEnv.c
+++ b/source/dnode/vnode/src/sma/smaEnv.c
@@ -62,7 +62,7 @@ int32_t smaInit() {
     }
 
     int32_t type = (8 == POINTER_BYTES) ? TSDB_DATA_TYPE_UBIGINT : TSDB_DATA_TYPE_UINT;
-    smaMgmt.refHash = taosHashInit(1, taosGetDefaultHashFunction(type), true, HASH_ENTRY_LOCK);
+    smaMgmt.refHash = taosHashInit(64, taosGetDefaultHashFunction(type), true, HASH_ENTRY_LOCK);
     if (!smaMgmt.refHash) {
       taosCloseRef(smaMgmt.rsetId);
       atomic_store_8(&smaMgmt.inited, 0);
@@ -107,6 +107,7 @@ void smaCleanUp() {
   if (old == 1) {
     taosCloseRef(smaMgmt.rsetId);
     taosHashCleanup(smaMgmt.refHash);
+    smaMgmt.refHash = NULL;
     taosTmrCleanUp(smaMgmt.tmrHandle);
     smaInfo("sma mgmt env is cleaned up, rsetId:%d, tmrHandle:%p", smaMgmt.rsetId, smaMgmt.tmrHandle);
     atomic_store_8(&smaMgmt.inited, 0);
diff --git a/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c b/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
index c8f3862071..a4b13e6a6b 100644
--- a/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
+++ b/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
@@ -1303,7 +1303,7 @@ _err:
 int32_t tsdbWriteBlockL(SDataFWriter *pWriter, SArray *aBlockL) {
   int32_t    code = 0;
   SLastFile *pLastFile = &pWriter->fLast;
-  int64_t    size;
+  int64_t    size = 0;
   int64_t    n;
 
   // check
diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c
index 037bf543b1..136b2de596 100644
--- a/source/libs/executor/src/executor.c
+++ b/source/libs/executor/src/executor.c
@@ -30,6 +30,8 @@ static void cleanupRefPool() {
   taosCloseRef(ref);
 }
 
+static FORCE_INLINE void streamInputBlockDataDestory(void* pBlock) { blockDataDestroy((SSDataBlock*)pBlock); }
+
 static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, char* id) {
   ASSERT(pOperator != NULL);
   if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
@@ -53,7 +55,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
     // prevent setting a different type of block
     pInfo->validBlockIndex = 0;
     if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
-      taosArrayClearP(pInfo->pBlockLists, taosMemoryFree);
+      taosArrayClearP(pInfo->pBlockLists, streamInputBlockDataDestory);
     } else {
       taosArrayClear(pInfo->pBlockLists);
     }
@@ -107,11 +109,7 @@ void tdCleanupStreamInputDataBlock(qTaskInfo_t tinfo) {
   if (pOptrInfo->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
     SStreamScanInfo* pInfo = pOptrInfo->info;
     if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
-      for (int32_t i = 0; i < taosArrayGetSize(pInfo->pBlockLists); ++i) {
-        SSDataBlock* p = *(SSDataBlock**)taosArrayGet(pInfo->pBlockLists, i);
-        taosArrayDestroy(p->pDataBlock);
-        taosMemoryFreeClear(p);
-      }
+      taosArrayClearP(pInfo->pBlockLists, streamInputBlockDataDestory);
     } else {
       ASSERT(0);
     }
-- 
GitLab