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] 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;
   }
 
-- 
GitLab