wip

Author: dantengsky

src/query/storages/fuse/src/fuse_part.rs

@@ -48,6 +48,10 @@ pub struct FuseBlockPartInfo {
 
     pub sort_min_max: Option<(Scalar, Scalar)>,
     pub block_meta_index: Option<BlockMetaIndex>,
+
+    pub bloom_index_location: Option<Location>,
+    pub bloom_index_size: u64,
+    pub bloom_index_column_ids: Vec<ColumnId>,
 }
 
 #[typetag::serde(name = "fuse")]
@@ -84,6 +88,9 @@ impl FuseBlockPartInfo {
         sort_min_max: Option<(Scalar, Scalar)>,
         block_meta_index: Option<BlockMetaIndex>,
         create_on: Option<DateTime<Utc>>,
+        bloom_index_location: Option<Location>,
+        bloom_index_size: u64,
+        bloom_index_column_ids: Vec<ColumnId>,
     ) -> Arc<Box<dyn PartInfo>> {
         Arc::new(Box::new(FuseBlockPartInfo {
             location,
@@ -94,6 +101,9 @@ impl FuseBlockPartInfo {
             sort_min_max,
             block_meta_index,
             columns_stat,
+            bloom_index_location,
+            bloom_index_size,
+            bloom_index_column_ids,
         }))
     }
 

src/query/storages/fuse/src/operations/read/runtime_filter_prunner.rs

@@ -35,11 +35,15 @@ use databend_common_expression::Scalar;
 use databend_common_expression::TableSchema;
 use databend_common_functions::BUILTIN_FUNCTIONS;
 use databend_common_hashtable::FastHash;
+use databend_common_sql::BloomIndexColumns;
 use databend_storages_common_index::statistics_to_domain;
+use databend_storages_common_table_meta::meta::BlockMeta;
 use log::info;
+use opendal::Operator;
 use xorf::BinaryFuse16;
 use xorf::Filter;
 
+use crate::pruning::BloomPrunerCreator;
 use crate::FuseBlockPartInfo;
 
 pub fn runtime_filter_pruner(
@@ -52,6 +56,7 @@ pub fn runtime_filter_pruner(
         return Ok(false);
     }
     let part = FuseBlockPartInfo::from_part(part)?;
+    eprintln!("filters {:#?}", filters);
     let pruned = filters.iter().any(|filter| {
         let column_refs = filter.column_refs();
         // Currently only support filter with one column(probe key).
@@ -90,15 +95,54 @@ pub fn runtime_filter_pruner(
         false
     });
 
-    info!(
-        "Pruned partition with {:?} rows by runtime filter",
-        part.nums_rows
-    );
-    Profile::record_usize_profile(ProfileStatisticsName::RuntimeFilterPruneParts, 1);
+    if pruned {
+        info!(
+            "Pruned partition with {:?} rows by runtime filter",
+            part.nums_rows
+        );
+        Profile::record_usize_profile(ProfileStatisticsName::RuntimeFilterPruneParts, 1);
+    } else {
+        Profile::record_usize_profile(ProfileStatisticsName::RuntimeFilterPruneParts, 0);
+    }
 
     Ok(pruned)
 }
 
+pub async fn runtime_bloom_filter_pruner(
+    table_schema: Arc<TableSchema>,
+    part: &PartInfoPtr,
+    filters: &[Expr<String>],
+    func_ctx: &FunctionContext,
+    dal: Operator,
+    bloom_index_cols: BloomIndexColumns,
+) -> Result<bool> {
+    let part = FuseBlockPartInfo::from_part(part)?;
+
+    let block_meta: BlockMeta;
+    let index_location = part.bloom_filter_index_location.clone();
+    let index_size = part.bloom_filter_index_size;
+    let column_ids = part.columns_meta.keys().cloned().collect::<Vec<_>>();
+
+    for filter_expr in filters {
+        if let Some(bloom_pruner) = BloomPrunerCreator::create(
+            func_ctx.clone(),
+            &table_schema,
+            dal.clone(),
+            Some(filter_expr),
+            bloom_index_cols.clone(),
+        )? {
+            let should_keep = bloom_pruner
+                .should_keep(&index_location, index_size, column_ids.clone())
+                .await;
+            if !should_keep {
+                return Ok(true);
+            }
+        }
+    }
+
+    Ok(false)
+}
+
 pub(crate) fn update_bitmap_with_bloom_filter(
     column: Column,
     filter: &BinaryFuse16,

src/query/storages/fuse/src/operations/read_partitions.rs

@@ -27,14 +27,14 @@ use databend_common_catalog::plan::TopK;
 use databend_common_catalog::table::Table;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
-use databend_common_expression::Scalar;
+use databend_common_expression::{ColumnId, Scalar};
 use databend_common_expression::TableSchemaRef;
 use databend_common_sql::field_default_value;
 use databend_common_storage::ColumnNodes;
 use databend_storages_common_cache::CacheAccessor;
 use databend_storages_common_cache_manager::CachedObject;
 use databend_storages_common_pruner::BlockMetaIndex;
-use databend_storages_common_table_meta::meta::BlockMeta;
+use databend_storages_common_table_meta::meta::{BlockMeta, Location};
 use databend_storages_common_table_meta::meta::ColumnStatistics;
 use databend_storages_common_table_meta::table::ChangeType;
 use log::debug;
@@ -430,6 +430,7 @@ impl FuseTable {
     }
 
     fn all_columns_part(
+        &self,
         schema: Option<&TableSchemaRef>,
         block_meta_index: &Option<BlockMetaIndex>,
         top_k: &Option<(TopK, Scalar)>,
@@ -476,6 +477,9 @@ impl FuseTable {
             sort_min_max,
             block_meta_index.to_owned(),
             create_on,
+            meta.bloom_filter_index_location.clone(),
+            meta.bloom_filter_index_size,
+            self.bloom_index_cols.clone(),
         )
     }