diff --git a/crates/iceberg/src/arrow/reader/predicate_visitor.rs b/crates/iceberg/src/arrow/reader/predicate_visitor.rs index 264c34a9a3..d3ddd923d5 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}; @@ -206,31 +208,37 @@ 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() { + // Confirm the leaf is among the projected columns. + if !self.column_indices.contains(column_idx) { return Err(invalid_data!( - "Leaf column `{}` in predicates isn't a root column in Parquet schema.", + "Leaf column `{}` in predicates cannot be found in the required column indices.", reference.field().name )); } - // The leaf column's index in the required column indices. - let index = self - .column_indices + let path: Arc<[String]> = self + .parquet_schema + .column(*column_idx) + .path() + .parts() .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 - ))?; + .cloned() + .collect(); - Ok(Some(index)) + Ok(Some(path)) } else { Ok(None) } @@ -262,20 +270,59 @@ fn constant_bool_array(value: bool, len: usize) -> BooleanArray { BooleanArray::new(buffer, None) } -/// 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. `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, - 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(); + let mut current_name = root_name; + + 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 `{current_name}` but found {:?}.", + current.data_type() + )) + })?; + // `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; + } + + 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 { @@ -357,9 +404,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 { @@ -373,9 +420,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 { @@ -389,9 +436,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 { @@ -405,9 +452,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) })) @@ -423,11 +470,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()) })) @@ -443,11 +490,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()) })) @@ -463,11 +510,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()) })) @@ -483,11 +530,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()) })) @@ -503,11 +550,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()) })) @@ -523,11 +570,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()) })) @@ -543,11 +590,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()) })) @@ -563,11 +610,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())?) @@ -584,7 +631,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()) @@ -592,7 +639,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 = constant_bool_array(false, batch.num_rows()); for literal in &literals { @@ -614,7 +661,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()) @@ -622,7 +669,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 = constant_bool_array(true, batch.num_rows()); for literal in &literals { let literal = try_cast_literal(literal, left.data_type())?; @@ -836,4 +883,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 34b7391134..3098afff13 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1282,6 +1282,524 @@ mod tests { ); } + /// 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_with_id("age", DataType::Int32, 3); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + field_with_id("id", DataType::Int32, 1), + field_with_id( + "person", + DataType::Struct(Fields::from(vec![age_field.clone()])), + 2, + ), + ])); + + 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; + + // 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]); + } + + /// 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_with_id("zip", DataType::Int32, 4); + let address_field = field_with_id( + "address", + DataType::Struct(Fields::from(vec![zip_field.clone()])), + 3, + ); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + field_with_id("id", DataType::Int32, 1), + field_with_id( + "person", + DataType::Struct(Fields::from(vec![address_field.clone()])), + 2, + ), + ])); + + 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; + + // 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(); + 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(); + + 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) + .with_data_file_format(DataFileFormat::Parquet) + .with_schema(iceberg_schema) + .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(); + batches + .iter() + .flat_map(|b| { + b.column(0) + .as_primitive::() + .values() + .to_vec() + }) + .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.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.score").less_than(Datum::int(1000)), + ) + .await; + 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 + /// `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]); + } + + /// 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 + /// 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" + ); + } + // 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