laskoviymishka commented on code in PR #2398:
URL: https://github.com/apache/iceberg-rust/pull/2398#discussion_r3972369361
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -692,6 +721,69 @@ 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: &crate::expr::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 {
Review Comment:
Every `(row_group, column)` pair here ends in a separate sequential `.await`
on `get_row_group_column_bloom_filter`. For 100 row groups and 3 predicate
columns that's 300 round trips in series before a single group is skipped — on
object storage that latency can easily swamp whatever I/O the pruning saves,
which undercuts the reason to enable this at all.
I'd fetch the filters concurrently — `join_all` over the `(rg, col)` pairs,
or at least per-column within a row group — so the round trips overlap.
Thoughts?
##########
crates/iceberg/src/arrow/reader/row_filter.rs:
##########
@@ -1280,4 +1280,175 @@ mod tests {
"positional deletes must be applied correctly even when page
indexes are absent"
);
}
+
+ /// Tests that bloom filter pushdown correctly prunes row groups.
+ #[tokio::test]
+ async fn test_bloom_filter_pushdown_prunes_row_groups() {
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("id", DataType::Int32,
false).with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "1".to_string(),
+ )])),
+ ]));
+
+ let tmp_dir = TempDir::new().unwrap();
+ let file_path = format!("{}/bloom_test.parquet",
tmp_dir.path().to_str().unwrap());
+
+ // Write a Parquet file with 3 row groups, each containing distinct
values,
+ // with bloom filters enabled.
+ // Row group 0: ids 0..100
+ // Row group 1: ids 100..200
+ // Row group 2: ids 200..300
+ let props = WriterProperties::builder()
+ .set_compression(Compression::SNAPPY)
+ .set_max_row_group_row_count(Some(100))
+ .set_bloom_filter_enabled(true)
+ .build();
+
+ let file = File::create(&file_path).unwrap();
+ let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(),
Some(props)).unwrap();
+
+ for batch_start in [0, 100, 200] {
+ let batch = RecordBatch::try_new(arrow_schema.clone(),
vec![Arc::new(
+ Int32Array::from((batch_start..batch_start +
100).collect::<Vec<i32>>()),
+ )])
+ .unwrap();
+ writer.write(&batch).unwrap();
+ }
+ writer.close().unwrap();
+
+ let file_io = FileIO::new_with_fs();
+
+ // Query for id = 150, which is only in row group 1.
+ // With bloom filter pushdown, row groups 0 and 2 should be pruned.
+ let predicate = Reference::new("id").equal_to(Datum::int(150));
+
+ let reader = ArrowReaderBuilder::new(file_io.clone(),
Runtime::current())
+ .with_bloom_filter_enabled(true)
+ .build();
+
+ let task = FileScanTask::builder()
+
.with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len())
+ .with_start(0)
+ .with_length(0)
+ .with_data_file_path(file_path.clone())
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(schema.clone())
+ .with_project_field_ids(vec![1])
+ .with_predicate(Some(predicate.bind(schema.clone(),
true).unwrap()))
+ .with_case_sensitive(false)
+ .build()
+ .unwrap();
+
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as
FileScanTaskStream;
+
+ let result = reader
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect::<Vec<RecordBatch>>()
+ .await
+ .unwrap();
+
+ // Only row group 1 (ids 100..200) should be read. The row filter
+ // then further filters to just id=150.
+ let total_rows: usize = result.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(total_rows, 1, "Should find exactly one row matching
id=150");
Review Comment:
Both integration tests only assert the final row count, which the existing
row filter already guarantees — a pruner that read every row group and skipped
nothing would pass both identically. So they don't actually prove bloom filter
pruning happened.
I'd assert on something that only changes when a group is skipped:
`metrics().bytes_read()` against a `with_bloom_filter_enabled(false)` baseline,
or a per-row-group read counter. Once there's a metric that observes pruning,
it'd also be worth reaching past `eq`-on-`INT32` to cover the `IN` path and a
string column end to end, since those are only unit-tested today.
##########
crates/iceberg/src/expr/visitors/bloom_filter_evaluator.rs:
##########
@@ -0,0 +1,1325 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Evaluates predicates against Parquet bloom filters to determine whether
+//! a row group can be skipped.
+
+use std::collections::{HashMap, HashSet};
+
+use fnv::FnvHashSet;
+use parquet::basic::Type as PhysicalType;
+use parquet::bloom_filter::Sbbf;
+use parquet::data_type::ByteArray;
+
+use crate::Result;
+use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor,
visit};
+use crate::expr::{BoundPredicate, BoundReference};
+use crate::spec::decimal_utils::decimal_to_fixed_length_bytes_exact;
+use crate::spec::{Datum, PrimitiveLiteral};
+
+const ROW_GROUP_MIGHT_MATCH: Result<bool> = Ok(true);
+const ROW_GROUP_CANT_MATCH: Result<bool> = Ok(false);
+
+/// A column's bloom filter for one row group, together with the file's
physical
+/// encoding of that column. A probe must be encoded the way the writer encoded
+/// the values it inserted, so the encoding travels with the filter.
+pub(crate) struct ColumnBloomFilter {
+ sbbf: Sbbf,
+ physical_type: PhysicalType,
+ /// `type_length` from the file's column descriptor. Only meaningful for
+ /// `FIXED_LEN_BYTE_ARRAY`.
+ type_length: i32,
+}
+
+impl ColumnBloomFilter {
+ pub(crate) fn new(sbbf: Sbbf, physical_type: PhysicalType, type_length:
i32) -> Self {
+ Self {
+ sbbf,
+ physical_type,
+ type_length,
+ }
+ }
+}
+
+pub(crate) struct BloomFilterEvaluator<'a> {
+ /// Maps Iceberg field_id -> bloom filter for this row group
+ bloom_filters: &'a HashMap<i32, ColumnBloomFilter>,
+}
+
+impl<'a> BloomFilterEvaluator<'a> {
+ /// Evaluate the predicate against the provided bloom filters.
+ /// Returns `false` if the row group definitely does not match,
+ /// `true` if it might match.
+ pub(crate) fn eval(
+ filter: &BoundPredicate,
+ bloom_filters: &HashMap<i32, ColumnBloomFilter>,
+ ) -> Result<bool> {
+ if bloom_filters.is_empty() {
+ return ROW_GROUP_MIGHT_MATCH;
+ }
+
+ let mut evaluator = BloomFilterEvaluator { bloom_filters };
+ visit(&mut evaluator, filter)
+ }
+
+ fn check_datum(&self, reference: &BoundReference, datum: &Datum) -> bool {
+ let field_id = reference.field().id;
+ let Some(column) = self.bloom_filters.get(&field_id) else {
+ // No bloom filter for this column — conservatively might match
+ return true;
+ };
+
+ check_in_bloom_filter(column, datum)
+ }
+}
+
+/// Collects field IDs that appear in `eq` or `in` predicates — the only
+/// predicate types that benefit from bloom filter checks.
+pub(crate) fn collect_bloom_filter_field_ids(predicate: &BoundPredicate) ->
Result<HashSet<i32>> {
Review Comment:
This collects field IDs from `eq`/`in` nodes even when they sit under a
`NOT`, so `NOT(id = 5)` and `NOT(id IN {...})` both end up loading a bloom
filter that can never prune anything — the evaluator's `not()` always returns
might-match and throws the inner result away.
That's not just a missed optimization: the equality-delete path builds
`(scan_pred) AND NOT(del_col IN {del_vals})`, so any scan with equality deletes
fetches `del_col`'s bloom filter on every candidate row group and discards it.
With bloom filters on, that's slower than with them off, in exactly the case
they'd most often be combined.
Java sidesteps this by running `rewriteNot` before binding, so its visitor
never sees a `NOT` (the `not()` handler throws). I'd do the same here —
normalize with `rewrite_not` before `collect_bloom_filter_field_ids`/`eval` —
or, if we'd rather not, track NOT-depth in the collector so `eq`/`in` under a
`NOT` don't insert. The rewrite path also buys back `NOT(NOT(x = v))` pruning
for free. wdyt?
##########
crates/iceberg/src/expr/visitors/bloom_filter_evaluator.rs:
##########
@@ -0,0 +1,1325 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Evaluates predicates against Parquet bloom filters to determine whether
+//! a row group can be skipped.
+
+use std::collections::{HashMap, HashSet};
+
+use fnv::FnvHashSet;
+use parquet::basic::Type as PhysicalType;
+use parquet::bloom_filter::Sbbf;
+use parquet::data_type::ByteArray;
+
+use crate::Result;
+use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor,
visit};
+use crate::expr::{BoundPredicate, BoundReference};
+use crate::spec::decimal_utils::decimal_to_fixed_length_bytes_exact;
+use crate::spec::{Datum, PrimitiveLiteral};
+
+const ROW_GROUP_MIGHT_MATCH: Result<bool> = Ok(true);
+const ROW_GROUP_CANT_MATCH: Result<bool> = Ok(false);
+
+/// A column's bloom filter for one row group, together with the file's
physical
+/// encoding of that column. A probe must be encoded the way the writer encoded
+/// the values it inserted, so the encoding travels with the filter.
+pub(crate) struct ColumnBloomFilter {
+ sbbf: Sbbf,
+ physical_type: PhysicalType,
+ /// `type_length` from the file's column descriptor. Only meaningful for
+ /// `FIXED_LEN_BYTE_ARRAY`.
+ type_length: i32,
+}
+
+impl ColumnBloomFilter {
+ pub(crate) fn new(sbbf: Sbbf, physical_type: PhysicalType, type_length:
i32) -> Self {
+ Self {
+ sbbf,
+ physical_type,
+ type_length,
+ }
+ }
+}
+
+pub(crate) struct BloomFilterEvaluator<'a> {
+ /// Maps Iceberg field_id -> bloom filter for this row group
+ bloom_filters: &'a HashMap<i32, ColumnBloomFilter>,
+}
+
+impl<'a> BloomFilterEvaluator<'a> {
+ /// Evaluate the predicate against the provided bloom filters.
+ /// Returns `false` if the row group definitely does not match,
+ /// `true` if it might match.
+ pub(crate) fn eval(
+ filter: &BoundPredicate,
+ bloom_filters: &HashMap<i32, ColumnBloomFilter>,
+ ) -> Result<bool> {
+ if bloom_filters.is_empty() {
+ return ROW_GROUP_MIGHT_MATCH;
+ }
+
+ let mut evaluator = BloomFilterEvaluator { bloom_filters };
+ visit(&mut evaluator, filter)
+ }
+
+ fn check_datum(&self, reference: &BoundReference, datum: &Datum) -> bool {
+ let field_id = reference.field().id;
+ let Some(column) = self.bloom_filters.get(&field_id) else {
+ // No bloom filter for this column — conservatively might match
+ return true;
+ };
+
+ check_in_bloom_filter(column, datum)
+ }
+}
+
+/// Collects field IDs that appear in `eq` or `in` predicates — the only
+/// predicate types that benefit from bloom filter checks.
+pub(crate) fn collect_bloom_filter_field_ids(predicate: &BoundPredicate) ->
Result<HashSet<i32>> {
+ let mut visitor = BloomFilterFieldIdCollector {
+ field_ids: HashSet::new(),
+ };
+ visit(&mut visitor, predicate)?;
+ Ok(visitor.field_ids)
+}
+
+struct BloomFilterFieldIdCollector {
+ field_ids: HashSet<i32>,
+}
+
+impl BoundPredicateVisitor for BloomFilterFieldIdCollector {
+ type T = ();
+
+ fn always_true(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ fn always_false(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ fn and(&mut self, _lhs: (), _rhs: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn or(&mut self, _lhs: (), _rhs: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn not(&mut self, _inner: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn is_null(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn not_null(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn is_nan(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn not_nan(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn less_than(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn less_than_or_eq(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn greater_than(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn greater_than_or_eq(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn eq(&mut self, r: &BoundReference, _l: &Datum, _p: &BoundPredicate) ->
Result<()> {
+ self.field_ids.insert(r.field().id);
+ Ok(())
+ }
+
+ fn not_eq(&mut self, _r: &BoundReference, _l: &Datum, _p: &BoundPredicate)
-> Result<()> {
+ Ok(())
+ }
+
+ fn starts_with(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn not_starts_with(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn r#in(
+ &mut self,
+ r: &BoundReference,
+ _literals: &FnvHashSet<Datum>,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ self.field_ids.insert(r.field().id);
+ Ok(())
+ }
+
+ fn not_in(
+ &mut self,
+ _r: &BoundReference,
+ _literals: &FnvHashSet<Datum>,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+}
+
+/// Check whether a datum value might be present in the bloom filter.
+///
+/// The value must be checked using the same physical encoding the Parquet
+/// writer used when inserting into the bloom filter. We use the actual
+/// physical type from the column metadata to ensure correctness regardless
+/// of which writer produced the file.
+fn check_in_bloom_filter(column: &ColumnBloomFilter, datum: &Datum) -> bool {
+ let ColumnBloomFilter {
+ sbbf,
+ physical_type,
+ type_length,
+ } = column;
+ let physical_type = *physical_type;
+
+ match datum.literal() {
+ PrimitiveLiteral::Boolean(v) => sbbf.check(v),
+ // A promoted column (int -> long, float -> double) keeps its original
+ // physical width in files written before the promotion, and the writer
+ // hashed that width, so probe at the file's width rather than the
+ // predicate's.
+ PrimitiveLiteral::Int(v) => match physical_type {
+ PhysicalType::INT32 => sbbf.check(v),
+ PhysicalType::INT64 => sbbf.check(&i64::from(*v)),
+ _ => true,
+ },
+ PrimitiveLiteral::Long(v) => match physical_type {
+ PhysicalType::INT64 => sbbf.check(v),
+ PhysicalType::INT32 => match i32::try_from(*v) {
+ Ok(narrowed) => sbbf.check(&narrowed),
+ // Out of range for the column, so it cannot be present.
Review Comment:
The comment says the value "cannot be present," which reads like we should
prune (return `false`), but the arm returns `true` (might-match). The behavior
is right — conservatively keeping the group — it's just the comment that's
backwards, and it's exactly the sort of thing that could talk a future reader
into flipping the arm.
I'd reword to something like "value can't fit in an INT32 column, so we
can't probe it — conservatively keep the row group."
##########
crates/iceberg/src/expr/visitors/bloom_filter_evaluator.rs:
##########
@@ -0,0 +1,1325 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Evaluates predicates against Parquet bloom filters to determine whether
+//! a row group can be skipped.
+
+use std::collections::{HashMap, HashSet};
+
+use fnv::FnvHashSet;
+use parquet::basic::Type as PhysicalType;
+use parquet::bloom_filter::Sbbf;
+use parquet::data_type::ByteArray;
+
+use crate::Result;
+use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor,
visit};
+use crate::expr::{BoundPredicate, BoundReference};
+use crate::spec::decimal_utils::decimal_to_fixed_length_bytes_exact;
+use crate::spec::{Datum, PrimitiveLiteral};
+
+const ROW_GROUP_MIGHT_MATCH: Result<bool> = Ok(true);
+const ROW_GROUP_CANT_MATCH: Result<bool> = Ok(false);
+
+/// A column's bloom filter for one row group, together with the file's
physical
+/// encoding of that column. A probe must be encoded the way the writer encoded
+/// the values it inserted, so the encoding travels with the filter.
+pub(crate) struct ColumnBloomFilter {
+ sbbf: Sbbf,
+ physical_type: PhysicalType,
+ /// `type_length` from the file's column descriptor. Only meaningful for
+ /// `FIXED_LEN_BYTE_ARRAY`.
+ type_length: i32,
+}
+
+impl ColumnBloomFilter {
+ pub(crate) fn new(sbbf: Sbbf, physical_type: PhysicalType, type_length:
i32) -> Self {
+ Self {
+ sbbf,
+ physical_type,
+ type_length,
+ }
+ }
+}
+
+pub(crate) struct BloomFilterEvaluator<'a> {
+ /// Maps Iceberg field_id -> bloom filter for this row group
+ bloom_filters: &'a HashMap<i32, ColumnBloomFilter>,
+}
+
+impl<'a> BloomFilterEvaluator<'a> {
+ /// Evaluate the predicate against the provided bloom filters.
+ /// Returns `false` if the row group definitely does not match,
+ /// `true` if it might match.
+ pub(crate) fn eval(
+ filter: &BoundPredicate,
+ bloom_filters: &HashMap<i32, ColumnBloomFilter>,
+ ) -> Result<bool> {
+ if bloom_filters.is_empty() {
+ return ROW_GROUP_MIGHT_MATCH;
+ }
+
+ let mut evaluator = BloomFilterEvaluator { bloom_filters };
+ visit(&mut evaluator, filter)
+ }
+
+ fn check_datum(&self, reference: &BoundReference, datum: &Datum) -> bool {
+ let field_id = reference.field().id;
+ let Some(column) = self.bloom_filters.get(&field_id) else {
+ // No bloom filter for this column — conservatively might match
+ return true;
+ };
+
+ check_in_bloom_filter(column, datum)
+ }
+}
+
+/// Collects field IDs that appear in `eq` or `in` predicates — the only
+/// predicate types that benefit from bloom filter checks.
+pub(crate) fn collect_bloom_filter_field_ids(predicate: &BoundPredicate) ->
Result<HashSet<i32>> {
+ let mut visitor = BloomFilterFieldIdCollector {
+ field_ids: HashSet::new(),
+ };
+ visit(&mut visitor, predicate)?;
+ Ok(visitor.field_ids)
+}
+
+struct BloomFilterFieldIdCollector {
+ field_ids: HashSet<i32>,
+}
+
+impl BoundPredicateVisitor for BloomFilterFieldIdCollector {
+ type T = ();
+
+ fn always_true(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ fn always_false(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ fn and(&mut self, _lhs: (), _rhs: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn or(&mut self, _lhs: (), _rhs: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn not(&mut self, _inner: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn is_null(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn not_null(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn is_nan(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn not_nan(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn less_than(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn less_than_or_eq(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn greater_than(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn greater_than_or_eq(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn eq(&mut self, r: &BoundReference, _l: &Datum, _p: &BoundPredicate) ->
Result<()> {
+ self.field_ids.insert(r.field().id);
+ Ok(())
+ }
+
+ fn not_eq(&mut self, _r: &BoundReference, _l: &Datum, _p: &BoundPredicate)
-> Result<()> {
+ Ok(())
+ }
+
+ fn starts_with(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn not_starts_with(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn r#in(
+ &mut self,
+ r: &BoundReference,
+ _literals: &FnvHashSet<Datum>,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ self.field_ids.insert(r.field().id);
+ Ok(())
+ }
+
+ fn not_in(
+ &mut self,
+ _r: &BoundReference,
+ _literals: &FnvHashSet<Datum>,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+}
+
+/// Check whether a datum value might be present in the bloom filter.
+///
+/// The value must be checked using the same physical encoding the Parquet
+/// writer used when inserting into the bloom filter. We use the actual
+/// physical type from the column metadata to ensure correctness regardless
+/// of which writer produced the file.
+fn check_in_bloom_filter(column: &ColumnBloomFilter, datum: &Datum) -> bool {
+ let ColumnBloomFilter {
+ sbbf,
+ physical_type,
+ type_length,
+ } = column;
+ let physical_type = *physical_type;
+
+ match datum.literal() {
+ PrimitiveLiteral::Boolean(v) => sbbf.check(v),
+ // A promoted column (int -> long, float -> double) keeps its original
+ // physical width in files written before the promotion, and the writer
+ // hashed that width, so probe at the file's width rather than the
+ // predicate's.
+ PrimitiveLiteral::Int(v) => match physical_type {
+ PhysicalType::INT32 => sbbf.check(v),
+ PhysicalType::INT64 => sbbf.check(&i64::from(*v)),
+ _ => true,
+ },
+ PrimitiveLiteral::Long(v) => match physical_type {
+ PhysicalType::INT64 => sbbf.check(v),
+ PhysicalType::INT32 => match i32::try_from(*v) {
+ Ok(narrowed) => sbbf.check(&narrowed),
+ // Out of range for the column, so it cannot be present.
+ Err(_) => true,
+ },
+ _ => true,
+ },
+ PrimitiveLiteral::Float(v) => match physical_type {
+ PhysicalType::FLOAT => sbbf.check(&v.0),
+ PhysicalType::DOUBLE => sbbf.check(&f64::from(v.0)),
+ _ => true,
+ },
+ PrimitiveLiteral::Double(v) => match physical_type {
+ PhysicalType::DOUBLE => sbbf.check(&v.0),
+ PhysicalType::FLOAT => {
+ let narrowed = v.0 as f32;
+ // Only an exactly representable value can equal a widened f32.
+ if f64::from(narrowed) == v.0 {
+ sbbf.check(&narrowed)
+ } else {
+ true
+ }
+ }
+ _ => true,
+ },
+ PrimitiveLiteral::String(v) => sbbf.check(v.as_str()),
+ PrimitiveLiteral::Binary(v) => sbbf.check(v.as_slice()),
+ PrimitiveLiteral::Int128(v) => {
+ // Decimal: dispatch based on the actual Parquet physical type
+ // from the file, not inferred from precision.
+ match physical_type {
+ PhysicalType::INT32 => sbbf.check(&(*v as i32)),
Review Comment:
These `*v as i32` / `*v as i64` casts truncate silently, while the `Long ->
INT32` arm just above uses `i32::try_from` with a conservative fallback. It's
safe here only because an i128 that doesn't fit can't be in the narrower column
— but that reasoning isn't obvious, and the mismatch with the arm above is the
kind of thing someone later "cleans up" into a real false negative.
I'd match the `Long` arm and use `try_from` with `Err(_) => true` for both
INT32 and INT64 so the intent is self-documenting.
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -692,6 +721,69 @@ 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: &crate::expr::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(_) => {
+ // If we can't read the bloom filter, conservatively
include the row group
Review Comment:
Both `Err(_)` arms — the bloom-filter read here and the `eval` error just
below — swallow the error with no trace. If reads start failing systematically
(corrupt footer, wrong offset, a network blip), the reader silently falls back
to a full scan and there's no way to tell that from "this file just has no
bloom filters."
A `tracing::debug!`/`warn!` with the error and the row-group/column
coordinates would make that degradation visible without changing the safe
fallback.
##########
crates/iceberg/src/spec/values/decimal_utils.rs:
##########
@@ -196,6 +196,33 @@ pub fn i128_to_be_bytes_min(value: i128) -> Vec<u8> {
bytes[start..].to_vec()
}
+/// Encode an i128 as exactly `len` big-endian two's complement bytes,
matching a
+/// Parquet `FIXED_LEN_BYTE_ARRAY` column of that declared `type_length`.
+///
+/// Returns `None` if `value` does not fit in `len` bytes, since a truncated
+/// encoding would represent a different number.
+pub fn decimal_to_fixed_length_bytes_exact(value: i128, len: usize) ->
Option<Vec<u8>> {
Review Comment:
This is only called from within the crate, but it's `pub`, so it lands on
the public API surface under semver. `pub(crate)` covers the call site and the
same-file test module still sees it.
##########
crates/iceberg/src/expr/visitors/bloom_filter_evaluator.rs:
##########
@@ -0,0 +1,1325 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Evaluates predicates against Parquet bloom filters to determine whether
+//! a row group can be skipped.
+
+use std::collections::{HashMap, HashSet};
+
+use fnv::FnvHashSet;
+use parquet::basic::Type as PhysicalType;
+use parquet::bloom_filter::Sbbf;
+use parquet::data_type::ByteArray;
+
+use crate::Result;
+use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor,
visit};
+use crate::expr::{BoundPredicate, BoundReference};
+use crate::spec::decimal_utils::decimal_to_fixed_length_bytes_exact;
+use crate::spec::{Datum, PrimitiveLiteral};
+
+const ROW_GROUP_MIGHT_MATCH: Result<bool> = Ok(true);
+const ROW_GROUP_CANT_MATCH: Result<bool> = Ok(false);
+
+/// A column's bloom filter for one row group, together with the file's
physical
+/// encoding of that column. A probe must be encoded the way the writer encoded
+/// the values it inserted, so the encoding travels with the filter.
+pub(crate) struct ColumnBloomFilter {
+ sbbf: Sbbf,
+ physical_type: PhysicalType,
+ /// `type_length` from the file's column descriptor. Only meaningful for
+ /// `FIXED_LEN_BYTE_ARRAY`.
+ type_length: i32,
+}
+
+impl ColumnBloomFilter {
+ pub(crate) fn new(sbbf: Sbbf, physical_type: PhysicalType, type_length:
i32) -> Self {
+ Self {
+ sbbf,
+ physical_type,
+ type_length,
+ }
+ }
+}
+
+pub(crate) struct BloomFilterEvaluator<'a> {
+ /// Maps Iceberg field_id -> bloom filter for this row group
+ bloom_filters: &'a HashMap<i32, ColumnBloomFilter>,
+}
+
+impl<'a> BloomFilterEvaluator<'a> {
+ /// Evaluate the predicate against the provided bloom filters.
+ /// Returns `false` if the row group definitely does not match,
+ /// `true` if it might match.
+ pub(crate) fn eval(
+ filter: &BoundPredicate,
+ bloom_filters: &HashMap<i32, ColumnBloomFilter>,
+ ) -> Result<bool> {
+ if bloom_filters.is_empty() {
+ return ROW_GROUP_MIGHT_MATCH;
+ }
+
+ let mut evaluator = BloomFilterEvaluator { bloom_filters };
+ visit(&mut evaluator, filter)
+ }
+
+ fn check_datum(&self, reference: &BoundReference, datum: &Datum) -> bool {
+ let field_id = reference.field().id;
+ let Some(column) = self.bloom_filters.get(&field_id) else {
+ // No bloom filter for this column — conservatively might match
+ return true;
+ };
+
+ check_in_bloom_filter(column, datum)
+ }
+}
+
+/// Collects field IDs that appear in `eq` or `in` predicates — the only
+/// predicate types that benefit from bloom filter checks.
+pub(crate) fn collect_bloom_filter_field_ids(predicate: &BoundPredicate) ->
Result<HashSet<i32>> {
+ let mut visitor = BloomFilterFieldIdCollector {
+ field_ids: HashSet::new(),
+ };
+ visit(&mut visitor, predicate)?;
+ Ok(visitor.field_ids)
+}
+
+struct BloomFilterFieldIdCollector {
+ field_ids: HashSet<i32>,
+}
+
+impl BoundPredicateVisitor for BloomFilterFieldIdCollector {
+ type T = ();
+
+ fn always_true(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ fn always_false(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ fn and(&mut self, _lhs: (), _rhs: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn or(&mut self, _lhs: (), _rhs: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn not(&mut self, _inner: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn is_null(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn not_null(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn is_nan(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn not_nan(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn less_than(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn less_than_or_eq(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn greater_than(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn greater_than_or_eq(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn eq(&mut self, r: &BoundReference, _l: &Datum, _p: &BoundPredicate) ->
Result<()> {
+ self.field_ids.insert(r.field().id);
+ Ok(())
+ }
+
+ fn not_eq(&mut self, _r: &BoundReference, _l: &Datum, _p: &BoundPredicate)
-> Result<()> {
+ Ok(())
+ }
+
+ fn starts_with(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn not_starts_with(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn r#in(
+ &mut self,
+ r: &BoundReference,
+ _literals: &FnvHashSet<Datum>,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ self.field_ids.insert(r.field().id);
+ Ok(())
+ }
+
+ fn not_in(
+ &mut self,
+ _r: &BoundReference,
+ _literals: &FnvHashSet<Datum>,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+}
+
+/// Check whether a datum value might be present in the bloom filter.
+///
+/// The value must be checked using the same physical encoding the Parquet
+/// writer used when inserting into the bloom filter. We use the actual
+/// physical type from the column metadata to ensure correctness regardless
+/// of which writer produced the file.
+fn check_in_bloom_filter(column: &ColumnBloomFilter, datum: &Datum) -> bool {
+ let ColumnBloomFilter {
+ sbbf,
+ physical_type,
+ type_length,
+ } = column;
+ let physical_type = *physical_type;
+
+ match datum.literal() {
+ PrimitiveLiteral::Boolean(v) => sbbf.check(v),
Review Comment:
Parquet doesn't define bloom filter semantics for BOOLEAN — Java's hash only
covers int/long/float/double/binary — so this arm probes something the spec
says nothing about. In practice no writer builds a bloom filter for a bool
column, so `bloom_filter_offset().is_none()` short-circuits it and the code is
dead.
Since it's dead and off-spec, I'd either drop the arm and let it fall
through to a conservative default, or leave a one-line comment noting it's
undefined. Minor, but no reason to carry a probe the spec doesn't back.
##########
crates/iceberg/src/expr/visitors/bloom_filter_evaluator.rs:
##########
@@ -0,0 +1,1325 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Evaluates predicates against Parquet bloom filters to determine whether
+//! a row group can be skipped.
+
+use std::collections::{HashMap, HashSet};
+
+use fnv::FnvHashSet;
+use parquet::basic::Type as PhysicalType;
+use parquet::bloom_filter::Sbbf;
+use parquet::data_type::ByteArray;
+
+use crate::Result;
+use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor,
visit};
+use crate::expr::{BoundPredicate, BoundReference};
+use crate::spec::decimal_utils::decimal_to_fixed_length_bytes_exact;
+use crate::spec::{Datum, PrimitiveLiteral};
+
+const ROW_GROUP_MIGHT_MATCH: Result<bool> = Ok(true);
+const ROW_GROUP_CANT_MATCH: Result<bool> = Ok(false);
+
+/// A column's bloom filter for one row group, together with the file's
physical
+/// encoding of that column. A probe must be encoded the way the writer encoded
+/// the values it inserted, so the encoding travels with the filter.
+pub(crate) struct ColumnBloomFilter {
+ sbbf: Sbbf,
+ physical_type: PhysicalType,
+ /// `type_length` from the file's column descriptor. Only meaningful for
+ /// `FIXED_LEN_BYTE_ARRAY`.
+ type_length: i32,
+}
+
+impl ColumnBloomFilter {
+ pub(crate) fn new(sbbf: Sbbf, physical_type: PhysicalType, type_length:
i32) -> Self {
+ Self {
+ sbbf,
+ physical_type,
+ type_length,
+ }
+ }
+}
+
+pub(crate) struct BloomFilterEvaluator<'a> {
+ /// Maps Iceberg field_id -> bloom filter for this row group
+ bloom_filters: &'a HashMap<i32, ColumnBloomFilter>,
+}
+
+impl<'a> BloomFilterEvaluator<'a> {
+ /// Evaluate the predicate against the provided bloom filters.
+ /// Returns `false` if the row group definitely does not match,
+ /// `true` if it might match.
+ pub(crate) fn eval(
+ filter: &BoundPredicate,
+ bloom_filters: &HashMap<i32, ColumnBloomFilter>,
+ ) -> Result<bool> {
+ if bloom_filters.is_empty() {
+ return ROW_GROUP_MIGHT_MATCH;
+ }
+
+ let mut evaluator = BloomFilterEvaluator { bloom_filters };
+ visit(&mut evaluator, filter)
+ }
+
+ fn check_datum(&self, reference: &BoundReference, datum: &Datum) -> bool {
+ let field_id = reference.field().id;
+ let Some(column) = self.bloom_filters.get(&field_id) else {
+ // No bloom filter for this column — conservatively might match
+ return true;
+ };
+
+ check_in_bloom_filter(column, datum)
+ }
+}
+
+/// Collects field IDs that appear in `eq` or `in` predicates — the only
+/// predicate types that benefit from bloom filter checks.
+pub(crate) fn collect_bloom_filter_field_ids(predicate: &BoundPredicate) ->
Result<HashSet<i32>> {
+ let mut visitor = BloomFilterFieldIdCollector {
+ field_ids: HashSet::new(),
+ };
+ visit(&mut visitor, predicate)?;
+ Ok(visitor.field_ids)
+}
+
+struct BloomFilterFieldIdCollector {
+ field_ids: HashSet<i32>,
+}
+
+impl BoundPredicateVisitor for BloomFilterFieldIdCollector {
+ type T = ();
+
+ fn always_true(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ fn always_false(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ fn and(&mut self, _lhs: (), _rhs: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn or(&mut self, _lhs: (), _rhs: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn not(&mut self, _inner: ()) -> Result<()> {
+ Ok(())
+ }
+
+ fn is_null(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn not_null(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn is_nan(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn not_nan(&mut self, _r: &BoundReference, _p: &BoundPredicate) ->
Result<()> {
+ Ok(())
+ }
+
+ fn less_than(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn less_than_or_eq(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn greater_than(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn greater_than_or_eq(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn eq(&mut self, r: &BoundReference, _l: &Datum, _p: &BoundPredicate) ->
Result<()> {
+ self.field_ids.insert(r.field().id);
+ Ok(())
+ }
+
+ fn not_eq(&mut self, _r: &BoundReference, _l: &Datum, _p: &BoundPredicate)
-> Result<()> {
+ Ok(())
+ }
+
+ fn starts_with(&mut self, _r: &BoundReference, _l: &Datum, _p:
&BoundPredicate) -> Result<()> {
+ Ok(())
+ }
+
+ fn not_starts_with(
+ &mut self,
+ _r: &BoundReference,
+ _l: &Datum,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+
+ fn r#in(
+ &mut self,
+ r: &BoundReference,
+ _literals: &FnvHashSet<Datum>,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ self.field_ids.insert(r.field().id);
+ Ok(())
+ }
+
+ fn not_in(
+ &mut self,
+ _r: &BoundReference,
+ _literals: &FnvHashSet<Datum>,
+ _p: &BoundPredicate,
+ ) -> Result<()> {
+ Ok(())
+ }
+}
+
+/// Check whether a datum value might be present in the bloom filter.
+///
+/// The value must be checked using the same physical encoding the Parquet
+/// writer used when inserting into the bloom filter. We use the actual
+/// physical type from the column metadata to ensure correctness regardless
+/// of which writer produced the file.
+fn check_in_bloom_filter(column: &ColumnBloomFilter, datum: &Datum) -> bool {
+ let ColumnBloomFilter {
+ sbbf,
+ physical_type,
+ type_length,
+ } = column;
+ let physical_type = *physical_type;
+
+ match datum.literal() {
+ PrimitiveLiteral::Boolean(v) => sbbf.check(v),
+ // A promoted column (int -> long, float -> double) keeps its original
+ // physical width in files written before the promotion, and the writer
+ // hashed that width, so probe at the file's width rather than the
+ // predicate's.
+ PrimitiveLiteral::Int(v) => match physical_type {
+ PhysicalType::INT32 => sbbf.check(v),
+ PhysicalType::INT64 => sbbf.check(&i64::from(*v)),
+ _ => true,
+ },
+ PrimitiveLiteral::Long(v) => match physical_type {
+ PhysicalType::INT64 => sbbf.check(v),
+ PhysicalType::INT32 => match i32::try_from(*v) {
+ Ok(narrowed) => sbbf.check(&narrowed),
+ // Out of range for the column, so it cannot be present.
+ Err(_) => true,
+ },
+ _ => true,
+ },
+ PrimitiveLiteral::Float(v) => match physical_type {
+ PhysicalType::FLOAT => sbbf.check(&v.0),
+ PhysicalType::DOUBLE => sbbf.check(&f64::from(v.0)),
+ _ => true,
+ },
+ PrimitiveLiteral::Double(v) => match physical_type {
+ PhysicalType::DOUBLE => sbbf.check(&v.0),
+ PhysicalType::FLOAT => {
+ let narrowed = v.0 as f32;
+ // Only an exactly representable value can equal a widened f32.
+ if f64::from(narrowed) == v.0 {
+ sbbf.check(&narrowed)
+ } else {
+ true
+ }
+ }
+ _ => true,
+ },
+ PrimitiveLiteral::String(v) => sbbf.check(v.as_str()),
+ PrimitiveLiteral::Binary(v) => sbbf.check(v.as_slice()),
+ PrimitiveLiteral::Int128(v) => {
+ // Decimal: dispatch based on the actual Parquet physical type
+ // from the file, not inferred from precision.
+ match physical_type {
+ PhysicalType::INT32 => sbbf.check(&(*v as i32)),
+ PhysicalType::INT64 => sbbf.check(&(*v as i64)),
+ PhysicalType::FIXED_LEN_BYTE_ARRAY => {
+ // Encode to the file's declared length, not one derived
from
+ // the Iceberg precision: a widened precision would change
the
+ // length and miss every entry the writer inserted.
+ match usize::try_from(*type_length)
+ .ok()
+ .and_then(|len|
decimal_to_fixed_length_bytes_exact(*v, len))
+ {
+ Some(bytes) => sbbf.check(&ByteArray::from(bytes)),
+ // Unusable length, or a value too large for the
column to
+ // hold — conservatively might match.
+ None => true,
+ }
+ }
+ _ => true, // Unexpected physical type — conservatively might
match
Review Comment:
Decimals encoded as `BYTE_ARRAY` fall into this `_ => true` and skip the
probe entirely. It's valid per the Parquet spec (and some older Spark writers
do emit it), and Java handles `BINARY` alongside `FIXED_LEN_BYTE_ARRAY` for
decimals.
No correctness risk since we just keep the group, but it's a silent parity
gap — decimal predicates get no pruning against those files. Fine as a
follow-up if we note it, but I'd at least leave a comment here so it reads as a
known gap rather than an oversight.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
