add rt bloom filter pruning to async native data source

Author: dantengsky

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

@@ -38,7 +38,7 @@ use crate::io::BlockReader;
 use crate::io::TableMetaLocationGenerator;
 use crate::io::VirtualColumnReader;
 use crate::operations::read::data_source_with_meta::DataSourceWithMeta;
-use crate::operations::read::runtime_filter_prunner::runtime_filter_pruner;
+use crate::operations::read::runtime_filter_prunner::{runtime_bloom_filter_pruner, runtime_range_filter_pruner};
 use crate::FuseBlockPartInfo;
 
 pub struct ReadNativeDataSource<const BLOCKING_IO: bool> {
@@ -139,7 +139,7 @@ impl SyncSource for ReadNativeDataSource<true> {
                         .ctx
                         .get_min_max_runtime_filter_with_id(self.table_index),
                 );
-                if runtime_filter_pruner(
+                if runtime_range_filter_pruner(
                     self.table_schema.clone(),
                     &part,
                     &filters,
@@ -246,7 +246,7 @@ impl Processor for ReadNativeDataSource<false> {
             );
             let mut native_part_infos = Vec::with_capacity(parts.len());
             for part in parts.into_iter() {
-                if runtime_filter_pruner(
+                if runtime_range_filter_pruner(
                     self.table_schema.clone(),
                     &part,
                     &filters,
@@ -255,6 +255,18 @@ impl Processor for ReadNativeDataSource<false> {
                     continue;
                 }
 
+                if runtime_bloom_filter_pruner(
+                    self.table_schema.clone(),
+                    &part,
+                    &filters,
+                    &self.func_ctx,
+                    &self.block_reader.operator,
+                )
+                .await?
+                {
+                    continue;
+                }
+
                 native_part_infos.push(part.clone());
                 let block_reader = self.block_reader.clone();
                 let index_reader = self.index_reader.clone();

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

@@ -41,7 +41,7 @@ use crate::io::TableMetaLocationGenerator;
 use crate::io::VirtualColumnReader;
 use crate::operations::read::data_source_with_meta::DataSourceWithMeta;
 use crate::operations::read::runtime_filter_prunner::runtime_bloom_filter_pruner;
-use crate::operations::read::runtime_filter_prunner::runtime_filter_pruner;
+use crate::operations::read::runtime_filter_prunner::runtime_range_filter_pruner;
 
 pub struct ReadParquetDataSource<const BLOCKING_IO: bool> {
     func_ctx: FunctionContext,
@@ -128,7 +128,7 @@ impl SyncSource for ReadParquetDataSource<true> {
                         .ctx
                         .get_min_max_runtime_filter_with_id(self.table_index),
                 );
-                if runtime_filter_pruner(
+                if runtime_range_filter_pruner(
                     self.table_schema.clone(),
                     &part,
                     &filters,
@@ -244,7 +244,7 @@ impl Processor for ReadParquetDataSource<false> {
             );
             let mut fuse_part_infos = Vec::with_capacity(parts.len());
             for part in parts.into_iter() {
-                if runtime_filter_pruner(
+                if runtime_range_filter_pruner(
                     self.table_schema.clone(),
                     &part,
                     &filters,

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

@@ -44,7 +44,7 @@ use xorf::Filter;
 use crate::pruning::BloomPrunerCreator;
 use crate::FuseBlockPartInfo;
 
-pub fn runtime_filter_pruner(
+pub fn runtime_range_filter_pruner(
     table_schema: Arc<TableSchema>,
     part: &PartInfoPtr,
     filters: &[Expr<String>],
@@ -92,7 +92,7 @@ pub fn runtime_filter_pruner(
 
     if pruned {
         info!(
-            "Pruned partition  with {:?} rows by runtime range filter",
+            "Pruned partition with {:?} rows by runtime range filter",
             part.nums_rows
         );
         Profile::record_usize_profile(ProfileStatisticsName::RuntimeRangeFilterPrunedParts, 1);
@@ -139,7 +139,7 @@ pub async fn runtime_bloom_filter_pruner(
 
     if pruned {
         info!(
-            "Pruned partition  with {:?} rows by runtime bloom filter",
+            "Pruned partition with {:?} rows by runtime bloom filter",
             part.nums_rows
         );
         Profile::record_usize_profile(ProfileStatisticsName::RuntimeBloomFilterPrunedParts, 1);

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

@@ -551,6 +551,7 @@ impl FuseTable {
         // row_count should be a hint value of  LIMIT,
         // not the count the rows in this partition
 
+        // duplicated code (TODO)
         let bloom_desc = bloom_index_cols.map(|v| BloomIndexDescriptor {
             bloom_index_location: meta.bloom_filter_index_location.clone(),
             bloom_index_size: meta.bloom_filter_index_size,