Merge branch 'unstable' into feat/set-ifeq-ifne-ifdeq-ifdne

kirito632 · web-flow · commit b6d4d2542c95 · 2026-04-11T16:20:45.000+08:00
diff --git a/.github/workflows/kvrocks.yaml b/.github/workflows/kvrocks.yaml
@@ -40,7 +40,7 @@ jobs:
       docs_only: ${{ steps.result.outputs.docs_only }}
     steps:
       - uses: actions/checkout@v6
-      - uses: dorny/paths-filter@de90cc6fb38fc0963ad72b210f1f284cd68cea36
+      - uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d
         id: changes
         with:
           filters: .github/config/changes.yml
@@ -61,7 +61,7 @@ jobs:
         uses: crate-ci/typos@v1.45.0
         with:
           config: .github/config/typos.toml
-      - uses: apache/skywalking-eyes/header@v0.7.0
+      - uses: apache/skywalking-eyes/header@v0.8.0
         with:
           config: .github/config/licenserc.yml
 
@@ -297,14 +297,14 @@ jobs:
         with:
           path: |
             ~/local/bin/redis-cli
-          key: ${{ runner.os }}-${{ runner.arch }}-redis-cli
+          key: ${{ matrix.os }}-redis-cli
       - name: Cache redis server
         id: cache-redis-server
         uses: actions/cache@v5
         with:
           path: |
             ~/local/bin/redis-server
-          key: ${{ runner.os }}-${{ runner.arch }}-redis-server
+          key: ${{ matrix.os }}-redis-server
       - name: Install redis
         if: ${{ steps.cache-redis.outputs.cache-hit != 'true' || steps.cache-redis-server.outputs.cache-hit != 'true' }}
         run: |
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -30,16 +30,11 @@ option(ASAN_WITH_LSAN "enable leak sanitizer while address sanitizer is enabled"
 option(ENABLE_STATIC_LIBSTDCXX "link kvrocks with static library of libstd++ instead of shared library" ON)
 option(ENABLE_LUAJIT "enable use of luaJIT instead of lua" ON)
 option(ENABLE_OPENSSL "enable openssl to support tls connection" OFF)
-option(ENABLE_LTO "enable link-time optimization" ON)
+option(ENABLE_LTO "enable link-time optimization" OFF)
 set(SYMBOLIZE_BACKEND "" CACHE STRING "symbolization backend library for cpptrace (libbacktrace, libdwarf, or empty)")
 set(PORTABLE 0 CACHE STRING "build a portable binary (disable arch-specific optimizations)")
 option(ENABLE_NEW_ENCODING "enable new encoding (#1033) for storing 64bit size and expire time in milliseconds" ON)
 
-# to save build time in debug mode
-if (CMAKE_BUILD_TYPE STREQUAL "Debug")
-    set(ENABLE_LTO OFF)
-endif()
-
 if (CMAKE_VERSION VERSION_GREATER_EQUAL "3.24.0")
     cmake_policy(SET CMP0135 NEW)
 endif()
@@ -144,7 +139,6 @@ endif()
 
 if (CMAKE_HOST_APPLE)
     set(DISABLE_JEMALLOC ON)
-    set(ENABLE_LTO OFF)
 endif ()
 
 if(NOT DISABLE_JEMALLOC)
@@ -298,11 +292,6 @@ else()
     target_compile_definitions(kvrocks_objs PUBLIC METADATA_ENCODING_VERSION=0)
 endif()
 
-# disable LTO on GCC <= 9 due to an ICE
-if((CMAKE_CXX_COMPILER_ID STREQUAL "GNU") AND (CMAKE_CXX_COMPILER_VERSION VERSION_LESS 10))
-    set(ENABLE_LTO OFF)
-endif()
-
 if(ENABLE_LTO)
     include(CheckIPOSupported)
     check_ipo_supported(RESULT ipo_result OUTPUT ipo_output LANGUAGES CXX)
diff --git a/cmake/lua.cmake b/cmake/lua.cmake
@@ -20,8 +20,8 @@ include_guard()
 include(cmake/utils.cmake)
 
 FetchContent_DeclareGitHubWithMirror(lua
-  RocksLabs/lua f458c3d797db31155fa0c156d5301716df48cb8c
-  MD5=c7c4deb9f750d8f2bef0044a701df85c
+  RocksLabs/lua c77781d0c23df021a903a9a6199d301ec559e9cd
+  MD5=8e9f463af29c0fe17baea52f21e5b790
 )
 
 FetchContent_GetProperties(lua)
diff --git a/kvrocks.conf b/kvrocks.conf
@@ -923,6 +923,19 @@ rocksdb.compression snappy
 #   compression library as mentioned above)
 rocksdb.compression_level 32767
 
+# Maximum per-SST dictionary size for block compression. Set to 0 to disable
+# dictionary compression.
+#
+# Default: 0
+rocksdb.compression_max_dict_bytes 0
+
+# Maximum amount of training data used by the ZSTD dictionary trainer. This is
+# most useful when rocksdb.compression is set to zstd and rocksdb.compression_max_dict_bytes
+# is non-zero.
+#
+# Default: 0
+rocksdb.compression_zstd_max_train_bytes 0
+
 # If non-zero, we perform bigger reads when doing compaction. If you're
 # running RocksDB on spinning disks, you should set this to at least 2MB.
 # That way RocksDB's compaction is doing sequential instead of random reads.
diff --git a/src/commands/cmd_stream.cc b/src/commands/cmd_stream.cc
@@ -566,17 +566,13 @@ class CommandXGroup : public Commander {
     }
 
     if (subcommand_ == "destroy") {
-      uint64_t delete_cnt = 0;
-      auto s = stream_db.DestroyGroup(ctx, stream_name_, group_name_, &delete_cnt);
+      bool destroyed = false;
+      auto s = stream_db.DestroyGroup(ctx, stream_name_, group_name_, &destroyed);
       if (!s.ok()) {
         return {Status::RedisExecErr, s.ToString()};
       }
 
-      if (delete_cnt > 0) {
-        *output = redis::Integer(1);
-      } else {
-        *output = redis::Integer(0);
-      }
+      *output = redis::Integer(destroyed ? 1 : 0);
     }
 
     if (subcommand_ == "createconsumer") {
@@ -1891,7 +1887,7 @@ REDIS_REGISTER_COMMANDS(Stream, MakeCmdAttr<CommandXAck>("xack", -4, "write no-d
                         MakeCmdAttr<CommandXInfo>("xinfo", -2, "read-only", NO_KEY),
                         MakeCmdAttr<CommandXPending>("xpending", -3, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandXRange>("xrange", -4, "read-only", 1, 1, 1),
-                        MakeCmdAttr<CommandXRevRange>("xrevrange", -2, "read-only", 1, 1, 1),
+                        MakeCmdAttr<CommandXRevRange>("xrevrange", -4, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandXRead>("xread", -4, "read-only blocking", CommandXRead::keyRangeGen),
                         MakeCmdAttr<CommandXReadGroup>("xreadgroup", -7, "write blocking",
                                                        CommandXReadGroup::keyRangeGen),
diff --git a/src/config/config.cc b/src/config/config.cc
@@ -254,6 +254,10 @@ Config::Config() {
        new EnumField<rocksdb::CompressionType>(&rocks_db.compression, compression_types,
                                                rocksdb::CompressionType::kNoCompression)},
       {"rocksdb.compression_level", true, new IntField(&rocks_db.compression_level, 32767, INT_MIN, INT_MAX)},
+      {"rocksdb.compression_max_dict_bytes", true,
+       new UInt32Field(&rocks_db.compression_max_dict_bytes, 0, 0, UINT32_MAX)},
+      {"rocksdb.compression_zstd_max_train_bytes", true,
+       new UInt32Field(&rocks_db.compression_zstd_max_train_bytes, 0, 0, UINT32_MAX)},
       {"rocksdb.compression_start_level", false,
        new IntField(&rocks_db.compression_start_level, 2, 0, KVROCKS_MAX_LSM_LEVEL - 1)},
       {"rocksdb.block_size", true, new IntField(&rocks_db.block_size, 16384, 0, INT_MAX)},
diff --git a/src/config/config.h b/src/config/config.h
@@ -236,6 +236,8 @@ struct Config {
     rocksdb::CompressionType compression;
     int compression_start_level;
     int compression_level;
+    uint32_t compression_max_dict_bytes;
+    uint32_t compression_zstd_max_train_bytes;
     bool disable_auto_compactions;
     bool enable_blob_files;
     int min_blob_size;
diff --git a/src/storage/storage.cc b/src/storage/storage.cc
@@ -178,6 +178,8 @@ rocksdb::Options Storage::InitRocksDBOptions() {
   options.write_buffer_size = config_->rocks_db.write_buffer_size * MiB;
   options.num_levels = KVROCKS_MAX_LSM_LEVEL;
   options.compression_opts.level = config_->rocks_db.compression_level;
+  options.compression_opts.max_dict_bytes = config_->rocks_db.compression_max_dict_bytes;
+  options.compression_opts.zstd_max_train_bytes = config_->rocks_db.compression_zstd_max_train_bytes;
   options.compression_per_level.resize(options.num_levels);
   options.wal_compression = config_->rocks_db.wal_compression;
   for (int i = 0; i < options.num_levels; ++i) {
diff --git a/src/types/redis_geo.cc b/src/types/redis_geo.cc
@@ -421,6 +421,6 @@ bool Geo::appendIfWithinShape(std::vector<GeoPoint> *geo_points, const GeoShape
 
 bool Geo::sortGeoPointASC(const GeoPoint &gp1, const GeoPoint &gp2) { return gp1.dist < gp2.dist; }
 
-bool Geo::sortGeoPointDESC(const GeoPoint &gp1, const GeoPoint &gp2) { return gp1.dist >= gp2.dist; }
+bool Geo::sortGeoPointDESC(const GeoPoint &gp1, const GeoPoint &gp2) { return gp1.dist > gp2.dist; }
 
 }  // namespace redis
diff --git a/src/types/redis_hash.cc b/src/types/redis_hash.cc
@@ -28,6 +28,7 @@
 #include <random>
 #include <utility>
 
+#include "common/string_util.h"
 #include "db_util.h"
 #include "parse_util.h"
 #include "sample_helper.h"
@@ -144,7 +145,7 @@ rocksdb::Status Hash::IncrByFloat(engine::Context &ctx, const Slice &user_key, c
   WriteBatchLogData log_data(kRedisHash);
   s = batch->PutLogData(log_data.Encode());
   if (!s.ok()) return s;
-  s = batch->Put(sub_key, std::to_string(*new_value));
+  s = batch->Put(sub_key, util::Float2String(*new_value));
   if (!s.ok()) return s;
   if (!exists) {
     metadata.size += 1;
diff --git a/src/types/redis_set.cc b/src/types/redis_set.cc
@@ -78,6 +78,7 @@ rocksdb::Status Set::Add(engine::Context &ctx, const Slice &user_key, const std:
     std::string sub_key = InternalKey(ns_key, member, metadata.version, storage_->IsSlotIdEncoded()).Encode();
     s = storage_->Get(ctx, ctx.GetReadOptions(), sub_key, &value);
     if (s.ok()) continue;
+    if (!s.IsNotFound()) return s;
     s = batch->Put(sub_key, Slice());
     if (!s.ok()) return s;
     *added_cnt += 1;
@@ -114,7 +115,8 @@ rocksdb::Status Set::Remove(engine::Context &ctx, const Slice &user_key, const s
     }
     std::string sub_key = InternalKey(ns_key, member, metadata.version, storage_->IsSlotIdEncoded()).Encode();
     s = storage_->Get(ctx, ctx.GetReadOptions(), sub_key, &value);
-    if (!s.ok()) continue;
+    if (s.IsNotFound()) continue;
+    if (!s.ok()) return s;
     s = batch->Delete(sub_key);
     if (!s.ok()) return s;
     *removed_cnt += 1;
diff --git a/src/types/redis_stream.cc b/src/types/redis_stream.cc
@@ -27,6 +27,7 @@
 #include <vector>
 
 #include "db_util.h"
+#include "string_util.h"
 #include "time_util.h"
 
 namespace redis {
@@ -737,8 +738,8 @@ rocksdb::Status Stream::CreateGroup(engine::Context &ctx, const Slice &stream_na
 }
 
 rocksdb::Status Stream::DestroyGroup(engine::Context &ctx, const Slice &stream_name, const std::string &group_name,
-                                     uint64_t *delete_cnt) {
-  *delete_cnt = 0;
+                                     bool *destroyed) {
+  *destroyed = false;
   std::string ns_key = AppendNamespacePrefix(stream_name);
 
   StreamMetadata metadata;
@@ -751,42 +752,44 @@ rocksdb::Status Stream::DestroyGroup(engine::Context &ctx, const Slice &stream_n
     return rocksdb::Status::InvalidArgument(errXGroupSubcommandRequiresKeyExist);
   }
 
+  std::string group_key = internalKeyFromGroupName(ns_key, metadata, group_name);
+  std::string val;
+  s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_, group_key, &val);
+  if (s.IsNotFound()) {
+    return rocksdb::Status::OK();
+  }
+  if (!s.ok()) return s;
+
   auto batch = storage_->GetWriteBatchBase();
   WriteBatchLogData log_data(kRedisStream);
   s = batch->PutLogData(log_data.Encode());
   if (!s.ok()) return s;
 
-  std::string sub_key_prefix;
-  PutFixed64(&sub_key_prefix, group_name.size());
-  sub_key_prefix += group_name;
-  std::string next_version_prefix_key =
-      InternalKey(ns_key, sub_key_prefix, metadata.version + 1, storage_->IsSlotIdEncoded()).Encode();
-  std::string prefix_key = InternalKey(ns_key, sub_key_prefix, metadata.version, storage_->IsSlotIdEncoded()).Encode();
+  for (auto type : {StreamSubkeyType::StreamConsumerGroupMetadata, StreamSubkeyType::StreamConsumerMetadata,
+                    StreamSubkeyType::StreamPelEntry}) {
+    std::string sub_key_prefix;
+    PutFixed64(&sub_key_prefix, UINT64_MAX);
+    PutFixed8(&sub_key_prefix, static_cast<uint8_t>(type));
+    PutFixed64(&sub_key_prefix, group_name.size());
+    sub_key_prefix += group_name;
 
-  rocksdb::ReadOptions read_options = ctx.DefaultScanOptions();
-  rocksdb::Slice upper_bound(next_version_prefix_key);
-  read_options.iterate_upper_bound = &upper_bound;
-  rocksdb::Slice lower_bound(prefix_key);
-  read_options.iterate_lower_bound = &lower_bound;
+    std::string prefix_key =
+        InternalKey(ns_key, sub_key_prefix, metadata.version, storage_->IsSlotIdEncoded()).Encode();
+    std::string end_key =
+        InternalKey(ns_key, util::StringNext(sub_key_prefix), metadata.version, storage_->IsSlotIdEncoded()).Encode();
 
-  auto iter = util::UniqueIterator(ctx, read_options, stream_cf_handle_);
-  for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
-    s = batch->Delete(stream_cf_handle_, iter->key());
+    s = batch->DeleteRange(stream_cf_handle_, prefix_key, end_key);
     if (!s.ok()) return s;
-    *delete_cnt += 1;
-  }
-
-  if (auto s = iter->status(); !s.ok()) {
-    return s;
   }
 
-  if (*delete_cnt != 0) {
+  *destroyed = true;
+  if (metadata.group_number > 0) {
     metadata.group_number -= 1;
-    std::string metadata_bytes;
-    metadata.Encode(&metadata_bytes);
-    s = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes);
-    if (!s.ok()) return s;
   }
+  std::string metadata_bytes;
+  metadata.Encode(&metadata_bytes);
+  s = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes);
+  if (!s.ok()) return s;
 
   return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
 }
diff --git a/src/types/redis_stream.h b/src/types/redis_stream.h
@@ -42,7 +42,7 @@ class Stream : public SubKeyScanner {
   rocksdb::Status CreateGroup(engine::Context &ctx, const Slice &stream_name, const StreamXGroupCreateOptions &options,
                               const std::string &group_name);
   rocksdb::Status DestroyGroup(engine::Context &ctx, const Slice &stream_name, const std::string &group_name,
-                               uint64_t *delete_cnt);
+                               bool *destroyed);
   rocksdb::Status CreateConsumer(engine::Context &ctx, const Slice &stream_name, const std::string &group_name,
                                  const std::string &consumer_name, int *created_number);
   rocksdb::Status DestroyConsumer(engine::Context &ctx, const Slice &stream_name, const std::string &group_name,
diff --git a/src/types/redis_string.cc b/src/types/redis_string.cc
@@ -462,7 +462,7 @@ rocksdb::Status String::IncrByFloat(engine::Context &ctx, const std::string &use
   *new_value = n;
 
   raw_value = raw_value.substr(0, offset);
-  raw_value.append(std::to_string(n));
+  raw_value.append(util::Float2String(n));
   return updateRawValue(ctx, ns_key, raw_value);
 }
 
diff --git a/src/types/redis_zset.cc b/src/types/redis_zset.cc
@@ -706,6 +706,8 @@ rocksdb::Status ZSet::Inter(engine::Context &ctx, const std::vector<KeyWeight> &
 
 rocksdb::Status ZSet::InterCard(engine::Context &ctx, const std::vector<std::string> &user_keys, uint64_t limit,
                                 uint64_t *inter_cnt) {
+  *inter_cnt = 0;
+
   std::vector<MemberScores> mscores_list;
   mscores_list.reserve(user_keys.size());
   RangeScoreSpec spec;
diff --git a/tests/cppunit/config_test.cc b/tests/cppunit/config_test.cc
@@ -134,6 +134,8 @@ TEST(Config, GetAndSet) {
       {"rocksdb.row_cache_size", "100"},
       {"rocksdb.rate_limiter_auto_tuned", "yes"},
       {"rocksdb.compression_level", "32767"},
+      {"rocksdb.compression_max_dict_bytes", "16384"},
+      {"rocksdb.compression_zstd_max_train_bytes", "262144"},
       {"rocksdb.wal_compression", "no"},
       {"histogram-bucket-boundaries", "10,100,1000,10000"},
 
@@ -189,6 +191,32 @@ TEST(Config, Rewrite) {
   unlink(path);
 }
 
+TEST(Config, LoadRocksDBDictionaryCompressionOptions) {
+  const char *path = "test.conf";
+  unlink(path);
+
+  std::ofstream output_file(path, std::ios::out);
+  output_file << "rocksdb.compression_max_dict_bytes 16384"
+              << "\n";
+  output_file << "rocksdb.compression_zstd_max_train_bytes 262144"
+              << "\n";
+  output_file.close();
+
+  Config config;
+  ASSERT_TRUE(config.Load(CLIOptions(path)).IsOK());
+  ASSERT_EQ(config.rocks_db.compression_max_dict_bytes, 16384U);
+  ASSERT_EQ(config.rocks_db.compression_zstd_max_train_bytes, 262144U);
+
+  ASSERT_TRUE(config.Rewrite({}).IsOK());
+
+  Config rewritten_config;
+  ASSERT_TRUE(rewritten_config.Load(CLIOptions(path)).IsOK());
+  ASSERT_EQ(rewritten_config.rocks_db.compression_max_dict_bytes, 16384U);
+  ASSERT_EQ(rewritten_config.rocks_db.compression_zstd_max_train_bytes, 262144U);
+
+  unlink(path);
+}
+
 TEST(Config, ParseConfigLine) {
   ASSERT_EQ(*ParseConfigLine(""), ConfigKV{});
   ASSERT_EQ(*ParseConfigLine("# hello"), ConfigKV{});
diff --git a/tests/cppunit/storage_test.cc b/tests/cppunit/storage_test.cc
diff --git a/tests/cppunit/types/hash_test.cc b/tests/cppunit/types/hash_test.cc
diff --git a/tests/cppunit/types/set_test.cc b/tests/cppunit/types/set_test.cc
diff --git a/tests/cppunit/types/stream_test.cc b/tests/cppunit/types/stream_test.cc
diff --git a/tests/cppunit/types/string_test.cc b/tests/cppunit/types/string_test.cc
diff --git a/tests/gocase/unit/config/config_test.go b/tests/gocase/unit/config/config_test.go
diff --git a/tests/gocase/unit/geo/geo_test.go b/tests/gocase/unit/geo/geo_test.go
diff --git a/tests/gocase/unit/type/stream/stream_test.go b/tests/gocase/unit/type/stream/stream_test.go

Original file line number	Diff line number	Diff line change
`@@ -20,8 +20,8 @@ include_guard()`
`20`	`20`	`include(cmake/utils.cmake)`
`21`	`21`
`22`	`22`	`FetchContent_DeclareGitHubWithMirror(lua`
`23`		`- RocksLabs/lua f458c3d797db31155fa0c156d5301716df48cb8c`
`24`		`- MD5=c7c4deb9f750d8f2bef0044a701df85c`
	`23`	`+ RocksLabs/lua c77781d0c23df021a903a9a6199d301ec559e9cd`
	`24`	`+ MD5=8e9f463af29c0fe17baea52f21e5b790`
`25`	`25`	`)`
`26`	`26`
`27`	`27`	`FetchContent_GetProperties(lua)`