mbutrovich commented on code in PR #2398:
URL: https://github.com/apache/iceberg-rust/pull/2398#discussion_r3982500414
##########
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:
Not a new comment, a response to your question about deferring.
Deferring seems right to me. `get_row_group_column_bloom_filter` takes `&mut
self`, so this isn't a case of picking the wrong combinator, and
[arrow-rs#8462](https://github.com/apache/arrow-rs/pull/8462) is the actual
unblock. The DataFusion reference is fair, and since the option is off by
default, a user who opts in is accepting the current cost rather than silently
paying it.
The thing that would make me comfortable is the doc on
`with_bloom_filter_enabled` being explicit that the reads are serialized per
row group per column, so the cost model is visible at the call site instead of
only in the tracking issue. Right now it says extra I/O per column per row
group, which reads as a volume cost rather than a latency one, and latency is
what dominates on object storage.
##########
crates/iceberg/src/arrow/reader/row_filter.rs:
##########
@@ -1280,4 +1281,462 @@ mod tests {
"positional deletes must be applied correctly even when page
indexes are absent"
);
}
+
+ // Bloom filter pushdown: on-vs-off equivalence
+ // Pushdown must never change results. An encoding bug in the probe shows
up as
+ // rows the bloom filter drops and the row filter keeps, so every case
below
+ // reads the same file twice and compares. Fixtures spread each row group's
+ // values across the whole domain, so min/max statistics cannot prune and
any
+ // reduction in bytes read is attributable to the bloom filter.
+
+ const ROWS_PER_GROUP: i32 = 20_000;
+ const GROUPS: i32 = 3;
+ /// Gap between consecutive values in a row group, so a value can be
absent from
+ /// the file yet still sit inside every row group's min/max range.
+ const STRIDE: i32 = 30;
+
+ /// The `i`th value of row group `group`, spread across the whole domain.
+ fn interleaved(group: i32, i: i32) -> i32 {
+ i * STRIDE + group
+ }
+
+ fn field_with_id(name: &str, ty: DataType, id: i32) -> Field {
+ Field::new(name, ty, false).with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ id.to_string(),
+ )]))
+ }
+
+ /// Writes one row group per batch, optionally with bloom filters.
+ fn write_row_groups(
+ path: &str,
+ arrow_schema: Arc<ArrowSchema>,
+ row_groups: Vec<RecordBatch>,
+ with_bloom_filters: bool,
+ ) {
+ let mut props =
WriterProperties::builder().set_compression(Compression::UNCOMPRESSED);
+ if with_bloom_filters {
+ props = props
+ .set_bloom_filter_enabled(true)
+ .set_bloom_filter_max_ndv(ROWS_PER_GROUP as u64);
+ }
+
+ let file = File::create(path).unwrap();
+ let mut writer = ArrowWriter::try_new(file, arrow_schema,
Some(props.build())).unwrap();
+ for batch in row_groups {
+ writer.write(&batch).unwrap();
+ // Force a row group boundary so each batch is independently
prunable.
+ writer.flush().unwrap();
+ }
+ writer.close().unwrap();
+ }
+
+ async fn read_once(
+ file_path: &str,
+ schema: Arc<Schema>,
+ project_field_ids: Vec<i32>,
+ predicate: Predicate,
+ bloom_enabled: bool,
+ ) -> (Vec<RecordBatch>, u64) {
+ let file_io = FileIO::new_with_fs();
+ let reader = ArrowReaderBuilder::new(file_io, Runtime::current())
+ .with_bloom_filter_enabled(bloom_enabled)
+ // Keep the fixed footer prefetch small so the byte measurement
reflects
+ // row group I/O rather than a constant metadata read.
+ .with_metadata_size_hint(8 * 1024)
+ .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.to_string())
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(schema.clone())
+ .with_project_field_ids(project_field_ids)
+ .with_predicate(Some(predicate.bind(schema, 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();
+ let metrics = result.metrics().clone();
+ let batches: Vec<RecordBatch> =
result.stream().try_collect().await.unwrap();
+
+ (batches, metrics.bytes_read())
+ }
+
+ /// Batch boundaries differ when fewer row groups are read, so collapse to
a
+ /// single batch before comparing. `None` means no rows at all.
+ fn collapse(batches: &[RecordBatch]) -> Option<RecordBatch> {
+ let first = batches.first()?;
+ Some(arrow_select::concat::concat_batches(&first.schema(),
batches).unwrap())
+ }
+
+ /// Reads the file with pushdown on and off, asserts the rows are
identical, and
+ /// returns the rows plus (bytes_on, bytes_off).
+ async fn assert_pushdown_agrees(
+ case: &str,
+ file_path: &str,
+ schema: Arc<Schema>,
+ project_field_ids: Vec<i32>,
+ predicate: Predicate,
+ ) -> (Option<RecordBatch>, u64, u64) {
+ let (off, bytes_off) = read_once(
+ file_path,
+ schema.clone(),
+ project_field_ids.clone(),
+ predicate.clone(),
+ false,
+ )
+ .await;
+ let (on, bytes_on) = read_once(file_path, schema, project_field_ids,
predicate, true).await;
+
+ let off = collapse(&off);
+ let on = collapse(&on);
+ assert_eq!(
+ on, off,
+ "{case}: bloom filter pushdown changed the rows returned"
+ );
+
+ (on, bytes_on, bytes_off)
+ }
+
+ fn rows(batch: &Option<RecordBatch>) -> usize {
+ batch.as_ref().map_or(0, |b| b.num_rows())
+ }
+
+ fn value_schema(iceberg_type: Type) -> Arc<Schema> {
+ Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![NestedField::required(1, "v",
iceberg_type).into()])
+ .build()
+ .unwrap(),
+ )
+ }
+
+ /// Writes a single-column file, one row group per group index.
+ fn write_value_fixture(
+ path: &str,
+ data_type: DataType,
+ make_group: impl Fn(i32) -> ArrayRef,
+ with_bloom_filters: bool,
+ ) {
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![field_with_id("v",
data_type, 1)]));
+ let row_groups = (0..GROUPS)
+ .map(|g| RecordBatch::try_new(arrow_schema.clone(),
vec![make_group(g)]).unwrap())
+ .collect();
+ write_row_groups(path, arrow_schema, row_groups, with_bloom_filters);
+ }
+
+ /// The core check for one physical encoding: a value that is present must
+ /// survive pushdown, and one that is absent must prune every row group.
The
+ /// first catches a probe that encodes wrongly and drops real rows; the
second
+ /// catches a probe that silently never prunes.
+ async fn assert_prunes_and_agrees(
+ case: &str,
+ file_path: &str,
+ schema: Arc<Schema>,
+ present: Datum,
+ absent: Datum,
+ ) {
+ let (on, bytes_on, bytes_off) = assert_pushdown_agrees(
+ &format!("{case}/present"),
+ file_path,
+ schema.clone(),
+ vec![1],
+ Reference::new("v").equal_to(present),
+ )
+ .await;
+ assert_eq!(rows(&on), 1, "{case}/present: expected one matching row");
+ assert!(
+ bytes_on < bytes_off,
+ "{case}/present: pushdown must skip row groups: {bytes_on} vs
{bytes_off}"
+ );
+
+ let (on, bytes_on, bytes_off) = assert_pushdown_agrees(
+ &format!("{case}/absent"),
+ file_path,
+ schema,
+ vec![1],
+ Reference::new("v").equal_to(absent),
+ )
+ .await;
+ assert_eq!(rows(&on), 0, "{case}/absent: expected no matching rows");
+ assert!(
+ bytes_on * 2 < bytes_off,
+ "{case}/absent: pruning every row group should cut reads sharply: \
+ {bytes_on} vs {bytes_off}"
+ );
+ }
+
+ /// Value present in exactly one row group: `interleaved(1, 51)`.
+ const PRESENT_INT: i32 = 51 * STRIDE + 1;
+ /// Not congruent to any group index mod STRIDE, so absent from the file
while
+ /// still inside every row group's min/max range.
+ const ABSENT_INT: i32 = 7;
+
+ fn int_group(g: i32) -> ArrayRef {
+ Arc::new(Int32Array::from(
+ (0..ROWS_PER_GROUP)
+ .map(|i| interleaved(g, i))
+ .collect::<Vec<_>>(),
+ ))
+ }
+
+ #[tokio::test]
+ async fn test_bloom_pushdown_int32_eq() {
+ let tmp = TempDir::new().unwrap();
+ let path = format!("{}/int32.parquet", tmp.path().to_str().unwrap());
+ write_value_fixture(&path, DataType::Int32, int_group, true);
+
+ assert_prunes_and_agrees(
+ "int32",
+ &path,
+ value_schema(Type::Primitive(PrimitiveType::Int)),
+ Datum::int(PRESENT_INT),
+ Datum::int(ABSENT_INT),
+ )
+ .await;
+ }
+
+ /// `IN` takes a different evaluator path from `eq`: it must keep the row
group
+ /// when any literal might be present, and prune only when all are absent.
+ #[tokio::test]
+ async fn test_bloom_pushdown_int32_in() {
+ let tmp = TempDir::new().unwrap();
+ let path = format!("{}/int32_in.parquet",
tmp.path().to_str().unwrap());
+ write_value_fixture(&path, DataType::Int32, int_group, true);
+ let schema = value_schema(Type::Primitive(PrimitiveType::Int));
+
+ // One present literal keeps its row group; the absent one must not
suppress it.
+ let (on, bytes_on, bytes_off) = assert_pushdown_agrees(
+ "int32_in/mixed",
+ &path,
+ schema.clone(),
+ vec![1],
+ Reference::new("v").is_in([Datum::int(PRESENT_INT),
Datum::int(ABSENT_INT)]),
+ )
+ .await;
+ assert_eq!(rows(&on), 1);
+ assert!(bytes_on < bytes_off, "{bytes_on} vs {bytes_off}");
+
+ // All literals absent: every row group prunes.
+ let (on, bytes_on, bytes_off) = assert_pushdown_agrees(
+ "int32_in/all_absent",
+ &path,
+ schema,
+ vec![1],
+ Reference::new("v").is_in([Datum::int(ABSENT_INT),
Datum::int(ABSENT_INT + 1)]),
+ )
+ .await;
+ assert_eq!(rows(&on), 0);
+ assert!(bytes_on * 2 < bytes_off, "{bytes_on} vs {bytes_off}");
+ }
+
+ #[tokio::test]
+ async fn test_bloom_pushdown_string_eq() {
+ let tmp = TempDir::new().unwrap();
+ let path = format!("{}/string.parquet", tmp.path().to_str().unwrap());
+ write_value_fixture(
+ &path,
+ DataType::Utf8,
+ |g| {
+ Arc::new(StringArray::from(
+ (0..ROWS_PER_GROUP)
+ .map(|i| format!("{:016}", interleaved(g, i)))
+ .collect::<Vec<_>>(),
+ ))
+ },
+ true,
+ );
+
+ assert_prunes_and_agrees(
+ "string",
+ &path,
+ value_schema(Type::Primitive(PrimitiveType::String)),
+ Datum::string(format!("{PRESENT_INT:016}")),
+ Datum::string(format!("{ABSENT_INT:016}")),
+ )
+ .await;
+ }
+
+ /// Decimal precision decides the Parquet physical type, and the probe has
to be
+ /// encoded to match: precision 9 -> INT32, 15 -> INT64, 25 ->
FIXED_LEN_BYTE_ARRAY.
+ /// Each width is a separate arm of `check_in_bloom_filter`.
+ async fn check_decimal_width(name: &str, precision: u32) {
+ const SCALE: u32 = 2;
+ let tmp = TempDir::new().unwrap();
+ let path = format!("{}/{name}.parquet", tmp.path().to_str().unwrap());
+
+ write_value_fixture(
+ &path,
+ DataType::Decimal128(precision as u8, SCALE as i8),
+ |g| {
+ let values: Vec<i128> = (0..ROWS_PER_GROUP)
+ .map(|i| i128::from(interleaved(g, i)))
+ .collect();
+ Arc::new(
+ Decimal128Array::from(values)
+ .with_precision_and_scale(precision as u8, SCALE as i8)
+ .unwrap(),
+ )
+ },
+ true,
+ );
+
+ let datum = |mantissa: i32| {
+ Datum::decimal_with_precision(
+ crate::spec::decimal_utils::decimal_from_i128_with_scale(
+ i128::from(mantissa),
+ SCALE,
+ ),
+ precision,
+ )
+ .unwrap()
+ };
+
+ assert_prunes_and_agrees(
+ name,
+ &path,
+ value_schema(Type::Primitive(PrimitiveType::Decimal {
+ precision,
+ scale: SCALE,
+ })),
+ datum(PRESENT_INT),
+ datum(ABSENT_INT),
+ )
+ .await;
+ }
+
+ #[tokio::test]
+ async fn test_bloom_pushdown_decimal_int32() {
+ check_decimal_width("decimal_int32", 9).await;
+ }
+
+ #[tokio::test]
+ async fn test_bloom_pushdown_decimal_int64() {
+ check_decimal_width("decimal_int64", 15).await;
+ }
+
+ #[tokio::test]
+ async fn test_bloom_pushdown_decimal_fixed_len_byte_array() {
+ check_decimal_width("decimal_flba", 25).await;
+ }
+
+ /// Widening a decimal's precision does not rewrite existing files, so the
file
+ /// keeps its original physical width while the bound literal arrives at
the new
+ /// precision. Deriving the probe encoding from the schema instead of the
file
+ /// would miss every entry the writer inserted and prune matching row
groups.
+ #[tokio::test]
+ async fn test_bloom_pushdown_decimal_widened_precision() {
+ const SCALE: u32 = 2;
+ let tmp = TempDir::new().unwrap();
+ let path = format!("{}/decimal_widened.parquet",
tmp.path().to_str().unwrap());
+
+ // File written at precision 9 -> INT32.
+ write_value_fixture(
+ &path,
+ DataType::Decimal128(9, SCALE as i8),
+ |g| {
+ let values: Vec<i128> = (0..ROWS_PER_GROUP)
+ .map(|i| i128::from(interleaved(g, i)))
+ .collect();
+ Arc::new(
+ Decimal128Array::from(values)
+ .with_precision_and_scale(9, SCALE as i8)
+ .unwrap(),
+ )
+ },
+ true,
+ );
+
+ // Table schema has since widened to precision 30.
+ let schema = value_schema(Type::Primitive(PrimitiveType::Decimal {
+ precision: 30,
+ scale: SCALE,
+ }));
+ let datum = |mantissa: i32| {
+ Datum::decimal_with_precision(
+ crate::spec::decimal_utils::decimal_from_i128_with_scale(
+ i128::from(mantissa),
+ SCALE,
+ ),
+ 30,
+ )
+ .unwrap()
+ };
+
+ assert_prunes_and_agrees(
+ "decimal_widened",
+ &path,
+ schema,
+ datum(PRESENT_INT),
+ datum(ABSENT_INT),
+ )
+ .await;
+ }
+
+ /// `Sbbf` hashes raw IEEE bytes, so it treats `-0.0` and `0.0` as
distinct. The
+ /// row filter's `arrow_ord::cmp::eq` happens to agree today, but for
unrelated
+ /// reasons, so pin that the two stay consistent.
+ #[tokio::test]
+ async fn test_bloom_pushdown_float_signed_zero() {
+ let tmp = TempDir::new().unwrap();
+ let path = format!("{}/float_zero.parquet",
tmp.path().to_str().unwrap());
+
+ // Row group 0 holds only -0.0; row group 1 only +0.0; row group 2
neither.
+ write_value_fixture(
+ &path,
+ DataType::Float32,
+ |g| {
+ let fill = match g {
+ 0 => -0.0f32,
+ 1 => 0.0f32,
+ _ => 7.5f32,
+ };
+ Arc::new(Float32Array::from(vec![fill; ROWS_PER_GROUP as
usize]))
+ },
+ true,
+ );
Review Comment:
Thanks for adding this one. I think the fixture layout keeps it from testing
what it describes, though: row-group statistics prune the interesting group
before the bloom filter is consulted, so the test passes no matter how either
layer treats signed zero.
`row_group_filtering_enabled` defaults to true and no test here turns it
off, so `RowGroupMetricsEvaluator` runs first. Each group is filled with a
single value, so `min == max`, and Iceberg's float comparison is a total order
where `-0.0 < 0.0`. Probing `0.0` puts group 0 (`min = max = -0.0`) out of
range, so it's pruned on statistics alone; probing `-0.0` does the same to
group 1. In both iterations the group holding the *other* zero disappears
before any probe happens.
So the divergence this test is meant to catch cannot make it fail. If the
bloom check started treating `-0.0` and `+0.0` as equal, group 0 is already
gone. If `arrow_ord::cmp::eq` became IEEE-conformant, group 0 is still already
gone, so the extra rows never surface.
For the test to be able to fail, a row group needs to hold one zero but not
the other, and have a min/max range wide enough that statistics keep it. A
group of `-0.0` plus a larger filler value, probed with `+0.0`, would do it:
- statistics: `0.0` falls inside `[-0.0, 5.0]`, so the group survives
- bloom: `+0.0` was never inserted, so the group is pruned and no rows come
back
- pushdown off: `arrow_ord::cmp::eq` also rejects `-0.0`, so no rows come
back either, and the two agree today
- if that kernel ever became IEEE-conformant, the off path would return the
`-0.0` rows while the bloom path still pruned them, and
`assert_pushdown_agrees` would fail
Asserting the row count explicitly would also help, so the test records
which equality semantics is expected rather than only that the two paths match.
--
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]
