From fa054086a8122b2ace2220b855e08e8234098ebb Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Mon, 27 Jul 2026 13:53:58 +0800 Subject: [PATCH 1/5] fix(arrow): allow predicates on struct-nested leaf columns A predicate on a primitive leaf inside a struct (e.g. `person.age > 25`) failed while building the Arrow row filter with "Leaf column ... isn't a root column in Parquet schema". `PredicateConverter::bound_reference` resolved the field to its Parquet leaf index and rejected it when that leaf's column root was a group, and `project_column` only handled top-level columns. Resolve the referenced field to its Parquet column path (root to leaf) instead of a single leaf index, and have `project_column` walk that path through the projected `RecordBatch`, descending `StructArray` children by name to reach the primitive leaf. `ProjectionMask::leaves` preserves the Parquet nesting, so a predicate on `nested.value` sees `nested` as a `StructArray` holding the leaf. Only top-level and struct-nested primitive leaves can bind a predicate (list/map interiors have no accessor), so every path segment before the leaf is a struct. Tested: unit coverage for the nested-leaf conversion, plus end-to-end reads of real Parquet files for single- and double-nested struct leaves, and a bind-time check that predicates on list/map interior fields are rejected before reaching the row filter. Closes #2432 --- .../src/arrow/reader/predicate_visitor.rs | 242 +++++++++---- crates/iceberg/src/arrow/reader/row_filter.rs | 317 ++++++++++++++++++ 2 files changed, 500 insertions(+), 59 deletions(-) diff --git a/crates/iceberg/src/arrow/reader/predicate_visitor.rs b/crates/iceberg/src/arrow/reader/predicate_visitor.rs index 272de49390..c8a90c7110 100644 --- a/crates/iceberg/src/arrow/reader/predicate_visitor.rs +++ b/crates/iceberg/src/arrow/reader/predicate_visitor.rs @@ -25,7 +25,9 @@ use std::sync::Arc; use arrow_arith::boolean::{and, and_kleene, is_not_null, is_null, not, or, or_kleene}; use arrow_array::cast::AsArray; use arrow_array::types::{Float32Type, Float64Type}; -use arrow_array::{Array, ArrayRef, BooleanArray, Datum as ArrowDatum, RecordBatch, Scalar}; +use arrow_array::{ + Array, ArrayRef, BooleanArray, Datum as ArrowDatum, RecordBatch, Scalar, StructArray, +}; use arrow_buffer::BooleanBuffer; use arrow_cast::cast::cast; use arrow_ord::cmp::{eq, gt, gt_eq, lt, lt_eq, neq}; @@ -207,26 +209,21 @@ pub(super) struct PredicateConverter<'a> { } impl PredicateConverter<'_> { - /// When visiting a bound reference, we return index of the leaf column in the - /// required column indices which is used to project the column in the record batch. - /// Return None if the field id is not found in the column map, which is possible - /// due to schema evolution. - fn bound_reference(&mut self, reference: &BoundReference) -> Result> { + /// When visiting a bound reference, we return the Parquet column path (root to leaf) + /// of the referenced leaf column, which the predicate closures use to extract the + /// column from the record batch. Return None if the field id is not found in the + /// column map, which is possible due to schema evolution. + /// + /// The path lets predicates target a primitive leaf nested inside a struct: for a + /// top-level column the path is a single element, and for `nested.value` it is + /// `["nested", "value"]`. `ProjectionMask::leaves` preserves the Parquet nesting, so + /// the projected record batch exposes `nested` as a `StructArray` holding the leaf, + /// which `project_column` descends by name. + fn bound_reference(&mut self, reference: &BoundReference) -> Result>> { // The leaf column's index in Parquet schema. if let Some(column_idx) = self.column_map.get(&reference.field().id) { - if self.parquet_schema.get_column_root(*column_idx).is_group() { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "Leaf column `{}` in predicates isn't a root column in Parquet schema.", - reference.field().name - ), - )); - } - - // The leaf column's index in the required column indices. - let index = self - .column_indices + // Confirm the leaf is among the projected columns. + self.column_indices .iter() .position(|&idx| idx == *column_idx) .ok_or(Error::new( @@ -237,7 +234,16 @@ impl PredicateConverter<'_> { ), ))?; - Ok(Some(index)) + let path: Arc<[String]> = self + .parquet_schema + .column(*column_idx) + .path() + .parts() + .iter() + .cloned() + .collect(); + + Ok(Some(path)) } else { Ok(None) } @@ -258,20 +264,55 @@ impl PredicateConverter<'_> { } } -/// Gets the leaf column from the record batch for the required column index. Only -/// supports top-level columns for now. +/// Walks the Parquet column path (root to leaf) through the projected record batch to +/// reach a primitive leaf. A single-element path returns the matching top-level column; +/// a longer path descends through `StructArray` children by name. Predicates can only +/// bind to primitive leaves in top-level or struct-nested positions (list/map interiors +/// have no accessor), so every path segment before the leaf resolves to a struct. fn project_column( batch: &RecordBatch, - column_idx: usize, + path: &[String], ) -> std::result::Result { - let column = batch.column(column_idx); - - match column.data_type() { - DataType::Struct(_) => Err(ArrowError::SchemaError( - "Does not support struct column yet.".to_string(), - )), - _ => Ok(column.clone()), - } + let (root_name, rest) = path + .split_first() + .ok_or_else(|| ArrowError::SchemaError("Predicate column path is empty.".to_string()))?; + + let mut current = batch + .column_by_name(root_name) + .ok_or_else(|| { + ArrowError::SchemaError(format!( + "Predicate column root `{root_name}` not found in projected record batch." + )) + })? + .clone(); + + for part in rest { + let struct_array = current + .as_any() + .downcast_ref::() + .ok_or_else(|| { + ArrowError::SchemaError(format!( + "Predicate column path expected a struct at `{part}` but found {:?}.", + current.data_type() + )) + })?; + current = struct_array + .column_by_name(part) + .ok_or_else(|| { + ArrowError::SchemaError(format!( + "Predicate column nested field `{part}` not found in struct." + )) + })? + .clone(); + } + + if matches!(current.data_type(), DataType::Struct(_)) { + return Err(ArrowError::SchemaError( + "Predicate column path resolved to a struct, expected a primitive leaf.".to_string(), + )); + } + + Ok(current) } fn compute_is_nan(array: &ArrayRef) -> std::result::Result { @@ -353,9 +394,9 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { reference: &BoundReference, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { Ok(Box::new(move |batch| { - let column = project_column(&batch, idx)?; + let column = project_column(&batch, &path)?; is_null(&column) })) } else { @@ -369,9 +410,9 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { reference: &BoundReference, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { Ok(Box::new(move |batch| { - let column = project_column(&batch, idx)?; + let column = project_column(&batch, &path)?; is_not_null(&column) })) } else { @@ -385,9 +426,9 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { reference: &BoundReference, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { Ok(Box::new(move |batch| { - let column = project_column(&batch, idx)?; + let column = project_column(&batch, &path)?; compute_is_nan(&column) })) } else { @@ -401,9 +442,9 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { reference: &BoundReference, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { Ok(Box::new(move |batch| { - let column = project_column(&batch, idx)?; + let column = project_column(&batch, &path)?; let is_nan = compute_is_nan(&column)?; not(&is_nan) })) @@ -419,11 +460,11 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { let literal = get_arrow_datum(literal)?; Ok(Box::new(move |batch| { - let left = project_column(&batch, idx)?; + let left = project_column(&batch, &path)?; let literal = try_cast_literal(&literal, left.data_type())?; lt(&left, literal.as_ref()) })) @@ -439,11 +480,11 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { let literal = get_arrow_datum(literal)?; Ok(Box::new(move |batch| { - let left = project_column(&batch, idx)?; + let left = project_column(&batch, &path)?; let literal = try_cast_literal(&literal, left.data_type())?; lt_eq(&left, literal.as_ref()) })) @@ -459,11 +500,11 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { let literal = get_arrow_datum(literal)?; Ok(Box::new(move |batch| { - let left = project_column(&batch, idx)?; + let left = project_column(&batch, &path)?; let literal = try_cast_literal(&literal, left.data_type())?; gt(&left, literal.as_ref()) })) @@ -479,11 +520,11 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { let literal = get_arrow_datum(literal)?; Ok(Box::new(move |batch| { - let left = project_column(&batch, idx)?; + let left = project_column(&batch, &path)?; let literal = try_cast_literal(&literal, left.data_type())?; gt_eq(&left, literal.as_ref()) })) @@ -499,11 +540,11 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { let literal = get_arrow_datum(literal)?; Ok(Box::new(move |batch| { - let left = project_column(&batch, idx)?; + let left = project_column(&batch, &path)?; let literal = try_cast_literal(&literal, left.data_type())?; eq(&left, literal.as_ref()) })) @@ -519,11 +560,11 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { let literal = get_arrow_datum(literal)?; Ok(Box::new(move |batch| { - let left = project_column(&batch, idx)?; + let left = project_column(&batch, &path)?; let literal = try_cast_literal(&literal, left.data_type())?; neq(&left, literal.as_ref()) })) @@ -539,11 +580,11 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { let literal = get_arrow_datum(literal)?; Ok(Box::new(move |batch| { - let left = project_column(&batch, idx)?; + let left = project_column(&batch, &path)?; let literal = try_cast_literal(&literal, left.data_type())?; starts_with(&left, literal.as_ref()) })) @@ -559,11 +600,11 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { let literal = get_arrow_datum(literal)?; Ok(Box::new(move |batch| { - let left = project_column(&batch, idx)?; + let left = project_column(&batch, &path)?; let literal = try_cast_literal(&literal, left.data_type())?; // update here if arrow ever adds a native not_starts_with not(&starts_with(&left, literal.as_ref())?) @@ -580,7 +621,7 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { literals: &FnvHashSet, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { let literals: Vec<_> = literals .iter() .map(|lit| get_arrow_datum(lit).unwrap()) @@ -588,7 +629,7 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { Ok(Box::new(move |batch| { // update this if arrow ever adds a native is_in kernel - let left = project_column(&batch, idx)?; + let left = project_column(&batch, &path)?; let mut acc = BooleanArray::from(vec![false; batch.num_rows()]); for literal in &literals { @@ -610,7 +651,7 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { literals: &FnvHashSet, _predicate: &BoundPredicate, ) -> Result> { - if let Some(idx) = self.bound_reference(reference)? { + if let Some(path) = self.bound_reference(reference)? { let literals: Vec<_> = literals .iter() .map(|lit| get_arrow_datum(lit).unwrap()) @@ -618,7 +659,7 @@ impl BoundPredicateVisitor for PredicateConverter<'_> { Ok(Box::new(move |batch| { // update this if arrow ever adds a native not_in kernel - let left = project_column(&batch, idx)?; + let left = project_column(&batch, &path)?; let mut acc = BooleanArray::from(vec![true; batch.num_rows()]); for literal in &literals { let literal = try_cast_literal(literal, left.data_type())?; @@ -817,4 +858,87 @@ mod tests { ); assert!(!result.is_null(2)); } + + /// A predicate on a primitive leaf inside a struct must build and evaluate rather than + /// being rejected because the leaf's Parquet column root is a group. Regression test + /// for issue #2432. + #[test] + fn test_predicate_on_nested_leaf_column() { + use arrow_array::{ArrayRef, Int32Array, StructArray}; + use arrow_schema::Fields; + + use crate::spec::Datum; + + // Schema: id: int (1), person: struct (2) + let schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required( + 2, + "person", + Type::Struct(crate::spec::StructType::new(vec![ + NestedField::required(3, "age", Type::Primitive(PrimitiveType::Int)) + .into(), + ])), + ) + .into(), + ]) + .build() + .unwrap(), + ); + + // `person.age > 25`, bound to the nested leaf (field id 3). + let bound = Reference::new("person.age") + .greater_than(Datum::int(25)) + .bind(schema, true) + .unwrap(); + + // Arrow batch mirroring the schema: a struct column whose `age` child is the leaf. + let age = Arc::new(Int32Array::from(vec![30, 20, 40])) as ArrayRef; + let person = Arc::new(StructArray::from(vec![( + Arc::new(Field::new("age", DataType::Int32, false)), + age as ArrayRef, + )])); + let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new( + "person", + DataType::Struct(Fields::from(vec![Field::new( + "age", + DataType::Int32, + false, + )])), + false, + )])); + let batch = RecordBatch::try_new(arrow_schema, vec![person]).unwrap(); + + // Parquet schema: the nested leaf `person.age` sits under a group, so its column + // root is a group (the shape that used to be rejected). + let message_type = " + message schema { + required group person = 2 { + required int32 age = 3; + } + } + "; + let parquet_type = parse_message_type(message_type).expect("parse schema"); + let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_type)); + + // `person.age` is the single projected leaf (Parquet leaf index 0). + let column_map = HashMap::from([(3i32, 0usize)]); + let column_indices = vec![0usize]; + + let mut converter = PredicateConverter { + parquet_schema: &parquet_schema, + column_map: &column_map, + column_indices: &column_indices, + }; + + let mut predicate_fn = visit(&mut converter, &bound).unwrap(); + let result = predicate_fn(batch).unwrap(); + + assert_eq!([result.value(0), result.value(1), result.value(2)], [ + true, false, true + ]); + } } diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index a94c159d1d..258694ec6a 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1279,4 +1279,321 @@ mod tests { "positional deletes must be applied correctly even when page indexes are absent" ); } + + /// End-to-end regression for issue #2432: a predicate on a primitive leaf nested in a + /// struct (`person.age > 25`) must build a row filter and prune rows, rather than + /// failing because the leaf's Parquet column root is a group. Reads a real Parquet + /// file so the projected `RecordBatch` shape (a `StructArray` holding the leaf) comes + /// from arrow-rs, not a hand-built batch. + #[tokio::test] + async fn test_predicate_on_nested_struct_leaf_reads_real_parquet() { + use arrow_array::StructArray; + use arrow_schema::Fields; + + use crate::spec::StructType; + + // Schema: id: int (1), person: struct (2) + let iceberg_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required( + 2, + "person", + Type::Struct(StructType::new(vec![ + NestedField::required(3, "age", Type::Primitive(PrimitiveType::Int)) + .into(), + ])), + ) + .into(), + ]) + .build() + .unwrap(), + ); + + // Arrow schema carrying field ids so the reader resolves by id, not position. + let age_field = Field::new("age", DataType::Int32, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "3".to_string(), + )])); + 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(), + )])), + Field::new( + "person", + DataType::Struct(Fields::from(vec![age_field.clone()])), + false, + ) + .with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "2".to_string(), + )])), + ])); + + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap().to_string(); + let file_path = format!("{table_location}/1.parquet"); + + let id = Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef; + let person = Arc::new(StructArray::from(vec![( + Arc::new(age_field), + Arc::new(Int32Array::from(vec![30, 20, 40])) as ArrayRef, + )])) as ArrayRef; + let batch = RecordBatch::try_new(arrow_schema.clone(), vec![id, person]).unwrap(); + + let props = WriterProperties::builder() + .set_compression(Compression::SNAPPY) + .build(); + let file = File::create(&file_path).unwrap(); + let mut writer = ArrowWriter::try_new(file, arrow_schema, Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + + // Predicate targets the nested leaf; project only the top-level `id` so the + // predicate's own projection (not the output projection) drives the struct read. + let predicate = Reference::new("person.age") + .greater_than(Datum::int(25)) + .bind(iceberg_schema.clone(), false) + .unwrap(); + + let file_io = FileIO::new_with_fs(); + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).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(iceberg_schema.clone()) + .with_project_field_ids(vec![1]) + .with_predicate(Some(predicate)) + .with_case_sensitive(false) + .build(); + + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + let batches = reader + .read(tasks) + .unwrap() + .stream() + .try_collect::>() + .await + .unwrap(); + + // Rows with person.age > 25 are id=1 (30) and id=3 (40); id=2 (20) is pruned. + let ids: Vec = batches + .iter() + .flat_map(|b| { + b.column(0) + .as_primitive::() + .values() + .to_vec() + }) + .collect(); + assert_eq!(ids, vec![1, 3]); + } + + /// A predicate on a field nested two structs deep (`person.address.zip`) must resolve + /// through both struct levels. This exercises `project_column`'s descent loop more than + /// once, which the single-level `person.age` case does not. Regression for issue #2432. + #[tokio::test] + async fn test_predicate_on_doubly_nested_struct_leaf() { + use arrow_array::StructArray; + use arrow_schema::Fields; + + use crate::spec::StructType; + + // Schema: id: int (1), person: struct (3)> (2) + let iceberg_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required( + 2, + "person", + Type::Struct(StructType::new(vec![ + NestedField::required( + 3, + "address", + Type::Struct(StructType::new(vec![ + NestedField::required( + 4, + "zip", + Type::Primitive(PrimitiveType::Int), + ) + .into(), + ])), + ) + .into(), + ])), + ) + .into(), + ]) + .build() + .unwrap(), + ); + + let zip_field = Field::new("zip", DataType::Int32, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "4".to_string(), + )])); + let address_field = Field::new( + "address", + DataType::Struct(Fields::from(vec![zip_field.clone()])), + false, + ) + .with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "3".to_string(), + )])); + 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(), + )])), + Field::new( + "person", + DataType::Struct(Fields::from(vec![address_field.clone()])), + false, + ) + .with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "2".to_string(), + )])), + ])); + + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap().to_string(); + let file_path = format!("{table_location}/1.parquet"); + + let id = Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef; + let address = Arc::new(StructArray::from(vec![( + Arc::new(zip_field), + Arc::new(Int32Array::from(vec![10001, 20002, 30003])) as ArrayRef, + )])) as ArrayRef; + let person = + Arc::new(StructArray::from(vec![(Arc::new(address_field), address)])) as ArrayRef; + let batch = RecordBatch::try_new(arrow_schema.clone(), vec![id, person]).unwrap(); + + let props = WriterProperties::builder() + .set_compression(Compression::SNAPPY) + .build(); + let file = File::create(&file_path).unwrap(); + let mut writer = ArrowWriter::try_new(file, arrow_schema, Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + + // person.address.zip >= 20002 keeps id=2 (20002) and id=3 (30003). + let predicate = Reference::new("person.address.zip") + .greater_than_or_equal_to(Datum::int(20002)) + .bind(iceberg_schema.clone(), false) + .unwrap(); + + let file_io = FileIO::new_with_fs(); + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).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(iceberg_schema.clone()) + .with_project_field_ids(vec![1]) + .with_predicate(Some(predicate)) + .with_case_sensitive(false) + .build(); + + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + let batches = reader + .read(tasks) + .unwrap() + .stream() + .try_collect::>() + .await + .unwrap(); + + let ids: Vec = batches + .iter() + .flat_map(|b| { + b.column(0) + .as_primitive::() + .values() + .to_vec() + }) + .collect(); + assert_eq!(ids, vec![2, 3]); + } + + /// Fields inside a list or map have no accessor (see `Schema::build_accessors`), so a + /// predicate referencing one fails at bind time and never reaches the row-filter + /// conversion. This pins the assumption `project_column` relies on: every predicate + /// path segment before the leaf is a struct, never a list/map interior. + #[test] + fn test_predicate_on_list_and_map_interior_fails_to_bind() { + use crate::spec::{ListType, MapType, StructType}; + + let schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + // tags: list> (2, element 3) + NestedField::required( + 2, + "tags", + Type::List(ListType::new( + NestedField::required( + 3, + "element", + Type::Struct(StructType::new(vec![ + NestedField::required( + 4, + "name", + Type::Primitive(PrimitiveType::String), + ) + .into(), + ])), + ) + .into(), + )), + ) + .into(), + // props: map (5) + NestedField::required( + 5, + "props", + Type::Map(MapType::new( + NestedField::required(6, "key", Type::Primitive(PrimitiveType::String)) + .into(), + NestedField::required(7, "value", Type::Primitive(PrimitiveType::Int)) + .into(), + )), + ) + .into(), + ]) + .build() + .unwrap(), + ); + + // A field inside the list element and a map value: both must fail to bind. + assert!( + Reference::new("tags.element.name") + .is_null() + .bind(schema.clone(), false) + .is_err(), + "a predicate on a field inside a list must not bind" + ); + assert!( + Reference::new("props.value") + .greater_than(Datum::int(0)) + .bind(schema.clone(), false) + .is_err(), + "a predicate on a map value must not bind" + ); + } } From bfb3a2e674c300b1746a70fe56c8da5781695b8d Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Thu, 27 Aug 2026 14:11:00 +0800 Subject: [PATCH 2/5] fix(arrow): name the offending segment in project_column path errors The struct downcast failure named the next path segment instead of the array that was not a struct. --- .../iceberg/src/arrow/reader/predicate_visitor.rs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/crates/iceberg/src/arrow/reader/predicate_visitor.rs b/crates/iceberg/src/arrow/reader/predicate_visitor.rs index c8a90c7110..2b79a87ee4 100644 --- a/crates/iceberg/src/arrow/reader/predicate_visitor.rs +++ b/crates/iceberg/src/arrow/reader/predicate_visitor.rs @@ -266,9 +266,10 @@ impl PredicateConverter<'_> { /// Walks the Parquet column path (root to leaf) through the projected record batch to /// reach a primitive leaf. A single-element path returns the matching top-level column; -/// a longer path descends through `StructArray` children by name. Predicates can only -/// bind to primitive leaves in top-level or struct-nested positions (list/map interiors -/// have no accessor), so every path segment before the leaf resolves to a struct. +/// a longer path descends through `StructArray` children by name. `Schema::build_accessors` +/// builds accessors only for primitives and primitives nested in structs — never for a list +/// or map, nor for anything inside one — so `Reference::bind` rejects those predicates and +/// every path segment before the leaf here is a struct. fn project_column( batch: &RecordBatch, path: &[String], @@ -285,6 +286,7 @@ fn project_column( )) })? .clone(); + let mut current_name = root_name; for part in rest { let struct_array = current @@ -292,7 +294,7 @@ fn project_column( .downcast_ref::() .ok_or_else(|| { ArrowError::SchemaError(format!( - "Predicate column path expected a struct at `{part}` but found {:?}.", + "Predicate column path expected a struct at `{current_name}` but found {:?}.", current.data_type() )) })?; @@ -300,10 +302,11 @@ fn project_column( .column_by_name(part) .ok_or_else(|| { ArrowError::SchemaError(format!( - "Predicate column nested field `{part}` not found in struct." + "Predicate column nested field `{part}` not found in struct `{current_name}`." )) })? .clone(); + current_name = part; } if matches!(current.data_type(), DataType::Struct(_)) { From d61f4fbbd3b4e8f2bf4eaed3729373364c761da3 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 29 Sep 2026 03:07:57 +0800 Subject: [PATCH 3/5] fix(arrow): carry parent-struct nulls into nested predicate leaves Addresses review on #2906: - project_column now descends via StructArray::flatten, so a leaf under a null parent struct reads as null (spec: a null parent implies a null leaf). - bound_reference checks projection membership with contains, dropping the discarded position and the eagerly built ok_or error. - tests: add null-parent cases (required and optional leaf, doubly nested) and a shared write-then-read helper for the nested-leaf regressions. --- .../src/arrow/reader/predicate_visitor.rs | 30 +- crates/iceberg/src/arrow/reader/row_filter.rs | 331 ++++++++++++------ 2 files changed, 234 insertions(+), 127 deletions(-) diff --git a/crates/iceberg/src/arrow/reader/predicate_visitor.rs b/crates/iceberg/src/arrow/reader/predicate_visitor.rs index 5dc57db5cb..d3ddd923d5 100644 --- a/crates/iceberg/src/arrow/reader/predicate_visitor.rs +++ b/crates/iceberg/src/arrow/reader/predicate_visitor.rs @@ -222,13 +222,12 @@ impl PredicateConverter<'_> { // The leaf column's index in Parquet schema. if let Some(column_idx) = self.column_map.get(&reference.field().id) { // Confirm the leaf is among the projected columns. - self.column_indices - .iter() - .position(|&idx| idx == *column_idx) - .ok_or(invalid_data!( - "Leaf column `{}` in predicates cannot be found in the required column indices.", - reference.field().name - ))?; + if !self.column_indices.contains(column_idx) { + return Err(invalid_data!( + "Leaf column `{}` in predicates cannot be found in the required column indices.", + reference.field().name + )); + } let path: Arc<[String]> = self .parquet_schema @@ -305,14 +304,15 @@ fn project_column( current.data_type() )) })?; - current = struct_array - .column_by_name(part) - .ok_or_else(|| { - ArrowError::SchemaError(format!( - "Predicate column nested field `{part}` not found in struct `{current_name}`." - )) - })? - .clone(); + // `flatten` ANDs the struct's validity into each child, so a leaf under a null + // parent struct reads as null (spec: a null parent implies a null leaf). + let (fields, mut columns) = struct_array.flatten(); + let (idx, _) = fields.find(part).ok_or_else(|| { + ArrowError::SchemaError(format!( + "Predicate column nested field `{part}` not found in struct `{current_name}`." + )) + })?; + current = columns.swap_remove(idx); current_name = part; } diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index e1f749b991..3f69e1b1ea 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1315,87 +1315,30 @@ mod tests { ); // Arrow schema carrying field ids so the reader resolves by id, not position. - let age_field = Field::new("age", DataType::Int32, false).with_metadata(HashMap::from([( - PARQUET_FIELD_ID_META_KEY.to_string(), - "3".to_string(), - )])); + let age_field = field_with_id("age", DataType::Int32, 3); 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(), - )])), - Field::new( + field_with_id("id", DataType::Int32, 1), + field_with_id( "person", DataType::Struct(Fields::from(vec![age_field.clone()])), - false, - ) - .with_metadata(HashMap::from([( - PARQUET_FIELD_ID_META_KEY.to_string(), - "2".to_string(), - )])), + 2, + ), ])); - let tmp_dir = TempDir::new().unwrap(); - let table_location = tmp_dir.path().to_str().unwrap().to_string(); - let file_path = format!("{table_location}/1.parquet"); - let id = Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef; let person = Arc::new(StructArray::from(vec![( Arc::new(age_field), Arc::new(Int32Array::from(vec![30, 20, 40])) as ArrayRef, )])) as ArrayRef; - let batch = RecordBatch::try_new(arrow_schema.clone(), vec![id, person]).unwrap(); - - let props = WriterProperties::builder() - .set_compression(Compression::SNAPPY) - .build(); - let file = File::create(&file_path).unwrap(); - let mut writer = ArrowWriter::try_new(file, arrow_schema, Some(props)).unwrap(); - writer.write(&batch).unwrap(); - writer.close().unwrap(); - - // Predicate targets the nested leaf; project only the top-level `id` so the - // predicate's own projection (not the output projection) drives the struct read. - let predicate = Reference::new("person.age") - .greater_than(Datum::int(25)) - .bind(iceberg_schema.clone(), false) - .unwrap(); - - let file_io = FileIO::new_with_fs(); - let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).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(iceberg_schema.clone()) - .with_project_field_ids(vec![1]) - .with_predicate(Some(predicate)) - .with_case_sensitive(false) - .build() - .unwrap(); - - let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; - let batches = reader - .read(tasks) - .unwrap() - .stream() - .try_collect::>() - .await - .unwrap(); - // Rows with person.age > 25 are id=1 (30) and id=3 (40); id=2 (20) is pruned. - let ids: Vec = batches - .iter() - .flat_map(|b| { - b.column(0) - .as_primitive::() - .values() - .to_vec() - }) - .collect(); + // person.age > 25 keeps id=1 (30) and id=3 (40); id=2 (20) is pruned. + let ids = ids_kept_by_predicate( + iceberg_schema, + arrow_schema, + vec![id, person], + Reference::new("person.age").greater_than(Datum::int(25)), + ) + .await; assert_eq!(ids, vec![1, 3]); } @@ -1440,39 +1383,21 @@ mod tests { .unwrap(), ); - let zip_field = Field::new("zip", DataType::Int32, false).with_metadata(HashMap::from([( - PARQUET_FIELD_ID_META_KEY.to_string(), - "4".to_string(), - )])); - let address_field = Field::new( + let zip_field = field_with_id("zip", DataType::Int32, 4); + let address_field = field_with_id( "address", DataType::Struct(Fields::from(vec![zip_field.clone()])), - false, - ) - .with_metadata(HashMap::from([( - PARQUET_FIELD_ID_META_KEY.to_string(), - "3".to_string(), - )])); + 3, + ); 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(), - )])), - Field::new( + field_with_id("id", DataType::Int32, 1), + field_with_id( "person", DataType::Struct(Fields::from(vec![address_field.clone()])), - false, - ) - .with_metadata(HashMap::from([( - PARQUET_FIELD_ID_META_KEY.to_string(), - "2".to_string(), - )])), + 2, + ), ])); - let tmp_dir = TempDir::new().unwrap(); - let table_location = tmp_dir.path().to_str().unwrap().to_string(); - let file_path = format!("{table_location}/1.parquet"); - let id = Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef; let address = Arc::new(StructArray::from(vec![( Arc::new(zip_field), @@ -1480,8 +1405,30 @@ mod tests { )])) as ArrayRef; let person = Arc::new(StructArray::from(vec![(Arc::new(address_field), address)])) as ArrayRef; - let batch = RecordBatch::try_new(arrow_schema.clone(), vec![id, person]).unwrap(); + // person.address.zip >= 20002 keeps id=2 (20002) and id=3 (30003). + let ids = ids_kept_by_predicate( + iceberg_schema, + arrow_schema, + vec![id, person], + Reference::new("person.address.zip").greater_than_or_equal_to(Datum::int(20002)), + ) + .await; + assert_eq!(ids, vec![2, 3]); + } + + /// Writes `columns` as one Parquet file, pushes `predicate` down while reading it back, + /// and returns the surviving rows' `id`s (the file's first column). `predicate` binds + /// against `iceberg_schema`. Shared write-then-read scaffolding for the nested-leaf tests. + async fn ids_kept_by_predicate( + iceberg_schema: SchemaRef, + arrow_schema: Arc, + columns: Vec, + predicate: Predicate, + ) -> Vec { + let tmp_dir = TempDir::new().unwrap(); + let file_path = format!("{}/1.parquet", tmp_dir.path().to_str().unwrap()); + let batch = RecordBatch::try_new(arrow_schema.clone(), columns).unwrap(); let props = WriterProperties::builder() .set_compression(Compression::SNAPPY) .build(); @@ -1490,22 +1437,15 @@ mod tests { writer.write(&batch).unwrap(); writer.close().unwrap(); - // person.address.zip >= 20002 keeps id=2 (20002) and id=3 (30003). - let predicate = Reference::new("person.address.zip") - .greater_than_or_equal_to(Datum::int(20002)) - .bind(iceberg_schema.clone(), false) - .unwrap(); - - let file_io = FileIO::new_with_fs(); - let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); - + let predicate = predicate.bind(iceberg_schema.clone(), false).unwrap(); + let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).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_path(file_path) .with_data_file_format(DataFileFormat::Parquet) - .with_schema(iceberg_schema.clone()) + .with_schema(iceberg_schema) .with_project_field_ids(vec![1]) .with_predicate(Some(predicate)) .with_case_sensitive(false) @@ -1520,8 +1460,7 @@ mod tests { .try_collect::>() .await .unwrap(); - - let ids: Vec = batches + batches .iter() .flat_map(|b| { b.column(0) @@ -1529,8 +1468,176 @@ mod tests { .values() .to_vec() }) - .collect(); - assert_eq!(ids, vec![2, 3]); + .collect() + } + + /// A null parent struct implies null leaves (spec: "if a parent struct column is null it + /// implies the leaf column is null"). With `person` optional and a null row, `project_column` + /// must carry the parent's validity into the child, so a required leaf reads as null there. + /// Before the `flatten` fix these predicates over-returned the null row. + #[tokio::test] + async fn test_predicate_on_leaf_under_null_parent_struct() { + use arrow_array::StructArray; + use arrow_buffer::NullBuffer; + use arrow_schema::Fields; + + use crate::spec::StructType; + + // id: int (1), person: optional struct (2) + let iceberg_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::optional( + 2, + "person", + Type::Struct(StructType::new(vec![ + NestedField::required(3, "age", Type::Primitive(PrimitiveType::Int)) + .into(), + NestedField::optional(4, "score", Type::Primitive(PrimitiveType::Int)) + .into(), + ])), + ) + .into(), + ]) + .build() + .unwrap(), + ); + let age_field = field_with_id("age", DataType::Int32, 3); + let score_field = Field::new("score", DataType::Int32, true).with_metadata(HashMap::from( + [(PARQUET_FIELD_ID_META_KEY.to_string(), "4".to_string())], + )); + let person_field = Field::new( + "person", + DataType::Struct(Fields::from(vec![age_field.clone(), score_field.clone()])), + true, + ) + .with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "2".to_string(), + )])); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + field_with_id("id", DataType::Int32, 1), + person_field, + ])); + + // Row id=2 has a null `person`; its stored age/score are placeholders to be masked. + let id = Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef; + let person = Arc::new(StructArray::new( + Fields::from(vec![age_field, score_field]), + vec![ + Arc::new(Int32Array::from(vec![30, 0, 40])) as ArrayRef, + Arc::new(Int32Array::from(vec![Some(100), Some(0), Some(300)])) as ArrayRef, + ], + Some(NullBuffer::from(vec![true, false, true])), + )) as ArrayRef; + // person.age < 1000 holds for the stored 30/40 but the null row's age reads as null. + let ids = ids_kept_by_predicate( + iceberg_schema.clone(), + arrow_schema.clone(), + vec![id.clone(), person.clone()], + Reference::new("person.age").less_than(Datum::int(1000)), + ) + .await; + assert_eq!( + ids, + vec![1, 3], + "required leaf under a null parent must read as null" + ); + + // person.age != 30 keeps only id=3 (40); the null row is null, not "!= 30". + let ids = ids_kept_by_predicate( + iceberg_schema, + arrow_schema, + vec![id, person], + Reference::new("person.age").not_equal_to(Datum::int(30)), + ) + .await; + assert_eq!(ids, vec![3]); + } + + /// The parent-null rule holds through two struct levels: a null outer `person` makes + /// `person.address.zip` null, since each level merges its own validity into its children. + #[tokio::test] + async fn test_predicate_on_leaf_under_null_outer_struct_doubly_nested() { + use arrow_array::StructArray; + use arrow_buffer::NullBuffer; + use arrow_schema::Fields; + + use crate::spec::StructType; + + // id: int (1), person: optional struct (3)> (2) + let iceberg_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::optional( + 2, + "person", + Type::Struct(StructType::new(vec![ + NestedField::required( + 3, + "address", + Type::Struct(StructType::new(vec![ + NestedField::required( + 4, + "zip", + Type::Primitive(PrimitiveType::Int), + ) + .into(), + ])), + ) + .into(), + ])), + ) + .into(), + ]) + .build() + .unwrap(), + ); + let zip_field = field_with_id("zip", DataType::Int32, 4); + let address_field = field_with_id( + "address", + DataType::Struct(Fields::from(vec![zip_field.clone()])), + 3, + ); + let person_field = Field::new( + "person", + DataType::Struct(Fields::from(vec![address_field.clone()])), + true, + ) + .with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "2".to_string(), + )])); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + field_with_id("id", DataType::Int32, 1), + person_field, + ])); + + let id = Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef; + let address = Arc::new(StructArray::from(vec![( + Arc::new(zip_field), + Arc::new(Int32Array::from(vec![10001, 0, 30003])) as ArrayRef, + )])) as ArrayRef; + // Row id=2's `person` is null, so `address` and its `zip` under it read as null. + let person = Arc::new(StructArray::new( + Fields::from(vec![address_field]), + vec![address], + Some(NullBuffer::from(vec![true, false, true])), + )) as ArrayRef; + + // person.address.zip < 100000 holds for the stored 10001/30003 but not the null row. + let ids = ids_kept_by_predicate( + iceberg_schema, + arrow_schema, + vec![id, person], + Reference::new("person.address.zip").less_than(Datum::int(100000)), + ) + .await; + assert_eq!(ids, vec![1, 3]); } /// Fields inside a list or map have no accessor (see `Schema::build_accessors`), so a From c8f1a81719512bc785cb58c78ebf88d9419b897f Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 29 Sep 2026 03:27:39 +0800 Subject: [PATCH 4/5] test(arrow): cover an optional leaf under a null parent struct The null-parent test declared an optional `score` leaf but only asserted on the required `age` leaf. Add a `person.score < 1000` case so the optional leaf under a null parent is pinned too, matching the reviewed request. --- crates/iceberg/src/arrow/reader/row_filter.rs | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index 3f69e1b1ea..2ede7ab5cb 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1547,14 +1547,29 @@ mod tests { ); // person.age != 30 keeps only id=3 (40); the null row is null, not "!= 30". + let ids = ids_kept_by_predicate( + iceberg_schema.clone(), + arrow_schema.clone(), + vec![id.clone(), person.clone()], + Reference::new("person.age").not_equal_to(Datum::int(30)), + ) + .await; + assert_eq!(ids, vec![3]); + + // An optional leaf under the null parent reads as null too: person.score < 1000 + // holds for the stored 100/300 but not the null row. let ids = ids_kept_by_predicate( iceberg_schema, arrow_schema, vec![id, person], - Reference::new("person.age").not_equal_to(Datum::int(30)), + Reference::new("person.score").less_than(Datum::int(1000)), ) .await; - assert_eq!(ids, vec![3]); + assert_eq!( + ids, + vec![1, 3], + "optional leaf under a null parent must read as null" + ); } /// The parent-null rule holds through two struct levels: a null outer `person` makes From 2439ce24f735154059bd07646fafb831ae170527 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 29 Sep 2026 16:24:19 +0800 Subject: [PATCH 5/5] test(arrow): cover a predicate over two leaves of the same struct Every nested test so far referenced one leaf per struct, so ProjectionMask projected a single-child struct and project_column's by-name lookup always landed on index 0. Add a `person.age > 25 AND person.score < 250` read test so the struct is projected with two children; taking the child by position instead of by name would read `age` for the `score` comparison and keep an extra row. --- crates/iceberg/src/arrow/reader/row_filter.rs | 77 +++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index 2ede7ab5cb..3098afff13 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1655,6 +1655,83 @@ mod tests { assert_eq!(ids, vec![1, 3]); } + /// A predicate that references two leaves of the same struct + /// (`person.age > 25 AND person.score < 250`) makes `ProjectionMask::leaves` project the + /// struct with more than one child, so `project_column` must find each child by name and + /// not by position. The other nested tests project a single child per struct, where the + /// lookup always lands on index 0 and would still pass if the child were taken positionally. + #[tokio::test] + async fn test_predicate_on_two_leaves_of_same_struct() { + use arrow_array::StructArray; + use arrow_schema::Fields; + + use crate::spec::StructType; + + // id: int (1), person: optional struct (2) + let iceberg_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::optional( + 2, + "person", + Type::Struct(StructType::new(vec![ + NestedField::required(3, "age", Type::Primitive(PrimitiveType::Int)) + .into(), + NestedField::optional(4, "score", Type::Primitive(PrimitiveType::Int)) + .into(), + ])), + ) + .into(), + ]) + .build() + .unwrap(), + ); + let age_field = field_with_id("age", DataType::Int32, 3); + let score_field = Field::new("score", DataType::Int32, true).with_metadata(HashMap::from( + [(PARQUET_FIELD_ID_META_KEY.to_string(), "4".to_string())], + )); + let person_field = Field::new( + "person", + DataType::Struct(Fields::from(vec![age_field.clone(), score_field.clone()])), + true, + ) + .with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "2".to_string(), + )])); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + field_with_id("id", DataType::Int32, 1), + person_field, + ])); + + let id = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6])) as ArrayRef; + let person = Arc::new(StructArray::from(vec![ + ( + Arc::new(age_field), + Arc::new(Int32Array::from(vec![30, 10, 40, 50, 20, 35])) as ArrayRef, + ), + ( + Arc::new(score_field), + Arc::new(Int32Array::from(vec![100, 300, 300, 200, 50, 240])) as ArrayRef, + ), + ])) as ArrayRef; + + // age > 25 AND score < 250. If project_column took the child positionally instead of by + // name, the score comparison would read `age` and keep id=3 too (age 40, score 300). + let ids = ids_kept_by_predicate( + iceberg_schema, + arrow_schema, + vec![id, person], + Reference::new("person.age") + .greater_than(Datum::int(25)) + .and(Reference::new("person.score").less_than(Datum::int(250))), + ) + .await; + assert_eq!(ids, vec![1, 4, 6]); + } + /// Fields inside a list or map have no accessor (see `Schema::build_accessors`), so a /// predicate referencing one fails at bind time and never reaches the row-filter /// conversion. This pins the assumption `project_column` relies on: every predicate