feat: add bloom filter when write sst

Cargo.toml

@@ -69,6 +69,7 @@ common_types = { path = "common_types" }
 common_util = { path = "common_util" }
 df_operator = { path = "df_operator" }
 env_logger = "0.6"
+ethbloom = "0.13.0"
 futures = "0.3"
 lazy_static = "1.4.0"
 log = "0.4"

analytic_engine/Cargo.toml

@@ -20,6 +20,7 @@ bytes = { workspace = true }
 common_types = { workspace = true }
 common_util = { workspace = true }
 datafusion = { workspace = true }
+ethbloom = { workspace = true }
 futures = { workspace = true }
 lazy_static = { workspace = true }
 log = { workspace = true }

analytic_engine/src/compaction/picker.rs

@@ -594,7 +594,6 @@ mod tests {
             file::SstMetaData,
             manager::{tests::LevelsControllerMockBuilder, LevelsController},
         },
-        table_options::StorageFormatOptions,
     };
 
     fn build_sst_meta_data(time_range: TimeRange, size: u64) -> SstMetaData {
@@ -606,7 +605,8 @@ mod tests {
             schema: build_schema(),
             size,
             row_num: 2,
-            storage_format_opts: StorageFormatOptions::default(),
+            storage_format_opts: Default::default(),
+            bloom_filter: Default::default(),
         }
     }
 

analytic_engine/src/instance/flush_compaction.rs

@@ -620,6 +620,7 @@ impl Instance {
                 storage_format_opts: StorageFormatOptions::new(
                     table_data.table_options().storage_format,
                 ),
+                bloom_filter: Default::default(),
             };
 
             let store = self.space_store.clone();
@@ -725,6 +726,7 @@ impl Instance {
             size: 0,
             row_num: 0,
             storage_format_opts: StorageFormatOptions::new(table_data.storage_format()),
+            bloom_filter: Default::default(),
         };
 
         // Alloc file id for next sst file

analytic_engine/src/sst/file.rs

@@ -27,6 +27,7 @@ use common_util::{
     metric::Meter,
     runtime::{JoinHandle, Runtime},
 };
+use ethbloom::Bloom;
 use log::{debug, error, info};
 use object_store::ObjectStoreRef;
 use proto::{common as common_pb, sst as sst_pb};
@@ -56,6 +57,16 @@ pub enum Error {
     #[snafu(display("Storage format options are not found.\nBacktrace\n:{}", backtrace))]
     StorageFormatOptionsNotFound { backtrace: Backtrace },
 
+    #[snafu(display("Bloom filter options are not found.\nBacktrace\n:{}", backtrace))]
+    BloomFitlerNotFound { backtrace: Backtrace },
+
+    #[snafu(display(
+        "Bloom filter should be 256 byte, current:{}.\nBacktrace\n:{}",
+        size,
+        backtrace
+    ))]
+    InvalidBloomFilter { size: usize, backtrace: Backtrace },
+
     #[snafu(display("Failed to convert time range, err:{}", source))]
     ConvertTimeRange { source: common_types::time::Error },
 
@@ -425,6 +436,64 @@ impl FileMeta {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub struct BloomFilter {
+    // Two level vector means
+    // 1. row group
+    // 2. column
+    filters: Vec<Vec<Bloom>>,
+}
+
+impl BloomFilter {
+    pub fn new(filters: Vec<Vec<Bloom>>) -> Self {
+        Self { filters }
+    }
+
+    pub fn to_pb(&self) -> sst_pb::SstBloomFilter {
+        let filters = self
+            .filters
+            .iter()
+            .map(|row_group_filter| {
+                let bloom_filter = row_group_filter
+                    .iter()
+                    .map(|column_filter| column_filter.data().to_vec())
+                    .collect::<Vec<_>>();
+                sst_pb::sst_bloom_filter::RowGroupFilter { bloom_filter }
+            })
+            .collect::<Vec<_>>();
+
+        sst_pb::SstBloomFilter { filters }
+    }
+}
+
+impl TryFrom<sst_pb::SstBloomFilter> for BloomFilter {
+    type Error = Error;
+
+    fn try_from(src: sst_pb::SstBloomFilter) -> Result<Self> {
+        let filters = src
+            .filters
+            .into_iter()
+            .map(|row_group_filter| {
+                row_group_filter
+                    .bloom_filter
+                    .into_iter()
+                    .map(|encoded_bytes| {
+                        let size = encoded_bytes.len();
+                        let bs: [u8; 256] = encoded_bytes
+                            .try_into()
+                            .ok()
+                            .with_context(|| InvalidBloomFilter { size })?;
+
+                        Ok(Bloom::from(bs))
+                    })
+                    .collect::<Result<Vec<_>>>()
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        Ok(BloomFilter { filters })
+    }
+}
+
 /// Meta data of a sst file
 #[derive(Debug, Clone, PartialEq)]
 pub struct SstMetaData {
@@ -440,6 +509,7 @@ pub struct SstMetaData {
     // total row number
     pub row_num: u64,
     pub storage_format_opts: StorageFormatOptions,
+    pub bloom_filter: BloomFilter,
 }
 
 impl SstMetaData {
@@ -459,6 +529,7 @@ impl From<SstMetaData> for sst_pb::SstMetaData {
             size: src.size,
             row_num: src.row_num,
             storage_format_opts: Some(src.storage_format_opts.into()),
+            bloom_filter: Some(src.bloom_filter.to_pb()),
         }
     }
 }
@@ -479,6 +550,11 @@ impl TryFrom<sst_pb::SstMetaData> for SstMetaData {
             src.storage_format_opts
                 .context(StorageFormatOptionsNotFound)?,
         );
+        let bloom_filter = {
+            let pb_filter = src.bloom_filter.context(BloomFitlerNotFound)?;
+            BloomFilter::try_from(pb_filter)?
+        };
+
         Ok(Self {
             min_key: src.min_key.into(),
             max_key: src.max_key.into(),
@@ -488,6 +564,7 @@ impl TryFrom<sst_pb::SstMetaData> for SstMetaData {
             size: src.size,
             row_num: src.row_num,
             storage_format_opts,
+            bloom_filter,
         })
     }
 }
@@ -670,6 +747,8 @@ pub fn merge_sst_meta(files: &[FileHandle], schema: Schema) -> SstMetaData {
         size: 0,
         row_num: 0,
         storage_format_opts: StorageFormatOptions::new(storage_format),
+        // bloom filter is rebuilt when write sst, so use default here
+        bloom_filter: Default::default(),
     }
 }
 
@@ -723,9 +802,10 @@ pub mod tests {
                 time_range: self.time_range,
                 max_sequence: self.max_sequence,
                 schema: self.schema.clone(),
-                size: 0,
                 row_num: 0,
-                storage_format_opts: StorageFormatOptions::default(),
+                size: 0,
+                storage_format_opts: Default::default(),
+                bloom_filter: Default::default(),
             }
         }
     }