Add prefix bloom filter support

Author: zaidoon1

src/compaction/worker.rs

@@ -364,6 +364,7 @@ fn merge_segments(
                 opts.tree_id,
                 opts.config.cache.clone(),
                 opts.config.descriptor_table.clone(),
+                opts.config.prefix_extractor.clone(),
                 payload.dest_level <= 1, // TODO: look at configuration
                 payload.dest_level <= 2, // TODO: look at configuration
                 #[cfg(feature = "metrics")]

src/config.rs

@@ -2,7 +2,10 @@
 // This source code is licensed under both the Apache 2.0 and MIT License
 // (found in the LICENSE-* files in the repository)
 
-use crate::{path::absolute_path, BlobTree, Cache, CompressionType, DescriptorTable, Tree};
+use crate::{
+    path::absolute_path, prefix::SharedPrefixExtractor, BlobTree, Cache, CompressionType,
+    DescriptorTable, Tree,
+};
 use std::{
     path::{Path, PathBuf},
     sync::Arc,
@@ -94,6 +97,10 @@ pub struct Config {
     /// Descriptor table to use
     #[doc(hidden)]
     pub descriptor_table: Arc<DescriptorTable>,
+
+    /// Prefix extractor for bloom filters
+    #[doc(hidden)]
+    pub prefix_extractor: Option<SharedPrefixExtractor>,
 }
 
 impl Default for Config {
@@ -115,6 +122,7 @@ impl Default for Config {
             compression: CompressionType::None,
             blob_compression: CompressionType::None,
             bloom_bits_per_key: 10,
+            prefix_extractor: None,
 
             blob_file_target_size: /* 64 MiB */ 64 * 1_024 * 1_024,
             blob_file_separation_threshold: /* 4 KiB */ 4 * 1_024,
@@ -312,6 +320,30 @@ impl Config {
         self
     }
 
+    /// Sets the prefix extractor for bloom filters.
+    ///
+    /// A prefix extractor allows bloom filters to index prefixes of keys
+    /// instead of (or in addition to) the full keys. This enables efficient
+    /// filtering for prefix-based queries.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// # use lsm_tree::Config;
+    /// use lsm_tree::prefix::FixedPrefixExtractor;
+    /// use std::sync::Arc;
+    ///
+    /// # let path = tempfile::tempdir()?;
+    /// let config = Config::new(path)
+    ///     .prefix_extractor(Arc::new(FixedPrefixExtractor::new(8)));
+    /// # Ok::<(), Box<dyn std::error::Error>>(())
+    /// ```
+    #[must_use]
+    pub fn prefix_extractor(mut self, extractor: SharedPrefixExtractor) -> Self {
+        self.prefix_extractor = Some(extractor);
+        self
+    }
+
     /// Opens a tree using the config.
     ///
     /// # Errors

src/lib.rs

@@ -182,6 +182,9 @@ pub mod mvcc_stream;
 
 mod path;
 
+/// Prefix extraction for bloom filters
+pub mod prefix;
+
 #[doc(hidden)]
 pub mod range;
 

src/metrics.rs

@@ -45,4 +45,14 @@ impl Metrics {
         let hits = self.bloom_filter_hits.load(Relaxed) as f64;
         hits / queries
     }
+
+    /// Number of bloom filter queries performed.
+    pub fn bloom_filter_queries(&self) -> usize {
+        self.bloom_filter_queries.load(Relaxed)
+    }
+
+    /// Number of bloom filter hits (queries that avoided disk I/O).
+    pub fn bloom_filter_hits(&self) -> usize {
+        self.bloom_filter_hits.load(Relaxed)
+    }
 }

src/multi_reader.rs

@@ -89,7 +89,9 @@ mod tests {
             let mut readers: VecDeque<_> = VecDeque::new();
 
             for segment in &segments {
-                readers.push_back(segment.iter());
+                if let Some(iter) = segment.iter() {
+                    readers.push_back(iter);
+                }
             }
 
             let multi_reader = MultiReader::new(readers);
@@ -115,7 +117,9 @@ mod tests {
             let mut readers: VecDeque<_> = VecDeque::new();
 
             for segment in &segments {
-                readers.push_back(segment.iter());
+                if let Some(iter) = segment.iter() {
+                    readers.push_back(iter);
+                }
             }
 
             let multi_reader = MultiReader::new(readers);
@@ -141,7 +145,9 @@ mod tests {
             let mut readers: VecDeque<_> = VecDeque::new();
 
             for segment in &segments {
-                readers.push_back(segment.iter());
+                if let Some(iter) = segment.iter() {
+                    readers.push_back(iter);
+                }
             }
 
             let multi_reader = MultiReader::new(readers);