Support random seed per bloom object by default (configurable)

Signed-off-by: Karthik Subbarao <[email protected]>
valkey-io · Dec 5, 2024 · 25fedaf · 25fedaf
1 parent 34d6fb0
commit 25fedaf
Show file tree

Hide file tree

Showing 14 changed files with 394 additions and 286 deletions.
diff --git a/src/bloom/command_handler.rs b/src/bloom/command_handler.rs
@@ -112,10 +112,12 @@ pub fn bloom_filter_add_value(
             let fp_rate = configs::BLOOM_FP_RATE_DEFAULT;
             let capacity = configs::BLOOM_CAPACITY.load(Ordering::Relaxed) as u32;
             let expansion = configs::BLOOM_EXPANSION.load(Ordering::Relaxed) as u32;
+            let use_random_seed = configs::BLOOM_USE_RANDOM_SEED.load(Ordering::Relaxed);
             let mut bloom = match BloomFilterType::new_reserved(
                 fp_rate,
                 capacity,
                 expansion,
+                use_random_seed,
                 validate_size_limit,
             ) {
                 Ok(bf) => bf,
@@ -274,12 +276,14 @@ pub fn bloom_filter_reserve(ctx: &Context, input_args: &[ValkeyString]) -> Valke
     match value {
         Some(_) => Err(ValkeyError::Str(utils::ITEM_EXISTS)),
         None => {
+            let use_random_seed = configs::BLOOM_USE_RANDOM_SEED.load(Ordering::Relaxed);
             // Skip bloom filter size validation on replicated cmds.
             let validate_size_limit = !ctx.get_flags().contains(ContextFlags::REPLICATED);
             let bloom = match BloomFilterType::new_reserved(
                 fp_rate,
                 capacity,
                 expansion,
+                use_random_seed,
                 validate_size_limit,
             ) {
                 Ok(bf) => bf,
@@ -403,10 +407,12 @@ pub fn bloom_filter_insert(ctx: &Context, input_args: &[ValkeyString]) -> Valkey
             if nocreate {
                 return Err(ValkeyError::Str(utils::NOT_FOUND));
             }
+            let use_random_seed = configs::BLOOM_USE_RANDOM_SEED.load(Ordering::Relaxed);
             let mut bloom = match BloomFilterType::new_reserved(
                 fp_rate,
                 capacity,
                 expansion,
+                use_random_seed,
                 validate_size_limit,
             ) {
                 Ok(bf) => bf,

diff --git a/src/bloom/data_type.rs b/src/bloom/data_type.rs
@@ -1,5 +1,6 @@
 use crate::bloom::utils::BloomFilter;
 use crate::bloom::utils::BloomFilterType;
+use crate::configs;
 use crate::metrics::BLOOM_NUM_OBJECTS;
 use crate::metrics::BLOOM_OBJECT_TOTAL_MEMORY_BYTES;
 use crate::wrapper::bloom_callback;
@@ -27,7 +28,6 @@ pub static BLOOM_FILTER_TYPE: ValkeyType = ValkeyType::new(
         digest: Some(bloom_callback::bloom_digest),
 
         mem_usage: Some(bloom_callback::bloom_mem_usage),
-        // TODO
         free: Some(bloom_callback::bloom_free),
 
         aux_load: Some(bloom_callback::bloom_aux_load),
@@ -72,7 +72,10 @@ impl ValkeyDataType for BloomFilterType {
             return None;
         };
         let mut filters: Vec<BloomFilter> = Vec::with_capacity(num_filters as usize);
-
+        let Ok(is_seed_random_u64) = raw::load_unsigned(rdb) else {
+            return None;
+        };
+        let is_seed_random = is_seed_random_u64 == 1;
         for i in 0..num_filters {
             let Ok(bitmap) = raw::load_string_buffer(rdb) else {
                 return None;
@@ -90,7 +93,7 @@ impl ValkeyDataType for BloomFilterType {
                 }
             };
             if !BloomFilter::validate_size(capacity as u32, new_fp_rate) {
-                logging::log_warning("Failed to restore bloom object because it contains a filter larger than the max allowed size limit.");
+                logging::log_warning("Failed to restore bloom object: Contains a filter larger than the max allowed size limit.");
                 return None;
             }
             // Only load num_items when it's the last filter
@@ -104,6 +107,10 @@ impl ValkeyDataType for BloomFilterType {
             };
             let filter =
                 BloomFilter::from_existing(bitmap.as_ref(), num_items as u32, capacity as u32);
+            if !is_seed_random && filter.seed() != configs::FIXED_SEED {
+                logging::log_warning("Failed to restore bloom object: Object in fixed seed mode, but seed does not match fixed seed.");
+                return None;
+            }
             filters.push(filter);
         }
         BLOOM_OBJECT_TOTAL_MEMORY_BYTES.fetch_add(
@@ -114,6 +121,7 @@ impl ValkeyDataType for BloomFilterType {
         let item = BloomFilterType {
             expansion: expansion as u32,
             fp_rate,
+            is_seed_random,
             filters,
         };
         Some(item)