Add flags for setting save format for SBF filters

Added a new flag `rdb_sbf_chunked` which determines the save format of
SBFs. Also, separate functions for saving SBFs were added.

Signed-off-by: Eric <[email protected]>

Author: EricHayter

src/server/rdb_extensions.h

@@ -13,11 +13,12 @@ constexpr uint8_t RDB_TYPE_JSON = 30;
 constexpr uint8_t RDB_TYPE_HASH_WITH_EXPIRY = 31;
 constexpr uint8_t RDB_TYPE_SET_WITH_EXPIRY = 32;
 constexpr uint8_t RDB_TYPE_SBF = 33;
+constexpr uint8_t RDB_TYPE_SBF2 = 34;
 
 constexpr bool rdbIsObjectTypeDF(uint8_t type) {
   return __rdbIsObjectType(type) || (type == RDB_TYPE_JSON) ||
          (type == RDB_TYPE_HASH_WITH_EXPIRY) || (type == RDB_TYPE_SET_WITH_EXPIRY) ||
-         (type == RDB_TYPE_SBF);
+         (type == RDB_TYPE_SBF) || (type == RDB_TYPE_SBF2);
 }
 
 //  Opcodes: Range 200-240 is used by DF extensions.

src/server/rdb_load.cc

@@ -54,6 +54,7 @@ extern "C" {
 ABSL_DECLARE_FLAG(int32_t, list_max_listpack_size);
 ABSL_DECLARE_FLAG(int32_t, list_compress_depth);
 ABSL_DECLARE_FLAG(uint32_t, dbnum);
+ABSL_DECLARE_FLAG(bool, rdb_sbf_chunked);
 ABSL_FLAG(bool, rdb_load_dry_run, false, "Dry run RDB load without applying changes");
 ABSL_FLAG(bool, rdb_ignore_expiry, false, "Ignore Key Expiry when loding from RDB snapshot");
 
@@ -188,7 +189,7 @@ string ModuleTypeName(uint64_t module_id) {
 bool RdbTypeAllowedEmpty(int type) {
   return type == RDB_TYPE_STRING || type == RDB_TYPE_JSON || type == RDB_TYPE_SBF ||
          type == RDB_TYPE_STREAM_LISTPACKS || type == RDB_TYPE_SET_WITH_EXPIRY ||
-         type == RDB_TYPE_HASH_WITH_EXPIRY;
+         type == RDB_TYPE_HASH_WITH_EXPIRY || type == RDB_TYPE_SBF2;
 }
 
 DbSlice& GetCurrentDbSlice() {
@@ -1316,6 +1317,9 @@ error_code RdbLoaderBase::ReadObj(int rdbtype, OpaqueObj* dest) {
     case RDB_TYPE_SBF:
       iores = ReadSBF();
       break;
+    case RDB_TYPE_SBF2:
+      iores = ReadSBF2();
+      break;
     default:
       LOG(ERROR) << "Unsupported rdb type " << rdbtype;
 
@@ -1851,7 +1855,7 @@ auto RdbLoaderBase::ReadRedisJson() -> io::Result<OpaqueObj> {
   return OpaqueObj{std::move(dest), RDB_TYPE_JSON};
 }
 
-auto RdbLoaderBase::ReadSBF() -> io::Result<OpaqueObj> {
+auto RdbLoaderBase::ReadSBFImpl(bool chunking) -> io::Result<OpaqueObj> {
   RdbSBF res;
   uint64_t options;
   SET_OR_UNEXPECT(LoadLen(nullptr), options);
@@ -1875,20 +1879,24 @@ auto RdbLoaderBase::ReadSBF() -> io::Result<OpaqueObj> {
     string filter_data;
     SET_OR_UNEXPECT(LoadLen(nullptr), hash_cnt);
 
-    unsigned total_size = 0;
-    SET_OR_UNEXPECT(LoadLen(nullptr), total_size);
+    if (absl::GetFlag(FLAGS_rdb_sbf_chunked)) {
+      unsigned total_size = 0;
+      SET_OR_UNEXPECT(LoadLen(nullptr), total_size);
 
-    filter_data.resize(total_size);
-    size_t offset = 0;
-    while (offset < total_size) {
-      unsigned chunk_size = 0;
-      SET_OR_UNEXPECT(LoadLen(nullptr), chunk_size);
-      error_code ec = FetchBuf(chunk_size, filter_data.data() + offset);
-      if (ec) {
-        return make_unexpected(ec);
-      }
+      filter_data.resize(total_size);
+      size_t offset = 0;
+      while (offset < total_size) {
+        unsigned chunk_size = 0;
+        SET_OR_UNEXPECT(LoadLen(nullptr), chunk_size);
+        error_code ec = FetchBuf(chunk_size, filter_data.data() + offset);
+        if (ec) {
+          return make_unexpected(ec);
+        }
 
-      offset += chunk_size;
+        offset += chunk_size;
+      }
+    } else {
+        SET_OR_UNEXPECT(FetchGenericString(), filter_data);
     }
 
     size_t bit_len = filter_data.size() * 8;
@@ -1900,6 +1908,15 @@ auto RdbLoaderBase::ReadSBF() -> io::Result<OpaqueObj> {
   return OpaqueObj{std::move(res), RDB_TYPE_SBF};
 }
 
+
+auto RdbLoaderBase::ReadSBF() -> io::Result<OpaqueObj> {
+    return ReadSBFImpl(false);
+}
+
+auto RdbLoaderBase::ReadSBF2() -> io::Result<OpaqueObj> {
+    return ReadSBFImpl(true);
+}
+
 template <typename T> io::Result<T> RdbLoaderBase::FetchInt() {
   auto ec = EnsureRead(sizeof(T));
   if (ec)

src/server/rdb_load.h

@@ -169,7 +169,9 @@ class RdbLoaderBase {
   ::io::Result<OpaqueObj> ReadListQuicklist(int rdbtype);
   ::io::Result<OpaqueObj> ReadStreams(int rdbtype);
   ::io::Result<OpaqueObj> ReadRedisJson();
+  ::io::Result<OpaqueObj> ReadSBFImpl(bool chunking);
   ::io::Result<OpaqueObj> ReadSBF();
+  ::io::Result<OpaqueObj> ReadSBF2();
 
   std::error_code SkipModuleData();
   std::error_code HandleCompressedBlob(int op_type);

src/server/rdb_save.cc

@@ -49,6 +49,10 @@ ABSL_FLAG(dfly::CompressionMode, compression_mode, dfly::CompressionMode::MULTI_
 ABSL_RETIRED_FLAG(bool, stream_rdb_encode_v2, true,
                   "Retired. Uses format, compatible with redis 7.2 and Dragonfly v1.26+");
 
+// Flip this value to 'true' in March 2026.
+ABSL_FLAG(bool, rdb_sbf_chunked, false,
+        "Enable new save format for saving SBFs in chunks.");
+
 namespace dfly {
 
 using namespace std;
@@ -623,13 +627,15 @@ std::error_code RdbSerializer::SaveSBFObject(const PrimeValue& pv) {
     RETURN_ON_ERR(SaveLen(sbf->hashfunc_cnt(i)));
 
     string_view blob = sbf->data(i);
-    size_t num_chunks = (blob.size() + kFilterChunkSize - 1) / kFilterChunkSize;
-    RETURN_ON_ERR(SaveLen(blob.size()));
+    if (absl::GetFlag(FLAGS_rdb_sbf_chunked)) {
+      RETURN_ON_ERR(SaveLen(blob.size()));
 
-    for (size_t chunk_idx = 0; chunk_idx < num_chunks; ++chunk_idx) {
-      size_t offset = chunk_idx * kFilterChunkSize;
-      size_t chunk_len = std::min(kFilterChunkSize, blob.size() - offset);
-      RETURN_ON_ERR(SaveString(blob.substr(offset, chunk_len)));
+      for (size_t offset = 0; offset< blob.size(); offset += kFilterChunkSize) {
+        size_t chunk_len = std::min(kFilterChunkSize, blob.size() - offset);
+        RETURN_ON_ERR(SaveString(blob.substr(offset, chunk_len)));
+      }
+    } else {
+        RETURN_ON_ERR(SaveString(blob));
     }
 
     FlushState flush_state = FlushState::kFlushMidEntry;

src/server/rdb_save.h

@@ -20,7 +20,6 @@ extern "C" {
 #include "server/journal/types.h"
 #include "server/table.h"
 
-constexpr size_t kFilterChunkSize = 1ULL << 26;
 
 typedef struct rax rax;
 typedef struct streamCG streamCG;
@@ -211,6 +210,7 @@ class SerializerBase {
   io::IoBuf mem_buf_;
   std::unique_ptr<detail::CompressorImpl> compressor_impl_;
 
+  static constexpr size_t kFilterChunkSize = 1ULL << 26;
   static constexpr size_t kMinStrSizeToCompress = 256;
   static constexpr size_t kMaxStrSizeToCompress = 1 * 1024 * 1024;
   static constexpr double kMinCompressionReductionPrecentage = 0.95;

src/server/rdb_test.cc

@@ -35,6 +35,7 @@ ABSL_DECLARE_FLAG(int32, list_max_listpack_size);
 ABSL_DECLARE_FLAG(dfly::CompressionMode, compression_mode);
 ABSL_DECLARE_FLAG(bool, rdb_ignore_expiry);
 ABSL_DECLARE_FLAG(uint32_t, num_shards);
+ABSL_DECLARE_FLAG(bool, rdb_sbf_chunked);
 
 namespace dfly {
 
@@ -671,6 +672,7 @@ TEST_F(RdbTest, SBF) {
 }
 
 TEST_F(RdbTest, SBFLargeFilterChunking) {
+  absl::SetFlag(&FLAGS_rdb_sbf_chunked, true);
   max_memory_limit = 200000000;
 
   // Using this set of parameters for the BF.RESERVE command resulted in a