fix: Implement chunking for large SBF filters

This commit adds chunking functionalities for load/save operations of
bloom filters. Additional information is added in the serialization of
each filter. Specifically, when saving each filter the total size of the
filter is written followed by chunks of the filter (max size of 64 MB
per chunk).

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

Author: EricHayter

src/server/rdb_load.cc

@@ -1874,7 +1874,23 @@ auto RdbLoaderBase::ReadSBF() -> io::Result<OpaqueObj> {
     unsigned hash_cnt;
     string filter_data;
     SET_OR_UNEXPECT(LoadLen(nullptr), hash_cnt);
-    SET_OR_UNEXPECT(FetchGenericString(), filter_data);
+
+    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);
+      }
+
+      offset += chunk_size;
+    }
+
     size_t bit_len = filter_data.size() * 8;
     if (!is_power2(bit_len)) {  // must be power of two
       return Unexpected(errc::rdb_file_corrupted);

src/server/rdb_save.cc

@@ -623,11 +623,17 @@ std::error_code RdbSerializer::SaveSBFObject(const PrimeValue& pv) {
     RETURN_ON_ERR(SaveLen(sbf->hashfunc_cnt(i)));
 
     string_view blob = sbf->data(i);
-    RETURN_ON_ERR(SaveString(blob));
-    FlushState flush_state = FlushState::kFlushMidEntry;
-    if ((i + 1) == sbf->num_filters())
-      flush_state = FlushState::kFlushEndEntry;
+    size_t num_chunks = (blob.size() + kFilterChunkSize - 1) / kFilterChunkSize;
+    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)));
+    }
+
+    FlushState flush_state =
+        (i + 1 == sbf->num_filters()) ? FlushState::kFlushEndEntry : FlushState::kFlushMidEntry;
     FlushIfNeeded(flush_state);
   }
 

src/server/rdb_save.h

@@ -20,6 +20,8 @@ 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;
 typedef struct quicklistNode quicklistNode;