From b84f97c8fb5310fbd8a7bb7bd5ea64da4feb3ecd Mon Sep 17 00:00:00 2001 From: Rafael Escrich Date: Thu, 4 Jun 2026 19:35:52 -0300 Subject: [PATCH] Add RedisBloom-compatible BF.* commands for address bloom filters. Implements BF.RESERVE, BF.ADD, BF.MADD, BF.EXISTS, and BF.MEXISTS using fastbloom so Forja's multichain indexer can use RedVector instead of Redis Stack. --- command/src/command.rs | 244 ++++++++++++++++++++++++++++++++++++++++- database/Cargo.toml | 1 + database/src/bloom.rs | 76 +++++++++++++ database/src/lib.rs | 52 +++++++++ 4 files changed, 368 insertions(+), 5 deletions(-) create mode 100644 database/src/bloom.rs diff --git a/command/src/command.rs b/command/src/command.rs index 1183be7..6fa6bf9 100644 --- a/command/src/command.rs +++ b/command/src/command.rs @@ -655,6 +655,7 @@ fn dbtype(parser: &mut ParsedCommand, db: &Database, dbindex: usize) -> Response Some(Value::Set(_)) => Response::Data("set".to_owned().into_bytes()), Some(Value::SortedSet(_)) => Response::Data("zset".to_owned().into_bytes()), Some(Value::Hash(_)) => Response::Data("hash".to_owned().into_bytes()), + Some(Value::Bloom(_)) => Response::Data("MBbloom--".to_owned().into_bytes()), None => Response::Data("none".to_owned().into_bytes()), } } @@ -1018,6 +1019,100 @@ fn pfmerge(parser: &ParsedCommand, db: &mut Database, dbindex: usize) -> Respons r } +fn bf_reserve(parser: &mut ParsedCommand, db: &mut Database, dbindex: usize) -> Response { + validate_arguments_exact!(parser, 4); + let key = try_validate!(parser.get_vec(1), "ERR invalid key"); + let error_rate = try_validate!(parser.get_f64(2), "ERR invalid error rate"); + let capacity = try_validate!(parser.get_i64(3), "ERR invalid capacity"); + + if capacity <= 0 { + return Response::Error("ERR invalid bloom filter parameters".to_owned()); + } + + match db + .get_or_create(dbindex, &key) + .bf_reserve(error_rate, capacity as usize) + { + Ok(()) => { + db.key_updated(dbindex, &key); + Response::Status("OK".to_owned()) + } + Err(e) => Response::Error(e.to_string()), + } +} + +fn bf_add(parser: &mut ParsedCommand, db: &mut Database, dbindex: usize) -> Response { + validate_arguments_exact!(parser, 3); + let key = try_validate!(parser.get_vec(1), "ERR invalid key"); + let item = try_validate!(parser.get_slice(2), "ERR invalid item").to_vec(); + + match db.get_or_create(dbindex, &key).bf_add(&item) { + Ok(added) => { + db.key_updated(dbindex, &key); + Response::Integer(if added { 1 } else { 0 }) + } + Err(e) => Response::Error(e.to_string()), + } +} + +fn bf_madd(parser: &mut ParsedCommand, db: &mut Database, dbindex: usize) -> Response { + validate_arguments_gte!(parser, 3); + let key = try_validate!(parser.get_vec(1), "ERR invalid key"); + let mut items = Vec::with_capacity(parser.argv.len() - 2); + for i in 2..parser.argv.len() { + items.push(try_validate!(parser.get_slice(i), "ERR invalid item").to_vec()); + } + + let value = db.get_or_create(dbindex, &key); + let mut result = Vec::with_capacity(items.len()); + for item in &items { + match value.bf_add(item) { + Ok(added) => result.push(Response::Integer(if added { 1 } else { 0 })), + Err(e) => return Response::Error(e.to_string()), + } + } + + db.key_updated(dbindex, &key); + Response::Array(result) +} + +fn bf_exists(parser: &mut ParsedCommand, db: &Database, dbindex: usize) -> Response { + validate_arguments_exact!(parser, 3); + let key = try_validate!(parser.get_vec(1), "ERR invalid key"); + let item = try_validate!(parser.get_slice(2), "ERR invalid item"); + + match db.get(dbindex, &key) { + Some(value) => match value.bf_exists(item) { + Ok(exists) => Response::Integer(if exists { 1 } else { 0 }), + Err(e) => Response::Error(e.to_string()), + }, + None => Response::Integer(0), + } +} + +fn bf_mexists(parser: &mut ParsedCommand, db: &Database, dbindex: usize) -> Response { + validate_arguments_gte!(parser, 3); + let key = try_validate!(parser.get_vec(1), "ERR invalid key"); + let mut items = Vec::with_capacity(parser.argv.len() - 2); + for i in 2..parser.argv.len() { + items.push(try_validate!(parser.get_slice(i), "ERR invalid item").to_vec()); + } + + match db.get(dbindex, &key) { + Some(value) => { + let mut result = Vec::with_capacity(items.len()); + for item in &items { + match value.bf_exists(item) { + Ok(exists) => result.push(Response::Integer(if exists { 1 } else { 0 })), + Err(e) => return Response::Error(e.to_string()), + } + } + Response::Array(result) + } + None => Response::Array(items.iter().map(|_| Response::Integer(0)).collect()), + } +} + fn generic_push( parser: &mut ParsedCommand, db: &mut Database, @@ -3081,6 +3176,7 @@ fn object(parser: &mut ParsedCommand, db: &mut Database, dbindex: usize) -> Resp ValueHash::ZipList(_) => "ziplist", ValueHash::HashMap(_) => "hashtable", }, + Value::Bloom(_) => "bloom", Value::Nil => return Response::Nil, }; Response::Data(encoding.to_string().into_bytes()) @@ -4249,7 +4345,8 @@ fn command_cmd(parser: &mut ParsedCommand, _db: &Database) -> Response { "ttl", "pttl", "persist", "slaveof", "role", "config", "subscribe", "unsubscribe", "psubscribe", "punsubscribe", "publish", "pubsub", "watch", "unwatch", "restore", "dump", "object", "client", "time", "bitop", "bitcount", "bitpos", "wait", "command", - "pfadd", "pfcount", "pfmerge", + "pfadd", "pfcount", "pfmerge", "bf.reserve", "bf.add", "bf.madd", "bf.exists", + "bf.mexists", ]; #[cfg(feature = "vector-search")] let mut commands = commands; @@ -4718,6 +4815,11 @@ fn command_properties(command_name: &str) -> CommandProperties { "pfadd" => (-2, wmf, 1, 1, 1), "pfcount" => (-2, READONLY, 1, -1, 1), "pfmerge" => (-2, wm, 1, -1, 1), + "bf.reserve" => (4, wmf, 1, 1, 1), + "bf.add" => (3, wmf, 1, 1, 1), + "bf.madd" => (-3, wmf, 1, 1, 1), + "bf.exists" => (3, fr, 1, 1, 1), + "bf.mexists" => (-3, fr, 1, 1, 1), "pfdebug" => (-3, WRITE, 0, 0, 0), "latency" => (-2, ars | ls, 0, 0, 0), // RediSearch (FT.*) commands (only when vector-search feature is enabled) @@ -4771,17 +4873,17 @@ fn execute_command( let raw_command = try_opt_validate!(parser.get_str(0), "Invalid command"); let lower_command = raw_command.to_ascii_lowercase(); - // For FT.* commands, bypass mapped_command to ensure they're allowed - // Check both lowercase and original case to handle any edge cases + // For module-style dotted commands, bypass mapped_command to ensure they're allowed. let is_ft_command = lower_command.starts_with("ft."); + let is_bf_command = lower_command.starts_with("bf."); // Debug logging if raw_command.to_uppercase().starts_with("FT.") || lower_command.starts_with("ft.") { eprintln!("DEBUG: FT command detected: raw='{}', lower='{}', is_ft={}", raw_command, lower_command, is_ft_command); } - let command_name = if is_ft_command { - // FT.* commands - use directly, don't go through mapped_command + let command_name = if is_ft_command || is_bf_command { + // Module commands - use directly, don't go through mapped_command. // Normalize to lowercase for consistency &*lower_command } else { @@ -4906,6 +5008,11 @@ fn execute_command( "pfadd" => pfadd(parser, db, dbindex), "pfcount" => pfcount(parser, db, dbindex), "pfmerge" => pfmerge(parser, db, dbindex), + "bf.reserve" => bf_reserve(parser, db, dbindex), + "bf.add" => bf_add(parser, db, dbindex), + "bf.madd" => bf_madd(parser, db, dbindex), + "bf.exists" => bf_exists(parser, db, dbindex), + "bf.mexists" => bf_mexists(parser, db, dbindex), "pfselftest" => pfselftest(parser, db, dbindex), "pfdebug" => pfdebug(parser, db, dbindex), "exists" => exists(parser, db, dbindex), @@ -5805,6 +5912,133 @@ mod test_command { ); } + #[test] + fn bf_reserve_add_exists_command() { + let mut db = Database::new(Config::new(Logger::new(Level::Warning))); + assert_eq!( + command( + parser!(b"BF.RESERVE wallet_bloom 0.01 1000"), + &mut db, + &mut Client::mock() + ) + .unwrap(), + Response::Status("OK".to_owned()) + ); + assert_eq!( + command( + parser!(b"BF.EXISTS wallet_bloom 0x742d35Cc6634C0532925a3b8D4C9db96C4b4d8b6"), + &mut db, + &mut Client::mock() + ) + .unwrap(), + Response::Integer(0) + ); + assert_eq!( + command( + parser!(b"BF.ADD wallet_bloom 0x742d35Cc6634C0532925a3b8D4C9db96C4b4d8b6"), + &mut db, + &mut Client::mock() + ) + .unwrap(), + Response::Integer(1) + ); + assert_eq!( + command( + parser!(b"BF.ADD wallet_bloom 0x742d35Cc6634C0532925a3b8D4C9db96C4b4d8b6"), + &mut db, + &mut Client::mock() + ) + .unwrap(), + Response::Integer(0) + ); + assert_eq!( + command( + parser!(b"BF.EXISTS wallet_bloom 0x742d35Cc6634C0532925a3b8D4C9db96C4b4d8b6"), + &mut db, + &mut Client::mock() + ) + .unwrap(), + Response::Integer(1) + ); + } + + #[test] + fn bf_madd_and_mexists_command() { + let mut db = Database::new(Config::new(Logger::new(Level::Warning))); + assert_eq!( + command( + parser!(b"BF.MADD wallet_bloom evm_address tron_address evm_address"), + &mut db, + &mut Client::mock() + ) + .unwrap(), + Response::Array(vec![ + Response::Integer(1), + Response::Integer(1), + Response::Integer(0), + ]) + ); + assert_eq!( + command( + parser!(b"BF.MEXISTS wallet_bloom evm_address missing tron_address"), + &mut db, + &mut Client::mock() + ) + .unwrap(), + Response::Array(vec![ + Response::Integer(1), + Response::Integer(0), + Response::Integer(1), + ]) + ); + } + + #[test] + fn bf_exists_missing_key_returns_zero() { + let mut db = Database::new(Config::new(Logger::new(Level::Warning))); + assert_eq!( + command( + parser!(b"BF.EXISTS missing_bloom value"), + &mut db, + &mut Client::mock() + ) + .unwrap(), + Response::Integer(0) + ); + assert_eq!( + command( + parser!(b"BF.MEXISTS missing_bloom a b"), + &mut db, + &mut Client::mock() + ) + .unwrap(), + Response::Array(vec![Response::Integer(0), Response::Integer(0)]) + ); + } + + #[test] + fn bf_wrong_type_and_existing_key_errors() { + let mut db = Database::new(Config::new(Logger::new(Level::Warning))); + assert_eq!( + command(parser!(b"SET key value"), &mut db, &mut Client::mock()).unwrap(), + Response::Status("OK".to_owned()) + ); + assert!( + command(parser!(b"BF.ADD key value"), &mut db, &mut Client::mock()) + .unwrap() + .is_error() + ); + assert!( + command( + parser!(b"BF.RESERVE key 0.01 1000"), + &mut db, + &mut Client::mock() + ) + .unwrap() + .is_error() + ); + } + #[test] fn pfcount1_command() { let mut db = Database::new(Config::new(Logger::new(Level::Warning))); diff --git a/database/Cargo.toml b/database/Cargo.toml index 72698ec..6a0994a 100644 --- a/database/Cargo.toml +++ b/database/Cargo.toml @@ -5,6 +5,7 @@ version = "0.1.0" [dependencies] crc64 = "0.2" +fastbloom = { version = "0.17", default-features = false } rand = "0.3" rehashinghashmap = "0.1" skiplist = "0.3" diff --git a/database/src/bloom.rs b/database/src/bloom.rs new file mode 100644 index 0000000..3fdaaa5 --- /dev/null +++ b/database/src/bloom.rs @@ -0,0 +1,76 @@ +use fastbloom::BloomFilter; + +use crate::error::OperationError; + +const DEFAULT_ERROR_RATE: f64 = 0.01; +const DEFAULT_CAPACITY: usize = 100; + +#[derive(PartialEq, Debug, Clone)] +pub struct ValueBloom { + filter: BloomFilter, + error_rate: f64, + capacity: usize, + insertions: usize, +} + +impl ValueBloom { + pub fn new(error_rate: f64, capacity: usize) -> Result { + if !(error_rate > 0.0 && error_rate < 1.0) || capacity == 0 { + return Err(OperationError::ValueError( + "ERR invalid bloom filter parameters".to_owned(), + )); + } + + Ok(ValueBloom { + filter: BloomFilter::with_false_pos(error_rate).expected_items(capacity), + error_rate, + capacity, + insertions: 0, + }) + } + + pub fn default_filter() -> Self { + ValueBloom::new(DEFAULT_ERROR_RATE, DEFAULT_CAPACITY) + .expect("default bloom filter parameters are valid") + } + + pub fn add(&mut self, item: &[u8]) -> bool { + let may_have_existed = self.filter.insert(item); + if !may_have_existed { + self.insertions += 1; + } + !may_have_existed + } + + pub fn exists(&self, item: &[u8]) -> bool { + self.filter.contains(item) + } + + pub fn capacity(&self) -> usize { + self.capacity + } + + pub fn error_rate(&self) -> f64 { + self.error_rate + } + + pub fn insertions(&self) -> usize { + self.insertions + } + + pub fn memory_bytes(&self) -> u64 { + (self.filter.as_slice().len() * std::mem::size_of::()) as u64 + } + + pub fn debug_object(&self) -> String { + format!( + "Value at:0x0000000000 refcount:0 encoding:bloom capacity:{} error_rate:{} \ + insertions:{} bits:{} hashes:{}", + self.capacity, + self.error_rate, + self.insertions, + self.filter.num_bits(), + self.filter.num_hashes() + ) + } +} diff --git a/database/src/lib.rs b/database/src/lib.rs index 940e3a7..dd99d9f 100644 --- a/database/src/lib.rs +++ b/database/src/lib.rs @@ -3,6 +3,7 @@ extern crate config; #[macro_use(log)] extern crate logger; extern crate crc64; +extern crate fastbloom; extern crate parser; extern crate persistence; extern crate rand; @@ -16,6 +17,7 @@ extern crate util; extern crate hnsw_rs; pub mod dbutil; +pub mod bloom; pub mod error; pub mod hash; pub mod list; @@ -47,6 +49,7 @@ use response::Response; use util::{get_random_hex_chars, glob_match, mstime}; use error::OperationError; +use bloom::ValueBloom; use hash::ValueHash; use list::ValueList; use rdbutil::encode_u64_to_slice_u8; @@ -83,6 +86,7 @@ pub enum Value { Set(ValueSet), SortedSet(ValueSortedSet), Hash(ValueHash), + Bloom(ValueBloom), } /// Events relevant for clients in pubsub mode @@ -263,6 +267,13 @@ impl Value { } } + pub fn is_bloom(&self) -> bool { + match self { + Value::Bloom(_) => true, + _ => false, + } + } + /// Sets the value to a string. /// /// # Examples @@ -578,6 +589,37 @@ impl Value { } } + pub fn bf_reserve(&mut self, error_rate: f64, capacity: usize) -> Result<(), OperationError> { + match self { + Value::Nil => { + *self = Value::Bloom(ValueBloom::new(error_rate, capacity)?); + Ok(()) + } + _ => Err(OperationError::ValueError("ERR item exists".to_owned())), + } + } + + pub fn bf_add(&mut self, item: &[u8]) -> Result { + match self { + Value::Nil => *self = Value::Bloom(ValueBloom::default_filter()), + Value::Bloom(_) => {} + _ => return Err(OperationError::WrongTypeError), + }; + + match self { + Value::Bloom(value) => Ok(value.add(item)), + _ => panic!("Expected value to be a bloom filter"), + } + } + + pub fn bf_exists(&self, item: &[u8]) -> Result { + match self { + Value::Nil => Ok(false), + Value::Bloom(value) => Ok(value.exists(item)), + _ => Err(OperationError::WrongTypeError), + } + } + /// Adds an element to a list. /// Returns the size of the list. /// @@ -1928,6 +1970,11 @@ impl Value { Value::Set(s) => s.dump(&mut data)?, Value::SortedSet(s) => s.dump(&mut data)?, Value::Hash(h) => h.dump(&mut data)?, + Value::Bloom(_) => { + return Err(OperationError::ValueError( + "ERR DUMP is not supported for bloom filters".to_owned(), + )); + } }; let crc = crc64(0, &*data); encode_u64_to_slice_u8(crc, &mut data).unwrap(); @@ -1944,6 +1991,7 @@ impl Value { Value::Set(s) => s.debug_object(), Value::SortedSet(s) => s.debug_object(), Value::Hash(h) => h.debug_object(), + Value::Bloom(b) => b.debug_object(), } } @@ -1955,6 +2003,7 @@ impl Value { Value::Set(s) => s.scard() == 0, Value::SortedSet(s) => s.zcard() == 0, Value::Hash(h) => h.is_empty(), + Value::Bloom(b) => b.insertions() == 0, } } } @@ -2213,6 +2262,9 @@ impl Database { } }; } + Value::Bloom(b) => { + size += b.memory_bytes() + 32; + } Value::Nil => size += 0, }