From 56782a5d41eca1b85bf66ec1c48362188c351c43 Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Thu, 1 Jun 2023 18:29:16 +0800
Subject: [PATCH 01/19] fix:open task case & modify mqRebVgReq encode/decode
 style

---
 include/common/tmsg.h                      | 60 +++++++++++++---------
 source/dnode/mnode/impl/src/mndSubscribe.c | 19 +++++--
 source/dnode/vnode/src/tq/tq.c             | 19 ++++---
 tests/parallel_test/cases.task             |  2 +-
 4 files changed, 65 insertions(+), 35 deletions(-)

diff --git a/include/common/tmsg.h b/include/common/tmsg.h
index b4674aba54..d78e771fcf 100644
--- a/include/common/tmsg.h
+++ b/include/common/tmsg.h
@@ -2807,39 +2807,49 @@ typedef struct {
   int64_t suid;
 } SMqRebVgReq;
 
-static FORCE_INLINE int32_t tEncodeSMqRebVgReq(void** buf, const SMqRebVgReq* pReq) {
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
-  tlen += taosEncodeFixedI32(buf, pReq->vgId);
-  tlen += taosEncodeFixedI64(buf, pReq->oldConsumerId);
-  tlen += taosEncodeFixedI64(buf, pReq->newConsumerId);
-  tlen += taosEncodeString(buf, pReq->subKey);
-  tlen += taosEncodeFixedI8(buf, pReq->subType);
-  tlen += taosEncodeFixedI8(buf, pReq->withMeta);
+static FORCE_INLINE int tEncodeSMqRebVgReq(SEncoder *pCoder, const SMqRebVgReq* pReq) {
+  if (tStartEncode(pCoder) < 0) return -1;
+  if (tEncodeI64(pCoder, pReq->leftForVer) < 0) return -1;
+  if (tEncodeI32(pCoder, pReq->vgId) < 0) return -1;
+  if (tEncodeI64(pCoder, pReq->oldConsumerId) < 0) return -1;
+  if (tEncodeI64(pCoder, pReq->newConsumerId) < 0) return -1;
+  if (tEncodeCStr(pCoder, pReq->subKey) < 0) return -1;
+  if (tEncodeI8(pCoder, pReq->subType) < 0) return -1;
+  if (tEncodeI8(pCoder, pReq->withMeta) < 0) return -1;
+
   if (pReq->subType == TOPIC_SUB_TYPE__COLUMN) {
-    tlen += taosEncodeString(buf, pReq->qmsg);
+    if (tEncodeCStr(pCoder, pReq->qmsg) < 0) return -1;
   } else if (pReq->subType == TOPIC_SUB_TYPE__TABLE) {
-    tlen += taosEncodeFixedI64(buf, pReq->suid);
-    tlen += taosEncodeString(buf, pReq->qmsg);
+    if (tEncodeI64(pCoder, pReq->suid) < 0) return -1;
+    if (tEncodeCStr(pCoder, pReq->qmsg) < 0) return -1;
   }
-  return tlen;
+  tEndEncode(pCoder);
+  return 0;
 }
 
-static FORCE_INLINE void* tDecodeSMqRebVgReq(const void* buf, SMqRebVgReq* pReq) {
-  buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
-  buf = taosDecodeFixedI32(buf, &pReq->vgId);
-  buf = taosDecodeFixedI64(buf, &pReq->oldConsumerId);
-  buf = taosDecodeFixedI64(buf, &pReq->newConsumerId);
-  buf = taosDecodeStringTo(buf, pReq->subKey);
-  buf = taosDecodeFixedI8(buf, &pReq->subType);
-  buf = taosDecodeFixedI8(buf, &pReq->withMeta);
+static FORCE_INLINE int tDecodeSMqRebVgReq(SDecoder *pCoder, SMqRebVgReq* pReq) {
+  if (tStartDecode(pCoder) < 0) return -1;
+
+  if (tDecodeI64(pCoder, &pReq->leftForVer) < 0) return -1;
+
+  if (tDecodeI32(pCoder, &pReq->vgId) < 0) return -1;
+  if (tDecodeI64(pCoder, &pReq->oldConsumerId) < 0) return -1;
+  if (tDecodeI64(pCoder, &pReq->newConsumerId) < 0) return -1;
+  if (tDecodeCStrTo(pCoder, pReq->subKey) < 0) return -1;
+  if (tDecodeI8(pCoder, &pReq->subType) < 0) return -1;
+  if (tDecodeI8(pCoder, &pReq->withMeta) < 0) return -1;
+
   if (pReq->subType == TOPIC_SUB_TYPE__COLUMN) {
-    buf = taosDecodeString(buf, &pReq->qmsg);
+    if (tDecodeCStr(pCoder, &pReq->qmsg) < 0) return -1;
   } else if (pReq->subType == TOPIC_SUB_TYPE__TABLE) {
-    buf = taosDecodeFixedI64(buf, &pReq->suid);
-    buf = taosDecodeString(buf, &pReq->qmsg);
+    if (tDecodeI64(pCoder, &pReq->suid) < 0) return -1;
+    if (!tDecodeIsEnd(pCoder)){
+      if (tDecodeCStr(pCoder, &pReq->qmsg) < 0) return -1;
+    }
   }
-  return (void*)buf;
+
+  tEndDecode(pCoder);
+  return 0;
 }
 
 typedef struct {
diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c
index e62102fa77..74421afa33 100644
--- a/source/dnode/mnode/impl/src/mndSubscribe.c
+++ b/source/dnode/mnode/impl/src/mndSubscribe.c
@@ -111,7 +111,14 @@ static int32_t mndBuildSubChangeReq(void **pBuf, int32_t *pLen, const SMqSubscri
   req.suid = pSub->stbUid;
   tstrncpy(req.subKey, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);
 
-  int32_t tlen = sizeof(SMsgHead) + tEncodeSMqRebVgReq(NULL, &req);
+  int32_t tlen = 0;
+  int32_t ret = 0;
+  tEncodeSize(tEncodeSMqRebVgReq, &req, tlen, ret);
+  if (ret < 0) {
+    return -1;
+  }
+
+  tlen += sizeof(SMsgHead);
   void   *buf = taosMemoryMalloc(tlen);
   if (buf == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -123,8 +130,14 @@ static int32_t mndBuildSubChangeReq(void **pBuf, int32_t *pLen, const SMqSubscri
   pMsgHead->contLen = htonl(tlen);
   pMsgHead->vgId = htonl(pRebVg->pVgEp->vgId);
 
-  void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
-  tEncodeSMqRebVgReq(&abuf, &req);
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, POINTER_SHIFT(buf, sizeof(SMsgHead)), tlen);
+  if (tEncodeSMqRebVgReq(&encoder, &req) < 0) {
+    taosMemoryFreeClear(buf);
+    tEncoderClear(&encoder);
+    return -1;
+  }
+  tEncoderClear(&encoder);
   *pBuf = buf;
   *pLen = tlen;
 
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 11760b5fd0..35055a00e3 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -633,7 +633,16 @@ int32_t tqProcessDelCheckInfoReq(STQ* pTq, int64_t sversion, char* msg, int32_t
 int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
   int ret = 0;
   SMqRebVgReq req = {0};
-  tDecodeSMqRebVgReq(msg, &req);
+  SDecoder       dc = {0};
+
+  tDecoderInit(&dc, msg, msgLen);
+
+  // decode req
+  if (tDecodeSMqRebVgReq(&dc, &req) < 0) {
+    terrno = TSDB_CODE_INVALID_MSG;
+    tDecoderClear(&dc);
+    return -1;
+  }
 
   SVnode* pVnode = pTq->pVnode;
   int32_t vgId = TD_VID(pVnode);
@@ -680,8 +689,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
     pHandle->snapshotVer = ver;
 
     if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
-      pHandle->execHandle.execCol.qmsg = req.qmsg;
-      req.qmsg = NULL;
+      pHandle->execHandle.execCol.qmsg = taosStrdup(req.qmsg);;
 
       pHandle->execHandle.task = qCreateQueueExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle, vgId,
                                                           &pHandle->execHandle.numOfCols, req.newConsumerId);
@@ -701,8 +709,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
     } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
       pHandle->pWalReader = walOpenReader(pVnode->pWal, NULL);
       pHandle->execHandle.execTb.suid = req.suid;
-      pHandle->execHandle.execTb.qmsg = req.qmsg;
-      req.qmsg = NULL;
+      pHandle->execHandle.execTb.qmsg = taosStrdup(req.qmsg);
 
       if(strcmp(pHandle->execHandle.execTb.qmsg, "") != 0) {
         if (nodesStringToNode(pHandle->execHandle.execTb.qmsg, &pHandle->execHandle.execTb.node) != 0) {
@@ -766,7 +773,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
 
 end:
   taosWUnLockLatch(&pTq->lock);
-  taosMemoryFree(req.qmsg);
+  tDecoderClear(&dc);
   return ret;
 }
 
diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task
index 6a9a2b609a..0b235d0dc8 100644
--- a/tests/parallel_test/cases.task
+++ b/tests/parallel_test/cases.task
@@ -564,7 +564,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/user_privilege.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/fsync.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/multilevel.py
-#,,n,system-test,python3 ./test.py -f 0-others/compatibility.py
+,,n,system-test,python3 ./test.py -f 0-others/compatibility.py
 ,,n,system-test,python3 ./test.py -f 0-others/tag_index_basic.py
 ,,n,system-test,python3 ./test.py -f 0-others/udfpy_main.py
 ,,n,system-test,python3 ./test.py -N 3 -f 0-others/walRetention.py

From becc5c51a40c1d1e9f14b65f54eef9e3f939f33a Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 2 Jun 2023 02:35:15 +0000
Subject: [PATCH 02/19] fix compatibility

---
 source/libs/stream/src/streamBackendRocksdb.c | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c
index c8a6597bad..c54f055e6a 100644
--- a/source/libs/stream/src/streamBackendRocksdb.c
+++ b/source/libs/stream/src/streamBackendRocksdb.c
@@ -1924,17 +1924,17 @@ int32_t streamStateGetParName_rocksdb(SStreamState* pState, int64_t groupId, voi
 
 int32_t streamDefaultPut_rocksdb(SStreamState* pState, const void* key, void* pVal, int32_t pVLen) {
   int code = 0;
-  STREAM_STATE_PUT_ROCKSDB(pState, "default", &key, pVal, pVLen);
+  STREAM_STATE_PUT_ROCKSDB(pState, "default", key, pVal, pVLen);
   return code;
 }
 int32_t streamDefaultGet_rocksdb(SStreamState* pState, const void* key, void** pVal, int32_t* pVLen) {
   int code = 0;
-  STREAM_STATE_GET_ROCKSDB(pState, "default", &key, pVal, pVLen);
+  STREAM_STATE_GET_ROCKSDB(pState, "default", key, pVal, pVLen);
   return code;
 }
 int32_t streamDefaultDel_rocksdb(SStreamState* pState, const void* key) {
   int code = 0;
-  STREAM_STATE_DEL_ROCKSDB(pState, "default", &key);
+  STREAM_STATE_DEL_ROCKSDB(pState, "default", key);
   return code;
 }
 

From 5c64a730a9668f2768a13c9255f4a73e6b087046 Mon Sep 17 00:00:00 2001
From: xleili <xlli@taosdata.com>
Date: Fri, 2 Jun 2023 11:17:38 +0800
Subject: [PATCH 03/19] enchance: change goes as static link

---
 cmake/cmake.options        |  2 +-
 cmake/cmake.platform       | 10 ----------
 packaging/tools/makepkg.sh |  3 +++
 3 files changed, 4 insertions(+), 11 deletions(-)

diff --git a/cmake/cmake.options b/cmake/cmake.options
index b00ae14715..555b72cbdf 100644
--- a/cmake/cmake.options
+++ b/cmake/cmake.options
@@ -80,7 +80,7 @@ ENDIF ()
 option(
         BUILD_GEOS
         "If build geos on Windows"
-        OFF
+        ON
     )
 
 option(
diff --git a/cmake/cmake.platform b/cmake/cmake.platform
index f9faf7316c..ba747c6134 100644
--- a/cmake/cmake.platform
+++ b/cmake/cmake.platform
@@ -56,17 +56,7 @@ IF (${CMAKE_SYSTEM_NAME} MATCHES "Linux" OR ${CMAKE_SYSTEM_NAME} MATCHES "Darwin
 
         SET(TD_DARWIN TRUE)
         SET(OSTYPE "macOS")
-        execute_process(COMMAND geos-config --cflags OUTPUT_VARIABLE GEOS_CFLAGS)
-        execute_process(COMMAND geos-config --ldflags OUTPUT_VARIABLE GEOS_LDFLAGS)
-        string(SUBSTRING ${GEOS_CFLAGS} 2 -1 GEOS_CFLAGS)
-        string(REGEX REPLACE "\n" "" GEOS_CFLAGS ${GEOS_CFLAGS})
-        string(SUBSTRING ${GEOS_LDFLAGS} 2 -1 GEOS_LDFLAGS)
-        string(REGEX REPLACE "\n" "" GEOS_LDFLAGS ${GEOS_LDFLAGS})
-        MESSAGE("GEOS_CFLAGS "${GEOS_CFLAGS})
-        MESSAGE("GEOS_LDFLAGS "${GEOS_LDFLAGS})
         ADD_DEFINITIONS("-DDARWIN -Wno-tautological-pointer-compare")
-        INCLUDE_DIRECTORIES(${GEOS_CFLAGS})
-        LINK_DIRECTORIES(${GEOS_LDFLAGS})
 
         IF (${CMAKE_SYSTEM_PROCESSOR} MATCHES "arm64")
             MESSAGE("Current system arch is arm64")
diff --git a/packaging/tools/makepkg.sh b/packaging/tools/makepkg.sh
index 2f1e803689..7cfd9aa664 100755
--- a/packaging/tools/makepkg.sh
+++ b/packaging/tools/makepkg.sh
@@ -112,9 +112,11 @@ fi
 if [ "$osType" == "Darwin" ]; then
     lib_files="${build_dir}/lib/libtaos.${version}.dylib"
     wslib_files="${build_dir}/lib/libtaosws.dylib"
+    rocksdb_lib_files="${build_dir}/lib/librocksdb.dylib.8.1.1"
 else
     lib_files="${build_dir}/lib/libtaos.so.${version}"
     wslib_files="${build_dir}/lib/libtaosws.so"
+    rocksdb_lib_files="${build_dir}/lib/librocksdb.so.8.1.1"
 fi
 header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h ${code_dir}/include/libs/function/taosudf.h"
 
@@ -337,6 +339,7 @@ fi
 # Copy driver
 mkdir -p ${install_dir}/driver && cp ${lib_files} ${install_dir}/driver && echo "${versionComp}" >${install_dir}/driver/vercomp.txt
 [ -f ${wslib_files} ] && cp ${wslib_files} ${install_dir}/driver || :
+[ -f ${rocksdb_lib_files} ] && cp ${rocksdb_lib_files} ${install_dir}/driver || :
 
 # Copy connector
 if [ "$verMode" == "cluster" ]; then

From 63d9a5dcf37220262e11db2c3a96c2626fa798d7 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 2 Jun 2023 04:08:08 +0000
Subject: [PATCH 04/19] avoid mem leak

---
 source/libs/stream/src/tstreamFileState.c | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/source/libs/stream/src/tstreamFileState.c b/source/libs/stream/src/tstreamFileState.c
index f531f65565..54832b042a 100644
--- a/source/libs/stream/src/tstreamFileState.c
+++ b/source/libs/stream/src/tstreamFileState.c
@@ -16,12 +16,12 @@
 #include "tstreamFileState.h"
 
 #include "query.h"
+#include "storageapi.h"
 #include "streamBackendRocksdb.h"
 #include "taos.h"
 #include "tcommon.h"
 #include "thash.h"
 #include "tsimplehash.h"
-#include "storageapi.h"
 
 #define FLUSH_RATIO                    0.5
 #define FLUSH_NUM                      4
@@ -420,6 +420,7 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
       return TSDB_CODE_FAILED;
     }
     sscanf(val, "%" PRId64 "", &maxCheckPointId);
+    taosMemoryFree(val);
   }
   for (int64_t i = maxCheckPointId; i > 0; i--) {
     char    buf[128] = {0};
@@ -435,9 +436,11 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
     if (ts < mark) {
       // statekey winkey.ts < mark
       forceRemoveCheckpoint(pFileState, i);
+      taosMemoryFreeClear(val);
       break;
     } else {
     }
+    taosMemoryFree(val);
   }
   return code;
 }

From 3b69027325f59d03ed7893fdbb326e3392244c13 Mon Sep 17 00:00:00 2001
From: xleili <xlli@taosdata.com>
Date: Fri, 2 Jun 2023 14:01:25 +0800
Subject: [PATCH 05/19] enhance: supply librocksdb.so install and uninstall

---
 packaging/tools/install.sh | 9 +++++++++
 packaging/tools/post.sh    | 6 ++++++
 packaging/tools/remove.sh  | 3 +++
 3 files changed, 18 insertions(+)

diff --git a/packaging/tools/install.sh b/packaging/tools/install.sh
index 1b47b10520..9452e48d33 100755
--- a/packaging/tools/install.sh
+++ b/packaging/tools/install.sh
@@ -250,18 +250,27 @@ function install_lib() {
   # Remove links
   ${csudo}rm -f ${lib_link_dir}/libtaos.* || :
   ${csudo}rm -f ${lib64_link_dir}/libtaos.* || :
+  ${csudo}rm -f ${lib_link_dir}/librocksdb.* || :
+  ${csudo}rm -f ${lib64_link_dir}/librocksdb.* || :
   #${csudo}rm -rf ${v15_java_app_dir}              || :
   ${csudo}cp -rf ${script_dir}/driver/* ${install_main_dir}/driver && ${csudo}chmod 777 ${install_main_dir}/driver/*
 
   ${csudo}ln -sf ${install_main_dir}/driver/libtaos.* ${lib_link_dir}/libtaos.so.1
   ${csudo}ln -sf ${lib_link_dir}/libtaos.so.1 ${lib_link_dir}/libtaos.so
 
+  ${csudo}ln -sf ${install_main_dir}/driver/librocksdb.* ${lib_link_dir}/librocksdb.so.8
+  ${csudo}ln -sf ${lib_link_dir}/librocksdb.so.8 ${lib_link_dir}/librocksdb.so
+
+
   [ -f ${install_main_dir}/driver/libtaosws.so ] && ${csudo}ln -sf ${install_main_dir}/driver/libtaosws.so ${lib_link_dir}/libtaosws.so || :
 
   if [[ -d ${lib64_link_dir} && ! -e ${lib64_link_dir}/libtaos.so ]]; then
     ${csudo}ln -sf ${install_main_dir}/driver/libtaos.* ${lib64_link_dir}/libtaos.so.1 || :
     ${csudo}ln -sf ${lib64_link_dir}/libtaos.so.1 ${lib64_link_dir}/libtaos.so || :
 
+    ${csudo}ln -sf ${install_main_dir}/driver/librocksdb.* ${lib64_link_dir}/librocksdb.so.8 || :
+    ${csudo}ln -sf ${lib64_link_dir}/librocksdb.so.8 ${lib64_link_dir}/librocksdb.so || :
+
     [ -f ${install_main_dir}/libtaosws.so ] && ${csudo}ln -sf ${install_main_dir}/libtaosws.so ${lib64_link_dir}/libtaosws.so || :
   fi
 
diff --git a/packaging/tools/post.sh b/packaging/tools/post.sh
index fc392c9684..4a8dda1d30 100755
--- a/packaging/tools/post.sh
+++ b/packaging/tools/post.sh
@@ -202,10 +202,15 @@ function install_lib() {
     log_print "start install lib from ${lib_dir} to ${lib_link_dir}"
     ${csudo}rm -f ${lib_link_dir}/libtaos* || :
     ${csudo}rm -f ${lib64_link_dir}/libtaos* || :
+    
+    #rocksdb
+    [ -f ${lib_link_dir}/librocksdb* ] && ${csudo}rm -f ${lib_link_dir}/librocksdb* || :
+    [ -f ${lib64_link_dir}/librocksdb* ] && ${csudo}rm -f ${lib64_link_dir}/librocksdb* || :
 
     [ -f ${lib_link_dir}/libtaosws.${lib_file_ext} ] && ${csudo}rm -f ${lib_link_dir}/libtaosws.${lib_file_ext} || :
     [ -f ${lib64_link_dir}/libtaosws.${lib_file_ext} ] && ${csudo}rm -f ${lib64_link_dir}/libtaosws.${lib_file_ext} || :
 
+    ${csudo}ln -s ${lib_dir}/librocksdb.* ${lib_link_dir}/librocksdb.${lib_file_ext_1} 2>>${install_log_path} || return 1
     ${csudo}ln -s ${lib_dir}/libtaos.* ${lib_link_dir}/libtaos.${lib_file_ext_1} 2>>${install_log_path} || return 1
     ${csudo}ln -s ${lib_link_dir}/libtaos.${lib_file_ext_1} ${lib_link_dir}/libtaos.${lib_file_ext} 2>>${install_log_path} || return 1
 
@@ -214,6 +219,7 @@ function install_lib() {
     if [[ -d ${lib64_link_dir} && ! -e ${lib64_link_dir}/libtaos.${lib_file_ext} ]]; then
       ${csudo}ln -s ${lib_dir}/libtaos.* ${lib64_link_dir}/libtaos.${lib_file_ext_1} 2>>${install_log_path} || return 1
       ${csudo}ln -s ${lib64_link_dir}/libtaos.${lib_file_ext_1} ${lib64_link_dir}/libtaos.${lib_file_ext}  2>>${install_log_path} || return 1
+      ${csudo}ln -s ${lib_dir}/librocksdb.* ${lib64_link_dir}/librocksdb.${lib_file_ext_1} 2>>${install_log_path} || return 1
 
       [ -f ${lib_dir}/libtaosws.${lib_file_ext} ] && ${csudo}ln -sf ${lib_dir}/libtaosws.${lib_file_ext} ${lib64_link_dir}/libtaosws.${lib_file_ext} 2>>${install_log_path}
     fi
diff --git a/packaging/tools/remove.sh b/packaging/tools/remove.sh
index 6c671473bf..a17b29983c 100755
--- a/packaging/tools/remove.sh
+++ b/packaging/tools/remove.sh
@@ -142,11 +142,14 @@ function clean_local_bin() {
 function clean_lib() {
   # Remove link
   ${csudo}rm -f ${lib_link_dir}/libtaos.* || :
+  ${csudo}rm -f ${lib_link_dir}/librocksdb.* || :
   [ -f ${lib_link_dir}/libtaosws.* ] && ${csudo}rm -f ${lib_link_dir}/libtaosws.* || :
 
   ${csudo}rm -f ${lib64_link_dir}/libtaos.* || :
+  ${csudo}rm -f ${lib64_link_dir}/librocksdb.* || :
   [ -f ${lib64_link_dir}/libtaosws.* ] && ${csudo}rm -f ${lib64_link_dir}/libtaosws.* || :
   #${csudo}rm -rf ${v15_java_app_dir}           || :
+  
 }
 
 function clean_header() {

From f59161464d7f596418b34915d3bd51368675e50c Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 2 Jun 2023 06:48:26 +0000
Subject: [PATCH 06/19] avoid mem leak

---
 source/libs/executor/src/scanoperator.c | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index 146552b058..6d17672513 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -13,6 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+// clang-format off
+
 #include "executorInt.h"
 #include "filter.h"
 #include "function.h"
@@ -2444,6 +2446,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
     int32_t len = 0;
     pAPI->stateStore.streamStateGetInfo(pTaskInfo->streamInfo.pState, STREAM_SCAN_OP_NAME, strlen(STREAM_SCAN_OP_NAME), &buff, &len);
     streamScanOperatorDecode(buff, len, pInfo);
+    taosMemoryFree(buff);
   }
 
   setOperatorInfo(pOperator, STREAM_SCAN_OP_NAME, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, false, OP_NOT_OPENED, pInfo,
@@ -3458,3 +3461,5 @@ static void destoryTableCountScanOperator(void* param) {
   taosArrayDestroy(pTableCountScanInfo->stbUidList);
   taosMemoryFreeClear(param);
 }
+
+// clang-format on

From 822f2673990766daf5d6403ecc054ea4c73e255f Mon Sep 17 00:00:00 2001
From: xleili <xlli@taosdata.com>
Date: Fri, 2 Jun 2023 15:36:36 +0800
Subject: [PATCH 07/19] enhance: install and uninstall rocksb.so in deb and rpm

---
 packaging/deb/DEBIAN/preinst | 1 +
 packaging/deb/DEBIAN/prerm   | 1 +
 packaging/deb/makedeb.sh     | 1 +
 packaging/rpm/tdengine.spec  | 2 ++
 4 files changed, 5 insertions(+)

diff --git a/packaging/deb/DEBIAN/preinst b/packaging/deb/DEBIAN/preinst
index 904a946e20..d9951fc4d4 100644
--- a/packaging/deb/DEBIAN/preinst
+++ b/packaging/deb/DEBIAN/preinst
@@ -80,4 +80,5 @@ fi
 
 # there can not libtaos.so*, otherwise ln -s  error
 ${csudo}rm -f ${install_main_dir}/driver/libtaos.*   || :
+${csudo}rm -f ${install_main_dir}/driver/librocksdb.*   || :
 [ -f ${install_main_dir}/driver/libtaosws.so ] && ${csudo}rm -f ${install_main_dir}/driver/libtaosws.so   || :
diff --git a/packaging/deb/DEBIAN/prerm b/packaging/deb/DEBIAN/prerm
index 0d63115a04..80799ae0e5 100644
--- a/packaging/deb/DEBIAN/prerm
+++ b/packaging/deb/DEBIAN/prerm
@@ -40,6 +40,7 @@ else
     ${csudo}rm -f ${inc_link_dir}/taosudf.h     || :
     [ -f ${inc_link_dir}/taosws.h ] && ${csudo}rm -f ${inc_link_dir}/taosws.h     || :
     ${csudo}rm -f ${lib_link_dir}/libtaos.*  || :
+    ${csudo}rm -f ${lib_link_dir}/librocksdb.*  || :
     [ -f ${lib_link_dir}/libtaosws.so ] && ${csudo}rm -f ${lib_link_dir}/libtaosws.so  || :
 
     ${csudo}rm -f ${log_link_dir}            || :
diff --git a/packaging/deb/makedeb.sh b/packaging/deb/makedeb.sh
index 9f49cf345a..624ccb0e3b 100755
--- a/packaging/deb/makedeb.sh
+++ b/packaging/deb/makedeb.sh
@@ -94,6 +94,7 @@ fi
 
 cp ${compile_dir}/build/bin/taos                    ${pkg_dir}${install_home_path}/bin
 cp ${compile_dir}/build/lib/${libfile}              ${pkg_dir}${install_home_path}/driver
+cp ${compile_dir}/build/lib/librocksdb.8              ${pkg_dir}${install_home_path}/driver
 [ -f ${compile_dir}/build/lib/${wslibfile} ] && cp ${compile_dir}/build/lib/${wslibfile}            ${pkg_dir}${install_home_path}/driver ||:
 cp ${compile_dir}/../include/client/taos.h          ${pkg_dir}${install_home_path}/include
 cp ${compile_dir}/../include/common/taosdef.h       ${pkg_dir}${install_home_path}/include
diff --git a/packaging/rpm/tdengine.spec b/packaging/rpm/tdengine.spec
index 52d5335003..31d597785e 100644
--- a/packaging/rpm/tdengine.spec
+++ b/packaging/rpm/tdengine.spec
@@ -174,6 +174,7 @@ fi
 
 # there can not libtaos.so*, otherwise ln -s  error
 ${csudo}rm -f %{homepath}/driver/libtaos*   || :
+${csudo}rm -f %{homepath}/driver/librocksdb*   || :
 
 #Scripts executed after installation
 %post
@@ -219,6 +220,7 @@ if [ $1 -eq 0 ];then
     ${csudo}rm -f ${inc_link_dir}/taoserror.h     || :
     ${csudo}rm -f ${inc_link_dir}/taosudf.h     || :    
     ${csudo}rm -f ${lib_link_dir}/libtaos.*  || :
+    ${csudo}rm -f ${lib_link_dir}/librocksdb.*  || :
 
     ${csudo}rm -f ${log_link_dir}            || :
     ${csudo}rm -f ${data_link_dir}           || :

From 53696caabf16b91c1e621d6d1e11f1083fd1f9df Mon Sep 17 00:00:00 2001
From: xleili <xlli@taosdata.com>
Date: Fri, 2 Jun 2023 16:52:36 +0800
Subject: [PATCH 08/19] fix: rpm and deb librocksdb.so install and unintall

---
 packaging/deb/DEBIAN/preinst | 2 +-
 packaging/deb/DEBIAN/prerm   | 2 +-
 packaging/deb/makedeb.sh     | 3 ++-
 packaging/rpm/tdengine.spec  | 2 ++
 4 files changed, 6 insertions(+), 3 deletions(-)

diff --git a/packaging/deb/DEBIAN/preinst b/packaging/deb/DEBIAN/preinst
index d9951fc4d4..d6558d5b3b 100644
--- a/packaging/deb/DEBIAN/preinst
+++ b/packaging/deb/DEBIAN/preinst
@@ -80,5 +80,5 @@ fi
 
 # there can not libtaos.so*, otherwise ln -s  error
 ${csudo}rm -f ${install_main_dir}/driver/libtaos.*   || :
-${csudo}rm -f ${install_main_dir}/driver/librocksdb.*   || :
+[ -f ${install_main_dir}/driver/librocksdb.* ] && ${csudo}rm -f ${install_main_dir}/driver/librocksdb.*  || :
 [ -f ${install_main_dir}/driver/libtaosws.so ] && ${csudo}rm -f ${install_main_dir}/driver/libtaosws.so   || :
diff --git a/packaging/deb/DEBIAN/prerm b/packaging/deb/DEBIAN/prerm
index 80799ae0e5..8f8d472867 100644
--- a/packaging/deb/DEBIAN/prerm
+++ b/packaging/deb/DEBIAN/prerm
@@ -40,7 +40,7 @@ else
     ${csudo}rm -f ${inc_link_dir}/taosudf.h     || :
     [ -f ${inc_link_dir}/taosws.h ] && ${csudo}rm -f ${inc_link_dir}/taosws.h     || :
     ${csudo}rm -f ${lib_link_dir}/libtaos.*  || :
-    ${csudo}rm -f ${lib_link_dir}/librocksdb.*  || :
+    [ -f ${lib_link_dir}/librocksdb.* ] && ${csudo}rm -f ${lib_link_dir}/librocksdb.*  || :
     [ -f ${lib_link_dir}/libtaosws.so ] && ${csudo}rm -f ${lib_link_dir}/libtaosws.so  || :
 
     ${csudo}rm -f ${log_link_dir}            || :
diff --git a/packaging/deb/makedeb.sh b/packaging/deb/makedeb.sh
index 624ccb0e3b..024c69deb1 100755
--- a/packaging/deb/makedeb.sh
+++ b/packaging/deb/makedeb.sh
@@ -31,6 +31,7 @@ cd ${pkg_dir}
 
 libfile="libtaos.so.${tdengine_ver}"
 wslibfile="libtaosws.so"
+rocksdblib="librocksdb.so.8"
 
 # create install dir
 install_home_path="/usr/local/taos"
@@ -94,7 +95,7 @@ fi
 
 cp ${compile_dir}/build/bin/taos                    ${pkg_dir}${install_home_path}/bin
 cp ${compile_dir}/build/lib/${libfile}              ${pkg_dir}${install_home_path}/driver
-cp ${compile_dir}/build/lib/librocksdb.8              ${pkg_dir}${install_home_path}/driver
+[ -f ${compile_dir}/build/lib/${rocksdblib} ] && cp ${compile_dir}/build/lib/${rocksdblib}            ${pkg_dir}${install_home_path}/driver ||:
 [ -f ${compile_dir}/build/lib/${wslibfile} ] && cp ${compile_dir}/build/lib/${wslibfile}            ${pkg_dir}${install_home_path}/driver ||:
 cp ${compile_dir}/../include/client/taos.h          ${pkg_dir}${install_home_path}/include
 cp ${compile_dir}/../include/common/taosdef.h       ${pkg_dir}${install_home_path}/include
diff --git a/packaging/rpm/tdengine.spec b/packaging/rpm/tdengine.spec
index 31d597785e..2b056c376a 100644
--- a/packaging/rpm/tdengine.spec
+++ b/packaging/rpm/tdengine.spec
@@ -45,6 +45,7 @@ echo buildroot: %{buildroot}
 
 libfile="libtaos.so.%{_version}"
 wslibfile="libtaosws.so"
+rocksdblib="librocksdb.so.8"
 
 # create install path, and cp file
 mkdir -p %{buildroot}%{homepath}/bin
@@ -92,6 +93,7 @@ if [ -f %{_compiledir}/build/bin/taosadapter ]; then
 fi
 cp %{_compiledir}/build/lib/${libfile}              %{buildroot}%{homepath}/driver
 [ -f %{_compiledir}/build/lib/${wslibfile} ] && cp %{_compiledir}/build/lib/${wslibfile}            %{buildroot}%{homepath}/driver ||:
+[ -f %{_compiledir}/build/lib/${rocksdblib} ] && cp %{_compiledir}/build/lib/${rocksdblib}            %{buildroot}%{homepath}/driver ||:
 cp %{_compiledir}/../include/client/taos.h          %{buildroot}%{homepath}/include
 cp %{_compiledir}/../include/common/taosdef.h       %{buildroot}%{homepath}/include
 cp %{_compiledir}/../include/util/taoserror.h       %{buildroot}%{homepath}/include

From 9f8f5374aaecd3f07f461240e2bc42ce35a56a8f Mon Sep 17 00:00:00 2001
From: yihaoDeng <cyang@taosdata.com>
Date: Fri, 2 Jun 2023 16:56:20 +0800
Subject: [PATCH 09/19] fix: num_of_records_per_req item in json config need
 less than 32768. current = 100000

---
 tests/system-test/1-insert/manyVgroups.json | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/system-test/1-insert/manyVgroups.json b/tests/system-test/1-insert/manyVgroups.json
index 3b0fa96b08..8c6f39cf96 100644
--- a/tests/system-test/1-insert/manyVgroups.json
+++ b/tests/system-test/1-insert/manyVgroups.json
@@ -11,7 +11,7 @@
     "confirm_parameter_prompt": "no",
     "insert_interval": 0,
     "interlace_rows": 0,
-    "num_of_records_per_req": 100000,
+    "num_of_records_per_req": 10000,
     "databases": [
         {
             "dbinfo": {
@@ -73,4 +73,4 @@
             ]
         }
     ]
-}
\ No newline at end of file
+}

From 27d8870e182a6f31a4590869f3a7ec3166e29fd2 Mon Sep 17 00:00:00 2001
From: t_max <1172915550@qq.com>
Date: Fri, 2 Jun 2023 17:09:03 +0800
Subject: [PATCH 10/19] fix: no data after seek

---
 source/client/src/clientTmq.c | 14 ++++++++++++--
 1 file changed, 12 insertions(+), 2 deletions(-)

diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c
index ccb09e3584..e1b2b9c48b 100644
--- a/source/client/src/clientTmq.c
+++ b/source/client/src/clientTmq.c
@@ -1868,7 +1868,10 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
         // update the local offset value only for the returned values, only when the local offset is NOT updated
         // by tmq_offset_seek function
         if (!pVg->seekUpdated) {
+          tscDebug("consumer:0x%" PRIx64" local offset is update, since seekupdate not set", tmq->consumerId);
           pVg->offsetInfo.currentOffset = pDataRsp->rspOffset;
+        } else {
+          tscDebug("consumer:0x%" PRIx64" local offset is NOT update, since seekupdate is set", tmq->consumerId);
         }
 
         // update the status
@@ -1952,8 +1955,15 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
           return NULL;
         }
 
-        if(pollRspWrapper->taosxRsp.rspOffset.type != 0){    // if offset is validate
-          pVg->offsetInfo.currentOffset = pollRspWrapper->taosxRsp.rspOffset;
+        // update the local offset value only for the returned values, only when the local offset is NOT updated
+        // by tmq_offset_seek function
+        if (!pVg->seekUpdated) {
+          if(pollRspWrapper->taosxRsp.rspOffset.type != 0) {    // if offset is validate
+            tscDebug("consumer:0x%" PRIx64" local offset is update, since seekupdate not set", tmq->consumerId);
+            pVg->offsetInfo.currentOffset = pollRspWrapper->taosxRsp.rspOffset;
+          }
+        } else {
+          tscDebug("consumer:0x%" PRIx64" local offset is NOT update, since seekupdate is set", tmq->consumerId);
         }
 
         atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);

From 449051cda67aff90ccb2e690b5521025803b78b0 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 2 Jun 2023 10:05:32 +0000
Subject: [PATCH 11/19] avoid mem leak

---
 source/libs/stream/src/streamBackendRocksdb.c | 4 ++--
 source/libs/stream/src/tstreamFileState.c     | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c
index b8af05c86d..30a7cefc9c 100644
--- a/source/libs/stream/src/streamBackendRocksdb.c
+++ b/source/libs/stream/src/streamBackendRocksdb.c
@@ -1056,7 +1056,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
     rocksdb_readoptions_t* opts = pState->pTdbState->readOpts;                                                         \
     size_t                 len = 0;                                                                                    \
     char*                  val = rocksdb_get_cf(db, opts, pHandle, (const char*)buf, klen, (size_t*)&len, &err);       \
-    if (val == NULL) {                                                                                                 \
+    if (val == NULL || len == 0) {                                                                                     \
       if (err == NULL) {                                                                                               \
         qTrace("streamState str: %s failed to read from %s_%s, err: not exist", toString, pState->pTdbState->idstr,    \
                funcname);                                                                                              \
@@ -1069,7 +1069,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
     } else {                                                                                                           \
       char*   p = NULL;                                                                                                \
       int32_t len = ginitDict[i].deValueFunc(val, len, NULL, (char**)pVal);                                            \
-      if (len < 0) {                                                                                                   \
+      if (len <= 0) {                                                                                                  \
         qError("streamState str: %s failed to read from %s_%s, err: already ttl ", toString, pState->pTdbState->idstr, \
                funcname);                                                                                              \
         code = -1;                                                                                                     \
diff --git a/source/libs/stream/src/tstreamFileState.c b/source/libs/stream/src/tstreamFileState.c
index 54832b042a..7ea1a41925 100644
--- a/source/libs/stream/src/tstreamFileState.c
+++ b/source/libs/stream/src/tstreamFileState.c
@@ -416,7 +416,7 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
     int32_t len = 0;
     memcpy(buf, taskKey, strlen(taskKey));
     code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);
-    if (code != 0) {
+    if (code != 0 || len == 0 || val == NULL) {
       return TSDB_CODE_FAILED;
     }
     sscanf(val, "%" PRId64 "", &maxCheckPointId);

From 17f2c3f88249b07ed79c5888bdf505c1322d9f20 Mon Sep 17 00:00:00 2001
From: liuyao <54liuyao@163.com>
Date: Fri, 2 Jun 2023 16:51:44 +0800
Subject: [PATCH 12/19] reset wind info

---
 source/libs/executor/inc/executorInt.h        |  1 +
 source/libs/executor/src/timewindowoperator.c | 25 +++++++++++++++++++
 source/libs/stream/src/tstreamFileState.c     |  2 +-
 3 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/source/libs/executor/inc/executorInt.h b/source/libs/executor/inc/executorInt.h
index 79391bc1c5..ffc63a22a8 100644
--- a/source/libs/executor/inc/executorInt.h
+++ b/source/libs/executor/inc/executorInt.h
@@ -456,6 +456,7 @@ typedef struct SStreamIntervalOperatorInfo {
   SSHashObj*         pUpdatedMap;
   int64_t            dataVersion;
   SStateStore        statestore;
+  bool               recvGetAll;
 } SStreamIntervalOperatorInfo;
 
 typedef struct SDataGroupInfo {
diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c
index fc08f827a5..3a83472079 100644
--- a/source/libs/executor/src/timewindowoperator.c
+++ b/source/libs/executor/src/timewindowoperator.c
@@ -2439,6 +2439,15 @@ static inline int winPosCmprImpl(const void* pKey1, const void* pKey2) {
   return 0;
 }
 
+static void resetUnCloseWinInfo(SSHashObj* winMap) {
+  void*   pIte = NULL;
+  int32_t iter = 0;
+  while ((pIte = tSimpleHashIterate(winMap, pIte, &iter)) != NULL) {
+    SRowBuffPos* pPos = *(SRowBuffPos**)pIte;
+    pPos->beUsed = true;
+  }
+}
+
 static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
   SStreamIntervalOperatorInfo* pInfo = pOperator->info;
   SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;
@@ -2472,6 +2481,11 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
       return pInfo->binfo.pRes;
     }
 
+    if (pInfo->recvGetAll) {
+      pInfo->recvGetAll = false;
+      resetUnCloseWinInfo(pInfo->aggSup.pResultRowHashTable);
+    }
+
     setOperatorCompleted(pOperator);
     if (!IS_FINAL_OP(pInfo)) {
       clearFunctionContext(&pOperator->exprSupp);
@@ -2565,6 +2579,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
 
       break;
     } else if (pBlock->info.type == STREAM_GET_ALL && IS_FINAL_OP(pInfo)) {
+      pInfo->recvGetAll = true;
       getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pInfo->pUpdatedMap);
       continue;
     } else if (pBlock->info.type == STREAM_RETRIEVE && !IS_FINAL_OP(pInfo)) {
@@ -2773,6 +2788,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
                                                   compareTs, pInfo->pState, pInfo->twAggSup.deleteMark);
   pInfo->dataVersion = 0;
   pInfo->statestore = pTaskInfo->storageAPI.stateStore;
+  pInfo->recvGetAll = false;
 
   pOperator->operatorType = pPhyNode->type;
   pOperator->blocking = true;
@@ -4751,6 +4767,12 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
       printDataBlock(pInfo->binfo.pRes, "single interval");
       return pInfo->binfo.pRes;
     }
+
+    if (pInfo->recvGetAll) {
+      pInfo->recvGetAll = false;
+      resetUnCloseWinInfo(pInfo->aggSup.pResultRowHashTable);
+    }
+
     setOperatorCompleted(pOperator);
     if (pInfo->twAggSup.maxTs > 0 &&
         pInfo->twAggSup.maxTs - pInfo->twAggSup.checkPointInterval > pInfo->twAggSup.checkPointTs) {
@@ -4790,6 +4812,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
       continue;
     } else if (pBlock->info.type == STREAM_GET_ALL) {
       qDebug("===stream===single interval recv|block type STREAM_GET_ALL");
+      pInfo->recvGetAll = true;
       getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pInfo->pUpdatedMap);
       continue;
     } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
@@ -4960,6 +4983,8 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
                                          destroyStreamFinalIntervalOperatorInfo, optrDefaultBufFn, NULL);
 
   pInfo->statestore = pTaskInfo->storageAPI.stateStore;
+  pInfo->recvGetAll = false;
+
   initIntervalDownStream(downstream, pPhyNode->type, pInfo);
   code = appendDownstream(pOperator, &downstream, 1);
   if (code != TSDB_CODE_SUCCESS) {
diff --git a/source/libs/stream/src/tstreamFileState.c b/source/libs/stream/src/tstreamFileState.c
index f531f65565..e13499d87f 100644
--- a/source/libs/stream/src/tstreamFileState.c
+++ b/source/libs/stream/src/tstreamFileState.c
@@ -137,7 +137,7 @@ void clearExpiredRowBuff(SStreamFileState* pFileState, TSKEY ts, bool all) {
   SListNode* pNode = NULL;
   while ((pNode = tdListNext(&iter)) != NULL) {
     SRowBuffPos* pPos = *(SRowBuffPos**)(pNode->data);
-    if (all || (pFileState->getTs(pPos->pKey) < ts)) {
+    if (all || (pFileState->getTs(pPos->pKey) < ts && !pPos->beUsed)) {
       ASSERT(pPos->pRowBuff != NULL);
       tdListAppend(pFileState->freeBuffs, &(pPos->pRowBuff));
       pPos->pRowBuff = NULL;

From 6ee68331b319aefe7578a0051285ad4a7069588f Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 2 Jun 2023 11:54:36 +0000
Subject: [PATCH 13/19] update DEFAULT_BUFF

---
 source/libs/stream/src/streamBackendRocksdb.c |  7 +++++--
 source/libs/stream/src/streamUpdate.c         | 12 ++++++------
 2 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c
index 30a7cefc9c..e07198f6ab 100644
--- a/source/libs/stream/src/streamBackendRocksdb.c
+++ b/source/libs/stream/src/streamBackendRocksdb.c
@@ -87,8 +87,6 @@ void* streamBackendInit(const char* path) {
   pHandle->cfInst = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
 
   rocksdb_env_t* env = rocksdb_create_default_env();  // rocksdb_envoptions_create();
-  rocksdb_env_set_low_priority_background_threads(env, 4);
-  rocksdb_env_set_high_priority_background_threads(env, 2);
 
   rocksdb_cache_t* cache = rocksdb_cache_create_lru(64 << 20);
 
@@ -577,6 +575,11 @@ int32_t decodeValueFunc(void* value, int32_t vlen, int64_t* ttl, char** dest) {
   int64_t now = taosGetTimestampMs();
   p = taosDecodeFixedI64(p, &key.unixTimestamp);
   p = taosDecodeFixedI32(p, &key.len);
+  if (vlen != sizeof(int64_t) + sizeof(int32_t) + key.len) {
+    *dest = NULL;
+    return -1;
+  }
+
   if (key.len == 0) {
     key.data = NULL;
   } else {
diff --git a/source/libs/stream/src/streamUpdate.c b/source/libs/stream/src/streamUpdate.c
index fff666ec9f..85be120dbd 100644
--- a/source/libs/stream/src/streamUpdate.c
+++ b/source/libs/stream/src/streamUpdate.c
@@ -20,9 +20,9 @@
 #include "ttime.h"
 
 #define DEFAULT_FALSE_POSITIVE   0.01
-#define DEFAULT_BUCKET_SIZE      1310720
-#define DEFAULT_MAP_CAPACITY     1310720
-#define DEFAULT_MAP_SIZE         (DEFAULT_MAP_CAPACITY * 10)
+#define DEFAULT_BUCKET_SIZE      131072
+#define DEFAULT_MAP_CAPACITY     131072
+#define DEFAULT_MAP_SIZE         (DEFAULT_MAP_CAPACITY * 100)
 #define ROWS_PER_MILLISECOND     1
 #define MAX_NUM_SCALABLE_BF      100000
 #define MIN_NUM_SCALABLE_BF      10
@@ -44,8 +44,8 @@ static void windowSBfAdd(SUpdateInfo *pInfo, uint64_t count) {
   }
 }
 
-static void clearItemHelper(void* p) {
-  SScalableBf** pBf = p;
+static void clearItemHelper(void *p) {
+  SScalableBf **pBf = p;
   tScalableBfDestroy(*pBf);
 }
 
@@ -274,7 +274,7 @@ void updateInfoDestoryColseWinSBF(SUpdateInfo *pInfo) {
 }
 
 int32_t updateInfoSerialize(void *buf, int32_t bufLen, const SUpdateInfo *pInfo) {
-  if(!pInfo) {
+  if (!pInfo) {
     return 0;
   }
 

From e48e485b6c71165a1546cbdf4ecef37b9e599225 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 2 Jun 2023 11:56:24 +0000
Subject: [PATCH 14/19] update default buffer

---
 source/libs/stream/src/streamBackendRocksdb.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c
index e07198f6ab..5be6d1b003 100644
--- a/source/libs/stream/src/streamBackendRocksdb.c
+++ b/source/libs/stream/src/streamBackendRocksdb.c
@@ -572,11 +572,10 @@ int32_t decodeValueFunc(void* value, int32_t vlen, int64_t* ttl, char** dest) {
     *dest = NULL;
     return -1;
   }
-  int64_t now = taosGetTimestampMs();
   p = taosDecodeFixedI64(p, &key.unixTimestamp);
   p = taosDecodeFixedI32(p, &key.len);
   if (vlen != sizeof(int64_t) + sizeof(int32_t) + key.len) {
-    *dest = NULL;
+    if (dest != NULL) *dest = NULL;
     return -1;
   }
 
@@ -587,6 +586,7 @@ int32_t decodeValueFunc(void* value, int32_t vlen, int64_t* ttl, char** dest) {
   }
 
   if (ttl != NULL) {
+    int64_t now = taosGetTimestampMs();
     *ttl = key.unixTimestamp == 0 ? 0 : key.unixTimestamp - now;
   }
   if (dest != NULL) {

From 3ec700e0f7c20d036a87c50d140c08dd17b2edda Mon Sep 17 00:00:00 2001
From: yihaoDeng <cyang@taosdata.com>
Date: Fri, 2 Jun 2023 22:25:38 +0800
Subject: [PATCH 15/19] fix mem problem

---
 source/libs/stream/src/streamBackendRocksdb.c | 67 ++++++++++---------
 1 file changed, 34 insertions(+), 33 deletions(-)

diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c
index 5be6d1b003..6309cb9df2 100644
--- a/source/libs/stream/src/streamBackendRocksdb.c
+++ b/source/libs/stream/src/streamBackendRocksdb.c
@@ -576,6 +576,7 @@ int32_t decodeValueFunc(void* value, int32_t vlen, int64_t* ttl, char** dest) {
   p = taosDecodeFixedI32(p, &key.len);
   if (vlen != sizeof(int64_t) + sizeof(int32_t) + key.len) {
     if (dest != NULL) *dest = NULL;
+    qError("vlen: %d, read len: %d", vlen, key.len);
     return -1;
   }
 
@@ -1008,35 +1009,35 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
                                     ((rocksdb_column_family_handle_t**)pState->pTdbState->pHandle)[idx]);
 }
 
-#define STREAM_STATE_PUT_ROCKSDB(pState, funcname, key, value, vLen)                                     \
-  do {                                                                                                   \
-    code = 0;                                                                                            \
-    char  buf[128] = {0};                                                                                \
-    char* err = NULL;                                                                                    \
-    int   i = streamGetInit(pState, funcname);                                                           \
-    if (i < 0) {                                                                                         \
-      qWarn("streamState failed to get cf name: %s", funcname);                                          \
-      code = -1;                                                                                         \
-      break;                                                                                             \
-    }                                                                                                    \
-    char toString[128] = {0};                                                                            \
-    if (qDebugFlag & DEBUG_TRACE) ginitDict[i].toStrFunc((void*)key, toString);                          \
-    int32_t                         klen = ginitDict[i].enFunc((void*)key, buf);                         \
-    rocksdb_column_family_handle_t* pHandle =                                                            \
-        ((rocksdb_column_family_handle_t**)pState->pTdbState->pHandle)[ginitDict[i].idx];                \
-    rocksdb_t*              db = pState->pTdbState->rocksdb;                                             \
-    rocksdb_writeoptions_t* opts = pState->pTdbState->writeOpts;                                         \
-    char*                   ttlV = NULL;                                                                 \
-    int32_t                 ttlVLen = ginitDict[i].enValueFunc((char*)value, vLen, 0, &ttlV);            \
-    rocksdb_put_cf(db, opts, pHandle, (const char*)buf, klen, (const char*)ttlV, (size_t)ttlVLen, &err); \
-    if (err != NULL) {                                                                                   \
-      taosMemoryFree(err);                                                                               \
-      qError("streamState str: %s failed to write to %s, err: %s", toString, funcname, err);             \
-      code = -1;                                                                                         \
-    } else {                                                                                             \
-      qTrace("streamState str:%s succ to write to %s, valLen:%d", toString, funcname, vLen);             \
-    }                                                                                                    \
-    taosMemoryFree(ttlV);                                                                                \
+#define STREAM_STATE_PUT_ROCKSDB(pState, funcname, key, value, vLen)                                                   \
+  do {                                                                                                                 \
+    code = 0;                                                                                                          \
+    char  buf[128] = {0};                                                                                              \
+    char* err = NULL;                                                                                                  \
+    int   i = streamGetInit(pState, funcname);                                                                         \
+    if (i < 0) {                                                                                                       \
+      qWarn("streamState failed to get cf name: %s", funcname);                                                        \
+      code = -1;                                                                                                       \
+      break;                                                                                                           \
+    }                                                                                                                  \
+    char toString[128] = {0};                                                                                          \
+    if (qDebugFlag & DEBUG_TRACE) ginitDict[i].toStrFunc((void*)key, toString);                                        \
+    int32_t                         klen = ginitDict[i].enFunc((void*)key, buf);                                       \
+    rocksdb_column_family_handle_t* pHandle =                                                                          \
+        ((rocksdb_column_family_handle_t**)pState->pTdbState->pHandle)[ginitDict[i].idx];                              \
+    rocksdb_t*              db = pState->pTdbState->rocksdb;                                                           \
+    rocksdb_writeoptions_t* opts = pState->pTdbState->writeOpts;                                                       \
+    char*                   ttlV = NULL;                                                                               \
+    int32_t                 ttlVLen = ginitDict[i].enValueFunc((char*)value, vLen, 0, &ttlV);                          \
+    rocksdb_put_cf(db, opts, pHandle, (const char*)buf, klen, (const char*)ttlV, (size_t)ttlVLen, &err);               \
+    if (err != NULL) {                                                                                                 \
+      taosMemoryFree(err);                                                                                             \
+      qError("streamState str: %s failed to write to %s, err: %s", toString, funcname, err);                           \
+      code = -1;                                                                                                       \
+    } else {                                                                                                           \
+      qTrace("streamState str:%s succ to write to %s, rowValLen:%d, ttlValLen:%d", toString, funcname, vLen, ttlVLen); \
+    }                                                                                                                  \
+    taosMemoryFree(ttlV);                                                                                              \
   } while (0);
 
 #define STREAM_STATE_GET_ROCKSDB(pState, funcname, key, pVal, vLen)                                                    \
@@ -1071,17 +1072,17 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
       code = -1;                                                                                                       \
     } else {                                                                                                           \
       char*   p = NULL;                                                                                                \
-      int32_t len = ginitDict[i].deValueFunc(val, len, NULL, (char**)pVal);                                            \
-      if (len <= 0) {                                                                                                  \
+      int32_t tlen = ginitDict[i].deValueFunc(val, len, NULL, (char**)pVal);                                           \
+      if (tlen <= 0) {                                                                                                 \
         qError("streamState str: %s failed to read from %s_%s, err: already ttl ", toString, pState->pTdbState->idstr, \
                funcname);                                                                                              \
         code = -1;                                                                                                     \
       } else {                                                                                                         \
         qTrace("streamState str: %s succ to read from %s_%s, valLen:%d", toString, pState->pTdbState->idstr, funcname, \
-               len);                                                                                                   \
+               tlen);                                                                                                  \
       }                                                                                                                \
       taosMemoryFree(val);                                                                                             \
-      if (vLen != NULL) *vLen = len;                                                                                   \
+      if (vLen != NULL) *vLen = tlen;                                                                                  \
     }                                                                                                                  \
     if (code == 0)                                                                                                     \
       qDebug("streamState str: %s succ to read from %s_%s", toString, pState->pTdbState->idstr, funcname);             \

From 5063c175f4ff62ceb210d5424d1d7bda5681d1b9 Mon Sep 17 00:00:00 2001
From: yihaoDeng <cyang@taosdata.com>
Date: Fri, 2 Jun 2023 23:47:14 +0800
Subject: [PATCH 16/19] fix mem problem

---
 source/libs/stream/src/streamBackendRocksdb.c |  2 +-
 source/libs/stream/src/tstreamFileState.c     | 13 ++++++++-----
 2 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c
index 6309cb9df2..b3995f020b 100644
--- a/source/libs/stream/src/streamBackendRocksdb.c
+++ b/source/libs/stream/src/streamBackendRocksdb.c
@@ -574,7 +574,7 @@ int32_t decodeValueFunc(void* value, int32_t vlen, int64_t* ttl, char** dest) {
   }
   p = taosDecodeFixedI64(p, &key.unixTimestamp);
   p = taosDecodeFixedI32(p, &key.len);
-  if (vlen != sizeof(int64_t) + sizeof(int32_t) + key.len) {
+  if (vlen != (sizeof(int64_t) + sizeof(int32_t) + key.len)) {
     if (dest != NULL) *dest = NULL;
     qError("vlen: %d, read len: %d", vlen, key.len);
     return -1;
diff --git a/source/libs/stream/src/tstreamFileState.c b/source/libs/stream/src/tstreamFileState.c
index 7ea1a41925..757b1de6b6 100644
--- a/source/libs/stream/src/tstreamFileState.c
+++ b/source/libs/stream/src/tstreamFileState.c
@@ -419,7 +419,9 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
     if (code != 0 || len == 0 || val == NULL) {
       return TSDB_CODE_FAILED;
     }
-    sscanf(val, "%" PRId64 "", &maxCheckPointId);
+    memcpy(val, buf, len);
+    buf[len] = 0;
+    maxCheckPointId = atol((char*)buf);
     taosMemoryFree(val);
   }
   for (int64_t i = maxCheckPointId; i > 0; i--) {
@@ -431,16 +433,17 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
     if (code != 0) {
       return TSDB_CODE_FAILED;
     }
+    memcpy(val, buf, len);
+    buf[len] = 0;
+    taosMemoryFree(val);
+
     TSKEY ts;
-    sscanf(val, "%" PRId64 "", &ts);
+    ts = atol((char*)buf);
     if (ts < mark) {
       // statekey winkey.ts < mark
       forceRemoveCheckpoint(pFileState, i);
-      taosMemoryFreeClear(val);
       break;
-    } else {
     }
-    taosMemoryFree(val);
   }
   return code;
 }

From 81fe174aa26b0dd75ed9f805698edf919ed34ef3 Mon Sep 17 00:00:00 2001
From: yihaoDeng <cyang@taosdata.com>
Date: Sat, 3 Jun 2023 08:19:59 +0800
Subject: [PATCH 17/19] fix mem leak

---
 source/libs/stream/src/streamExec.c | 39 ++++++++++++++++-------------
 1 file changed, 22 insertions(+), 17 deletions(-)

diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index 716b939e5f..1a78607770 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -16,11 +16,11 @@
 #include "streamInc.h"
 
 // maximum allowed processed block batches. One block may include several submit blocks
-#define MAX_STREAM_EXEC_BATCH_NUM 32
-#define MIN_STREAM_EXEC_BATCH_NUM 8
-#define MAX_STREAM_RESULT_DUMP_THRESHOLD  100
+#define MAX_STREAM_EXEC_BATCH_NUM        32
+#define MIN_STREAM_EXEC_BATCH_NUM        8
+#define MAX_STREAM_RESULT_DUMP_THRESHOLD 100
 
-static int32_t updateCheckPointInfo (SStreamTask* pTask);
+static int32_t updateCheckPointInfo(SStreamTask* pTask);
 
 bool streamTaskShouldStop(const SStreamStatus* pStatus) {
   int32_t status = atomic_load_8((int8_t*)&pStatus->taskStatus);
@@ -48,10 +48,11 @@ static int32_t doDumpResult(SStreamTask* pTask, SStreamQueueItem* pItem, SArray*
       return -1;
     }
 
-    qDebug("s-task:%s dump stream result data blocks, num:%d, size:%.2fMiB", pTask->id.idStr, numOfBlocks, size/1048576.0);
+    qDebug("s-task:%s dump stream result data blocks, num:%d, size:%.2fMiB", pTask->id.idStr, numOfBlocks,
+           size / 1048576.0);
 
     code = streamTaskOutputResultBlock(pTask, pStreamBlocks);
-    if (code == TSDB_CODE_UTIL_QUEUE_OUT_OF_MEMORY) { // back pressure and record position
+    if (code == TSDB_CODE_UTIL_QUEUE_OUT_OF_MEMORY) {  // back pressure and record position
       destroyStreamDataBlock(pStreamBlocks);
       return -1;
     }
@@ -65,7 +66,8 @@ static int32_t doDumpResult(SStreamTask* pTask, SStreamQueueItem* pItem, SArray*
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, int64_t* totalSize, int32_t* totalBlocks) {
+static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, int64_t* totalSize,
+                                  int32_t* totalBlocks) {
   int32_t code = TSDB_CODE_SUCCESS;
   void*   pExecutor = pTask->exec.pExecutor;
 
@@ -82,7 +84,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, i
     }
 
     if (streamTaskShouldStop(&pTask->status)) {
-      taosArrayDestroy(pRes); // memory leak
+      taosArrayDestroy(pRes);  // memory leak
       return 0;
     }
 
@@ -101,7 +103,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, i
       if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
         SSDataBlock block = {0};
 
-        const SStreamDataBlock* pRetrieveBlock = (const SStreamDataBlock*) pItem;
+        const SStreamDataBlock* pRetrieveBlock = (const SStreamDataBlock*)pItem;
         ASSERT(taosArrayGetSize(pRetrieveBlock->blocks) == 1);
 
         assignOneDataBlock(&block, taosArrayGet(pRetrieveBlock->blocks, 0));
@@ -153,7 +155,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, i
     ASSERT(numOfBlocks == taosArrayGetSize(pRes));
     code = doDumpResult(pTask, pItem, pRes, size, totalSize, totalBlocks);
   } else {
-    taosArrayDestroy(pRes);
+    taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
   }
 
   return code;
@@ -286,7 +288,7 @@ int32_t streamBatchExec(SStreamTask* pTask, int32_t batchLimit) {
 }
 #endif
 
-int32_t updateCheckPointInfo (SStreamTask* pTask) {
+int32_t updateCheckPointInfo(SStreamTask* pTask) {
   int64_t ckId = 0;
   int64_t dataVer = 0;
   qGetCheckpointVersion(pTask->exec.pExecutor, &dataVer, &ckId);
@@ -294,7 +296,8 @@ int32_t updateCheckPointInfo (SStreamTask* pTask) {
   SCheckpointInfo* pCkInfo = &pTask->chkInfo;
   if (ckId > pCkInfo->id) {  // save it since the checkpoint is updated
     qDebug("s-task:%s exec end, start to update check point, ver from %" PRId64 " to %" PRId64
-           ", checkPoint id:%" PRId64 " -> %" PRId64, pTask->id.idStr, pCkInfo->version, dataVer, pCkInfo->id, ckId);
+           ", checkPoint id:%" PRId64 " -> %" PRId64,
+           pTask->id.idStr, pCkInfo->version, dataVer, pCkInfo->id, ckId);
 
     pTask->chkInfo = (SCheckpointInfo){.version = dataVer, .id = ckId, .currentVer = pCkInfo->currentVer};
 
@@ -417,14 +420,15 @@ int32_t streamExecForAll(SStreamTask* pTask) {
         ASSERT(pTask->taskLevel == TASK_LEVEL__SOURCE);
         const SStreamDataSubmit* pSubmit = (const SStreamDataSubmit*)pInput;
         qSetMultiStreamInput(pExecutor, &pSubmit->submit, 1, STREAM_INPUT__DATA_SUBMIT);
-        qDebug("s-task:%s set submit blocks as source block completed, %p %p len:%d ver:%" PRId64, pTask->id.idStr, pSubmit,
-               pSubmit->submit.msgStr, pSubmit->submit.msgLen, pSubmit->submit.ver);
+        qDebug("s-task:%s set submit blocks as source block completed, %p %p len:%d ver:%" PRId64, pTask->id.idStr,
+               pSubmit, pSubmit->submit.msgStr, pSubmit->submit.msgLen, pSubmit->submit.ver);
       } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
         const SStreamDataBlock* pBlock = (const SStreamDataBlock*)pInput;
 
         SArray* pBlockList = pBlock->blocks;
         int32_t numOfBlocks = taosArrayGetSize(pBlockList);
-        qDebug("s-task:%s set sdata blocks as input num:%d, ver:%" PRId64, pTask->id.idStr, numOfBlocks, pBlock->sourceVer);
+        qDebug("s-task:%s set sdata blocks as input num:%d, ver:%" PRId64, pTask->id.idStr, numOfBlocks,
+               pBlock->sourceVer);
         qSetMultiStreamInput(pExecutor, pBlockList->pData, numOfBlocks, STREAM_INPUT__DATA_BLOCK);
       } else if (pItem->type == STREAM_INPUT__MERGED_SUBMIT) {
         const SStreamMergedSubmit* pMerged = (const SStreamMergedSubmit*)pInput;
@@ -445,8 +449,9 @@ int32_t streamExecForAll(SStreamTask* pTask) {
     int32_t totalBlocks = 0;
     streamTaskExecImpl(pTask, pInput, &resSize, &totalBlocks);
 
-    double  el = (taosGetTimestampMs() - st) / 1000.0;
-    qDebug("s-task:%s exec end, elapsed time:%.2fs, result size:%.2fMiB, numOfBlocks:%d", pTask->id.idStr, el, resSize / 1048576.0, totalBlocks);
+    double el = (taosGetTimestampMs() - st) / 1000.0;
+    qDebug("s-task:%s exec end, elapsed time:%.2fs, result size:%.2fMiB, numOfBlocks:%d", pTask->id.idStr, el,
+           resSize / 1048576.0, totalBlocks);
     streamFreeQitem(pInput);
   }
 

From 7244ab1f6bfd28c1d5d2a9cbc4f4442a8bb50c94 Mon Sep 17 00:00:00 2001
From: yihaoDeng <cyang@taosdata.com>
Date: Sat, 3 Jun 2023 08:28:21 +0800
Subject: [PATCH 18/19] fix mem leak

---
 source/libs/stream/src/streamExec.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index 1a78607770..aa2c8050de 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -84,7 +84,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, i
     }
 
     if (streamTaskShouldStop(&pTask->status)) {
-      taosArrayDestroy(pRes);  // memory leak
+      taosArrayDestroyEx(pRes, (FDelete)blockDataFreeRes);
       return 0;
     }
 

From 9e0392ff59a1db9bcee81c01dc7b30cc2fdda8ff Mon Sep 17 00:00:00 2001
From: yihaoDeng <cyang@taosdata.com>
Date: Sat, 3 Jun 2023 09:17:05 +0800
Subject: [PATCH 19/19] fix mem leak

---
 source/libs/stream/src/streamExec.c | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index aa2c8050de..49b35ce8db 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -101,8 +101,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, i
 
     if (output == NULL) {
       if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
-        SSDataBlock block = {0};
-
+        SSDataBlock             block = {0};
         const SStreamDataBlock* pRetrieveBlock = (const SStreamDataBlock*)pItem;
         ASSERT(taosArrayGetSize(pRetrieveBlock->blocks) == 1);