From 8c6dc3753382a36d129052948212bb590f23ea3e Mon Sep 17 00:00:00 2001
From: 54liuyao <54liuyao@163.com>
Date: Thu, 26 Sep 2024 14:13:39 +0800
Subject: [PATCH] fix(stream):set buff size for selectivity function

---
 source/libs/stream/src/streamBackendRocksdb.c |  4 ++--
 source/libs/stream/src/streamState.c          | 10 ++++++----
 2 files changed, 8 insertions(+), 6 deletions(-)

diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c
index d55bf19c87..d358850e75 100644
--- a/source/libs/stream/src/streamBackendRocksdb.c
+++ b/source/libs/stream/src/streamBackendRocksdb.c
@@ -4166,7 +4166,7 @@ int32_t streamStateSessionAddIfNotExist_rocksdb(SStreamState* pState, SSessionKe
 
   if (code == 0) {
     if (sessionRangeKeyCmpr(&searchKey, key) == 0) {
-      memcpy(tmp, *pVal, valSize);
+      memcpy(tmp, *pVal, *pVLen);
       taosMemoryFreeClear(*pVal);
       goto _end;
     }
@@ -4182,7 +4182,7 @@ int32_t streamStateSessionAddIfNotExist_rocksdb(SStreamState* pState, SSessionKe
   code = streamStateSessionGetKVByCur_rocksdb(pCur, key, pVal, pVLen);
   if (code == 0) {
     if (sessionRangeKeyCmpr(&searchKey, key) == 0) {
-      memcpy(tmp, *pVal, valSize);
+      memcpy(tmp, *pVal, *pVLen);
       goto _end;
     }
   }
diff --git a/source/libs/stream/src/streamState.c b/source/libs/stream/src/streamState.c
index eb7254acba..41ff8f3c24 100644
--- a/source/libs/stream/src/streamState.c
+++ b/source/libs/stream/src/streamState.c
@@ -170,11 +170,12 @@ int32_t streamStateFuncPut(SStreamState* pState, const SWinKey* key, const void*
   int32_t lino = 0;
   void*   pVal = NULL;
   int32_t len = getRowStateRowSize(pState->pFileState);
-  code = getFunctionRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), &pVal, &len);
+  int32_t  tmpLen = len;
+  code = getFunctionRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), &pVal, &tmpLen);
   QUERY_CHECK_CODE(code, lino, _end);
 
   char*    buf = ((SRowBuffPos*)pVal)->pRowBuff;
-  uint32_t rowSize = streamFileStateGetSelectRowSize(pState->pFileState);
+  int32_t  rowSize = streamFileStateGetSelectRowSize(pState->pFileState);
   memcpy(buf + len - rowSize, value, vLen);
 
 _end:
@@ -188,11 +189,12 @@ int32_t streamStateFuncGet(SStreamState* pState, const SWinKey* key, void** ppVa
   int32_t lino = 0;
   void*   pVal = NULL;
   int32_t len = getRowStateRowSize(pState->pFileState);
-  code = getFunctionRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), (void**)(&pVal), &len);
+  int32_t  tmpLen = len;
+  code = getFunctionRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), (void**)(&pVal), &tmpLen);
   QUERY_CHECK_CODE(code, lino, _end);
 
   char*    buf = ((SRowBuffPos*)pVal)->pRowBuff;
-  uint32_t rowSize = streamFileStateGetSelectRowSize(pState->pFileState);
+  int32_t  rowSize = streamFileStateGetSelectRowSize(pState->pFileState);
   *ppVal = buf + len - rowSize;
   streamStateReleaseBuf(pState, pVal, false);