feat: bloom filter pushdown

crates/iceberg/public-api.txt

@@ -72,6 +72,7 @@ impl iceberg::arrow::ArrowReaderBuilder
 pub fn iceberg::arrow::ArrowReaderBuilder::build(self) -> iceberg::arrow::ArrowReader
 pub fn iceberg::arrow::ArrowReaderBuilder::new(file_io: iceberg::io::FileIO, runtime: iceberg::Runtime) -> Self
 pub fn iceberg::arrow::ArrowReaderBuilder::with_batch_size(self, batch_size: usize) -> Self
+pub fn iceberg::arrow::ArrowReaderBuilder::with_bloom_filter_enabled(self, bloom_filter_enabled: bool) -> Self
 pub fn iceberg::arrow::ArrowReaderBuilder::with_data_file_concurrency_limit(self, val: usize) -> Self
 pub fn iceberg::arrow::ArrowReaderBuilder::with_metadata_size_hint(self, metadata_size_hint: usize) -> Self
 pub fn iceberg::arrow::ArrowReaderBuilder::with_range_coalesce_bytes(self, range_coalesce_bytes: u64) -> Self
@@ -1363,6 +1364,7 @@ pub fn iceberg::scan::TableScanBuilder<'a>::select_all(self) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::select_empty(self) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::snapshot_id(self, snapshot_id: i64) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::with_batch_size(self, batch_size: core::option::Option<usize>) -> Self
+pub fn iceberg::scan::TableScanBuilder<'a>::with_bloom_filter_enabled(self, bloom_filter_enabled: bool) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::with_case_sensitive(self, case_sensitive: bool) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::with_concurrency_limit(self, limit: usize) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::with_data_file_concurrency_limit(self, limit: usize) -> Self

crates/iceberg/src/arrow/reader/mod.rs

@@ -57,6 +57,7 @@ pub struct ArrowReaderBuilder {
     concurrency_limit_data_files: usize,
     row_group_filtering_enabled: bool,
     row_selection_enabled: bool,
+    bloom_filter_enabled: bool,
     parquet_read_options: ParquetReadOptions,
     runtime: Runtime,
 }
@@ -72,6 +73,7 @@ impl ArrowReaderBuilder {
             concurrency_limit_data_files: num_cpus,
             row_group_filtering_enabled: true,
             row_selection_enabled: false,
+            bloom_filter_enabled: false,
             parquet_read_options: ParquetReadOptions::builder().build(),
             runtime,
         }
@@ -102,6 +104,21 @@ impl ArrowReaderBuilder {
         self
     }
 
+    /// Determines whether to enable bloom filter-based row group filtering.
+    ///
+    /// When enabled, if a read is performed with an equality or IN predicate,
+    /// the bloom filter for relevant columns in each row group is read and
+    /// checked. Row groups where the bloom filter proves the value is absent
+    /// are skipped entirely.
+    ///
+    /// Defaults to disabled. Each bloom filter is a separate read, and they are
+    /// issued serially — one round trip per relevant column per row group, before
+    /// any data is read. TODO(#3191)
+    pub fn with_bloom_filter_enabled(mut self, bloom_filter_enabled: bool) -> Self {
+        self.bloom_filter_enabled = bloom_filter_enabled;
+        self
+    }
+
     /// Provide a hint as to the number of bytes to prefetch for parsing the Parquet metadata
     ///
     /// This hint can help reduce the number of fetch requests. For more details see the
@@ -141,6 +158,7 @@ impl ArrowReaderBuilder {
             concurrency_limit_data_files: self.concurrency_limit_data_files,
             row_group_filtering_enabled: self.row_group_filtering_enabled,
             row_selection_enabled: self.row_selection_enabled,
+            bloom_filter_enabled: self.bloom_filter_enabled,
             parquet_read_options: self.parquet_read_options,
         }
     }
@@ -158,5 +176,6 @@ pub struct ArrowReader {
 
     row_group_filtering_enabled: bool,
     row_selection_enabled: bool,
+    bloom_filter_enabled: bool,
     parquet_read_options: ParquetReadOptions,
 }

crates/iceberg/src/arrow/reader/pipeline.rs

@@ -44,6 +44,10 @@ use crate::arrow::record_batch_transformer::RecordBatchTransformerBuilder;
 use crate::arrow::scan_metrics::{CountingFileRead, ScanMetrics, ScanResult};
 use crate::encryption::StandardKeyMetadata;
 use crate::error::Result;
+use crate::expr::BoundPredicate;
+use crate::expr::visitors::bloom_filter_evaluator::{
+    BloomFilterEvaluator, ColumnBloomFilter, collect_bloom_filter_field_ids,
+};
 use crate::io::{FileIO, FileMetadata, FileRead};
 use crate::metadata_columns::{
     RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER, RESERVED_COL_NAME_POS,
@@ -70,6 +74,7 @@ impl ArrowReader {
                 .with_scan_metrics(scan_metrics.clone()),
             row_group_filtering_enabled: self.row_group_filtering_enabled,
             row_selection_enabled: self.row_selection_enabled,
+            bloom_filter_enabled: self.bloom_filter_enabled,
             parquet_read_options: self.parquet_read_options,
             scan_metrics: scan_metrics.clone(),
         };
@@ -123,6 +128,7 @@ struct FileScanTaskReader {
     delete_file_loader: CachingDeleteFileLoader,
     row_group_filtering_enabled: bool,
     row_selection_enabled: bool,
+    bloom_filter_enabled: bool,
     parquet_read_options: ParquetReadOptions,
     scan_metrics: ScanMetrics,
 }
@@ -632,6 +638,30 @@ impl FileScanTaskReader {
                 };
             }
 
+            if self.bloom_filter_enabled {
+                let all_rgs;
+                let candidate_rgs = match &selected_row_group_indices {
+                    Some(indices) => indices.as_slice(),
+                    None => {
+                        all_rgs = (0..record_batch_stream_builder.metadata().num_row_groups())
+                            .collect::<Vec<_>>();
+                        &all_rgs
+                    }
+                };
+
+                let bloom_filtered = Self::filter_row_groups_by_bloom_filter(
+                    &predicate,
+                    &mut record_batch_stream_builder,
+                    candidate_rgs,
+                    &field_id_map,
+                )
+                .await?;
+
+                if bloom_filtered.len() < candidate_rgs.len() {
+                    selected_row_group_indices = Some(bloom_filtered);
+                }
+            }
+
             if self.row_selection_enabled {
                 row_selection = ArrowReader::get_row_selection_for_filter_predicate(
                     &predicate,
@@ -692,6 +722,78 @@ impl FileScanTaskReader {
 
         Ok(Box::pin(record_batch_stream) as ArrowRecordBatchStream)
     }
+
+    /// Reads bloom filters for relevant columns and evaluates the predicate
+    /// against them to filter out row groups that definitely don't match.
+    async fn filter_row_groups_by_bloom_filter(
+        predicate: &BoundPredicate,
+        builder: &mut ParquetRecordBatchStreamBuilder<ArrowFileReader>,
+        candidate_row_groups: &[usize],
+        field_id_map: &HashMap<i32, usize>,
+    ) -> Result<Vec<usize>> {
+        // Only collect field IDs from eq/in predicates — the only types
+        // bloom filters can help with. Skip columns not in the parquet schema.
+        let bloom_filter_field_ids: Vec<i32> = collect_bloom_filter_field_ids(predicate)?
+            .into_iter()
+            .filter(|id| field_id_map.contains_key(id))
+            .collect();
+
+        if bloom_filter_field_ids.is_empty() {
+            return Ok(candidate_row_groups.to_vec());
+        }
+
+        let mut result = Vec::with_capacity(candidate_row_groups.len());
+
+        for &rg_idx in candidate_row_groups {
+            let mut bloom_filters: HashMap<i32, ColumnBloomFilter> = HashMap::new();
+
+            for &field_id in &bloom_filter_field_ids {
+                let col_idx = field_id_map[&field_id];
+                let col_meta = builder.metadata().row_group(rg_idx).column(col_idx);
+
+                // Only attempt to load if this column chunk actually has a bloom filter
+                if col_meta.bloom_filter_offset().is_none() {
+                    continue;
+                }
+
+                let physical_type = col_meta.column_type();
+                let type_length = col_meta.column_descr().type_length();
+
+                match builder
+                    .get_row_group_column_bloom_filter(rg_idx, col_idx)
+                    .await
+                {
+                    Ok(Some(sbbf)) => {
+                        bloom_filters.insert(
+                            field_id,
+                            ColumnBloomFilter::new(sbbf, physical_type, type_length),
+                        );
+                    }
+                    Ok(None) => {}
+                    Err(e) => {
+                        // Left absent from the map, so the evaluator treats the column
+                        // as might-match and the row group survives.
+                        tracing::debug!(
+                            "Bloom filter for field {field_id} in row group {rg_idx} could not be read: {e}"
+                        );
+                    }
+                }
+            }
+
+            match BloomFilterEvaluator::eval(predicate, &bloom_filters) {
+                Ok(true) => result.push(rg_idx),
+                Ok(false) => { /* Row group pruned by bloom filter */ }
+                Err(e) => {
+                    tracing::debug!(
+                        "Bloom filter evaluation failed for row group {rg_idx}, including it: {e}"
+                    );
+                    result.push(rg_idx);
+                }
+            }
+        }
+
+        Ok(result)
+    }
 }
 
 impl ArrowReader {