From c54e4aeaadc69e0104ce31d091826653a70dc081 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 23 Feb 2024 09:30:53 +0000
Subject: [PATCH 01/24] add test case

---
 source/libs/index/src/indexComm.c             | 11 ++-
 source/libs/stream/inc/streamBackendRocksdb.h |  7 +-
 source/libs/stream/src/streamMeta.c           | 28 +++---
 source/libs/stream/test/CMakeLists.txt        | 94 ++++++++++++++-----
 source/libs/stream/test/backendTest.cpp       | 38 ++++++++
 source/libs/stream/test/tstreamUpdateTest.cpp | 13 ++-
 6 files changed, 137 insertions(+), 54 deletions(-)
 create mode 100644 source/libs/stream/test/backendTest.cpp

diff --git a/source/libs/index/src/indexComm.c b/source/libs/index/src/indexComm.c
index 1313221952..b7b9f1cc9f 100644
--- a/source/libs/index/src/indexComm.c
+++ b/source/libs/index/src/indexComm.c
@@ -76,8 +76,8 @@ char* idxInt2str(int64_t val, char* dst, int radix) {
   return dst - 1;
 }
 __compar_fn_t idxGetCompar(int8_t type) {
-  if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_VARBINARY ||
-      type == TSDB_DATA_TYPE_NCHAR || type == TSDB_DATA_TYPE_GEOMETRY) {
+  if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_VARBINARY || type == TSDB_DATA_TYPE_NCHAR ||
+      type == TSDB_DATA_TYPE_GEOMETRY) {
     return (__compar_fn_t)strcmp;
   }
   return getComparFunc(type, 0);
@@ -108,8 +108,8 @@ static FORCE_INLINE TExeCond tCompareEqual(void* a, void* b, int8_t type) {
   return tCompare(func, QUERY_TERM, a, b, type);
 }
 TExeCond tCompare(__compar_fn_t func, int8_t cmptype, void* a, void* b, int8_t dtype) {
-  if (dtype == TSDB_DATA_TYPE_BINARY || dtype == TSDB_DATA_TYPE_NCHAR ||
-      dtype == TSDB_DATA_TYPE_VARBINARY || dtype == TSDB_DATA_TYPE_GEOMETRY) {
+  if (dtype == TSDB_DATA_TYPE_BINARY || dtype == TSDB_DATA_TYPE_NCHAR || dtype == TSDB_DATA_TYPE_VARBINARY ||
+      dtype == TSDB_DATA_TYPE_GEOMETRY) {
     return tDoCompare(func, cmptype, a, b);
   }
 #if 1
@@ -290,6 +290,7 @@ int idxUidCompare(const void* a, const void* b) {
   uint64_t r = *(uint64_t*)b;
   return l - r;
 }
+#ifdef BUILD_NO_CALL
 int32_t idxConvertData(void* src, int8_t type, void** dst) {
   int tlen = -1;
   switch (type) {
@@ -372,6 +373,8 @@ int32_t idxConvertData(void* src, int8_t type, void** dst) {
   // indexMayFillNumbericData(*dst, tlen);
   return tlen;
 }
+#endif
+
 int32_t idxConvertDataToStr(void* src, int8_t type, void** dst) {
   if (src == NULL) {
     *dst = strndup(INDEX_DATA_NULL_STR, (int)strlen(INDEX_DATA_NULL_STR));
diff --git a/source/libs/stream/inc/streamBackendRocksdb.h b/source/libs/stream/inc/streamBackendRocksdb.h
index 03f70604b7..1f8e99cd27 100644
--- a/source/libs/stream/inc/streamBackendRocksdb.h
+++ b/source/libs/stream/inc/streamBackendRocksdb.h
@@ -17,7 +17,7 @@
 #define _STREAM_BACKEDN_ROCKSDB_H_
 
 #include "rocksdb/c.h"
-//#include "streamInt.h"
+// #include "streamInt.h"
 #include "streamState.h"
 #include "tcommon.h"
 
@@ -244,11 +244,6 @@ int32_t streamBackendDelInUseChkp(void* arg, int64_t chkpId);
 
 int32_t taskDbBuildSnap(void* arg, SArray* pSnap);
 
-// int32_t streamDefaultIter_rocksdb(SStreamState* pState, const void* start, const void* end, SArray* result);
-
-// STaskDbWrapper* taskDbOpen(char* path, char* key, int64_t chkpId);
-// void            taskDbDestroy(void* pDb, bool flush);
-
 int32_t taskDbDoCheckpoint(void* arg, int64_t chkpId);
 
 SBkdMgt* bkdMgtCreate(char* path);
diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c
index b35f401cb9..a09b940a19 100644
--- a/source/libs/stream/src/streamMeta.c
+++ b/source/libs/stream/src/streamMeta.c
@@ -70,7 +70,7 @@ static void streamMetaEnvInit() {
   streamTimerInit();
 }
 
-void streamMetaInit() { taosThreadOnce(&streamMetaModuleInit, streamMetaEnvInit);}
+void streamMetaInit() { taosThreadOnce(&streamMetaModuleInit, streamMetaEnvInit); }
 
 void streamMetaCleanup() {
   taosCloseRef(streamBackendId);
@@ -1104,14 +1104,14 @@ static int32_t metaHeartbeatToMnodeImpl(SStreamMeta* pMeta) {
         .inputQUsed = SIZE_IN_MiB(streamQueueGetItemSize((*pTask)->inputq.queue)),
     };
 
-    entry.inputRate = entry.inputQUsed * 100.0 / (2*STREAM_TASK_QUEUE_CAPACITY_IN_SIZE);
+    entry.inputRate = entry.inputQUsed * 100.0 / (2 * STREAM_TASK_QUEUE_CAPACITY_IN_SIZE);
     if ((*pTask)->info.taskLevel == TASK_LEVEL__SINK) {
       entry.sinkQuota = (*pTask)->outputInfo.pTokenBucket->quotaRate;
       entry.sinkDataSize = SIZE_IN_MiB((*pTask)->execInfo.sink.dataSize);
     }
 
     if ((*pTask)->chkInfo.checkpointingId != 0) {
-      entry.checkpointFailed = ((*pTask)->chkInfo.failedId >= (*pTask)->chkInfo.checkpointingId)? 1:0;
+      entry.checkpointFailed = ((*pTask)->chkInfo.failedId >= (*pTask)->chkInfo.checkpointingId) ? 1 : 0;
       entry.checkpointId = (*pTask)->chkInfo.checkpointingId;
       entry.chkpointTransId = (*pTask)->chkInfo.transId;
 
@@ -1172,7 +1172,7 @@ static int32_t metaHeartbeatToMnodeImpl(SStreamMeta* pMeta) {
     stDebug("vgId:%d no tasks and no mnd epset, not send stream hb to mnode", pMeta->vgId);
   }
 
-  _end:
+_end:
   streamMetaClearHbMsg(&hbMsg);
   return TSDB_CODE_SUCCESS;
 }
@@ -1304,28 +1304,28 @@ void streamMetaResetStartInfo(STaskStartInfo* pStartInfo) {
 }
 
 void streamMetaRLock(SStreamMeta* pMeta) {
-//  stTrace("vgId:%d meta-rlock", pMeta->vgId);
+  //  stTrace("vgId:%d meta-rlock", pMeta->vgId);
   taosThreadRwlockRdlock(&pMeta->lock);
 }
 
 void streamMetaRUnLock(SStreamMeta* pMeta) {
-//  stTrace("vgId:%d meta-runlock", pMeta->vgId);
+  //  stTrace("vgId:%d meta-runlock", pMeta->vgId);
   int32_t code = taosThreadRwlockUnlock(&pMeta->lock);
   if (code != TSDB_CODE_SUCCESS) {
     stError("vgId:%d meta-runlock failed, code:%d", pMeta->vgId, code);
   } else {
-//    stTrace("vgId:%d meta-runlock completed", pMeta->vgId);
+    //    stTrace("vgId:%d meta-runlock completed", pMeta->vgId);
   }
 }
 
 void streamMetaWLock(SStreamMeta* pMeta) {
-//  stTrace("vgId:%d meta-wlock", pMeta->vgId);
+  //  stTrace("vgId:%d meta-wlock", pMeta->vgId);
   taosThreadRwlockWrlock(&pMeta->lock);
-//  stTrace("vgId:%d meta-wlock completed", pMeta->vgId);
+  //  stTrace("vgId:%d meta-wlock completed", pMeta->vgId);
 }
 
 void streamMetaWUnLock(SStreamMeta* pMeta) {
-//  stTrace("vgId:%d meta-wunlock", pMeta->vgId);
+  //  stTrace("vgId:%d meta-wunlock", pMeta->vgId);
   taosThreadRwlockUnlock(&pMeta->lock);
 }
 
@@ -1395,7 +1395,7 @@ void streamMetaUpdateStageRole(SStreamMeta* pMeta, int64_t stage, bool isLeader)
     pMeta->sendMsgBeforeClosing = true;
   }
 
-  pMeta->role = (isLeader)? NODE_ROLE_LEADER:NODE_ROLE_FOLLOWER;
+  pMeta->role = (isLeader) ? NODE_ROLE_LEADER : NODE_ROLE_FOLLOWER;
   streamMetaWUnLock(pMeta);
 
   if (isLeader) {
@@ -1531,8 +1531,8 @@ int32_t streamMetaStopAllTasks(SStreamMeta* pMeta) {
 
 bool streamMetaAllTasksReady(const SStreamMeta* pMeta) {
   int32_t num = taosArrayGetSize(pMeta->pTaskList);
-  for(int32_t i = 0; i < num; ++i) {
-    STaskId* pTaskId = taosArrayGet(pMeta->pTaskList, i);
+  for (int32_t i = 0; i < num; ++i) {
+    STaskId*      pTaskId = taosArrayGet(pMeta->pTaskList, i);
     SStreamTask** ppTask = taosHashGet(pMeta->pTasksMap, pTaskId, sizeof(*pTaskId));
     if (ppTask == NULL) {
       continue;
@@ -1633,7 +1633,7 @@ int32_t streamMetaAddTaskLaunchResult(SStreamMeta* pMeta, int64_t streamId, int3
     pStartInfo->elapsedTime = (pStartInfo->startTs != 0) ? pStartInfo->readyTs - pStartInfo->startTs : 0;
 
     stDebug("vgId:%d all %d task(s) check downstream completed, last completed task:0x%x (succ:%d) startTs:%" PRId64
-                ", readyTs:%" PRId64 " total elapsed time:%.2fs",
+            ", readyTs:%" PRId64 " total elapsed time:%.2fs",
             pMeta->vgId, numOfTotal, taskId, ready, pStartInfo->startTs, pStartInfo->readyTs,
             pStartInfo->elapsedTime / 1000.0);
 
diff --git a/source/libs/stream/test/CMakeLists.txt b/source/libs/stream/test/CMakeLists.txt
index c90e05bcf6..c18372a493 100644
--- a/source/libs/stream/test/CMakeLists.txt
+++ b/source/libs/stream/test/CMakeLists.txt
@@ -1,40 +1,82 @@
 
-MESSAGE(STATUS "build stream unit test")
-
-# GoogleTest requires at least C++11
-SET(CMAKE_CXX_STANDARD 11)
-AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
 
 # bloomFilterTest
-ADD_EXECUTABLE(streamUpdateTest "tstreamUpdateTest.cpp")
 
-TARGET_LINK_LIBRARIES(streamUpdateTest
-        PUBLIC os util common gtest gtest_main stream executor index
-        )
+#TARGET_LINK_LIBRARIES(streamUpdateTest
+        #PUBLIC os util common gtest gtest_main stream executor index
+        #)
 
-TARGET_INCLUDE_DIRECTORIES(
-  streamUpdateTest
+#TARGET_INCLUDE_DIRECTORIES(
+  #streamUpdateTest
+  #PUBLIC "${TD_SOURCE_DIR}/include/libs/stream/"
+  #PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
+#)
+
+#ADD_EXECUTABLE(checkpointTest checkpointTest.cpp)
+#TARGET_LINK_LIBRARIES(
+        #checkpointTest
+        #PUBLIC os common gtest stream executor qcom index transport util
+#)
+
+#TARGET_INCLUDE_DIRECTORIES(
+        #checkpointTest
+        #PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
+#)
+
+#add_executable(backendTest "")
+
+#target_sources(backendTest 
+        #PRIVATE
+        #"backendTest.cpp"
+#)
+
+#TARGET_LINK_LIBRARIES(
+        #backendTest
+        #PUBLIC rocksdb
+        #PUBLIC os common gtest stream executor qcom index transport util
+#)
+
+#TARGET_INCLUDE_DIRECTORIES(
+        #backendTest
+        #PUBLIC "${TD_SOURCE_DIR}/include/libs/stream/"
+        #PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
+#)
+
+#add_test(
+  #NAME streamUpdateTest
+  #COMMAND streamUpdateTest
+#)
+
+#add_test(
+  #NAME checkpointTest
+  #COMMAND checkpointTest
+#)
+#add_test(
+  #NAME backendTest
+  #COMMAND backendTest
+#)
+
+
+add_executable(backendTest "")
+
+target_sources(backendTest
+  PRIVATE
+  "backendTest.cpp"
+)
+
+target_include_directories(
+  backendTest
   PUBLIC "${TD_SOURCE_DIR}/include/libs/stream/"
   PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
 )
 
-ADD_EXECUTABLE(checkpointTest checkpointTest.cpp)
-TARGET_LINK_LIBRARIES(
-        checkpointTest
+target_link_libraries(
+        backendTest
+        PUBLIC rocksdb
         PUBLIC os common gtest stream executor qcom index transport util
 )
 
-TARGET_INCLUDE_DIRECTORIES(
-        checkpointTest
-        PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
-)
-
 add_test(
-  NAME streamUpdateTest
-  COMMAND streamUpdateTest
-)
-
-add_test(
-  NAME checkpointTest
-  COMMAND checkpointTest
+  NAME backendTest 
+  COMMAND backendTest 
 )
\ No newline at end of file
diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
new file mode 100644
index 0000000000..5051337aee
--- /dev/null
+++ b/source/libs/stream/test/backendTest.cpp
@@ -0,0 +1,38 @@
+#include <gtest/gtest.h>
+
+#include <taoserror.h>
+#include <tglobal.h>
+#include <iostream>
+#include "streamBackendRocksdb.h"
+
+class BackendEnv : public ::testing::Test {
+ protected:
+  virtual void SetUp() {}
+  virtual void TearDown() {}
+};
+
+void *backendCreate() {
+  const char *streamPath = "/tmp";
+
+  char *absPath = NULL;
+  void *p = NULL;
+  // SBackendWrapper *p = streamBackendInit(streamPath, -1, 2);
+  // p = taskDbOpen((char *)streamPath, (char *)"test", -1);
+  // p = bkdMgtCreate((char *)streamPath);
+
+  ASSERT(p != NULL);
+  return p;
+}
+void backendOpen() {
+  void *p = backendCreate();
+  ASSERT(p != NULL);
+}
+
+TEST_F(BackendEnv, checkOpen) { backendOpen(); }
+TEST_F(BackendEnv, backendOpt) {}
+TEST_F(BackendEnv, backendDestroy) {}
+
+int main(int argc, char **argv) {
+  testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
\ No newline at end of file
diff --git a/source/libs/stream/test/tstreamUpdateTest.cpp b/source/libs/stream/test/tstreamUpdateTest.cpp
index 1b999e5fb0..59171876ff 100644
--- a/source/libs/stream/test/tstreamUpdateTest.cpp
+++ b/source/libs/stream/test/tstreamUpdateTest.cpp
@@ -14,10 +14,7 @@ class StreamStateEnv : public ::testing::Test {
     streamMetaInit();
     backend = streamBackendInit(path, 0, 0);
   }
-  virtual void TearDown() {
-    streamMetaCleanup();
-    // indexClose(index);
-  }
+  virtual void TearDown() { streamMetaCleanup(); }
 
   const char *path = TD_TMP_DIR_PATH "stream";
   void       *backend;
@@ -50,6 +47,14 @@ bool equalSBF(SScalableBf *left, SScalableBf *right) {
 }
 
 TEST(TD_STREAM_UPDATE_TEST, update) {
+  const char *streamPath = "/tmp";
+
+  char *absPath = NULL;
+  void *p = NULL;
+  // SBackendWrapper *p = streamBackendInit(streamPath, -1, 2);
+  // p = taskDbOpen((char *)streamPath, (char *)"test", -1);
+  p = bkdMgtCreate((char *)streamPath);
+
   // const int64_t interval = 20 * 1000;
   // const int64_t watermark = 10 * 60 * 1000;
   // SUpdateInfo  *pSU = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);

From e41e10bf35534fb6b6804f8f79ca356f31fc6c3e Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Mon, 26 Feb 2024 11:51:46 +0000
Subject: [PATCH 02/24] add  stream backend  test

---
 source/libs/stream/inc/streamBackendRocksdb.h | 11 +++-
 source/libs/stream/test/CMakeLists.txt        | 60 +++++++++++++------
 source/libs/stream/test/backendTest.cpp       | 30 ++++++++--
 source/libs/stream/test/checkpointTest.cpp    | 28 ++++-----
 4 files changed, 89 insertions(+), 40 deletions(-)

diff --git a/source/libs/stream/inc/streamBackendRocksdb.h b/source/libs/stream/inc/streamBackendRocksdb.h
index 1f8e99cd27..6e6f85d8f9 100644
--- a/source/libs/stream/inc/streamBackendRocksdb.h
+++ b/source/libs/stream/inc/streamBackendRocksdb.h
@@ -21,6 +21,10 @@
 #include "streamState.h"
 #include "tcommon.h"
 
+#ifdef __cplusplus
+extern "C" {
+#endif
+
 typedef struct SCfComparator {
   rocksdb_comparator_t** comp;
   int32_t                numOfComp;
@@ -253,4 +257,9 @@ int32_t  bkdMgtDumpTo(SBkdMgt* bm, char* taskId, char* dname);
 void     bkdMgtDestroy(SBkdMgt* bm);
 
 int32_t taskDbGenChkpUploadData(void* arg, void* bkdMgt, int64_t chkpId, int8_t type, char** path, SArray* list);
-#endif
\ No newline at end of file
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
diff --git a/source/libs/stream/test/CMakeLists.txt b/source/libs/stream/test/CMakeLists.txt
index c18372a493..c472207b27 100644
--- a/source/libs/stream/test/CMakeLists.txt
+++ b/source/libs/stream/test/CMakeLists.txt
@@ -57,26 +57,48 @@
 #)
 
 
-add_executable(backendTest "")
+#add_executable(backendTest "")
 
-target_sources(backendTest
-  PRIVATE
-  "backendTest.cpp"
-)
+#target_sources(backendTest
+  #PUBLIC 
+  #"backendTest.cpp"
+#)
 
-target_include_directories(
-  backendTest
-  PUBLIC "${TD_SOURCE_DIR}/include/libs/stream/"
-  PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
-)
+#target_include_directories(
+  #backendTest
+  #PUBLIC "${TD_SOURCE_DIR}/include/libs/stream/"
+  #PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
+#)
 
-target_link_libraries(
-        backendTest
-        PUBLIC rocksdb
-        PUBLIC os common gtest stream executor qcom index transport util
-)
+#target_link_libraries(
+        #backendTest
+        #PUBLIC rocksdb
+        #PUBLIC os common gtest stream executor qcom index transport util
+#)
 
-add_test(
-  NAME backendTest 
-  COMMAND backendTest 
-)
\ No newline at end of file
+
+MESSAGE(STATUS "build parser unit test")
+
+IF(NOT TD_DARWIN)
+        # GoogleTest requires at least C++11
+        SET(CMAKE_CXX_STANDARD 11)
+        AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
+
+        ADD_EXECUTABLE(backendTest ${SOURCE_LIST})
+        TARGET_LINK_LIBRARIES(
+          backendTest
+          PUBLIC rocksdb
+          PUBLIC os common gtest stream executor qcom index transport util vnode
+        )
+
+        TARGET_INCLUDE_DIRECTORIES(
+                backendTest 
+                PUBLIC "${TD_SOURCE_DIR}/include/libs/stream/"
+                PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
+        )
+
+        ADD_TEST(
+          NAME backendTest 
+          COMMAND backendTest 
+        )
+ENDIF ()
\ No newline at end of file
diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
index 5051337aee..d70f16b461 100644
--- a/source/libs/stream/test/backendTest.cpp
+++ b/source/libs/stream/test/backendTest.cpp
@@ -4,6 +4,11 @@
 #include <tglobal.h>
 #include <iostream>
 #include "streamBackendRocksdb.h"
+#include "streamSnapshot.h"
+#include "streamState.h"
+#include "tstream.h"
+#include "tstreamFileState.h"
+#include "tstreamUpdate.h"
 
 class BackendEnv : public ::testing::Test {
  protected:
@@ -13,22 +18,35 @@ class BackendEnv : public ::testing::Test {
 
 void *backendCreate() {
   const char *streamPath = "/tmp";
+  void       *p = NULL;
 
-  char *absPath = NULL;
-  void *p = NULL;
-  // SBackendWrapper *p = streamBackendInit(streamPath, -1, 2);
-  // p = taskDbOpen((char *)streamPath, (char *)"test", -1);
-  // p = bkdMgtCreate((char *)streamPath);
+  // char *absPath = NULL;
+  // // SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(streamPath, -1, 2);
+  // STaskDbWrapper *p = taskDbOpen((char *)streamPath, (char *)"stream-backend", -1);
+  // ASSERT(p != NULL);
+  return p;
+}
 
+SStreamState *stateCreate(void *pBackend, char *keyidr) {
+  const char  *streamPath = "/tmp";
+  SStreamTask *pTask = (SStreamTask *)taosMemoryCalloc(1, sizeof(SStreamTask));
+  pTask->ver = 1024;
+  pTask->id.streamId = 1023;
+  pTask->id.taskId = 1111111;
+
+  SStreamState *p = streamStateOpen((char *)streamPath, pTask, true, 32, 32 * 1024);
   ASSERT(p != NULL);
   return p;
 }
 void backendOpen() {
   void *p = backendCreate();
   ASSERT(p != NULL);
+  taskDbDestroy(p, true);
 }
 
-TEST_F(BackendEnv, checkOpen) { backendOpen(); }
+TEST_F(BackendEnv, checkOpen) {
+  backendOpen();
+}
 TEST_F(BackendEnv, backendOpt) {}
 TEST_F(BackendEnv, backendDestroy) {}
 
diff --git a/source/libs/stream/test/checkpointTest.cpp b/source/libs/stream/test/checkpointTest.cpp
index 0dc2cc13f5..eb7f38744f 100644
--- a/source/libs/stream/test/checkpointTest.cpp
+++ b/source/libs/stream/test/checkpointTest.cpp
@@ -25,24 +25,24 @@
 #pragma GCC diagnostic ignored "-Wunused-variable"
 #pragma GCC diagnostic ignored "-Wsign-compare"
 
+#include "cos.h"
 #include "rsync.h"
 #include "streamInt.h"
-#include "cos.h"
 
-int main(int argc, char **argv) {
-  testing::InitGoogleTest(&argc, argv);
+// int main(int argc, char **argv) {
+//   testing::InitGoogleTest(&argc, argv);
 
-  if (taosInitCfg("/etc/taos/", NULL, NULL, NULL, NULL, 0) != 0) {
-    printf("error");
-  }
-  if (s3Init() < 0) {
-    return -1;
-  }
-  strcpy(tsSnodeAddress, "127.0.0.1");
-  int ret = RUN_ALL_TESTS();
-  s3CleanUp();
-  return ret;
-}
+//   if (taosInitCfg("/etc/taos/", NULL, NULL, NULL, NULL, 0) != 0) {
+//     printf("error");
+//   }
+//   if (s3Init() < 0) {
+//     return -1;
+//   }
+//   strcpy(tsSnodeAddress, "127.0.0.1");
+//   int ret = RUN_ALL_TESTS();
+//   s3CleanUp();
+//   return ret;
+// }
 
 TEST(testCase, checkpointUpload_Test) {
   stopRsync();

From 8d36445bb58b89e8c219fe8a844be7c9997d7c96 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Tue, 27 Feb 2024 05:29:58 +0000
Subject: [PATCH 03/24] add  stream backend  test

---
 source/libs/stream/test/backendTest.cpp | 53 +++++++++++++++++++++----
 1 file changed, 45 insertions(+), 8 deletions(-)

diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
index d70f16b461..df0ab9fe7e 100644
--- a/source/libs/stream/test/backendTest.cpp
+++ b/source/libs/stream/test/backendTest.cpp
@@ -3,6 +3,7 @@
 #include <taoserror.h>
 #include <tglobal.h>
 #include <iostream>
+#include <vector>
 #include "streamBackendRocksdb.h"
 #include "streamSnapshot.h"
 #include "streamState.h"
@@ -27,26 +28,62 @@ void *backendCreate() {
   return p;
 }
 
-SStreamState *stateCreate(void *pBackend, char *keyidr) {
-  const char  *streamPath = "/tmp";
+SStreamState *stateCreate(const char *path) {
   SStreamTask *pTask = (SStreamTask *)taosMemoryCalloc(1, sizeof(SStreamTask));
   pTask->ver = 1024;
   pTask->id.streamId = 1023;
   pTask->id.taskId = 1111111;
+  SStreamMeta *pMeta = streamMetaOpen((path), NULL, NULL, 0, 0, NULL);
+  pTask->pMeta = pMeta;
 
-  SStreamState *p = streamStateOpen((char *)streamPath, pTask, true, 32, 32 * 1024);
+  SStreamState *p = streamStateOpen((char *)path, pTask, true, 32, 32 * 1024);
   ASSERT(p != NULL);
   return p;
 }
 void backendOpen() {
-  void *p = backendCreate();
+  const char   *path = "/tmp/backend";
+  SStreamState *p = stateCreate(path);
   ASSERT(p != NULL);
-  taskDbDestroy(p, true);
+
+  // write bacth
+  // default/state/fill/sess/func/parname/partag
+  int32_t              size = 100;
+  std::vector<int64_t> tsArray;
+  for (int i = 0; i < size; i++) {
+    int64_t     ts = taosGetTimestampMs();
+    SWinKey     key = {.groupId = (uint64_t)(i), .ts = ts};
+    const char *val = "value data";
+    int32_t     vlen = strlen(val);
+    streamStatePut_rocksdb(p, &key, (char *)val, vlen);
+
+    tsArray.push_back(ts);
+  }
+  for (int i = 0; i < size; i++) {
+    int64_t ts = tsArray[i];
+    SWinKey key = {.groupId = (uint64_t)(i), .ts = ts};
+
+    const char *val = "value data";
+    int32_t     len = 0;
+    char       *newVal = NULL;
+    streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
+    ASSERT(len == strlen(val));
+  }
+  int64_t ts = tsArray[0];
+  SWinKey key = {.groupId = (uint64_t)(0), .ts = ts};
+  streamStateDel_rocksdb(p, &key);
+
+  
+
+  // read
+  // iterator
+  // rebuild chkp, reload from chkp
+  // sync
+  //
+  streamStateClose((SStreamState *)p, true);
+  // taskDbDestroy(p, true);
 }
 
-TEST_F(BackendEnv, checkOpen) {
-  backendOpen();
-}
+TEST_F(BackendEnv, checkOpen) { backendOpen(); }
 TEST_F(BackendEnv, backendOpt) {}
 TEST_F(BackendEnv, backendDestroy) {}
 

From 6b6f5aa5d1d8d132ff7635b091926ff395b0f0d2 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Tue, 27 Feb 2024 05:51:32 +0000
Subject: [PATCH 04/24] add test case

---
 source/libs/stream/test/backendTest.cpp | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
index df0ab9fe7e..1ccc0b4285 100644
--- a/source/libs/stream/test/backendTest.cpp
+++ b/source/libs/stream/test/backendTest.cpp
@@ -72,7 +72,18 @@ void backendOpen() {
   SWinKey key = {.groupId = (uint64_t)(0), .ts = ts};
   streamStateDel_rocksdb(p, &key);
 
-  
+  streamStateClear_rocksdb(p);
+
+  for (int i = 0; i < size; i++) {
+    int64_t     ts = tsArray[i];
+    SWinKey     key = {.groupId = (uint64_t)(i), .ts = ts};
+    const char *val = "value data";
+    int32_t     len = 0;
+    char       *newVal = NULL;
+    int32_t     code = streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
+    ASSERT(code != 0);
+  }
+  tsArray.clear();
 
   // read
   // iterator

From d62b82c295b2db35017cd8383baef70ea8cce680 Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Tue, 27 Feb 2024 16:02:15 +0800
Subject: [PATCH 05/24] opti:deal with escape in the end of field in schemaless

---
 source/client/inc/clientSml.h     |  2 +-
 source/client/src/clientSmlLine.c | 59 ++++++++++---------
 utils/test/c/sml_test.c           | 94 +++++++++++++++++++++++++++++--
 3 files changed, 122 insertions(+), 33 deletions(-)

diff --git a/source/client/inc/clientSml.h b/source/client/inc/clientSml.h
index b732abffb1..122914fd34 100644
--- a/source/client/inc/clientSml.h
+++ b/source/client/inc/clientSml.h
@@ -80,7 +80,7 @@ extern "C" {
 #define IS_SAME_KEY (maxKV->type == kv->type && maxKV->keyLen == kv->keyLen && memcmp(maxKV->key, kv->key, kv->keyLen) == 0)
 
 #define IS_SLASH_LETTER_IN_MEASUREMENT(sql)                                                           \
-  (*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE))
+  (*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE || *(sql) == SLASH))
 
 #define MOVE_FORWARD_ONE(sql, len) (memmove((void *)((sql)-1), (sql), len))
 
diff --git a/source/client/src/clientSmlLine.c b/source/client/src/clientSmlLine.c
index 0c610a4611..8bdbec0f2b 100644
--- a/source/client/src/clientSmlLine.c
+++ b/source/client/src/clientSmlLine.c
@@ -20,14 +20,14 @@
 
 #include "clientSml.h"
 
-#define IS_COMMA(sql) (*(sql) == COMMA && *((sql)-1) != SLASH)
-#define IS_SPACE(sql) (*(sql) == SPACE && *((sql)-1) != SLASH)
-#define IS_EQUAL(sql) (*(sql) == EQUAL && *((sql)-1) != SLASH)
+#define IS_COMMA(sql,escapeChar) (*(sql) == COMMA && (*((sql)-1) != SLASH || ((sql)-1 == escapeChar)))
+#define IS_SPACE(sql,escapeChar) (*(sql) == SPACE && (*((sql)-1) != SLASH || ((sql)-1 == escapeChar)))
+#define IS_EQUAL(sql,escapeChar) (*(sql) == EQUAL && (*((sql)-1) != SLASH || ((sql)-1 == escapeChar)))
 
 #define IS_SLASH_LETTER_IN_FIELD_VALUE(sql) (*((sql)-1) == SLASH && (*(sql) == QUOTE || *(sql) == SLASH))
 
 #define IS_SLASH_LETTER_IN_TAG_FIELD_KEY(sql) \
-  (*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE || *(sql) == EQUAL))
+  (*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE || *(sql) == EQUAL || *(sql) == SLASH))
 
 #define PROCESS_SLASH_IN_FIELD_VALUE(key, keyLen)  \
   for (int i = 1; i < keyLen; ++i) {               \
@@ -198,7 +198,7 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
   int     cnt = 0;
 
   while (*sql < sqlEnd) {
-    if (unlikely(IS_SPACE(*sql))) {
+    if (unlikely(IS_SPACE(*sql,NULL))) {
       break;
     }
 
@@ -207,18 +207,21 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
     size_t      keyLen = 0;
     bool        keyEscaped = false;
     size_t      keyLenEscaped = 0;
+    const char *escapeChar = NULL;
+
     while (*sql < sqlEnd) {
-      if (unlikely(IS_SPACE(*sql) || IS_COMMA(*sql))) {
+      if (unlikely(IS_SPACE(*sql,escapeChar) || IS_COMMA(*sql,escapeChar))) {
         smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
         terrno = TSDB_CODE_SML_INVALID_DATA;
         return -1;
       }
-      if (unlikely(IS_EQUAL(*sql))) {
+      if (unlikely(IS_EQUAL(*sql,escapeChar))) {
         keyLen = *sql - key;
         (*sql)++;
         break;
       }
       if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(*sql)) {
+        escapeChar = *sql;
         keyLenEscaped++;
         keyEscaped = true;
       }
@@ -238,15 +241,16 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
     size_t      valueLenEscaped = 0;
     while (*sql < sqlEnd) {
       // parse value
-      if (unlikely(IS_SPACE(*sql) || IS_COMMA(*sql))) {
+      if (unlikely(IS_SPACE(*sql,escapeChar) || IS_COMMA(*sql,escapeChar))) {
         break;
-      } else if (unlikely(IS_EQUAL(*sql))) {
+      } else if (unlikely(IS_EQUAL(*sql,escapeChar))) {
         smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
         terrno = TSDB_CODE_SML_INVALID_DATA;
         return -1;
       }
 
       if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(*sql)) {
+        escapeChar = *sql;
         valueLenEscaped++;
         valueEscaped = true;
       }
@@ -293,7 +297,7 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
     }
 
     cnt++;
-    if (IS_SPACE(*sql)) {
+    if (IS_SPACE(*sql,escapeChar)) {
       break;
     }
     (*sql)++;
@@ -326,7 +330,7 @@ static int32_t smlParseTagLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlL
 static int32_t smlParseColLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLineInfo *currElement) {
   int cnt = 0;
   while (*sql < sqlEnd) {
-    if (unlikely(IS_SPACE(*sql))) {
+    if (unlikely(IS_SPACE(*sql,NULL))) {
       break;
     }
 
@@ -335,17 +339,19 @@ static int32_t smlParseColLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlL
     size_t      keyLen = 0;
     bool        keyEscaped = false;
     size_t      keyLenEscaped = 0;
+    const char *escapeChar = NULL;
     while (*sql < sqlEnd) {
-      if (unlikely(IS_SPACE(*sql) || IS_COMMA(*sql))) {
+      if (unlikely(IS_SPACE(*sql,escapeChar) || IS_COMMA(*sql,escapeChar))) {
         smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
         return TSDB_CODE_SML_INVALID_DATA;
       }
-      if (unlikely(IS_EQUAL(*sql))) {
+      if (unlikely(IS_EQUAL(*sql,escapeChar))) {
         keyLen = *sql - key;
         (*sql)++;
         break;
       }
       if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(*sql)) {
+        escapeChar = *sql;
         keyLenEscaped++;
         keyEscaped = true;
       }
@@ -363,7 +369,6 @@ static int32_t smlParseColLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlL
     bool        valueEscaped = false;
     size_t      valueLenEscaped = 0;
     int         quoteNum = 0;
-    const char *escapeChar = NULL;
     while (*sql < sqlEnd) {
       // parse value
       if (unlikely(*(*sql) == QUOTE && (*(*sql - 1) != SLASH || (*sql - 1) == escapeChar))) {
@@ -374,7 +379,7 @@ static int32_t smlParseColLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlL
         }
         continue;
       }
-      if (quoteNum % 2 == 0 && (unlikely(IS_SPACE(*sql) || IS_COMMA(*sql)))) {
+      if (quoteNum % 2 == 0 && (unlikely(IS_SPACE(*sql,escapeChar) || IS_COMMA(*sql,escapeChar)))) {
         break;
       }
       if (IS_SLASH_LETTER_IN_FIELD_VALUE(*sql) && (*sql - 1) != escapeChar) {
@@ -437,7 +442,7 @@ static int32_t smlParseColLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlL
     }
 
     cnt++;
-    if (IS_SPACE(*sql)) {
+    if (IS_SPACE(*sql,escapeChar)) {
       break;
     }
     (*sql)++;
@@ -453,19 +458,18 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
   elements->measure = sql;
   // parse measure
   size_t measureLenEscaped = 0;
+  const char *escapeChar = NULL;
   while (sql < sqlEnd) {
-    if (unlikely((sql != elements->measure) && IS_SLASH_LETTER_IN_MEASUREMENT(sql))) {
-      elements->measureEscaped = true;
-      measureLenEscaped++;
-      sql++;
-      continue;
-    }
-    if (unlikely(IS_COMMA(sql))) {
+    if (unlikely(IS_COMMA(sql,escapeChar) || IS_SPACE(sql,escapeChar))) {
       break;
     }
 
-    if (unlikely(IS_SPACE(sql))) {
-      break;
+    if (unlikely((sql != elements->measure) && IS_SLASH_LETTER_IN_MEASUREMENT(sql))) {
+      elements->measureEscaped = true;
+      escapeChar = sql;
+      measureLenEscaped++;
+      sql++;
+      continue;
     }
     sql++;
   }
@@ -478,9 +482,12 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
   // to get measureTagsLen before
   const char *tmp = sql;
   while (tmp < sqlEnd) {
-    if (unlikely(IS_SPACE(tmp))) {
+    if (unlikely(IS_SPACE(tmp,escapeChar))) {
       break;
     }
+    if(unlikely(IS_SLASH_LETTER_IN_TAG_FIELD_KEY(sql))){
+      escapeChar = sql;
+    }
     tmp++;
   }
   elements->measureTagsLen = tmp - elements->measure;
diff --git a/utils/test/c/sml_test.c b/utils/test/c/sml_test.c
index 2c334eb67b..f81f13fc68 100644
--- a/utils/test/c/sml_test.c
+++ b/utils/test/c/sml_test.c
@@ -1018,7 +1018,7 @@ int sml_escape_Test() {
   ASSERT(numFields == 5);
   ASSERT(strncmp(fields[1].name, "inode\"i,= s_used", sizeof("inode\"i,= s_used") - 1) == 0);
   ASSERT(strncmp(fields[2].name, "total", sizeof("total") - 1) == 0);
-  ASSERT(strncmp(fields[3].name, "inode\"i,= s_f\\\\ree", sizeof("inode\"i,= s_f\\\\ree") - 1) == 0);
+  ASSERT(strncmp(fields[3].name, "inode\"i,= s_f\\ree", sizeof("inode\"i,= s_f\\ree") - 1) == 0);
   ASSERT(strncmp(fields[4].name, "dev\"i,= ce", sizeof("dev\"i,= ce") - 1) == 0);
 
   TAOS_ROW row = NULL;
@@ -1044,6 +1044,91 @@ int sml_escape_Test() {
   return code;
 }
 
+// test field with end of escape
+int sml_escape1_Test() {
+  TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
+
+  TAOS_RES *pRes = taos_query(taos, "create database if not exists db_escape");
+  taos_free_result(pRes);
+
+  pRes = taos_query(taos, "use db_escape");
+  taos_free_result(pRes);
+
+  const char *sql[] = {
+      "stab,t1\\=1 c1=3,c2=\"32fw\" 1661943970000000000",
+      "stab,t1=1\\ c1=3,c2=\"32fw\" 1661943980000000000",
+      "stab,t1=1 c1\\=3,c2=\"32fw\" 1661943990000000000",
+  };
+  for(int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++){
+    pRes = taos_schemaless_insert(taos, (char**)&sql[i], 1, TSDB_SML_LINE_PROTOCOL, 0);
+    int code = taos_errno(pRes);
+    ASSERT(code);
+  }
+
+  const char *sql1[] = {
+      "stab\\,t1=1 c1=3,c2=\"32fw\" 1661943960000000000",
+      "stab\\\\,t1=1 c1=3,c2=\"32fw\" 1661943960000000000",
+      "stab,t1\\\\=1 c1=3,c2=\"32fw\" 1661943970000000000",
+      "stab,t1=1\\\\ c1=3,c2=\"32fw\" 1661943980000000000",
+      "stab,t1=1 c1\\\\=3,c2=\"32fw\" 1661943990000000000",
+  };
+  pRes = taos_schemaless_insert(taos, (char **)sql1, sizeof(sql1) / sizeof(sql1[0]), TSDB_SML_LINE_PROTOCOL, 0);
+  printf("%s result:%s, rows:%d\n", __FUNCTION__, taos_errstr(pRes), taos_affected_rows(pRes));
+  int code = taos_errno(pRes);
+  ASSERT(!code);
+  ASSERT(taos_affected_rows(pRes) == 5);
+  taos_free_result(pRes);
+
+  pRes = taos_query(taos, "select * from stab order by _ts");     //check stable name
+  ASSERT(pRes);
+  int fieldNum = taos_field_count(pRes);
+  ASSERT(fieldNum == 6);
+  printf("fieldNum:%d\n", fieldNum);
+
+  int         numFields = taos_num_fields(pRes);
+  TAOS_FIELD *fields = taos_fetch_fields(pRes);
+  ASSERT(numFields == 6);
+  ASSERT(strncmp(fields[1].name, "c1", sizeof("c1") - 1) == 0);
+  ASSERT(strncmp(fields[2].name, "c2", sizeof("c2") - 1) == 0);
+  ASSERT(strncmp(fields[3].name, "c1\\", sizeof("c1\\") - 1) == 0);
+  ASSERT(strncmp(fields[4].name, "t1\\", sizeof("t1\\") - 1) == 0);
+  ASSERT(strncmp(fields[5].name, "t1", sizeof("t1") - 1) == 0);
+
+  TAOS_ROW row = NULL;
+  int32_t  rowIndex = 0;
+  while ((row = taos_fetch_row(pRes)) != NULL) {
+    int64_t ts = *(int64_t *)row[0];
+
+    if (rowIndex == 0) {
+      ASSERT(ts == 1661943970000);
+      ASSERT(*(double *)row[1] == 3);
+      ASSERT(strncmp(row[2], "32fw", sizeof("32fw") - 1) == 0);
+      ASSERT(row[3] == NULL);
+      ASSERT(strncmp(row[4], "1", sizeof("1") - 1) == 0);
+      ASSERT(row[5] == NULL);
+    }else if (rowIndex == 1) {
+      ASSERT(ts == 1661943980000);
+      ASSERT(*(double *)row[1] == 3);
+      ASSERT(strncmp(row[2], "32fw", sizeof("32fw") - 1) == 0);
+      ASSERT(row[3] == NULL);
+      ASSERT(row[4] == NULL);
+      ASSERT(strncmp(row[5], "1\\", sizeof("1\\") - 1) == 0);
+    }else if (rowIndex == 2) {
+      ASSERT(ts == 1661943990000);
+      ASSERT(row[1] == NULL);
+      ASSERT(strncmp(row[2], "32fw", sizeof("32fw") - 1) == 0);
+      ASSERT(*(double *)row[3] == 3);
+      ASSERT(row[4] == NULL);
+      ASSERT(strncmp(row[5], "1", sizeof("1") - 1) == 0);
+    }
+    rowIndex++;
+  }
+  taos_free_result(pRes);
+  taos_close(taos);
+
+  return code;
+}
+
 int sml_19221_Test() {
   TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
 
@@ -1775,17 +1860,14 @@ int main(int argc, char *argv[]) {
   ASSERT(ret);
   ret = sml_escape_Test();
   ASSERT(!ret);
+  ret = sml_escape1_Test();
+  ASSERT(!ret);
   ret = sml_ts3116_Test();
   ASSERT(!ret);
   ret = sml_ts2385_Test();    // this test case need config sml table name using ./sml_test config_file
   ASSERT(!ret);
   ret = sml_ts3303_Test();
   ASSERT(!ret);
-
-  //  for(int i = 0; i < sizeof(str)/sizeof(str[0]); i++){
-  //    printf("str:%s \t %d\n", str[i], smlCalTypeSum(str[i], strlen(str[i])));
-  //  }
-  //  int ret = 0;
   ret = sml_ttl_Test();
   ASSERT(!ret);
   ret = sml_ts2164_Test();

From 6bc12c872818e8b1bdf137c19a1c625fdc8c82e6 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Tue, 27 Feb 2024 16:27:59 +0800
Subject: [PATCH 06/24] refactor: do some internal refactor.

---
 include/util/tqueue.h                       | 36 ++-----------------
 source/client/src/clientTmq.c               |  7 ++--
 source/dnode/mgmt/mgmt_vnode/src/vmInt.c    | 10 +++---
 source/dnode/mgmt/mgmt_vnode/src/vmWorker.c | 10 +++---
 source/util/src/tqueue.c                    | 40 +++++++++++++++++++++
 source/util/src/tworker.c                   |  6 ++--
 6 files changed, 59 insertions(+), 50 deletions(-)

diff --git a/include/util/tqueue.h b/include/util/tqueue.h
index 9f09bd2930..eb887596d0 100644
--- a/include/util/tqueue.h
+++ b/include/util/tqueue.h
@@ -72,40 +72,6 @@ struct STaosQnode {
   char        item[];
 };
 
-struct STaosQueue {
-  STaosQnode   *head;
-  STaosQnode   *tail;
-  STaosQueue   *next;     // for queue set
-  STaosQset    *qset;     // for queue set
-  void         *ahandle;  // for queue set
-  FItem         itemFp;
-  FItems        itemsFp;
-  TdThreadMutex mutex;
-  int64_t       memOfItems;
-  int32_t       numOfItems;
-  int64_t       threadId;
-  int64_t       memLimit;
-  int64_t       itemLimit;
-};
-
-struct STaosQset {
-  STaosQueue   *head;
-  STaosQueue   *current;
-  TdThreadMutex mutex;
-  tsem_t        sem;
-  int32_t       numOfQueues;
-  int32_t       numOfItems;
-};
-
-struct STaosQall {
-  STaosQnode *current;
-  STaosQnode *start;
-  int32_t     numOfItems;
-  int64_t     memOfItems;
-  int32_t     unAccessedNumOfItems;
-  int64_t     unAccessMemOfItems;
-};
-
 STaosQueue *taosOpenQueue();
 void        taosCloseQueue(STaosQueue *queue);
 void        taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp);
@@ -140,6 +106,8 @@ int32_t    taosGetQueueNumber(STaosQset *qset);
 int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, SQueueInfo *qinfo);
 int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, SQueueInfo *qinfo);
 void    taosResetQsetThread(STaosQset *qset, void *pItem);
+void    taosQueueSetThreadId(STaosQueue *pQueue, int64_t threadId);
+int64_t taosQueueGetThreadId(STaosQueue *pQueue);
 
 #ifdef __cplusplus
 }
diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c
index 9b74456da2..a893b27896 100644
--- a/source/client/src/clientTmq.c
+++ b/source/client/src/clientTmq.c
@@ -876,12 +876,13 @@ int32_t tmqHandleAllDelayedTask(tmq_t* pTmq) {
   STaosQall* qall = taosAllocateQall();
   taosReadAllQitems(pTmq->delayedTask, qall);
 
-  if (qall->numOfItems == 0) {
+  int32_t numOfItems = taosQallItemSize(qall);
+  if (numOfItems == 0) {
     taosFreeQall(qall);
     return TSDB_CODE_SUCCESS;
   }
 
-  tscDebug("consumer:0x%" PRIx64 " handle delayed %d tasks before poll data", pTmq->consumerId, qall->numOfItems);
+  tscDebug("consumer:0x%" PRIx64 " handle delayed %d tasks before poll data", pTmq->consumerId, numOfItems);
   int8_t* pTaskType = NULL;
   taosGetQitem(qall, (void**)&pTaskType);
 
@@ -1839,7 +1840,7 @@ static void updateVgInfo(SMqClientVg* pVg, STqOffsetVal* reqOffset, STqOffsetVal
 }
 
 static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout) {
-  tscDebug("consumer:0x%" PRIx64 " start to handle the rsp, total:%d", tmq->consumerId, tmq->qall->numOfItems);
+  tscDebug("consumer:0x%" PRIx64 " start to handle the rsp, total:%d", tmq->consumerId, taosQallItemSize(tmq->qall));
 
   while (1) {
     SMqRspWrapper* pRspWrapper = NULL;
diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
index be88e8b3fd..3dfc2bd96f 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
@@ -194,26 +194,26 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal)
   while (pVnode->refCount > 0) taosMsleep(10);
 
   dInfo("vgId:%d, wait for vnode write queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pWriteW.queue,
-        pVnode->pWriteW.queue->threadId);
+        taosQueueGetThreadId(pVnode->pWriteW.queue));
   tMultiWorkerCleanup(&pVnode->pWriteW);
 
   dInfo("vgId:%d, wait for vnode sync queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncW.queue,
-        pVnode->pSyncW.queue->threadId);
+        taosQueueGetThreadId(pVnode->pSyncW.queue));
   tMultiWorkerCleanup(&pVnode->pSyncW);
 
   dInfo("vgId:%d, wait for vnode sync rd queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncRdW.queue,
-        pVnode->pSyncRdW.queue->threadId);
+        taosQueueGetThreadId(pVnode->pSyncRdW.queue));
   tMultiWorkerCleanup(&pVnode->pSyncRdW);
 
   dInfo("vgId:%d, wait for vnode apply queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue,
-        pVnode->pApplyW.queue->threadId);
+        taosQueueGetThreadId(pVnode->pApplyW.queue));
   tMultiWorkerCleanup(&pVnode->pApplyW);
 
   dInfo("vgId:%d, wait for vnode query queue:%p is empty", pVnode->vgId, pVnode->pQueryQ);
   while (!taosQueueEmpty(pVnode->pQueryQ)) taosMsleep(10);
 
   dInfo("vgId:%d, wait for vnode fetch queue:%p is empty, thread:%08" PRId64, pVnode->vgId, pVnode->pFetchQ,
-        pVnode->pFetchQ->threadId);
+        taosQueueGetThreadId(pVnode->pFetchQ));
   while (!taosQueueEmpty(pVnode->pFetchQ)) taosMsleep(10);
 
   tqNotifyClose(pVnode->pImpl->pTq);
diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
index 8b80527447..a6abe5ab4d 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
@@ -365,16 +365,16 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
   }
 
   dInfo("vgId:%d, write-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pWriteW.queue,
-        pVnode->pWriteW.queue->threadId);
+        taosQueueGetThreadId(pVnode->pWriteW.queue));
   dInfo("vgId:%d, sync-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncW.queue,
-        pVnode->pSyncW.queue->threadId);
+        taosQueueGetThreadId(pVnode->pSyncW.queue));
   dInfo("vgId:%d, sync-rd-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pSyncRdW.queue,
-        pVnode->pSyncRdW.queue->threadId);
+        taosQueueGetThreadId(pVnode->pSyncRdW.queue));
   dInfo("vgId:%d, apply-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pApplyW.queue,
-        pVnode->pApplyW.queue->threadId);
+        taosQueueGetThreadId(pVnode->pApplyW.queue));
   dInfo("vgId:%d, query-queue:%p is alloced", pVnode->vgId, pVnode->pQueryQ);
   dInfo("vgId:%d, fetch-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pFetchQ,
-        pVnode->pFetchQ->threadId);
+        taosQueueGetThreadId(pVnode->pFetchQ));
   dInfo("vgId:%d, stream-queue:%p is alloced", pVnode->vgId, pVnode->pStreamQ);
   return 0;
 }
diff --git a/source/util/src/tqueue.c b/source/util/src/tqueue.c
index 2cc13be6ba..7a4eb09b99 100644
--- a/source/util/src/tqueue.c
+++ b/source/util/src/tqueue.c
@@ -21,6 +21,40 @@
 int64_t tsRpcQueueMemoryAllowed = 0;
 int64_t tsRpcQueueMemoryUsed = 0;
 
+struct STaosQueue {
+  STaosQnode   *head;
+  STaosQnode   *tail;
+  STaosQueue   *next;     // for queue set
+  STaosQset    *qset;     // for queue set
+  void         *ahandle;  // for queue set
+  FItem         itemFp;
+  FItems        itemsFp;
+  TdThreadMutex mutex;
+  int64_t       memOfItems;
+  int32_t       numOfItems;
+  int64_t       threadId;
+  int64_t       memLimit;
+  int64_t       itemLimit;
+};
+
+struct STaosQset {
+  STaosQueue   *head;
+  STaosQueue   *current;
+  TdThreadMutex mutex;
+  tsem_t        sem;
+  int32_t       numOfQueues;
+  int32_t       numOfItems;
+};
+
+struct STaosQall {
+  STaosQnode *current;
+  STaosQnode *start;
+  int32_t     numOfItems;
+  int64_t     memOfItems;
+  int32_t     unAccessedNumOfItems;
+  int64_t     unAccessMemOfItems;
+};
+
 void taosSetQueueMemoryCapacity(STaosQueue *queue, int64_t cap) { queue->memLimit = cap; }
 void taosSetQueueCapacity(STaosQueue *queue, int64_t size) { queue->itemLimit = size; }
 
@@ -497,6 +531,12 @@ int64_t taosQallUnAccessedMemSize(STaosQall *qall) { return qall->unAccessMemOfI
 void    taosResetQitems(STaosQall *qall) { qall->current = qall->start; }
 int32_t taosGetQueueNumber(STaosQset *qset) { return qset->numOfQueues; }
 
+void taosQueueSetThreadId(STaosQueue* pQueue, int64_t threadId) {
+  pQueue->threadId = threadId;
+}
+
+int64_t taosQueueGetThreadId(STaosQueue *pQueue) { return pQueue->threadId; }
+
 #if 0
 
 void taosResetQsetThread(STaosQset *qset, void *pItem) {
diff --git a/source/util/src/tworker.c b/source/util/src/tworker.c
index 3e591c7d7f..138d4bc1f4 100644
--- a/source/util/src/tworker.c
+++ b/source/util/src/tworker.c
@@ -417,9 +417,9 @@ _OVER:
     return NULL;
   } else {
     while (worker->pid <= 0) taosMsleep(10);
-    queue->threadId = worker->pid;
-    uInfo("worker:%s, queue:%p is allocated, ahandle:%p thread:%08" PRId64, pool->name, queue, ahandle,
-          queue->threadId);
+
+    taosQueueSetThreadId(queue, worker->pid);
+    uInfo("worker:%s, queue:%p is allocated, ahandle:%p thread:%08" PRId64, pool->name, queue, ahandle, worker->pid);
     return queue;
   }
 }

From 0e522fbf5a4fdc50786b966695c9383986fd130e Mon Sep 17 00:00:00 2001
From: 54liuyao <54liuyao>
Date: Tue, 27 Feb 2024 16:30:38 +0800
Subject: [PATCH 07/24] set history task param

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

diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c
index 2534c5e9f0..831fd4e883 100644
--- a/source/libs/executor/src/executor.c
+++ b/source/libs/executor/src/executor.c
@@ -1009,6 +1009,22 @@ int32_t qSetStreamOperatorOptionForScanHistory(qTaskInfo_t tinfo) {
       pSup->deleteMark = INT64_MAX;
       pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;
       pInfo->ignoreExpiredData = false;
+    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT) {
+      SStreamCountAggOperatorInfo* pInfo = pOperator->info;
+      STimeWindowAggSupp* pSup = &pInfo->twAggSup;
+
+      ASSERT(pSup->calTrigger == STREAM_TRIGGER_AT_ONCE || pSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE);
+      ASSERT(pSup->calTriggerSaved == 0 && pSup->deleteMarkSaved == 0);
+
+      qInfo("save stream param for state: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);
+
+      pSup->calTriggerSaved = pSup->calTrigger;
+      pSup->deleteMarkSaved = pSup->deleteMark;
+      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;
+      pSup->deleteMark = INT64_MAX;
+      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;
+      pInfo->ignoreExpiredData = false;
+      qInfo("save stream task:%s, param for state: %d", GET_TASKID(pTaskInfo), pInfo->ignoreExpiredData);
     }
 
     // iterate operator tree

From 5be0cc7c8483a5f4baebaa1ba3efdc1baab1da5e Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Tue, 27 Feb 2024 18:07:04 +0800
Subject: [PATCH 08/24] opti:deal with escape in the end of field in schemaless

---
 source/client/src/clientSmlLine.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/source/client/src/clientSmlLine.c b/source/client/src/clientSmlLine.c
index 8bdbec0f2b..7535cbfd0c 100644
--- a/source/client/src/clientSmlLine.c
+++ b/source/client/src/clientSmlLine.c
@@ -485,8 +485,8 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
     if (unlikely(IS_SPACE(tmp,escapeChar))) {
       break;
     }
-    if(unlikely(IS_SLASH_LETTER_IN_TAG_FIELD_KEY(sql))){
-      escapeChar = sql;
+    if(unlikely(IS_SLASH_LETTER_IN_TAG_FIELD_KEY(tmp))){
+      escapeChar = tmp;
     }
     tmp++;
   }

From 828889a6d8ca7ab7dab065db91ef5ca93b9d1963 Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Tue, 27 Feb 2024 19:16:11 +0800
Subject: [PATCH 09/24] opti:deal with escape in the end of field in schemaless

---
 utils/test/c/sml_test.c | 11 ++++-------
 1 file changed, 4 insertions(+), 7 deletions(-)

diff --git a/utils/test/c/sml_test.c b/utils/test/c/sml_test.c
index f81f13fc68..01619decc5 100644
--- a/utils/test/c/sml_test.c
+++ b/utils/test/c/sml_test.c
@@ -1079,7 +1079,7 @@ int sml_escape1_Test() {
   ASSERT(taos_affected_rows(pRes) == 5);
   taos_free_result(pRes);
 
-  pRes = taos_query(taos, "select * from stab order by _ts");     //check stable name
+  pRes = taos_query(taos, "select * from stab");     //check stable name
   ASSERT(pRes);
   int fieldNum = taos_field_count(pRes);
   ASSERT(fieldNum == 6);
@@ -1099,22 +1099,19 @@ int sml_escape1_Test() {
   while ((row = taos_fetch_row(pRes)) != NULL) {
     int64_t ts = *(int64_t *)row[0];
 
-    if (rowIndex == 0) {
-      ASSERT(ts == 1661943970000);
+    if (ts == 1661943970000) {
       ASSERT(*(double *)row[1] == 3);
       ASSERT(strncmp(row[2], "32fw", sizeof("32fw") - 1) == 0);
       ASSERT(row[3] == NULL);
       ASSERT(strncmp(row[4], "1", sizeof("1") - 1) == 0);
       ASSERT(row[5] == NULL);
-    }else if (rowIndex == 1) {
-      ASSERT(ts == 1661943980000);
+    }else if (ts == 1661943980000) {
       ASSERT(*(double *)row[1] == 3);
       ASSERT(strncmp(row[2], "32fw", sizeof("32fw") - 1) == 0);
       ASSERT(row[3] == NULL);
       ASSERT(row[4] == NULL);
       ASSERT(strncmp(row[5], "1\\", sizeof("1\\") - 1) == 0);
-    }else if (rowIndex == 2) {
-      ASSERT(ts == 1661943990000);
+    }else if (ts == 1661943990000) {
       ASSERT(row[1] == NULL);
       ASSERT(strncmp(row[2], "32fw", sizeof("32fw") - 1) == 0);
       ASSERT(*(double *)row[3] == 3);

From b3c18c31da19f3cb5b72001351eae62273067f57 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Tue, 27 Feb 2024 12:19:37 +0000
Subject: [PATCH 10/24] add test case to stream  backend

---
 source/libs/stream/inc/streamBackendRocksdb.h |   1 +
 source/libs/stream/src/streamBackendRocksdb.c |  12 +-
 source/libs/stream/test/backendTest.cpp       | 309 +++++++++++++++++-
 3 files changed, 305 insertions(+), 17 deletions(-)

diff --git a/source/libs/stream/inc/streamBackendRocksdb.h b/source/libs/stream/inc/streamBackendRocksdb.h
index 6e6f85d8f9..1dc1db8e9c 100644
--- a/source/libs/stream/inc/streamBackendRocksdb.h
+++ b/source/libs/stream/inc/streamBackendRocksdb.h
@@ -258,6 +258,7 @@ void     bkdMgtDestroy(SBkdMgt* bm);
 
 int32_t taskDbGenChkpUploadData(void* arg, void* bkdMgt, int64_t chkpId, int8_t type, char** path, SArray* list);
 
+uint32_t nextPow2(uint32_t x); 
 #ifdef __cplusplus
 }
 #endif
diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c
index f173157da6..910fd93989 100644
--- a/source/libs/stream/src/streamBackendRocksdb.c
+++ b/source/libs/stream/src/streamBackendRocksdb.c
@@ -2788,7 +2788,6 @@ SStreamStateCur* streamStateSeekToLast_rocksdb(SStreamState* pState) {
   STREAM_STATE_DEL_ROCKSDB(pState, "state", &maxStateKey);
   return pCur;
 }
-#ifdef BUILD_NO_CALL
 SStreamStateCur* streamStateGetCur_rocksdb(SStreamState* pState, const SWinKey* key) {
   stDebug("streamStateGetCur_rocksdb");
   STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
@@ -2838,7 +2837,6 @@ int32_t streamStateFuncDel_rocksdb(SStreamState* pState, const STupleKey* key) {
   STREAM_STATE_DEL_ROCKSDB(pState, "func", key);
   return 0;
 }
-#endif
 
 // session cf
 int32_t streamStateSessionPut_rocksdb(SStreamState* pState, const SSessionKey* key, const void* value, int32_t vLen) {
@@ -3432,7 +3430,6 @@ int32_t streamStateStateAddIfNotExist_rocksdb(SStreamState* pState, SSessionKey*
   SSessionKey tmpKey = *key;
   int32_t     valSize = *pVLen;
   void*       tmp = taosMemoryMalloc(valSize);
-  // tdbRealloc(NULL, valSize);
   if (!tmp) {
     return -1;
   }
@@ -3506,13 +3503,11 @@ int32_t streamStateGetParName_rocksdb(SStreamState* pState, int64_t groupId, voi
   return code;
 }
 
-#ifdef BUILD_NO_CALL
 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);
   return code;
 }
-#endif
 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);
@@ -3535,10 +3530,10 @@ int32_t streamDefaultIterGet_rocksdb(SStreamState* pState, const void* start, co
   if (pIter == NULL) {
     return -1;
   }
-
+  size_t klen = 0;
   rocksdb_iter_seek(pIter, start, strlen(start));
   while (rocksdb_iter_valid(pIter)) {
-    const char* key = rocksdb_iter_key(pIter, NULL);
+    const char* key = rocksdb_iter_key(pIter, &klen);
     int32_t     vlen = 0;
     const char* vval = rocksdb_iter_value(pIter, (size_t*)&vlen);
     char*       val = NULL;
@@ -3700,6 +3695,8 @@ uint32_t nextPow2(uint32_t x) {
   x = x | (x >> 16);
   return x + 1;
 }
+
+#ifdef BUILD_NO_CALL
 int32_t copyFiles(const char* src, const char* dst) {
   int32_t code = 0;
   // opt later, just hard link
@@ -3739,6 +3736,7 @@ _err:
   taosCloseDir(&pDir);
   return code >= 0 ? 0 : -1;
 }
+#endif
 
 int32_t isBkdDataMeta(char* name, int32_t len) {
   const char* pCurrent = "CURRENT";
diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
index 1ccc0b4285..83adaeb21c 100644
--- a/source/libs/stream/test/backendTest.cpp
+++ b/source/libs/stream/test/backendTest.cpp
@@ -40,7 +40,8 @@ SStreamState *stateCreate(const char *path) {
   ASSERT(p != NULL);
   return p;
 }
-void backendOpen() {
+void *backendOpen() {
+  streamMetaInit();
   const char   *path = "/tmp/backend";
   SStreamState *p = stateCreate(path);
   ASSERT(p != NULL);
@@ -85,18 +86,306 @@ void backendOpen() {
   }
   tsArray.clear();
 
-  // read
-  // iterator
-  // rebuild chkp, reload from chkp
-  // sync
-  //
+  for (int i = 0; i < size; i++) {
+    int64_t ts = taosGetTimestampMs();
+    tsArray.push_back(ts);
+
+    SWinKey     key = {.groupId = (uint64_t)(i), .ts = ts};
+    const char *val = "value data";
+    int32_t     vlen = strlen(val);
+    streamStatePut_rocksdb(p, &key, (char *)val, vlen);
+  }
+
+  SWinKey winkey;
+  int32_t code = streamStateGetFirst_rocksdb(p, &key);
+  ASSERT(code == 0);
+  ASSERT(key.ts == tsArray[0]);
+
+  SStreamStateCur *pCurr = streamStateSeekToLast_rocksdb(p);
+  ASSERT(pCurr != NULL);
+  streamStateFreeCur(pCurr);
+
+  winkey.groupId = 0;
+  winkey.ts = tsArray[0];
+  char   *val = NULL;
+  int32_t len = 0;
+
+  pCurr = streamStateSeekKeyNext_rocksdb(p, &winkey);
+  ASSERT(pCurr != NULL);
+
+  streamStateFreeCur(pCurr);
+
+  tsArray.clear();
+  for (int i = 0; i < size; i++) {
+    int64_t ts = taosGetTimestampMs();
+    tsArray.push_back(ts);
+    STupleKey key = {.groupId = (uint64_t)(0), .ts = ts, .exprIdx = i};
+
+    const char *val = "Value";
+    int32_t     len = strlen(val);
+    streamStateFuncPut_rocksdb(p, &key, val, len);
+  }
+  for (int i = 0; i < size; i++) {
+    STupleKey key = {.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
+    char     *val = NULL;
+    int32_t   len = 0;
+    streamStateFuncGet_rocksdb(p, &key, (void **)&val, &len);
+    ASSERT(len == strlen("Value"));
+  }
+  for (int i = 0; i < size; i++) {
+    STupleKey key = {.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
+    char     *val = NULL;
+    int32_t   len = 0;
+    streamStateFuncDel_rocksdb(p, &key);
+  }
+
+  // session put
+  tsArray.clear();
+
+  for (int i = 0; i < size; i++) {
+    SSessionKey key = {.win = {.skey = i, .ekey = i}, .groupId = (uint64_t)(0)};
+    tsArray.push_back(i);
+
+    const char *val = "Value";
+    int32_t     len = strlen(val);
+    streamStateSessionPut_rocksdb(p, &key, val, len);
+
+    char *pval = NULL;
+    ASSERT(0 == streamStateSessionGet_rocksdb(p, &key, (void **)&pval, &len));
+    ASSERT(strncmp(pval, val, len) == 0);
+  }
+
+  for (int i = 0; i < size; i++) {
+    SSessionKey key = {.win = {.skey = tsArray[i], .ekey = tsArray[i]}, .groupId = (uint64_t)(0)};
+
+    const char *val = "Value";
+    int32_t     len = strlen(val);
+
+    char *pval = NULL;
+    ASSERT(0 == streamStateSessionGet_rocksdb(p, &key, (void **)&pval, &len));
+    ASSERT(strncmp(pval, val, len) == 0);
+    taosMemoryFreeClear(pval);
+  }
+
+  pCurr = streamStateSessionSeekToLast_rocksdb(p, 0);
+  ASSERT(pCurr != NULL);
+
+  {
+    SSessionKey key;
+    char       *val = NULL;
+    int32_t     vlen = 0;
+    code = streamStateSessionGetKVByCur_rocksdb(pCurr, &key, (void **)&val, &vlen);
+    ASSERT(code == 0);
+    pCurr = streamStateSessionSeekKeyPrev_rocksdb(p, &key);
+
+    code = streamStateSessionGetKVByCur_rocksdb(pCurr, &key, (void **)&val, &vlen);
+    ASSERT(code == 0);
+
+    ASSERT(key.groupId == 0 && key.win.ekey == tsArray[tsArray.size() - 2]);
+
+    pCurr = streamStateSessionSeekKeyNext_rocksdb(p, &key);
+    code = streamStateSessionGetKVByCur_rocksdb(pCurr, &key, (void **)&val, &vlen);
+    ASSERT(code == 0);
+    ASSERT(vlen == strlen("Value"));
+    ASSERT(key.groupId == 0 && key.win.skey == tsArray[tsArray.size() - 1]);
+
+    ASSERT(0 == streamStateSessionAddIfNotExist_rocksdb(p, &key, 10, (void **)&val, &len));
+
+    ASSERT(0 ==
+           streamStateStateAddIfNotExist_rocksdb(p, &key, (char *)"key", strlen("key"), NULL, (void **)&val, &len));
+  }
+  for (int i = 0; i < size; i++) {
+    SSessionKey key = {.win = {.skey = tsArray[i], .ekey = tsArray[i]}, .groupId = (uint64_t)(0)};
+
+    const char *val = "Value";
+    int32_t     len = strlen(val);
+
+    char *pval = NULL;
+    ASSERT(0 == streamStateSessionDel_rocksdb(p, &key));
+  }
+
+  for (int i = 0; i < size; i++) {
+    SWinKey     key = {.groupId = (uint64_t)(i), .ts = tsArray[i]};
+    const char *val = "Value";
+    int32_t     vlen = strlen(val);
+    ASSERT(streamStateFillPut_rocksdb(p, &key, val, vlen) == 0);
+  }
+  for (int i = 0; i < size; i++) {
+    SWinKey key = {.groupId = (uint64_t)(i), .ts = tsArray[i]};
+    char   *val = NULL;
+    int32_t vlen = 0;
+    ASSERT(streamStateFillGet_rocksdb(p, &key, (void **)&val, &vlen) == 0);
+    taosMemoryFreeClear(val);
+  }
+  {
+    SWinKey          key = {.groupId = (uint64_t)(0), .ts = tsArray[0]};
+    SStreamStateCur *pCurr = streamStateFillGetCur_rocksdb(p, &key);
+    ASSERT(pCurr != NULL);
+
+    char   *val = NULL;
+    int32_t vlen = 0;
+    ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));
+    ASSERT(vlen == strlen("Value"));
+    streamStateFreeCur(pCurr);
+
+    pCurr = streamStateFillSeekKeyNext_rocksdb(p, &key);
+    ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));
+    ASSERT(vlen == strlen("Value") && key.groupId == 1 && key.ts == tsArray[1]);
+
+    key.groupId = 1;
+    key.ts = tsArray[1];
+
+    pCurr = streamStateFillSeekKeyPrev_rocksdb(p, &key);
+    ASSERT(pCurr != NULL);
+    ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));
+
+    ASSERT(vlen == strlen("Value") && key.groupId == 0 && key.ts == tsArray[0]);
+  }
+
+  for (int i = 0; i < size - 1; i++) {
+    SWinKey key = {.groupId = (uint64_t)(i), .ts = tsArray[i]};
+    char   *val = NULL;
+    int32_t vlen = 0;
+    ASSERT(streamStateFillDel_rocksdb(p, &key) == 0);
+    taosMemoryFreeClear(val);
+  }
+  streamStateSessionClear_rocksdb(p);
+
+  for (int i = 0; i < size; i++) {
+    char tbname[TSDB_TABLE_NAME_LEN] = {0};
+    sprintf(tbname, "%s_%d", "tbname", i);
+    ASSERT(0 == streamStatePutParName_rocksdb(p, i, tbname));
+  }
+  for (int i = 0; i < size; i++) {
+    char *val = NULL;
+    ASSERT(0 == streamStateGetParName_rocksdb(p, i, (void **)&val));
+    ASSERT(strncmp(val, "tbname", strlen("tbname")) == 0);
+    taosMemoryFree(val);
+  }
+
+  for (int i = 0; i < size; i++) {
+    char tbname[TSDB_TABLE_NAME_LEN] = {0};
+    sprintf(tbname, "%s_%d", "tbname", i);
+    ASSERT(0 == streamStatePutParName_rocksdb(p, i, tbname));
+  }
+  for (int i = 0; i < size; i++) {
+    char *val = NULL;
+    ASSERT(0 == streamStateGetParName_rocksdb(p, i, (void **)&val));
+    ASSERT(strncmp(val, "tbname", strlen("tbname")) == 0);
+    taosMemoryFree(val);
+  }
+  for (int i = 0; i < size; i++) {
+    char key[128] = {0};
+    sprintf(key, "tbname_%d", i);
+    char val[128] = {0};
+    sprintf(val, "val_%d", i);
+    code = streamDefaultPut_rocksdb(p, key, val, strlen(val));
+    ASSERT(code == 0);
+  }
+  for (int i = 0; i < size; i++) {
+    char key[128] = {0};
+    sprintf(key, "tbname_%d", i);
+
+    char   *val = NULL;
+    int32_t len = 0;
+    code = streamDefaultGet_rocksdb(p, key, (void **)&val, &len);
+    ASSERT(code == 0);
+  }
+  SArray *result = taosArrayInit(8, sizeof(void *));
+  streamDefaultIterGet_rocksdb(p, "tbname", "tbname_99", result);
+  ASSERT(taosArrayGetSize(result) >= 0);
+
+  return p;
+  // streamStateClose((SStreamState *)p, true);
+}
+TEST_F(BackendEnv, checkOpen) {
+  SStreamState *p = (SStreamState *)backendOpen();
+  int64_t       tsStart = taosGetTimestampMs();
+  {
+    void   *pBatch = streamStateCreateBatch();
+    int32_t size = 0;
+    for (int i = 0; i < size; i++) {
+      char key[128] = {0};
+      sprintf(key, "key_%d", i);
+      char val[128] = {0};
+      sprintf(val, "val_%d", i);
+      streamStatePutBatch(p, "default", (rocksdb_writebatch_t *)pBatch, (void *)key, (void *)val,
+                          (int32_t)(strlen(val)), tsStart + 100000);
+    }
+    streamStatePutBatch_rocksdb(p, pBatch);
+    streamStateDestroyBatch(pBatch);
+  }
+  {
+    void   *pBatch = streamStateCreateBatch();
+    int32_t size = 0;
+    char    valBuf[256] = {0};
+    for (int i = 0; i < size; i++) {
+      char key[128] = {0};
+      sprintf(key, "key_%d", i);
+      char val[128] = {0};
+      sprintf(val, "val_%d", i);
+      streamStatePutBatchOptimize(p, 0, (rocksdb_writebatch_t *)pBatch, (void *)key, (void *)val,
+                                  (int32_t)(strlen(val)), tsStart + 100000, (void *)valBuf);
+    }
+    streamStatePutBatch_rocksdb(p, pBatch);
+    streamStateDestroyBatch(pBatch);
+  }
+  // do checkpoint 2
+  taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 2);
+  {
+    void   *pBatch = streamStateCreateBatch();
+    int32_t size = 0;
+    char    valBuf[256] = {0};
+    for (int i = 0; i < size; i++) {
+      char key[128] = {0};
+      sprintf(key, "key_%d", i);
+      char val[128] = {0};
+      sprintf(val, "val_%d", i);
+      streamStatePutBatchOptimize(p, 0, (rocksdb_writebatch_t *)pBatch, (void *)key, (void *)val,
+                                  (int32_t)(strlen(val)), tsStart + 100000, (void *)valBuf);
+    }
+    streamStatePutBatch_rocksdb(p, pBatch);
+    streamStateDestroyBatch(pBatch);
+  }
+
+  taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 3);
+
+  const char *path = "/tmp/backend/stream";
+  const char *dump = "/tmp/backend/stream/dump";
+  // taosMkDir(dump);
+  taosMulMkDir(dump);
+  SBkdMgt *mgt = bkdMgtCreate((char *)path);
+  SArray  *result = taosArrayInit(4, sizeof(void *));
+  bkdMgtGetDelta(mgt, p->pTdbState->idstr, 3, result, (char *)dump);
+
+  bkdMgtDestroy(mgt);
   streamStateClose((SStreamState *)p, true);
-  // taskDbDestroy(p, true);
 }
 
-TEST_F(BackendEnv, checkOpen) { backendOpen(); }
-TEST_F(BackendEnv, backendOpt) {}
-TEST_F(BackendEnv, backendDestroy) {}
+TEST_F(BackendEnv, backendChkp) { const char *path = "/tmp"; }
+
+typedef struct BdKV {
+  uint32_t k;
+  uint32_t v;
+} BdKV;
+
+BdKV kvDict[] = {{0, 2},
+                 {1, 2},
+                 {.k = 15, .v = 16},
+                 {.k = 31, .v = 32},
+                 {.k = 56, .v = 64},
+                 {.k = 100, .v = 128},
+                 {.k = 200, .v = 256},
+                 {.k = 500, .v = 512},
+                 {.k = 1000, .v = 1024},
+                 {.k = 2000, .v = 2048},
+                 {.k = 3000, .v = 4096}};
+
+TEST_F(BackendEnv, backendUtil) {
+  for (int i = 0; i < sizeof(kvDict) / sizeof(kvDict[0]); i++) {
+    ASSERT_EQ(nextPow2((uint32_t)(kvDict[i].k)), kvDict[i].v);
+  }
+}
 
 int main(int argc, char **argv) {
   testing::InitGoogleTest(&argc, argv);

From e1f11d1639e1d3214706cf1272e9c0ea1d2f9dd2 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Wed, 28 Feb 2024 00:48:20 +0000
Subject: [PATCH 11/24] add test case to stream  backend

---
 source/libs/stream/test/checkpointTest.cpp | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/source/libs/stream/test/checkpointTest.cpp b/source/libs/stream/test/checkpointTest.cpp
index eb7f38744f..0caad479e5 100644
--- a/source/libs/stream/test/checkpointTest.cpp
+++ b/source/libs/stream/test/checkpointTest.cpp
@@ -25,6 +25,9 @@
 #pragma GCC diagnostic ignored "-Wunused-variable"
 #pragma GCC diagnostic ignored "-Wsign-compare"
 
+// tsSnodeAddress = "";
+// tsS3StreamEnabled = 0;
+
 #include "cos.h"
 #include "rsync.h"
 #include "streamInt.h"
@@ -45,26 +48,26 @@
 // }
 
 TEST(testCase, checkpointUpload_Test) {
-  stopRsync();
-  startRsync();
+  // stopRsync();
+  // startRsync();
 
   taosSsleep(5);
   char* id = "2013892036";
 
-  uploadCheckpoint(id, "/root/offset/");
+  // uploadCheckpoint(id, "/root/offset/");
 }
 
 TEST(testCase, checkpointDownload_Test) {
   char* id = "2013892036";
-  downloadCheckpoint(id, "/root/offset/download/");
+  // downloadCheckpoint(id, "/root/offset/download/");
 }
 
 TEST(testCase, checkpointDelete_Test) {
   char* id = "2013892036";
-  deleteCheckpoint(id);
+  // deleteCheckpoint(id);
 }
 
 TEST(testCase, checkpointDeleteFile_Test) {
   char* id = "2013892036";
-  deleteCheckpointFile(id, "offset-ver0");
+  // deleteCheckpointFile(id, "offset-ver0");
 }

From c1129bc8610a2e3035dce3b580596a38edfc29be Mon Sep 17 00:00:00 2001
From: 54liuyao <54liuyao>
Date: Wed, 28 Feb 2024 09:34:43 +0800
Subject: [PATCH 12/24] add ci

---
 tests/parallel_test/cases.task                |   1 +
 .../stream/distributeMultiLevelInterval0.sim  | 267 ++++++++++++++++++
 2 files changed, 268 insertions(+)
 create mode 100644 tests/script/tsim/stream/distributeMultiLevelInterval0.sim

diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task
index bdccf33c32..62f2d10525 100644
--- a/tests/parallel_test/cases.task
+++ b/tests/parallel_test/cases.task
@@ -1211,6 +1211,7 @@
 ,,y,script,./test.sh -f tsim/stream/deleteState.sim
 ,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim
 ,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
+,,y,script,./test.sh -f tsim/stream/distributeMultiLevelInterval0.sim
 ,,y,script,./test.sh -f tsim/stream/distributeSession0.sim
 ,,y,script,./test.sh -f tsim/stream/drop_stream.sim
 ,,y,script,./test.sh -f tsim/stream/event0.sim
diff --git a/tests/script/tsim/stream/distributeMultiLevelInterval0.sim b/tests/script/tsim/stream/distributeMultiLevelInterval0.sim
new file mode 100644
index 0000000000..784ab7f4a5
--- /dev/null
+++ b/tests/script/tsim/stream/distributeMultiLevelInterval0.sim
@@ -0,0 +1,267 @@
+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+
+system sh/cfg.sh -n dnode1 -c streamAggCnt  -v 2
+
+system sh/exec.sh -n dnode1 -s start 
+sleep 50
+sql connect
+
+
+
+print ===== step1
+sql drop stream if exists streams1;
+sql drop database if exists test;
+sql create database test  vgroups 4;
+sql use test;
+sql create stable st(ts timestamp, a int, b int , c int, d double) tags(ta int,tb int,tc int);
+sql create table ts1 using st tags(1,1,1);
+sql create table ts2 using st tags(2,2,2);
+sql create table ts3 using st tags(3,2,2);
+sql create table ts4 using st tags(4,2,2);
+sql create stream streams1 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 watermark 1d into streamt1 as select  _wstart, count(*) c1, sum(a) c3 , max(b) c4 from st interval(10s);
+
+sleep 1000
+
+sql insert into ts1 values(1648791213000,1,1,3,4.1);
+sql insert into ts1 values(1648791223000,2,2,3,1.1);
+sql insert into ts1 values(1648791233000,3,3,3,2.1);
+sql insert into ts1 values(1648791243000,4,4,3,3.1);
+
+sql insert into ts2 values(1648791213000,1,5,3,4.1);
+sql insert into ts2 values(1648791223000,2,6,3,1.1);
+sql insert into ts2 values(1648791233000,3,7,3,2.1);
+sql insert into ts2 values(1648791243000,4,8,3,3.1);
+
+
+$loop_count = 0
+loop0:
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+sleep 1000
+print 2 select * from streamt1;
+sql select * from streamt1;
+
+print $data00 $data01 $data02 $data03
+print $data10 $data11 $data12 $data13
+print $data20 $data21 $data22 $data23
+print $data30 $data31 $data32 $data33
+print $data40 $data41 $data42 $data43
+
+if $rows != 4 then
+  print =====rows=$rows
+  goto loop0
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop0
+endi
+
+if $data11 != 2 then
+  print =====data11=$data11
+  goto loop0
+endi
+
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop0
+endi
+
+if $data31 != 2 then
+  print =====data31=$data31
+  goto loop0
+endi
+
+
+sql insert into ts1 values(1648791213000,1,9,3,4.1);
+
+$loop_count = 0
+loop1:
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+sleep 1000
+print 2 select * from streamt1;
+sql select * from streamt1;
+
+print $data00 $data01 $data02 $data03
+print $data10 $data11 $data12 $data13
+print $data20 $data21 $data22 $data23
+print $data30 $data31 $data32 $data33
+print $data40 $data41 $data42 $data43
+
+if $rows != 4 then
+  print =====rows=$rows
+  goto loop1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop1
+endi
+
+if $data11 != 2 then
+  print =====data11=$data11
+  goto loop1
+endi
+
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop1
+endi
+
+if $data31 != 2 then
+  print =====data31=$data31
+  goto loop1
+endi
+
+sql delete from ts2 where ts = 1648791243000 ;
+
+$loop_count = 0
+loop2:
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+sleep 1000
+print 2 select * from streamt1;
+sql select * from streamt1;
+
+print $data00 $data01 $data02 $data03
+print $data10 $data11 $data12 $data13
+print $data20 $data21 $data22 $data23
+print $data30 $data31 $data32 $data33
+print $data40 $data41 $data42 $data43
+
+if $rows != 4 then
+  print =====rows=$rows
+  goto loop2
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop2
+endi
+
+if $data11 != 2 then
+  print =====data11=$data11
+  goto loop2
+endi
+
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop2
+endi
+
+if $data31 != 1 then
+  print =====data31=$data31
+  goto loop2
+endi
+
+sql delete from ts2 where ts = 1648791223000 ;
+
+$loop_count = 0
+loop3:
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+sleep 1000
+print 2 select * from streamt1;
+sql select * from streamt1;
+
+print $data00 $data01 $data02 $data03
+print $data10 $data11 $data12 $data13
+print $data20 $data21 $data22 $data23
+print $data30 $data31 $data32 $data33
+print $data40 $data41 $data42 $data43
+
+if $rows != 4 then
+  print =====rows=$rows
+  goto loop3
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop3
+endi
+
+if $data11 != 1 then
+  print =====data11=$data11
+  goto loop3
+endi
+
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop3
+endi
+
+if $data31 != 1 then
+  print =====data31=$data31
+  goto loop3
+endi
+
+
+sql insert into ts1 values(1648791233001,3,9,3,2.1);
+
+$loop_count = 0
+loop4:
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+sleep 1000
+print 2 select * from streamt1;
+sql select * from streamt1;
+
+print $data00 $data01 $data02 $data03
+print $data10 $data11 $data12 $data13
+print $data20 $data21 $data22 $data23
+print $data30 $data31 $data32 $data33
+print $data40 $data41 $data42 $data43
+
+if $rows != 4 then
+  print =====rows=$rows
+  goto loop4
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop4
+endi
+
+if $data11 != 1 then
+  print =====data11=$data11
+  goto loop4
+endi
+
+if $data21 != 3 then
+  print =====data21=$data21
+  goto loop4
+endi
+
+if $data31 != 1 then
+  print =====data31=$data31
+  goto loop4
+endi
+
+sql select  _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b)  c4, min(c) c5, avg(d) from st interval(10s);
+
+
+print ===== over
+
+system sh/stop_dnodes.sh

From 9f983a07d78f5548f623aa92319be5afe75704be Mon Sep 17 00:00:00 2001
From: yihaoDeng <yhdeng@taosdata.com>
Date: Wed, 28 Feb 2024 10:48:24 +0800
Subject: [PATCH 13/24] add test case

---
 source/libs/stream/test/backendTest.cpp | 33 +++++++++++++------------
 1 file changed, 17 insertions(+), 16 deletions(-)

diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
index 83adaeb21c..8143c6dbaf 100644
--- a/source/libs/stream/test/backendTest.cpp
+++ b/source/libs/stream/test/backendTest.cpp
@@ -19,7 +19,7 @@ class BackendEnv : public ::testing::Test {
 
 void *backendCreate() {
   const char *streamPath = "/tmp";
-  void       *p = NULL;
+  void *      p = NULL;
 
   // char *absPath = NULL;
   // // SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(streamPath, -1, 2);
@@ -42,7 +42,7 @@ SStreamState *stateCreate(const char *path) {
 }
 void *backendOpen() {
   streamMetaInit();
-  const char   *path = "/tmp/backend";
+  const char *  path = "/tmp/backend";
   SStreamState *p = stateCreate(path);
   ASSERT(p != NULL);
 
@@ -65,7 +65,7 @@ void *backendOpen() {
 
     const char *val = "value data";
     int32_t     len = 0;
-    char       *newVal = NULL;
+    char *      newVal = NULL;
     streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
     ASSERT(len == strlen(val));
   }
@@ -80,7 +80,7 @@ void *backendOpen() {
     SWinKey     key = {.groupId = (uint64_t)(i), .ts = ts};
     const char *val = "value data";
     int32_t     len = 0;
-    char       *newVal = NULL;
+    char *      newVal = NULL;
     int32_t     code = streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
     ASSERT(code != 0);
   }
@@ -107,7 +107,7 @@ void *backendOpen() {
 
   winkey.groupId = 0;
   winkey.ts = tsArray[0];
-  char   *val = NULL;
+  char *  val = NULL;
   int32_t len = 0;
 
   pCurr = streamStateSeekKeyNext_rocksdb(p, &winkey);
@@ -127,14 +127,14 @@ void *backendOpen() {
   }
   for (int i = 0; i < size; i++) {
     STupleKey key = {.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
-    char     *val = NULL;
+    char *    val = NULL;
     int32_t   len = 0;
     streamStateFuncGet_rocksdb(p, &key, (void **)&val, &len);
     ASSERT(len == strlen("Value"));
   }
   for (int i = 0; i < size; i++) {
     STupleKey key = {.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
-    char     *val = NULL;
+    char *    val = NULL;
     int32_t   len = 0;
     streamStateFuncDel_rocksdb(p, &key);
   }
@@ -172,7 +172,7 @@ void *backendOpen() {
 
   {
     SSessionKey key;
-    char       *val = NULL;
+    char *      val = NULL;
     int32_t     vlen = 0;
     code = streamStateSessionGetKVByCur_rocksdb(pCurr, &key, (void **)&val, &vlen);
     ASSERT(code == 0);
@@ -212,7 +212,7 @@ void *backendOpen() {
   }
   for (int i = 0; i < size; i++) {
     SWinKey key = {.groupId = (uint64_t)(i), .ts = tsArray[i]};
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t vlen = 0;
     ASSERT(streamStateFillGet_rocksdb(p, &key, (void **)&val, &vlen) == 0);
     taosMemoryFreeClear(val);
@@ -222,7 +222,7 @@ void *backendOpen() {
     SStreamStateCur *pCurr = streamStateFillGetCur_rocksdb(p, &key);
     ASSERT(pCurr != NULL);
 
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t vlen = 0;
     ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));
     ASSERT(vlen == strlen("Value"));
@@ -244,7 +244,7 @@ void *backendOpen() {
 
   for (int i = 0; i < size - 1; i++) {
     SWinKey key = {.groupId = (uint64_t)(i), .ts = tsArray[i]};
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t vlen = 0;
     ASSERT(streamStateFillDel_rocksdb(p, &key) == 0);
     taosMemoryFreeClear(val);
@@ -286,7 +286,7 @@ void *backendOpen() {
     char key[128] = {0};
     sprintf(key, "tbname_%d", i);
 
-    char   *val = NULL;
+    char *  val = NULL;
     int32_t len = 0;
     code = streamDefaultGet_rocksdb(p, key, (void **)&val, &len);
     ASSERT(code == 0);
@@ -302,7 +302,7 @@ TEST_F(BackendEnv, checkOpen) {
   SStreamState *p = (SStreamState *)backendOpen();
   int64_t       tsStart = taosGetTimestampMs();
   {
-    void   *pBatch = streamStateCreateBatch();
+    void *  pBatch = streamStateCreateBatch();
     int32_t size = 0;
     for (int i = 0; i < size; i++) {
       char key[128] = {0};
@@ -316,7 +316,7 @@ TEST_F(BackendEnv, checkOpen) {
     streamStateDestroyBatch(pBatch);
   }
   {
-    void   *pBatch = streamStateCreateBatch();
+    void *  pBatch = streamStateCreateBatch();
     int32_t size = 0;
     char    valBuf[256] = {0};
     for (int i = 0; i < size; i++) {
@@ -333,7 +333,7 @@ TEST_F(BackendEnv, checkOpen) {
   // do checkpoint 2
   taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 2);
   {
-    void   *pBatch = streamStateCreateBatch();
+    void *  pBatch = streamStateCreateBatch();
     int32_t size = 0;
     char    valBuf[256] = {0};
     for (int i = 0; i < size; i++) {
@@ -355,11 +355,12 @@ TEST_F(BackendEnv, checkOpen) {
   // taosMkDir(dump);
   taosMulMkDir(dump);
   SBkdMgt *mgt = bkdMgtCreate((char *)path);
-  SArray  *result = taosArrayInit(4, sizeof(void *));
+  SArray * result = taosArrayInit(4, sizeof(void *));
   bkdMgtGetDelta(mgt, p->pTdbState->idstr, 3, result, (char *)dump);
 
   bkdMgtDestroy(mgt);
   streamStateClose((SStreamState *)p, true);
+  taosRemoveDir(path);
 }
 
 TEST_F(BackendEnv, backendChkp) { const char *path = "/tmp"; }

From d7f14e916d65754b79e46f0d38ba8a0c74c8dfbf Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Wed, 28 Feb 2024 06:01:50 +0000
Subject: [PATCH 14/24] add test case

---
 source/libs/stream/test/backendTest.cpp | 38 ++++++++++++++-----------
 1 file changed, 22 insertions(+), 16 deletions(-)

diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
index 8143c6dbaf..3245979633 100644
--- a/source/libs/stream/test/backendTest.cpp
+++ b/source/libs/stream/test/backendTest.cpp
@@ -11,6 +11,12 @@
 #include "tstreamFileState.h"
 #include "tstreamUpdate.h"
 
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wwrite-strings"
+#pragma GCC diagnostic ignored "-Wunused-function"
+#pragma GCC diagnostic ignored "-Wunused-variable"
+#pragma GCC diagnostic ignored "-Wsign-compare"
+
 class BackendEnv : public ::testing::Test {
  protected:
   virtual void SetUp() {}
@@ -19,7 +25,7 @@ class BackendEnv : public ::testing::Test {
 
 void *backendCreate() {
   const char *streamPath = "/tmp";
-  void *      p = NULL;
+  void       *p = NULL;
 
   // char *absPath = NULL;
   // // SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(streamPath, -1, 2);
@@ -42,7 +48,7 @@ SStreamState *stateCreate(const char *path) {
 }
 void *backendOpen() {
   streamMetaInit();
-  const char *  path = "/tmp/backend";
+  const char   *path = "/tmp/backend";
   SStreamState *p = stateCreate(path);
   ASSERT(p != NULL);
 
@@ -65,7 +71,7 @@ void *backendOpen() {
 
     const char *val = "value data";
     int32_t     len = 0;
-    char *      newVal = NULL;
+    char       *newVal = NULL;
     streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
     ASSERT(len == strlen(val));
   }
@@ -80,7 +86,7 @@ void *backendOpen() {
     SWinKey     key = {.groupId = (uint64_t)(i), .ts = ts};
     const char *val = "value data";
     int32_t     len = 0;
-    char *      newVal = NULL;
+    char       *newVal = NULL;
     int32_t     code = streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
     ASSERT(code != 0);
   }
@@ -107,7 +113,7 @@ void *backendOpen() {
 
   winkey.groupId = 0;
   winkey.ts = tsArray[0];
-  char *  val = NULL;
+  char   *val = NULL;
   int32_t len = 0;
 
   pCurr = streamStateSeekKeyNext_rocksdb(p, &winkey);
@@ -127,14 +133,14 @@ void *backendOpen() {
   }
   for (int i = 0; i < size; i++) {
     STupleKey key = {.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
-    char *    val = NULL;
+    char     *val = NULL;
     int32_t   len = 0;
     streamStateFuncGet_rocksdb(p, &key, (void **)&val, &len);
     ASSERT(len == strlen("Value"));
   }
   for (int i = 0; i < size; i++) {
     STupleKey key = {.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
-    char *    val = NULL;
+    char     *val = NULL;
     int32_t   len = 0;
     streamStateFuncDel_rocksdb(p, &key);
   }
@@ -172,7 +178,7 @@ void *backendOpen() {
 
   {
     SSessionKey key;
-    char *      val = NULL;
+    char       *val = NULL;
     int32_t     vlen = 0;
     code = streamStateSessionGetKVByCur_rocksdb(pCurr, &key, (void **)&val, &vlen);
     ASSERT(code == 0);
@@ -212,7 +218,7 @@ void *backendOpen() {
   }
   for (int i = 0; i < size; i++) {
     SWinKey key = {.groupId = (uint64_t)(i), .ts = tsArray[i]};
-    char *  val = NULL;
+    char   *val = NULL;
     int32_t vlen = 0;
     ASSERT(streamStateFillGet_rocksdb(p, &key, (void **)&val, &vlen) == 0);
     taosMemoryFreeClear(val);
@@ -222,7 +228,7 @@ void *backendOpen() {
     SStreamStateCur *pCurr = streamStateFillGetCur_rocksdb(p, &key);
     ASSERT(pCurr != NULL);
 
-    char *  val = NULL;
+    char   *val = NULL;
     int32_t vlen = 0;
     ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));
     ASSERT(vlen == strlen("Value"));
@@ -244,7 +250,7 @@ void *backendOpen() {
 
   for (int i = 0; i < size - 1; i++) {
     SWinKey key = {.groupId = (uint64_t)(i), .ts = tsArray[i]};
-    char *  val = NULL;
+    char   *val = NULL;
     int32_t vlen = 0;
     ASSERT(streamStateFillDel_rocksdb(p, &key) == 0);
     taosMemoryFreeClear(val);
@@ -286,7 +292,7 @@ void *backendOpen() {
     char key[128] = {0};
     sprintf(key, "tbname_%d", i);
 
-    char *  val = NULL;
+    char   *val = NULL;
     int32_t len = 0;
     code = streamDefaultGet_rocksdb(p, key, (void **)&val, &len);
     ASSERT(code == 0);
@@ -302,7 +308,7 @@ TEST_F(BackendEnv, checkOpen) {
   SStreamState *p = (SStreamState *)backendOpen();
   int64_t       tsStart = taosGetTimestampMs();
   {
-    void *  pBatch = streamStateCreateBatch();
+    void   *pBatch = streamStateCreateBatch();
     int32_t size = 0;
     for (int i = 0; i < size; i++) {
       char key[128] = {0};
@@ -316,7 +322,7 @@ TEST_F(BackendEnv, checkOpen) {
     streamStateDestroyBatch(pBatch);
   }
   {
-    void *  pBatch = streamStateCreateBatch();
+    void   *pBatch = streamStateCreateBatch();
     int32_t size = 0;
     char    valBuf[256] = {0};
     for (int i = 0; i < size; i++) {
@@ -333,7 +339,7 @@ TEST_F(BackendEnv, checkOpen) {
   // do checkpoint 2
   taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 2);
   {
-    void *  pBatch = streamStateCreateBatch();
+    void   *pBatch = streamStateCreateBatch();
     int32_t size = 0;
     char    valBuf[256] = {0};
     for (int i = 0; i < size; i++) {
@@ -355,7 +361,7 @@ TEST_F(BackendEnv, checkOpen) {
   // taosMkDir(dump);
   taosMulMkDir(dump);
   SBkdMgt *mgt = bkdMgtCreate((char *)path);
-  SArray * result = taosArrayInit(4, sizeof(void *));
+  SArray  *result = taosArrayInit(4, sizeof(void *));
   bkdMgtGetDelta(mgt, p->pTdbState->idstr, 3, result, (char *)dump);
 
   bkdMgtDestroy(mgt);

From d5c5dc70269cb8fa498d799d6797361d73b09962 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Wed, 28 Feb 2024 06:45:35 +0000
Subject: [PATCH 15/24] add test case

---
 source/libs/stream/test/backendTest.cpp | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
index 3245979633..9165beaaec 100644
--- a/source/libs/stream/test/backendTest.cpp
+++ b/source/libs/stream/test/backendTest.cpp
@@ -178,8 +178,9 @@ void *backendOpen() {
 
   {
     SSessionKey key;
-    char       *val = NULL;
-    int32_t     vlen = 0;
+    memset(&key, 0, sizeof(key));
+    char   *val = NULL;
+    int32_t vlen = 0;
     code = streamStateSessionGetKVByCur_rocksdb(pCurr, &key, (void **)&val, &vlen);
     ASSERT(code == 0);
     pCurr = streamStateSessionSeekKeyPrev_rocksdb(p, &key);

From 64f2dc607a5b698bfbcc66b60d2b67cb94c77e52 Mon Sep 17 00:00:00 2001
From: Minglei Jin <mljin@taosdata.com>
Date: Wed, 28 Feb 2024 15:08:48 +0800
Subject: [PATCH 16/24] fix(tdb/pcache): not use lru page if already in hash

---
 source/libs/tdb/src/db/tdbPCache.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/libs/tdb/src/db/tdbPCache.c b/source/libs/tdb/src/db/tdbPCache.c
index 3ee65f11dd..455128e6ec 100644
--- a/source/libs/tdb/src/db/tdbPCache.c
+++ b/source/libs/tdb/src/db/tdbPCache.c
@@ -316,7 +316,7 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, TXN *pTxn)
   }
 
   // 3. Try to Recycle a page
-  if (!pPage && !pCache->lru.pLruPrev->isAnchor) {
+  if (!pPageH && !pPage && !pCache->lru.pLruPrev->isAnchor) {
     pPage = pCache->lru.pLruPrev;
     tdbPCacheRemovePageFromHash(pCache, pPage);
     tdbPCachePinPage(pCache, pPage);

From 249141a9a25ba5c04cef6a8ab4a6b1b8c9590983 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Wed, 28 Feb 2024 08:16:02 +0000
Subject: [PATCH 17/24] add compress dict

---
 source/libs/stream/test/backendTest.cpp | 35 ++++++++++++++++++-------
 1 file changed, 26 insertions(+), 9 deletions(-)

diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
index 9165beaaec..b07c860da8 100644
--- a/source/libs/stream/test/backendTest.cpp
+++ b/source/libs/stream/test/backendTest.cpp
@@ -16,6 +16,10 @@
 #pragma GCC diagnostic ignored "-Wunused-function"
 #pragma GCC diagnostic ignored "-Wunused-variable"
 #pragma GCC diagnostic ignored "-Wsign-compare"
+#pragma GCC diagnostic ignored "-Wsign-compare"
+#pragma GCC diagnostic ignored "-Wformat"
+#pragma GCC diagnostic ignored "-Wint-to-pointer-cast"
+#pragma GCC diagnostic ignored "-Wpointer-arith"
 
 class BackendEnv : public ::testing::Test {
  protected:
@@ -56,18 +60,22 @@ void *backendOpen() {
   // default/state/fill/sess/func/parname/partag
   int32_t              size = 100;
   std::vector<int64_t> tsArray;
-  for (int i = 0; i < size; i++) {
-    int64_t     ts = taosGetTimestampMs();
-    SWinKey     key = {.groupId = (uint64_t)(i), .ts = ts};
+  for (int32_t i = 0; i < size; i++) {
+    int64_t ts = taosGetTimestampMs();
+    SWinKey key;  // = {.groupId = (uint64_t)(i), .ts = ts};
+    key.groupId = (uint64_t)(i);
+    key.ts = ts;
     const char *val = "value data";
     int32_t     vlen = strlen(val);
     streamStatePut_rocksdb(p, &key, (char *)val, vlen);
 
     tsArray.push_back(ts);
   }
-  for (int i = 0; i < size; i++) {
+  for (int32_t i = 0; i < size; i++) {
     int64_t ts = tsArray[i];
-    SWinKey key = {.groupId = (uint64_t)(i), .ts = ts};
+    SWinKey key = {0};  //{.groupId = (uint64_t)(i), .ts = ts};
+    key.groupId = (uint64_t)(i);
+    key.ts = ts;
 
     const char *val = "value data";
     int32_t     len = 0;
@@ -76,14 +84,20 @@ void *backendOpen() {
     ASSERT(len == strlen(val));
   }
   int64_t ts = tsArray[0];
-  SWinKey key = {.groupId = (uint64_t)(0), .ts = ts};
+  SWinKey key = {0};  // {.groupId = (uint64_t)(0), .ts = ts};
+  key.groupId = (uint64_t)(0);
+  key.ts = ts;
+
   streamStateDel_rocksdb(p, &key);
 
   streamStateClear_rocksdb(p);
 
   for (int i = 0; i < size; i++) {
-    int64_t     ts = tsArray[i];
-    SWinKey     key = {.groupId = (uint64_t)(i), .ts = ts};
+    int64_t ts = tsArray[i];
+    SWinKey key = {0};  //{.groupId = (uint64_t)(i), .ts = ts};
+    key.groupId = (uint64_t)(i);
+    key.ts = ts;
+
     const char *val = "value data";
     int32_t     len = 0;
     char       *newVal = NULL;
@@ -96,7 +110,10 @@ void *backendOpen() {
     int64_t ts = taosGetTimestampMs();
     tsArray.push_back(ts);
 
-    SWinKey     key = {.groupId = (uint64_t)(i), .ts = ts};
+    SWinKey key = {0};  //{.groupId = (uint64_t)(i), .ts = ts};
+    key.groupId = (uint64_t)(i);
+    key.ts = ts;
+
     const char *val = "value data";
     int32_t     vlen = strlen(val);
     streamStatePut_rocksdb(p, &key, (char *)val, vlen);

From 00d6611ba77323bf4a403166f52b666d884d2e59 Mon Sep 17 00:00:00 2001
From: dmchen <cademfly@hotmail.com>
Date: Wed, 28 Feb 2024 08:31:45 +0000
Subject: [PATCH 18/24] fix/TD-28895

---
 source/dnode/mnode/impl/src/mndCompact.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/source/dnode/mnode/impl/src/mndCompact.c b/source/dnode/mnode/impl/src/mndCompact.c
index deaaf7f2af..de45e8fe9a 100644
--- a/source/dnode/mnode/impl/src/mndCompact.c
+++ b/source/dnode/mnode/impl/src/mndCompact.c
@@ -369,7 +369,7 @@ static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompa
     return -1;
   }
   mInfo("trans:%d, used to kill compact:%" PRId32, pTrans->id, pCompact->compactId);
-
+/*
   SSdbRaw *pCommitRaw = mndCompactActionEncode(pCompact);
   if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
     mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
@@ -377,7 +377,7 @@ static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompa
     return -1;
   }
   (void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
-
+*/
   void   *pIter = NULL;
   while (1) {
     SCompactDetailObj *pDetail = NULL;

From 8c46c71c98f0eb190be98e0bd556f0a6d848ff11 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Wed, 28 Feb 2024 09:02:55 +0000
Subject: [PATCH 19/24] add test case

---
 source/libs/stream/test/backendTest.cpp | 56 ++++++++++++++++++-------
 1 file changed, 42 insertions(+), 14 deletions(-)

diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
index b07c860da8..701ea0a75b 100644
--- a/source/libs/stream/test/backendTest.cpp
+++ b/source/libs/stream/test/backendTest.cpp
@@ -142,23 +142,34 @@ void *backendOpen() {
   for (int i = 0; i < size; i++) {
     int64_t ts = taosGetTimestampMs();
     tsArray.push_back(ts);
-    STupleKey key = {.groupId = (uint64_t)(0), .ts = ts, .exprIdx = i};
+    STupleKey key = {0};
+    key.groupId = (uint64_t)(0);  //= {.groupId = (uint64_t)(0), .ts = ts, .exprIdx = i};
+    key.ts = ts;
+    key.exprIdx = i;
 
     const char *val = "Value";
     int32_t     len = strlen(val);
     streamStateFuncPut_rocksdb(p, &key, val, len);
   }
   for (int i = 0; i < size; i++) {
-    STupleKey key = {.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
-    char     *val = NULL;
-    int32_t   len = 0;
+    STupleKey key = {0};  //{.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
+    key.groupId = (uint64_t)(0);
+    key.ts = tsArray[i];
+    key.exprIdx = i;
+
+    char   *val = NULL;
+    int32_t len = 0;
     streamStateFuncGet_rocksdb(p, &key, (void **)&val, &len);
     ASSERT(len == strlen("Value"));
   }
   for (int i = 0; i < size; i++) {
-    STupleKey key = {.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
-    char     *val = NULL;
-    int32_t   len = 0;
+    STupleKey key = {0};  //{.groupId = (uint64_t)(0), .ts = tsArray[i], .exprIdx = i};
+    key.groupId = (uint64_t)(0);
+    key.ts = tsArray[i];
+    key.exprIdx = i;
+
+    char   *val = NULL;
+    int32_t len = 0;
     streamStateFuncDel_rocksdb(p, &key);
   }
 
@@ -166,7 +177,10 @@ void *backendOpen() {
   tsArray.clear();
 
   for (int i = 0; i < size; i++) {
-    SSessionKey key = {.win = {.skey = i, .ekey = i}, .groupId = (uint64_t)(0)};
+    SSessionKey key = {0};  //{.win = {.skey = i, .ekey = i}, .groupId = (uint64_t)(0)};
+    key.win.skey = i;
+    key.win.ekey = i;
+    key.groupId = (uint64_t)(0);
     tsArray.push_back(i);
 
     const char *val = "Value";
@@ -179,7 +193,10 @@ void *backendOpen() {
   }
 
   for (int i = 0; i < size; i++) {
-    SSessionKey key = {.win = {.skey = tsArray[i], .ekey = tsArray[i]}, .groupId = (uint64_t)(0)};
+    SSessionKey key = {0};  //{.win = {.skey = tsArray[i], .ekey = tsArray[i]}, .groupId = (uint64_t)(0)};
+    key.win.skey = tsArray[i];
+    key.win.ekey = tsArray[i];
+    key.groupId = (uint64_t)(0);
 
     const char *val = "Value";
     int32_t     len = strlen(val);
@@ -219,7 +236,10 @@ void *backendOpen() {
            streamStateStateAddIfNotExist_rocksdb(p, &key, (char *)"key", strlen("key"), NULL, (void **)&val, &len));
   }
   for (int i = 0; i < size; i++) {
-    SSessionKey key = {.win = {.skey = tsArray[i], .ekey = tsArray[i]}, .groupId = (uint64_t)(0)};
+    SSessionKey key = {0};  //{.win = {.skey = tsArray[i], .ekey = tsArray[i]}, .groupId = (uint64_t)(0)};
+    key.win.skey = tsArray[i];
+    key.win.ekey = tsArray[i];
+    key.groupId = (uint64_t)(0);
 
     const char *val = "Value";
     int32_t     len = strlen(val);
@@ -229,20 +249,26 @@ void *backendOpen() {
   }
 
   for (int i = 0; i < size; i++) {
-    SWinKey     key = {.groupId = (uint64_t)(i), .ts = tsArray[i]};
+    SWinKey key = {0};  // {.groupId = (uint64_t)(i), .ts = tsArray[i]};
+    key.groupId = (uint64_t)(i);
+    key.ts = tsArray[i];
     const char *val = "Value";
     int32_t     vlen = strlen(val);
     ASSERT(streamStateFillPut_rocksdb(p, &key, val, vlen) == 0);
   }
   for (int i = 0; i < size; i++) {
-    SWinKey key = {.groupId = (uint64_t)(i), .ts = tsArray[i]};
+    SWinKey key = {0};  // {.groupId = (uint64_t)(i), .ts = tsArray[i]};
+    key.groupId = (uint64_t)(i);
+    key.ts = tsArray[i];
     char   *val = NULL;
     int32_t vlen = 0;
     ASSERT(streamStateFillGet_rocksdb(p, &key, (void **)&val, &vlen) == 0);
     taosMemoryFreeClear(val);
   }
   {
-    SWinKey          key = {.groupId = (uint64_t)(0), .ts = tsArray[0]};
+    SWinKey key = {0};  //{.groupId = (uint64_t)(0), .ts = tsArray[0]};
+    key.groupId = (uint64_t)(0);
+    key.ts = tsArray[0];
     SStreamStateCur *pCurr = streamStateFillGetCur_rocksdb(p, &key);
     ASSERT(pCurr != NULL);
 
@@ -267,7 +293,9 @@ void *backendOpen() {
   }
 
   for (int i = 0; i < size - 1; i++) {
-    SWinKey key = {.groupId = (uint64_t)(i), .ts = tsArray[i]};
+    SWinKey key = {0};  // {.groupId = (uint64_t)(i), .ts = tsArray[i]};
+    key.groupId = (uint64_t)(i);
+    key.ts = tsArray[i];
     char   *val = NULL;
     int32_t vlen = 0;
     ASSERT(streamStateFillDel_rocksdb(p, &key) == 0);

From 17e9f1d747c4e5371f2af813dabe70c120eafbea Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Wed, 28 Feb 2024 09:33:48 +0000
Subject: [PATCH 20/24] add test case

---
 source/libs/stream/test/backendTest.cpp | 13 ++-----------
 1 file changed, 2 insertions(+), 11 deletions(-)

diff --git a/source/libs/stream/test/backendTest.cpp b/source/libs/stream/test/backendTest.cpp
index 701ea0a75b..a949748eb5 100644
--- a/source/libs/stream/test/backendTest.cpp
+++ b/source/libs/stream/test/backendTest.cpp
@@ -422,17 +422,8 @@ typedef struct BdKV {
   uint32_t v;
 } BdKV;
 
-BdKV kvDict[] = {{0, 2},
-                 {1, 2},
-                 {.k = 15, .v = 16},
-                 {.k = 31, .v = 32},
-                 {.k = 56, .v = 64},
-                 {.k = 100, .v = 128},
-                 {.k = 200, .v = 256},
-                 {.k = 500, .v = 512},
-                 {.k = 1000, .v = 1024},
-                 {.k = 2000, .v = 2048},
-                 {.k = 3000, .v = 4096}};
+BdKV kvDict[] = {{0, 2},     {1, 2},     {15, 16},     {31, 32},     {56, 64},    {100, 128},
+                 {200, 256}, {500, 512}, {1000, 1024}, {2000, 2048}, {3000, 4096}};
 
 TEST_F(BackendEnv, backendUtil) {
   for (int i = 0; i < sizeof(kvDict) / sizeof(kvDict[0]); i++) {

From dd7f375d8e6e4b1c31d4eee6d22e9b97e12da85f Mon Sep 17 00:00:00 2001
From: Benguang Zhao <bgzhao@taosdata.com>
Date: Wed, 28 Feb 2024 19:01:22 +0800
Subject: [PATCH 21/24] enh: use reference id of sync node in timer callbacks
 of sync

---
 source/libs/sync/src/syncMain.c | 25 +++++++++++++++++--------
 1 file changed, 17 insertions(+), 8 deletions(-)

diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c
index e06ea70f70..7ff6116137 100644
--- a/source/libs/sync/src/syncMain.c
+++ b/source/libs/sync/src/syncMain.c
@@ -1343,7 +1343,7 @@ ESyncStrategy syncNodeStrategy(SSyncNode* pSyncNode) { return pSyncNode->raftCfg
 int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
   int32_t ret = 0;
   if (syncIsInit()) {
-    taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, pSyncNode, syncEnv()->pTimerManager,
+    taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, (void*)pSyncNode->rid, syncEnv()->pTimerManager,
                  &pSyncNode->pPingTimer);
     atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
   } else {
@@ -1415,8 +1415,8 @@ void syncNodeResetElectTimer(SSyncNode* pSyncNode) {
 static int32_t syncNodeDoStartHeartbeatTimer(SSyncNode* pSyncNode) {
   int32_t ret = 0;
   if (syncIsInit()) {
-    taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, pSyncNode, syncEnv()->pTimerManager,
-                 &pSyncNode->pHeartbeatTimer);
+    taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, (void*)pSyncNode->rid,
+                 syncEnv()->pTimerManager, &pSyncNode->pHeartbeatTimer);
     atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
   } else {
     sError("vgId:%d, start heartbeat timer error, sync env is stop", pSyncNode->vgId);
@@ -2153,7 +2153,11 @@ int32_t syncNodeGetPreIndexTerm(SSyncNode* pSyncNode, SyncIndex index, SyncIndex
 static void syncNodeEqPingTimer(void* param, void* tmrId) {
   if (!syncIsInit()) return;
 
-  SSyncNode* pNode = param;
+  int64_t    rid = (int64_t)param;
+  SSyncNode* pNode = syncNodeAcquire(rid);
+
+  if (pNode == NULL) return;
+
   if (atomic_load_64(&pNode->pingTimerLogicClockUser) <= atomic_load_64(&pNode->pingTimerLogicClock)) {
     SRpcMsg rpcMsg = {0};
     int32_t code = syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_PING, atomic_load_64(&pNode->pingTimerLogicClock),
@@ -2173,7 +2177,8 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) {
     }
 
   _out:
-    taosTmrReset(syncNodeEqPingTimer, pNode->pingTimerMS, pNode, syncEnv()->pTimerManager, &pNode->pPingTimer);
+    taosTmrReset(syncNodeEqPingTimer, pNode->pingTimerMS, (void*)pNode->rid, syncEnv()->pTimerManager,
+                 &pNode->pPingTimer);
   }
 }
 
@@ -2224,7 +2229,11 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) {
 static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
   if (!syncIsInit()) return;
 
-  SSyncNode* pNode = param;
+  int64_t    rid = (int64_t)param;
+  SSyncNode* pNode = syncNodeAcquire(rid);
+
+  if (pNode == NULL) return;
+
   if (pNode->totalReplicaNum > 1) {
     if (atomic_load_64(&pNode->heartbeatTimerLogicClockUser) <= atomic_load_64(&pNode->heartbeatTimerLogicClock)) {
       SRpcMsg rpcMsg = {0};
@@ -2245,7 +2254,7 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
       }
 
     _out:
-      taosTmrReset(syncNodeEqHeartbeatTimer, pNode->heartbeatTimerMS, pNode, syncEnv()->pTimerManager,
+      taosTmrReset(syncNodeEqHeartbeatTimer, pNode->heartbeatTimerMS, (void*)pNode->rid, syncEnv()->pTimerManager,
                    &pNode->pHeartbeatTimer);
 
     } else {
@@ -3385,4 +3394,4 @@ bool syncNodeCanChange(SSyncNode* pSyncNode) {
 
   return true;
 }
-#endif
\ No newline at end of file
+#endif

From e81afaf0a2ae14aca11dab5f778f8ba3ee3b8efd Mon Sep 17 00:00:00 2001
From: slzhou <shenglian_zhou@163.com>
Date: Thu, 29 Feb 2024 09:12:15 +0800
Subject: [PATCH 22/24] fix: no memory trim when no para tables scan

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

diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index 8cc2f72adb..9024f7a341 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -3763,8 +3763,9 @@ static int32_t stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {
 
     taosMemoryFree(pSubTblsInfo);
     pInfo->pSubTablesMergeInfo = NULL;
+
+    taosMemoryTrim(0);
   }
-  taosMemoryTrim(0);
   return TSDB_CODE_SUCCESS;
 }
 

From 85b78b0947285c26d85ec0e1580345e4203de4f6 Mon Sep 17 00:00:00 2001
From: dmchen <cademfly@hotmail.com>
Date: Thu, 29 Feb 2024 02:43:15 +0000
Subject: [PATCH 23/24] fix/TD-28910

---
 source/dnode/mnode/impl/src/mndCompact.c | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/source/dnode/mnode/impl/src/mndCompact.c b/source/dnode/mnode/impl/src/mndCompact.c
index de45e8fe9a..83254b0f73 100644
--- a/source/dnode/mnode/impl/src/mndCompact.c
+++ b/source/dnode/mnode/impl/src/mndCompact.c
@@ -363,13 +363,15 @@ static int32_t mndAddKillCompactAction(SMnode *pMnode, STrans *pTrans, SVgObj *p
 }
 
 static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompact) {
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq, "kill-compact");
+  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, pReq, "kill-compact");
   if (pTrans == NULL) {
     mError("compact:%" PRId32 ", failed to drop since %s" , pCompact->compactId, terrstr());
     return -1;
   }
   mInfo("trans:%d, used to kill compact:%" PRId32, pTrans->id, pCompact->compactId);
-/*
+
+  mndTransSetDbName(pTrans, pCompact->dbname, NULL);
+
   SSdbRaw *pCommitRaw = mndCompactActionEncode(pCompact);
   if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
     mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
@@ -377,8 +379,8 @@ static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompa
     return -1;
   }
   (void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
-*/
-  void   *pIter = NULL;
+
+  void *pIter = NULL;
   while (1) {
     SCompactDetailObj *pDetail = NULL;
     pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
@@ -612,15 +614,17 @@ static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) {
     return 0;
   }
 
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, NULL, "update-compact-progress");
+  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, NULL, "update-compact-progress");
   if (pTrans == NULL) {
     mError("trans:%" PRId32 ", failed to create since %s" , pTrans->id, terrstr());
     return -1;
   }
   mInfo("compact:%d, trans:%d, used to update compact progress.", compactId, pTrans->id);
-
+  
   SCompactObj   *pCompact = mndAcquireCompact(pMnode, compactId);
 
+  mndTransSetDbName(pTrans, pCompact->dbname, NULL);
+
   pIter = NULL;
   while (1) {
     SCompactDetailObj *pDetail = NULL;

From b744eabce0cd0cc6c5f50f14c01d35888fb5f624 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Thu, 29 Feb 2024 11:34:57 +0800
Subject: [PATCH 24/24] fix(util): fix the new size remaining bug.

---
 source/util/src/tarray.c | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/source/util/src/tarray.c b/source/util/src/tarray.c
index 88eb51d500..d9686d77f8 100644
--- a/source/util/src/tarray.c
+++ b/source/util/src/tarray.c
@@ -89,12 +89,14 @@ static int32_t taosArrayResize(SArray* pArray) {
 int32_t taosArrayEnsureCap(SArray* pArray, size_t newCap) {
   if (newCap > pArray->capacity) {
     float factor = BOUNDARY_BIG_FACTOR;
-    if(newCap * pArray->elemSize > BOUNDARY_SIZE){
+    if (newCap * pArray->elemSize > BOUNDARY_SIZE) {
       factor = BOUNDARY_SMALL_FACTOR;
     }
+
     size_t tsize = (pArray->capacity * factor);
     while (newCap > tsize) {
-      tsize = (tsize * factor);
+      size_t newSize = (tsize * factor);
+      tsize = (newSize == tsize) ? (tsize + 2) : newSize;
     }
 
     pArray->pData = taosMemoryRealloc(pArray->pData, tsize * pArray->elemSize);