diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 5582e4397f..529d90fed7 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -24,7 +24,7 @@ use futures::{StreamExt, TryStreamExt}; use tokio::sync::oneshot::{Receiver, channel}; use super::delete_filter::{DeleteFilter, PosDelLoadAction}; -use crate::arrow::delete_file_loader::BasicDeleteFileLoader; +use crate::arrow::delete_file_loader::{BasicDeleteFileLoader, equality_delete_projection}; use crate::arrow::scan_metrics::ScanMetrics; use crate::arrow::{arrow_primitive_to_literal, arrow_schema_to_schema}; use crate::delete_vector::DeleteVector; @@ -302,8 +302,9 @@ impl CachingDeleteFileLoader { del_filter.insert_equality_delete(task.file_path(), receiver); // Per the Iceberg spec, evolve schema for equality deletes but only for the - // equality_ids columns, not all table columns. + // columns holding the equality_ids, not all table columns. let equality_ids_vec = task.equality_ids().unwrap().to_vec(); + let projection = equality_delete_projection(&schema, &equality_ids_vec); let evolved_stream = BasicDeleteFileLoader::evolve_schema( basic_delete_file_loader .parquet_to_batch_stream( @@ -313,7 +314,7 @@ impl CachingDeleteFileLoader { ) .await?, schema, - &equality_ids_vec, + &projection, ) .await?; @@ -604,7 +605,7 @@ impl CachingDeleteFileLoader { let root_array: ArrayRef = Arc::new(StructArray::from(record_batch)); - let mut processor = EqDelColumnProcessor::new(&equality_ids); + let mut processor = EqDelColumnProcessor::new(&equality_ids, schema); visit_schema_with_partner(schema, &root_array, &mut processor, &accessor)?; let mut datum_columns_with_names = processor.finish()?; @@ -661,13 +662,15 @@ impl CachingDeleteFileLoader { struct EqDelColumnProcessor<'a> { equality_ids: &'a HashSet, + schema: &'a Schema, collected_columns: Vec<(ArrayRef, String, Type)>, } impl<'a> EqDelColumnProcessor<'a> { - fn new(equality_ids: &'a HashSet) -> Self { + fn new(equality_ids: &'a HashSet, schema: &'a Schema) -> Self { Self { equality_ids, + schema, collected_columns: Vec::with_capacity(equality_ids.len()), } } @@ -723,9 +726,17 @@ impl SchemaWithPartnerVisitor for EqDelColumnProcessor<'_> { fn field(&mut self, field: &NestedFieldRef, partner: &ArrayRef, _value: ()) -> Result<()> { if self.equality_ids.contains(&field.id) && field.field_type.as_primitive_type().is_some() { + // Use the full name so a nested field (`s.f`) binds to itself, not to a top-level + // column with the same leaf name. + let name = self.schema.name_by_field_id(field.id).ok_or_else(|| { + Error::new( + ErrorKind::Unexpected, + format!("equality delete field {} not found in schema", field.id), + ) + })?; self.collected_columns.push(( partner.clone(), - field.name.clone(), + name.to_string(), field.field_type.as_ref().clone(), )); } @@ -888,7 +899,7 @@ mod tests { .await .expect("error parsing batch stream"); - let expected = "((((((y IS NULL) OR (y != 1)) OR ((z IS NULL) OR (z != 100))) OR ((a IS NULL) OR (a != \"HELP\"))) OR ((sa IS NULL) OR (sa != 4))) OR ((b IS NULL) OR (b != 62696E6172795F64617461))) AND ((((((y IS NULL) OR (y != 2)) OR (z IS NOT NULL)) OR (a IS NOT NULL)) OR ((sa IS NULL) OR (sa != 5))) OR (b IS NOT NULL))".to_string(); + let expected = "((((((y IS NULL) OR (y != 1)) OR ((z IS NULL) OR (z != 100))) OR ((a IS NULL) OR (a != \"HELP\"))) OR ((s.sa IS NULL) OR (s.sa != 4))) OR ((b IS NULL) OR (b != 62696E6172795F64617461))) AND ((((((y IS NULL) OR (y != 2)) OR (z IS NOT NULL)) OR (a IS NOT NULL)) OR ((s.sa IS NULL) OR (s.sa != 5))) OR (b IS NOT NULL))".to_string(); assert_eq!(parsed_eq_delete.to_string(), expected); } @@ -1791,4 +1802,83 @@ mod tests { assert_eq!(err.kind(), ErrorKind::DataInvalid); assert!(err.message().contains("expected 2 from record_count")); } + + /// An equality delete on a field nested in a struct loads, and its predicate uses the full + /// field name. The table also has a top-level column with the same leaf name, which the + /// predicate must not refer to. + #[tokio::test] + async fn test_nested_field_equality_delete_predicate() { + use crate::arrow::delete_file_loader::equality_delete_projection; + use crate::expr::Bind; + use crate::spec::{Datum, StructType}; + + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "f", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional( + 2, + "s", + Type::Struct(StructType::new(vec![ + NestedField::optional(3, "f", Type::Primitive(PrimitiveType::Long)) + .into(), + ])), + ) + .into(), + ]) + .build() + .unwrap(), + ); + + // Delete file with `s.f = 7`. + let f_field = Arc::new(simple_field("f", DataType::Int64, true, "3")); + let s_field = simple_field( + "s", + DataType::Struct(Fields::from(vec![f_field.clone()])), + true, + "2", + ); + let delete_schema = Arc::new(arrow_schema::Schema::new(vec![s_field])); + let s = StructArray::from(vec![( + f_field, + Arc::new(Int64Array::from(vec![7])) as ArrayRef, + )]); + let batch = RecordBatch::try_new(delete_schema.clone(), vec![Arc::new(s)]).unwrap(); + let path = format!("{table_location}/nested-eq-delete.parquet"); + let mut writer = + ArrowWriter::try_new(File::create(&path).unwrap(), delete_schema, None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + + // Only the top-level struct holding field 3 is read. + assert_eq!(equality_delete_projection(&table_schema, &[3]), vec![2]); + + let loader = BasicDeleteFileLoader::new(FileIO::new_with_fs(), ScanMetrics::new()); + let stream = loader + .parquet_to_batch_stream(&path, std::fs::metadata(&path).unwrap().len(), None) + .await + .unwrap(); + let evolved = BasicDeleteFileLoader::evolve_schema(stream, table_schema.clone(), &[2]) + .await + .unwrap(); + let predicate = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( + evolved, + HashSet::from([3]), + ) + .await + .unwrap(); + + let s_f = Reference::new("s.f"); + assert_eq!( + predicate, + s_f.clone().is_null().or(s_f.not_equal_to(Datum::long(7))) + ); + // It binds to the nested field, not to the top-level `f`. + let bound = predicate.bind(table_schema, true).unwrap(); + assert!(format!("{bound:?}").contains("id: 3")); + assert!(!format!("{bound:?}").contains("id: 1,")); + } } diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index ec61710a42..8d628887f7 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -115,6 +115,30 @@ impl BasicDeleteFileLoader { } } +/// Returns the top-level field ids to read for an equality delete file. +/// +/// Equality ids may name fields nested in structs, but schema evolution of a delete file works on +/// top-level fields, so each id is replaced by the top-level field that contains it. +pub(crate) fn equality_delete_projection(schema: &Schema, equality_ids: &[i32]) -> Vec { + use crate::spec::{NestedField, Type}; + + fn contains(field: &NestedField, id: i32) -> bool { + field.id == id + || match field.field_type.as_ref() { + Type::Struct(s) => s.fields().iter().any(|child| contains(child, id)), + _ => false, + } + } + + schema + .as_struct() + .fields() + .iter() + .filter(|field| equality_ids.iter().any(|&id| contains(field, id))) + .map(|field| field.id) + .collect() +} + #[async_trait::async_trait] impl DeleteFileLoader for BasicDeleteFileLoader { async fn read_delete_file( @@ -130,10 +154,10 @@ impl DeleteFileLoader for BasicDeleteFileLoader { ) .await?; - // For equality deletes, only evolve the equality_ids columns. + // For equality deletes, only evolve the columns holding the equality_ids. // For positional deletes (equality_ids is None), use all field IDs. let field_ids = match task.equality_ids() { - Some(ids) => ids.to_vec(), + Some(ids) => equality_delete_projection(&schema, ids), None => schema.field_id_to_name_map().keys().cloned().collect(), };