refactor(fuse): runtime filter into two files: expr_bloom_filter.rs and expr_runtime_pruner.rs (#17804)

refactor(fuse): runtime filter into two files:expr_bloom_filter.rs and expr_runtime_pruner.rs

Author: BohuTANG

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

@@ -22,7 +22,6 @@ mod parquet_data_source;
 mod parquet_data_source_deserializer;
 mod parquet_data_transform_reader;
 mod parquet_rows_fetcher;
-mod runtime_filter_prunner;
 
 mod block_partition_meta;
 mod block_partition_receiver_source;

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

@@ -70,7 +70,7 @@ use crate::io::AggIndexReader;
 use crate::io::BlockReader;
 use crate::io::VirtualColumnReader;
 use crate::operations::read::data_source_with_meta::DataSourceWithMeta;
-use crate::operations::read::runtime_filter_prunner::update_bitmap_with_bloom_filter;
+use crate::pruning::ExprBloomFilter;
 use crate::DEFAULT_ROW_PER_PAGE;
 
 /// A helper struct to store the intermediate state while reading a native partition.
@@ -857,7 +857,9 @@ impl NativeDeserializeDataTransform {
                 let probe_block = self.block_reader.build_block(&[column], None)?;
                 let mut bitmap = MutableBitmap::from_len_zeroed(probe_block.num_rows());
                 let probe_column = probe_block.get_last_column().clone();
-                update_bitmap_with_bloom_filter(probe_column, filter, &mut bitmap)?;
+                // Apply the filter to the probe column.
+                ExprBloomFilter::new(filter.clone()).apply(probe_column, &mut bitmap)?;
+
                 let unset_bits = bitmap.null_count();
                 if unset_bits == bitmap.len() {
                     // skip current page.

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

@@ -38,7 +38,7 @@ use crate::io::TableMetaLocationGenerator;
 use crate::io::VirtualColumnReader;
 use crate::operations::read::block_partition_meta::BlockPartitionMeta;
 use crate::operations::read::data_source_with_meta::DataSourceWithMeta;
-use crate::operations::read::runtime_filter_prunner::runtime_filter_pruner;
+use crate::pruning::ExprRuntimePruner;
 use crate::FuseBlockPartInfo;
 
 pub struct ReadNativeDataTransform<const BLOCKING_IO: bool> {
@@ -123,14 +123,12 @@ impl Transform for ReadNativeDataTransform<true> {
                     self.context
                         .get_min_max_runtime_filter_with_id(self.scan_id),
                 );
-                if runtime_filter_pruner(
-                    self.table_schema.clone(),
-                    &part,
-                    &filters,
-                    &self.func_ctx,
-                )? {
+
+                let runtime_filter = ExprRuntimePruner::new(filters.clone());
+                if runtime_filter.prune(&self.func_ctx, self.table_schema.clone(), &part)? {
                     return Ok(DataBlock::empty());
                 }
+
                 if let Some(index_reader) = self.index_reader.as_ref() {
                     let fuse_part = FuseBlockPartInfo::from_part(&part)?;
                     let loc =
@@ -201,13 +199,9 @@ impl AsyncTransform for ReadNativeDataTransform<false> {
                             .get_min_max_runtime_filter_with_id(self.scan_id),
                     );
                     let mut native_part_infos = Vec::with_capacity(parts.len());
+                    let runtime_filter = ExprRuntimePruner::new(filters.clone());
                     for part in parts.into_iter() {
-                        if runtime_filter_pruner(
-                            self.table_schema.clone(),
-                            &part,
-                            &filters,
-                            &self.func_ctx,
-                        )? {
+                        if runtime_filter.prune(&self.func_ctx, self.table_schema.clone(), &part)? {
                             continue;
                         }
 

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

@@ -53,7 +53,7 @@ use crate::io::AggIndexReader;
 use crate::io::BlockReader;
 use crate::io::VirtualColumnReader;
 use crate::operations::read::data_source_with_meta::DataSourceWithMeta;
-use crate::operations::read::runtime_filter_prunner::update_bitmap_with_bloom_filter;
+use crate::pruning::ExprBloomFilter;
 
 pub struct DeserializeDataTransform {
     ctx: Arc<dyn TableContext>,
@@ -165,7 +165,9 @@ impl DeserializeDataTransform {
             let probe_column = probe_block_entry
                 .value
                 .convert_to_full_column(&probe_block_entry.data_type, data_block.num_rows());
-            update_bitmap_with_bloom_filter(probe_column, filter, &mut bitmap)?;
+
+            // Apply bloom filter
+            ExprBloomFilter::new(filter.clone()).apply(probe_column, &mut bitmap)?;
             bitmaps.push(bitmap);
         }
         if !bitmaps.is_empty() {

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

@@ -40,7 +40,7 @@ use crate::io::TableMetaLocationGenerator;
 use crate::io::VirtualColumnReader;
 use crate::operations::read::block_partition_meta::BlockPartitionMeta;
 use crate::operations::read::data_source_with_meta::DataSourceWithMeta;
-use crate::operations::read::runtime_filter_prunner::runtime_filter_pruner;
+use crate::pruning::ExprRuntimePruner;
 
 pub struct ReadParquetDataTransform<const BLOCKING_IO: bool> {
     func_ctx: FunctionContext,
@@ -124,12 +124,9 @@ impl Transform for ReadParquetDataTransform<true> {
                     self.context
                         .get_min_max_runtime_filter_with_id(self.scan_id),
                 );
-                if runtime_filter_pruner(
-                    self.table_schema.clone(),
-                    &part,
-                    &filters,
-                    &self.func_ctx,
-                )? {
+
+                let runtime_filter = ExprRuntimePruner::new(filters.clone());
+                if runtime_filter.prune(&self.func_ctx, self.table_schema.clone(), &part)? {
                     return Ok(DataBlock::empty());
                 }
 
@@ -207,13 +204,10 @@ impl AsyncTransform for ReadParquetDataTransform<false> {
                             .get_min_max_runtime_filter_with_id(self.scan_id),
                     );
                     let mut fuse_part_infos = Vec::with_capacity(parts.len());
+
+                    let runtime_filter = ExprRuntimePruner::new(filters.clone());
                     for part in parts.into_iter() {
-                        if runtime_filter_pruner(
-                            self.table_schema.clone(),
-                            &part,
-                            &filters,
-                            &self.func_ctx,
-                        )? {
+                        if runtime_filter.prune(&self.func_ctx, self.table_schema.clone(), &part)? {
                             continue;
                         }