diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index eb5e1ac4b0..f5d7d27e22 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -27,8 +27,6 @@ use crate::arrow::delete_file_loader::BasicDeleteFileLoader; use crate::arrow::scan_metrics::ScanMetrics; use crate::arrow::{arrow_primitive_to_literal, arrow_schema_to_schema}; use crate::delete_vector::DeleteVector; -use crate::expr::Predicate::AlwaysTrue; -use crate::expr::{Predicate, Reference}; use crate::io::FileIO; use crate::runtime::Runtime; use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile}; @@ -39,6 +37,58 @@ use crate::spec::{ }; use crate::{Error, ErrorKind, Result}; +/// A single composite equality-delete key: one entry per column, in the +/// column order of [`EqDeleteSet::fields`]. `None` encodes a SQL null. +#[derive(Hash, Eq, PartialEq, Debug, Clone)] +pub(crate) struct EqDeleteKey(pub(crate) Vec>); + +/// The parsed contents of equality-delete files sharing one column layout. +#[derive(Debug, Clone)] +pub(crate) struct EqDeleteSet { + pub(crate) keys: HashSet, + pub(crate) fields: Vec<(String, i32, Type)>, +} + +impl EqDeleteSet { + fn new(fields: Vec<(String, i32, Type)>) -> Self { + Self { + keys: HashSet::new(), + fields, + } + } + + pub(crate) fn is_empty(&self) -> bool { + self.keys.is_empty() + } + + /// Check if the layout of `other` matches `self`'s layout by Id and Type. + pub(crate) fn layout_matches(&self, other: &EqDeleteSet) -> bool { + self.fields.len() == other.fields.len() + && self + .fields + .iter() + .zip(other.fields.iter()) + .all(|((_, id, ty), (_, other_id, other_ty))| id == other_id && ty == other_ty) + } + + /// Merges `other`'s keys into this set if the layouts match, errors if + /// they do not match. + pub(crate) fn union(&mut self, other: &EqDeleteSet) -> Result<()> { + if !self.layout_matches(other) { + return Err(Error::new( + ErrorKind::Unexpected, + format!( + "Cannot union equality-delete sets with different layouts: \ + {:?} vs {:?}", + self.fields, other.fields + ), + )); + } + self.keys.extend(other.keys.iter().cloned()); + Ok(()) + } +} + #[derive(Clone, Debug)] pub(crate) struct CachingDeleteFileLoader { basic_delete_file_loader: BasicDeleteFileLoader, @@ -61,7 +111,7 @@ enum DeleteFileContext { FreshEqDel { batch_stream: ArrowRecordBatchStream, equality_ids: HashSet, - sender: tokio::sync::oneshot::Sender, + sender: tokio::sync::oneshot::Sender>, }, } @@ -115,16 +165,16 @@ impl CachingDeleteFileLoader { /// another concurrently processing data file scan task. If it is, we skip it. /// If not, the DeleteFilter is updated to contain a notifier to prevent other data file /// tasks from starting to load the same equality delete file. We spawn a task to load - /// the EQ delete's record batch stream, convert it to a predicate, update the delete filter, - /// and notify any task that was waiting for it. + /// the EQ delete's record batch stream, convert it to an equality-delete set, update the + /// delete filter, and notify any task that was waiting for it. /// * When this gets updated to add support for delete vectors, the load phase will return /// a PuffinReader for them. /// * The parse phase parses each record batch stream according to its associated data type. /// The result of this is a map of data file paths to delete vectors for the positional /// delete tasks (and in future for the delete vector tasks). For equality delete - /// file tasks, this results in an unbound Predicate. - /// * The unbound Predicates resulting from equality deletes are sent to their associated oneshot - /// channel to store them in the right place in the delete file managers state. + /// file tasks, this results in an equality-delete set. + /// * The equality-delete sets resulting from equality deletes are sent to their associated + /// oneshot channel to store them in the right place in the delete file managers state. /// * The results of all of these futures are awaited on in parallel with the specified /// level of concurrency and collected into a vec. We then combine all the delete /// vector maps that resulted from any positional delete or delete vector files into a @@ -146,7 +196,7 @@ impl CachingDeleteFileLoader { /// Pos Del Del Vec (Not yet Implemented) EQ Del /// | | | /// [parse pos del stream] [parse del vec puffin] [parse eq del] - /// HashMap HashMap (Predicate, Sender) + /// HashMap HashMap (EqDeleteSet, Sender) /// | | | /// | | [persist to state] /// | | () @@ -271,7 +321,15 @@ impl CachingDeleteFileLoader { // Per the Iceberg spec, evolve schema for equality deletes but only for the // equality_ids columns, not all table columns. - let equality_ids_vec = task.equality_ids.clone().unwrap(); + let equality_ids_vec = task.equality_ids.clone().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Equality delete file '{}' is missing equality_ids", + task.file_path + ), + ) + })?; let evolved_stream = BasicDeleteFileLoader::evolve_schema( basic_delete_file_loader .parquet_to_batch_stream( @@ -317,16 +375,16 @@ impl CachingDeleteFileLoader { batch_stream, equality_ids, } => { - let predicate = + let eq_delete_set = Self::parse_equality_deletes_record_batch_stream(batch_stream, equality_ids) .await?; sender - .send(predicate) - .map_err(|err| { + .send(Arc::new(eq_delete_set)) + .map_err(|_| { Error::new( ErrorKind::Unexpected, - "Could not send eq delete predicate to state", + "Could not send eq delete set to state", ) }) .map(|_| ParsedDeleteFileContext::EqDel) @@ -387,8 +445,8 @@ impl CachingDeleteFileLoader { async fn parse_equality_deletes_record_batch_stream( mut stream: ArrowRecordBatchStream, equality_ids: HashSet, - ) -> Result { - let mut row_predicates = Vec::new(); + ) -> Result { + let mut eq_delete_set: Option = None; let mut batch_schema_iceberg: Option = None; let accessor = EqDelRecordBatchPartnerAccessor; @@ -396,7 +454,7 @@ impl CachingDeleteFileLoader { let record_batch = record_batch?; if record_batch.num_columns() == 0 { - return Ok(AlwaysTrue); + return Ok(EqDeleteSet::new(Vec::new())); } let schema = match &batch_schema_iceberg { @@ -413,61 +471,42 @@ impl CachingDeleteFileLoader { let mut processor = EqDelColumnProcessor::new(&equality_ids); visit_schema_with_partner(schema, &root_array, &mut processor, &accessor)?; - let mut datum_columns_with_names = processor.finish()?; - if datum_columns_with_names.is_empty() { + let mut datum_columns = processor.finish()?; + if datum_columns.is_empty() { continue; } - // Iceberg spec (Equality Delete Files): a null data value never equals a non-null - // delete value, so a row with a null equality column must be kept. Build the keep - // predicate as `col IS NULL OR col != v` (`col IS NOT NULL` for a null delete value); - // a bare `col != v` drops nulls. + let eq_delete_set = eq_delete_set.get_or_insert_with(|| { + let fields = datum_columns + .iter() + .map(|(_, name, id, field_type)| (name.clone(), *id, field_type.clone())) + .collect(); + EqDeleteSet::new(fields) + }); + + // Each delete record is one composite key tuple over all equality columns, in + // `fields` order. A null cell is a first-class tuple element (`None`): per the + // Iceberg spec a null data value equals only a null delete value. #[allow(clippy::len_zero)] - while datum_columns_with_names[0].0.len() > 0 { - let mut row_keep_predicate = Predicate::AlwaysFalse; - for &mut (ref mut column, ref field_name) in &mut datum_columns_with_names { - if let Some(item) = column.next() { - let reference = Reference::new(field_name.clone()); - let cell_keep_predicate = if let Some(datum) = item? { - reference - .clone() - .is_null() - .or(reference.not_equal_to(datum.clone())) - } else { - reference.is_not_null() - }; - row_keep_predicate = row_keep_predicate.or(cell_keep_predicate); + while datum_columns[0].0.len() > 0 { + let mut key = Vec::with_capacity(datum_columns.len()); + for (column, _, _, _) in datum_columns.iter_mut() { + match column.next() { + Some(item) => key.push(item?), + None => key.push(None), } } - row_predicates.push(row_keep_predicate); + eq_delete_set.keys.insert(EqDeleteKey(key)); } } - // All row predicates are combined to a single predicate by creating a balanced binary tree. - // Using a simple fold would result in a deeply nested predicate that can cause a stack overflow. - while row_predicates.len() > 1 { - let mut next_level = Vec::with_capacity(row_predicates.len().div_ceil(2)); - let mut iter = row_predicates.into_iter(); - while let Some(p1) = iter.next() { - if let Some(p2) = iter.next() { - next_level.push(p1.and(p2)); - } else { - next_level.push(p1); - } - } - row_predicates = next_level; - } - - match row_predicates.pop() { - Some(p) => Ok(p), - None => Ok(AlwaysTrue), - } + Ok(eq_delete_set.unwrap_or_else(|| EqDeleteSet::new(Vec::new()))) } } struct EqDelColumnProcessor<'a> { equality_ids: &'a HashSet, - collected_columns: Vec<(ArrayRef, String, Type)>, + collected_columns: Vec<(ArrayRef, String, i32, Type)>, } impl<'a> EqDelColumnProcessor<'a> { @@ -485,11 +524,13 @@ impl<'a> EqDelColumnProcessor<'a> { Vec<( Box>>>, String, + i32, + Type, )>, > { self.collected_columns .into_iter() - .map(|(array, field_name, field_type)| { + .map(|(array, field_name, field_id, field_type)| { let primitive_type = field_type .as_primitive_type() .ok_or_else(|| { @@ -514,7 +555,7 @@ impl<'a> EqDelColumnProcessor<'a> { .transpose() })); - Ok((datum_iterator, field_name)) + Ok((datum_iterator, field_name, field_id, field_type)) }) .collect::>>() } @@ -532,6 +573,7 @@ impl SchemaWithPartnerVisitor for EqDelColumnProcessor<'_> { self.collected_columns.push(( partner.clone(), field.name.clone(), + field.id, field.field_type.as_ref().clone(), )); } @@ -643,7 +685,7 @@ mod tests { use super::*; use crate::arrow::delete_filter::tests::setup; use crate::scan::FileScanTaskDeleteFile; - use crate::spec::{DataContentType, Schema}; + use crate::spec::{DataContentType, Datum, PrimitiveType, Schema, Type}; #[tokio::test] async fn test_delete_file_loader_parse_equality_deletes() { @@ -673,9 +715,31 @@ 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(); - - assert_eq!(parsed_eq_delete.to_string(), expected); + assert_eq!(parsed_eq_delete.fields, vec![ + ("y".to_string(), 2, Type::Primitive(PrimitiveType::Long)), + ("z".to_string(), 3, Type::Primitive(PrimitiveType::Long)), + ("a".to_string(), 4, Type::Primitive(PrimitiveType::String)), + ("sa".to_string(), 6, Type::Primitive(PrimitiveType::Int)), + ("b".to_string(), 8, Type::Primitive(PrimitiveType::Binary)), + ]); + + let expected_keys = HashSet::from([ + EqDeleteKey(vec![ + Some(Datum::long(1)), + Some(Datum::long(100)), + Some(Datum::string("HELP")), + Some(Datum::int(4)), + Some(Datum::binary(b"binary_data".to_vec())), + ]), + EqDeleteKey(vec![ + Some(Datum::long(2)), + None, + None, + Some(Datum::int(5)), + None, + ]), + ]); + assert_eq!(parsed_eq_delete.keys, expected_keys); } // An equality delete keyed on a nullable column must not delete rows whose value in that @@ -683,7 +747,7 @@ mod tests { // delete value. Mirrors Iceberg-Java's // TestSparkReaderDeletes.testEqualityDeleteWithSchemaEvolution. #[tokio::test] - async fn test_equality_delete_predicate_preserves_null_rows() { + async fn test_equality_delete_set_preserves_null_rows() { let schema = Arc::new(arrow_schema::Schema::new(vec![simple_field( "status", DataType::Utf8, @@ -697,23 +761,28 @@ mod tests { .unwrap(); let stream: ArrowRecordBatchStream = futures::stream::iter(vec![Ok(batch)]).boxed(); - let predicate = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( + let set = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( stream, HashSet::from_iter(vec![3]), ) .await .expect("error parsing equality delete stream"); + assert_eq!(set.fields, vec![( + "status".to_string(), + 3, + Type::Primitive(PrimitiveType::String) + )]); assert_eq!( - predicate.to_string(), - "(status IS NULL) OR (status != \"INACTIVE\")" + set.keys, + HashSet::from([EqDeleteKey(vec![Some(Datum::string("INACTIVE"))])]) ); } // A delete row with a null value in the column matches only rows whose value is null (Iceberg // spec, Equality Delete Files), so the keep predicate is `col IS NOT NULL`. #[tokio::test] - async fn test_equality_delete_predicate_matches_null_delete_value() { + async fn test_equality_delete_set_matches_null_delete_value() { let schema = Arc::new(arrow_schema::Schema::new(vec![simple_field( "status", DataType::Utf8, @@ -726,20 +795,20 @@ mod tests { .unwrap(); let stream: ArrowRecordBatchStream = futures::stream::iter(vec![Ok(batch)]).boxed(); - let predicate = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( + let set = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( stream, HashSet::from_iter(vec![3]), ) .await .expect("error parsing equality delete stream"); - assert_eq!(predicate.to_string(), "status IS NOT NULL"); + assert_eq!(set.keys, HashSet::from([EqDeleteKey(vec![None])])); } // A delete row with several equality columns keeps a data row that differs in any one of them, // so the per-column keep predicates are OR-ed. #[tokio::test] - async fn test_equality_delete_predicate_multiple_columns() { + async fn test_equality_delete_set_multiple_columns() { let schema = Arc::new(arrow_schema::Schema::new(vec![ simple_field("id", DataType::Int64, true, "1"), simple_field("status", DataType::Utf8, true, "3"), @@ -751,23 +820,33 @@ mod tests { .unwrap(); let stream: ArrowRecordBatchStream = futures::stream::iter(vec![Ok(batch)]).boxed(); - let predicate = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( + let set = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( stream, HashSet::from_iter(vec![1, 3]), ) .await .expect("error parsing equality delete stream"); + assert_eq!(set.fields, vec![ + ("id".to_string(), 1, Type::Primitive(PrimitiveType::Long)), + ( + "status".to_string(), + 3, + Type::Primitive(PrimitiveType::String) + ), + ]); assert_eq!( - predicate.to_string(), - "((id IS NULL) OR (id != 1)) OR ((status IS NULL) OR (status != \"X\"))" + set.keys, + HashSet::from([EqDeleteKey(vec![ + Some(Datum::long(1)), + Some(Datum::string("X")), + ])]) ); } - // A data row is kept only if it matches none of the delete rows, so the per-row keep - // predicates are AND-ed. + // Each delete row is a distinct key; a data row is deleted if it matches any of them. #[tokio::test] - async fn test_equality_delete_predicate_multiple_delete_rows() { + async fn test_equality_delete_set_multiple_delete_rows() { let schema = Arc::new(arrow_schema::Schema::new(vec![simple_field( "status", DataType::Utf8, @@ -781,7 +860,7 @@ mod tests { .unwrap(); let stream: ArrowRecordBatchStream = futures::stream::iter(vec![Ok(batch)]).boxed(); - let predicate = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( + let set = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( stream, HashSet::from_iter(vec![3]), ) @@ -789,8 +868,11 @@ mod tests { .expect("error parsing equality delete stream"); assert_eq!( - predicate.to_string(), - "((status IS NULL) OR (status != \"A\")) AND ((status IS NULL) OR (status != \"B\"))" + set.keys, + HashSet::from([ + EqDeleteKey(vec![Some(Datum::string("A"))]), + EqDeleteKey(vec![Some(Datum::string("B"))]), + ]) ); } @@ -1113,11 +1195,11 @@ mod tests { // Verify both delete types can be processed together let result = delete_filter - .build_equality_delete_predicate(&file_scan_task) + .build_equality_delete_sets(&file_scan_task) .await; assert!( result.is_ok(), - "Failed to build equality delete predicate: {:?}", + "Failed to build equality delete sets: {:?}", result.err() ); } @@ -1161,13 +1243,14 @@ mod tests { let eq_ids = HashSet::from_iter(vec![2]); - let result = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( + let set = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( record_batch_stream, eq_ids, ) - .await; + .await + .expect("error parsing equality delete stream"); - assert!(result.is_ok()); + assert_eq!(set.keys.len(), 20_000); } #[tokio::test] diff --git a/crates/iceberg/src/arrow/delete_filter.rs b/crates/iceberg/src/arrow/delete_filter.rs index bcfcbd233d..b0ddce26d6 100644 --- a/crates/iceberg/src/arrow/delete_filter.rs +++ b/crates/iceberg/src/arrow/delete_filter.rs @@ -22,9 +22,8 @@ use tokio::sync::Notify; use tokio::sync::futures::OwnedNotified; use tokio::sync::oneshot::Receiver; +use crate::arrow::caching_delete_file_loader::EqDeleteSet; use crate::delete_vector::DeleteVector; -use crate::expr::Predicate::AlwaysTrue; -use crate::expr::{Bind, BoundPredicate, Predicate}; use crate::runtime::Runtime; use crate::scan::{FileScanTask, FileScanTaskDeleteFile}; use crate::spec::DataContentType; @@ -33,7 +32,7 @@ use crate::{Error, ErrorKind, Result}; #[derive(Debug)] enum EqDelState { Loading(Arc), - Loaded(Predicate), + Loaded(Arc), } /// State tracking for positional delete files. @@ -163,17 +162,17 @@ impl DeleteFilter { } } - /// Retrieve the equality delete predicate for a given eq delete file path - pub(crate) async fn get_equality_delete_predicate_for_delete_file_path( + /// Retrieve the equality delete set for a given eq delete file path + pub(crate) async fn get_equality_delete_set_for_delete_file_path( &self, file_path: &str, - ) -> Option { + ) -> Option> { let notifier = { match self.state.read().unwrap().equality_deletes.get(file_path) { None => return None, Some(EqDelState::Loading(notifier)) => notifier.clone(), - Some(EqDelState::Loaded(predicate)) => { - return Some(predicate.clone()); + Some(EqDelState::Loaded(set)) => { + return Some(set.clone()); } } }; @@ -181,50 +180,69 @@ impl DeleteFilter { notifier.notified().await; match self.state.read().unwrap().equality_deletes.get(file_path) { - Some(EqDelState::Loaded(predicate)) => Some(predicate.clone()), + Some(EqDelState::Loaded(set)) => Some(set.clone()), _ => unreachable!("Cannot be any other state than loaded"), } } - /// Builds eq delete predicate for the provided task. - pub(crate) async fn build_equality_delete_predicate( + /// Builds the equality-delete sets applicable to the given task, one per distinct + /// equality-column layout. + pub(crate) async fn build_equality_delete_sets( &self, file_scan_task: &FileScanTask, - ) -> Result> { - // * Filter the task's deletes into just the Equality deletes - // * Retrieve the unbound predicate for each from self.state.equality_deletes - // * Logical-AND them all together to get a single combined `Predicate` - // * Bind the predicate to the task's schema to get a `BoundPredicate` - - let mut combined_predicate = AlwaysTrue; + ) -> Result>> { + let mut groups: HashMap, Vec>> = HashMap::new(); for delete in &file_scan_task.deletes { if !is_equality_delete(delete) { continue; } - let Some(predicate) = self - .get_equality_delete_predicate_for_delete_file_path(&delete.file_path) + let Some(set) = self + .get_equality_delete_set_for_delete_file_path(&delete.file_path) .await else { return Err(Error::new( ErrorKind::Unexpected, format!( - "Missing predicate for equality delete file '{}'", + "Missing equality delete set for delete file '{}'", delete.file_path ), )); }; - combined_predicate = combined_predicate.and(predicate); + let layout = set.fields.iter().map(|(_, id, _)| *id).collect(); + groups.entry(layout).or_default().push(set); } - if combined_predicate == AlwaysTrue { - return Ok(None); + let mut result = Vec::with_capacity(groups.len()); + for mut sets in groups.into_values() { + if sets.len() == 1 { + result.push(sets.pop().unwrap()); + } else { + let base = sets + .iter() + .enumerate() + .max_by_key(|(_, set)| set.keys.len()) + .map(|(idx, _)| idx) + .expect("group is non-empty"); + let mut combined = (*sets.swap_remove(base)).clone(); + for other in &sets { + // The layouts always match here: every set in this group was parsed + // after `BasicDeleteFileLoader::evolve_schema` normalised its batch + // stream against the same `task.schema`, so one field id has one + // recorded type across the scan. `union` re-checks rather than + // trusting that, because the invariant lives in the caller: the + // equality-delete cache is keyed by delete-file path alone, with no + // record of the schema an entry was evolved against, and + // `ArrowReader` is `Clone` with a caller-supplied + // `FileScanTaskStream`. Two tasks with different schemas through one + // reader would reach this point, and an error beats a wrong read. + combined.union(other)?; + } + result.push(Arc::new(combined)); + } } - - let bound_predicate = combined_predicate - .bind(file_scan_task.schema.clone(), file_scan_task.case_sensitive)?; - Ok(Some(bound_predicate)) + Ok(result) } pub(crate) fn upsert_delete_vector( @@ -247,7 +265,7 @@ impl DeleteFilter { pub(crate) fn insert_equality_delete( &self, delete_file_path: &str, - eq_del: Receiver, + eq_del: Receiver>, ) { let notify = Arc::new(Notify::new()); { @@ -279,6 +297,7 @@ pub(crate) fn is_equality_delete(f: &FileScanTaskDeleteFile) -> bool { #[cfg(test)] pub(crate) mod tests { + use std::collections::HashSet; use std::fs::File; use std::path::Path; use std::sync::Arc; @@ -291,8 +310,9 @@ pub(crate) mod tests { use tempfile::TempDir; use super::*; - use crate::arrow::caching_delete_file_loader::CachingDeleteFileLoader; - use crate::expr::Reference; + use crate::arrow::caching_delete_file_loader::{ + CachingDeleteFileLoader, EqDeleteKey, EqDeleteSet, + }; use crate::io::FileIO; use crate::spec::{DataFileFormat, Datum, NestedField, PrimitiveType, Schema, Type}; @@ -517,54 +537,109 @@ pub(crate) mod tests { Arc::new(arrow_schema::Schema::new(fields)) } - #[tokio::test] - async fn test_build_equality_delete_predicate_case_sensitive() { + fn eq_delete_file(path: &str) -> FileScanTaskDeleteFile { + FileScanTaskDeleteFile::builder() + .with_file_path(path.to_string()) + .with_file_size_in_bytes(1) + .with_file_type(DataContentType::EqualityDeletes) + .with_partition_spec_id(0) + .build() + } + + fn task_with_eq_deletes(paths: &[&str]) -> FileScanTask { let schema = Arc::new( Schema::builder() .with_schema_id(1) .with_fields(vec![ - NestedField::required(1, "Id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(2, "id2", Type::Primitive(PrimitiveType::Long)).into(), ]) .build() .unwrap(), ); - - // ---------- fake FileScanTask ---------- - let task = FileScanTask::builder() + FileScanTask::builder() .with_file_size_in_bytes(0) .with_start(0) .with_length(0) .with_data_file_path("data.parquet".to_string()) .with_data_file_format(DataFileFormat::Parquet) - .with_schema(schema.clone()) + .with_schema(schema) .with_project_field_ids(vec![]) - .with_deletes(vec![ - FileScanTaskDeleteFile::builder() - .with_file_path("eq-del.parquet".to_string()) - .with_file_size_in_bytes(1) // never read; this test fails before opening the file - .with_file_type(DataContentType::EqualityDeletes) - .with_partition_spec_id(0) - .build(), - ]) - .with_case_sensitive(true) - .build(); + .with_deletes(paths.iter().map(|p| eq_delete_file(p)).collect()) + .with_case_sensitive(false) + .build() + } + + fn insert_set(filter: &DeleteFilter, path: &str, set: EqDeleteSet) { + let (tx, rx) = tokio::sync::oneshot::channel(); + filter.insert_equality_delete(path, rx); + tx.send(Arc::new(set)).unwrap(); + } + // Delete files that key on different equality columns must not be merged. + #[tokio::test] + async fn test_build_equality_delete_sets_mixed_ids_not_merged() { + let task = task_with_eq_deletes(&["eq-a.parquet", "eq-b.parquet"]); let filter = DeleteFilter::new(Runtime::current()); - // ---------- insert equality delete predicate ---------- - let pred = Reference::new("id").equal_to(Datum::long(10)); + insert_set(&filter, "eq-a.parquet", EqDeleteSet { + keys: HashSet::from([EqDeleteKey(vec![Some(Datum::long(1))])]), + fields: vec![("id".to_string(), 1, Type::Primitive(PrimitiveType::Long))], + }); + insert_set(&filter, "eq-b.parquet", EqDeleteSet { + keys: HashSet::from([EqDeleteKey(vec![Some(Datum::long(2))])]), + fields: vec![("id2".to_string(), 2, Type::Primitive(PrimitiveType::Long))], + }); - let (tx, rx) = tokio::sync::oneshot::channel(); - filter.insert_equality_delete("eq-del.parquet", rx); + let sets = filter.build_equality_delete_sets(&task).await.unwrap(); + assert_eq!(sets.len(), 2); + } + + // `union` must refuse a mismatch itself rather than trusting the caller to have grouped. + #[test] + fn test_union_rejects_mismatched_layout() { + let mut a = EqDeleteSet { + keys: HashSet::from([EqDeleteKey(vec![Some(Datum::long(1))])]), + fields: vec![("id".to_string(), 1, Type::Primitive(PrimitiveType::Long))], + }; + let b = EqDeleteSet { + keys: HashSet::from([EqDeleteKey(vec![Some(Datum::int(2))])]), + fields: vec![("id".to_string(), 1, Type::Primitive(PrimitiveType::Int))], + }; + assert!(!a.layout_matches(&b)); + let err = a.union(&b).expect_err("differing types must be rejected"); + assert!(err.to_string().contains("different layouts"), "{err}"); + assert_eq!( + a.keys.len(), + 1, + "a rejected union must not mutate the target" + ); + } - tx.send(pred).unwrap(); + // Delete files sharing an equality-column layout are unioned into a single set. + #[tokio::test] + async fn test_build_equality_delete_sets_same_layout_unioned() { + let task = task_with_eq_deletes(&["eq-a.parquet", "eq-b.parquet"]); + let filter = DeleteFilter::new(Runtime::current()); - // ---------- should FAIL ---------- - let result = filter.build_equality_delete_predicate(&task).await; + let fields = vec![("id".to_string(), 1, Type::Primitive(PrimitiveType::Long))]; + insert_set(&filter, "eq-a.parquet", EqDeleteSet { + keys: HashSet::from([EqDeleteKey(vec![Some(Datum::long(1))])]), + fields: fields.clone(), + }); + insert_set(&filter, "eq-b.parquet", EqDeleteSet { + keys: HashSet::from([EqDeleteKey(vec![Some(Datum::long(2))])]), + fields, + }); - assert!( - result.is_err(), - "case_sensitive=true should fail when column case mismatches" + let sets = filter.build_equality_delete_sets(&task).await.unwrap(); + assert_eq!(sets.len(), 1); + assert_eq!( + sets[0].keys, + HashSet::from([ + EqDeleteKey(vec![Some(Datum::long(1))]), + EqDeleteKey(vec![Some(Datum::long(2))]), + ]) ); } } diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs index e7bcdf5fb3..dad10e926e 100644 --- a/crates/iceberg/src/arrow/reader/pipeline.rs +++ b/crates/iceberg/src/arrow/reader/pipeline.rs @@ -26,7 +26,9 @@ use std::sync::atomic::AtomicU64; use arrow_schema::{DataType, Field}; use futures::{StreamExt, TryStreamExt}; -use parquet::arrow::arrow_reader::{ArrowReaderMetadata, ArrowReaderOptions}; +use parquet::arrow::arrow_reader::{ + ArrowPredicate, ArrowReaderMetadata, ArrowReaderOptions, RowFilter, +}; use parquet::arrow::{PARQUET_FIELD_ID_META_KEY, ParquetRecordBatchStreamBuilder, RowNumber}; use parquet::encryption::decrypt::FileDecryptionProperties; @@ -348,36 +350,24 @@ impl FileScanTaskReader { } let delete_filter = delete_filter_rx.await.unwrap()?; - let delete_predicate = delete_filter.build_equality_delete_predicate(&task).await?; - - // In addition to the optional predicate supplied in the `FileScanTask`, - // we also have an optional predicate resulting from equality delete files. - // If both are present, we logical-AND them together to form a single filter - // predicate that we can pass to the `RecordBatchStreamBuilder`. - let final_predicate = match (&task.predicate, delete_predicate) { - (None, None) => None, - (Some(predicate), None) => Some(predicate.clone()), - (None, Some(ref predicate)) => Some(predicate.clone()), - (Some(filter_predicate), Some(delete_predicate)) => { - Some(filter_predicate.clone().and(delete_predicate)) - } - }; + let eq_delete_sets = delete_filter.build_equality_delete_sets(&task).await?; + + // Equality deletes are applied as a row-filter predicate based on a hash-set (see below), + // so the scan-predicate carries only the optional predicate + let final_predicate = task.predicate.clone(); // There are three possible sources for potential lists of selected RowGroup indices, // and two for `RowSelection`s. - // Selected RowGroup index lists can come from three sources: + // Selected RowGroup index lists can come from two sources: // * When task.start and task.length specify a byte range (file splitting); - // * When there are equality delete files that are applicable; // * When there is a scan predicate and row_group_filtering_enabled = true. // `RowSelection`s can be created in either or both of the following cases: // * When there are positional delete files that are applicable; - // * When there is a scan predicate and row_selection_enabled = true - // Note that row group filtering from predicates only happens when - // there is a scan predicate AND row_group_filtering_enabled = true, - // but we perform row selection filtering if there are applicable - // equality delete files OR (there is a scan predicate AND row_selection_enabled), - // since the only implemented method of applying positional deletes is - // by using a `RowSelection`. + // * When there is a scan predicate and row_selection_enabled = true. + // Equality deletes appear in neither list: they are applied as `RowFilter` + // predicates during decoding (see `build_equality_delete_predicates`), so they + // never contribute row-group indices or a `RowSelection`. Positional deletes + // still do, since a `RowSelection` remains the only way they are applied. let mut selected_row_group_indices = None; let mut row_selection = None; @@ -392,6 +382,8 @@ impl FileScanTaskReader { selected_row_group_indices = Some(byte_range_filtered_row_groups); } + let mut row_filter_predicates: Vec> = Vec::new(); + if let Some(predicate) = final_predicate { let (iceberg_field_ids, field_id_map) = ArrowReader::build_field_id_set_and_map( record_batch_stream_builder.parquet_schema(), @@ -400,13 +392,12 @@ impl FileScanTaskReader { use_position_fallback, )?; - let row_filter = ArrowReader::get_row_filter( + row_filter_predicates.push(ArrowReader::build_scan_predicate( &predicate, record_batch_stream_builder.parquet_schema(), &iceberg_field_ids, &field_id_map, - )?; - record_batch_stream_builder = record_batch_stream_builder.with_row_filter(row_filter); + )?); if self.row_group_filtering_enabled { let predicate_filtered_row_groups = ArrowReader::get_selected_row_group_indices( @@ -442,6 +433,20 @@ impl FileScanTaskReader { } } + if !eq_delete_sets.is_empty() { + row_filter_predicates.extend(ArrowReader::build_equality_delete_predicates( + &eq_delete_sets, + record_batch_stream_builder.parquet_schema(), + record_batch_stream_builder.schema(), + use_position_fallback, + )?); + } + + if !row_filter_predicates.is_empty() { + record_batch_stream_builder = + record_batch_stream_builder.with_row_filter(RowFilter::new(row_filter_predicates)); + } + let positional_delete_indexes = delete_filter.get_delete_vector(&task); if let Some(positional_delete_indexes) = positional_delete_indexes { diff --git a/crates/iceberg/src/arrow/reader/projection.rs b/crates/iceberg/src/arrow/reader/projection.rs index 9fdc9fed65..6493f39961 100644 --- a/crates/iceberg/src/arrow/reader/projection.rs +++ b/crates/iceberg/src/arrow/reader/projection.rs @@ -49,17 +49,26 @@ impl ArrowReader { visit(&mut collector, predicate)?; let iceberg_field_ids = collector.field_ids(); + let field_id_map = + Self::resolve_field_id_map(parquet_schema, arrow_schema, use_position_fallback)?; - let field_id_map = match build_field_id_map(parquet_schema)? { + Ok((iceberg_field_ids, field_id_map)) + } + + /// Resolves the field-id-to-leaf-column-index map for a Parquet file + pub(super) fn resolve_field_id_map( + parquet_schema: &SchemaDescriptor, + arrow_schema: &ArrowSchemaRef, + use_position_fallback: bool, + ) -> Result> { + Ok(match build_field_id_map(parquet_schema)? { Some(map) => map, // No embedded field IDs and no name mapping: position-based fallback None if use_position_fallback => build_fallback_field_id_map(parquet_schema), // No embedded field IDs, but a name mapping assigned them to the Arrow // schema: resolve columns through the mapped Arrow field-id metadata None => build_field_id_map_from_arrow_schema(arrow_schema), - }; - - Ok((iceberg_field_ids, field_id_map)) + }) } /// Recursively extract leaf field IDs because Parquet projection works at the leaf column level. diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index 2e6b37ba30..53b3f18d62 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -23,26 +23,40 @@ use std::collections::{HashMap, HashSet}; use std::sync::Arc; +use arrow_array::{Array, BooleanArray, RecordBatch}; +use arrow_schema::{ArrowError, SchemaRef as ArrowSchemaRef}; use parquet::arrow::ProjectionMask; -use parquet::arrow::arrow_reader::{ArrowPredicateFn, RowFilter, RowSelection}; +use parquet::arrow::arrow_reader::{ArrowPredicate, ArrowPredicateFn, RowSelection}; use parquet::file::metadata::ParquetMetaData; use parquet::schema::types::SchemaDescriptor; use super::{ArrowReader, PredicateConverter}; +use crate::arrow::caching_delete_file_loader::{EqDeleteKey, EqDeleteSet}; +use crate::arrow::{arrow_primitive_to_literal, arrow_type_to_type}; use crate::error::Result; use crate::expr::BoundPredicate; use crate::expr::visitors::bound_predicate_visitor::visit; use crate::expr::visitors::page_index_evaluator::PageIndexEvaluator; use crate::expr::visitors::row_group_metrics_evaluator::RowGroupMetricsEvaluator; -use crate::spec::Schema; +use crate::spec::{Datum, Literal, PrimitiveType, Schema, Type}; + +/// One equality-delete key column of a decoded batch +struct KeyColumn { + literals: Vec>, + /// Iceberg type of the column as stored in this data file. + source_primitive: PrimitiveType, + /// Whether the file type differs from the table type, so values need `Datum::to`. + needs_promotion: bool, +} impl ArrowReader { - pub(super) fn get_row_filter( + /// Builds the Arrow row-filter predicate for a bound scan predicate. + pub(super) fn build_scan_predicate( predicates: &BoundPredicate, parquet_schema: &SchemaDescriptor, iceberg_field_ids: &HashSet, field_id_map: &HashMap, - ) -> Result { + ) -> Result> { // Collect Parquet column indices from field ids. // If the field id is not found in Parquet schema, it will be ignored due to schema evolution. let mut column_indices = iceberg_field_ids @@ -62,8 +76,134 @@ impl ArrowReader { // creates the projection mask for the Arrow predicates. let projection_mask = ProjectionMask::leaves(parquet_schema, column_indices.clone()); let predicate_func = visit(&mut converter, predicates)?; - let arrow_predicate = ArrowPredicateFn::new(projection_mask, predicate_func); - Ok(RowFilter::new(vec![Box::new(arrow_predicate)])) + Ok(Box::new(ArrowPredicateFn::new( + projection_mask, + predicate_func, + ))) + } + + /// Builds one Arrow row-filter predicate per equality-delete set. The predicate is based + /// on a hash-set lookup (see `EqDeleteSet`). It keeps a row unless its key tuple is present + /// in that set. A row is deleted when it matches any set (the predicates are AND-ed by the `RowFilter`). + pub(super) fn build_equality_delete_predicates( + sets: &[Arc], + parquet_schema: &SchemaDescriptor, + arrow_schema: &ArrowSchemaRef, + use_position_fallback: bool, + ) -> Result>> { + let field_id_map = + Self::resolve_field_id_map(parquet_schema, arrow_schema, use_position_fallback)?; + + let mut predicates: Vec> = Vec::new(); + for set in sets { + if set.is_empty() { + continue; + } + + // Parquet leaf index for each key column, in `fields` order; a column dropped + // from this file by schema evolution has no entry. + let leaf_indices: Vec> = set + .fields + .iter() + .map(|(_, id, _)| field_id_map.get(id).copied()) + .collect(); + + // Every key column may be absent from this file, leaving `column_indices` + // empty. `ProjectionMask::leaves(schema, [])` predicate with a zero-column + // batch carrying and correct row count, so the probe still returns one boolean + // per row and the selection. + let mut column_indices: Vec = leaf_indices.iter().flatten().copied().collect(); + column_indices.sort_unstable(); + column_indices.dedup(); + let projection_mask = ProjectionMask::leaves(parquet_schema, column_indices.clone()); + + // Position of each key column within the projected batch (parquet-rs presents the + // masked leaves in ascending leaf-index order). + let batch_positions: Vec> = leaf_indices + .iter() + .map(|leaf| leaf.and_then(|idx| column_indices.binary_search(&idx).ok())) + .collect(); + + let target_types: Vec = set.fields.iter().map(|(_, _, ty)| ty.clone()).collect(); + let num_cols = set.fields.len(); + let set = set.clone(); + + let predicate_func = + move |batch: RecordBatch| -> std::result::Result { + let num_rows = batch.num_rows(); + + let mut columns: Vec> = Vec::with_capacity(num_cols); + for (i, target_type) in target_types.iter().enumerate() { + // A column absent from this file (schema evolution) + let Some(pos) = batch_positions[i] else { + columns.push(None); + continue; + }; + let array = batch.column(pos); + let source_type = arrow_type_to_type(array.data_type()) + .map_err(|e| ArrowError::ComputeError(e.to_string()))?; + let source_primitive = source_type + .as_primitive_type() + .ok_or_else(|| { + ArrowError::ComputeError( + "equality delete key column is not a primitive type" + .to_string(), + ) + })? + .clone(); + columns.push(Some(KeyColumn { + // Promotion to the table type keeps these comparable with the + // parsed delete keys under schema evolution. + needs_promotion: source_type != *target_type, + literals: arrow_primitive_to_literal(array, &source_type) + .map_err(|e| ArrowError::ComputeError(e.to_string()))?, + source_primitive, + })); + } + + // One hash lookup per row. + let mut keep = Vec::with_capacity(num_rows); + let mut probe = EqDeleteKey(vec![None; num_cols]); + for row in 0..num_rows { + for (i, column) in columns.iter_mut().enumerate() { + let Some(column) = column else { + probe.0[i] = None; + continue; + }; + // we can `take` because each cell is probed once. + probe.0[i] = match column.literals[row].take() { + Some(Literal::Primitive(primitive)) => { + let datum = + Datum::new(column.source_primitive.clone(), primitive); + Some(if column.needs_promotion { + datum + .to(&target_types[i]) + .map_err(|e| ArrowError::ComputeError(e.to_string()))? + } else { + datum + }) + } + Some(other) => { + return Err(ArrowError::ComputeError(format!( + "equality delete key column {i} is not a primitive \ + literal: {other:?}" + ))); + } + None => None, + }; + } + keep.push(!set.keys.contains(&probe)); + } + Ok(BooleanArray::from(keep)) + }; + + predicates.push(Box::new(ArrowPredicateFn::new( + projection_mask, + predicate_func, + ))); + } + + Ok(predicates) } pub(super) fn get_selected_row_group_indices( @@ -217,7 +357,8 @@ mod tests { use arrow_array::cast::AsArray; use arrow_array::{ - ArrayRef, Int32Array, Int64Array, LargeStringArray, RecordBatch, StringArray, + ArrayRef, Decimal128Array, Float64Array, Int32Array, Int64Array, LargeStringArray, + RecordBatch, StringArray, }; use arrow_schema::{DataType, Field, Schema as ArrowSchema}; use futures::TryStreamExt; @@ -235,7 +376,8 @@ mod tests { use crate::io::FileIO; use crate::scan::{FileScanTask, FileScanTaskDeleteFile, FileScanTaskStream}; use crate::spec::{ - DataContentType, DataFileFormat, Datum, NestedField, PrimitiveType, Schema, SchemaRef, Type, + DataContentType, DataFileFormat, Datum, Literal, NestedField, PrimitiveType, Schema, + SchemaRef, Type, }; async fn test_perform_read( @@ -1285,4 +1427,816 @@ mod tests { "positional deletes must be applied correctly even when page indexes are absent" ); } + + fn eqd_field(name: &str, dt: DataType, id: i32, nullable: bool) -> Field { + Field::new(name, dt, nullable).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + id.to_string(), + )])) + } + + fn eqd_write(path: &str, schema: Arc, columns: Vec) { + let batch = RecordBatch::try_new(schema.clone(), columns).unwrap(); + let file = File::create(path).unwrap(); + let props = WriterProperties::builder() + .set_compression(Compression::SNAPPY) + .build(); + let mut writer = ArrowWriter::try_new(file, schema, Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + } + + fn eqd_delete_file(path: &str, equality_ids: Vec) -> FileScanTaskDeleteFile { + FileScanTaskDeleteFile::builder() + .with_file_path(path.to_string()) + .with_file_size_in_bytes(std::fs::metadata(path).unwrap().len()) + .with_file_type(DataContentType::EqualityDeletes) + .with_equality_ids(Some(equality_ids)) + .with_partition_spec_id(0) + .build() + } + + async fn eqd_read( + data_path: &str, + table_schema: SchemaRef, + project_field_ids: Vec, + deletes: Vec, + ) -> Vec { + eqd_read_with( + data_path, + table_schema, + project_field_ids, + deletes, + None, + false, + ) + .await + } + + /// `eqd_read` plus the two things that interact with equality-delete predicates: a + /// scan predicate (pushed as a second `ArrowPredicate`) and the row-selection path. + async fn eqd_read_with( + data_path: &str, + table_schema: SchemaRef, + project_field_ids: Vec, + deletes: Vec, + predicate: Option, + row_selection_enabled: bool, + ) -> Vec { + let file_io = FileIO::new_with_fs(); + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()) + .with_row_selection_enabled(row_selection_enabled) + .build(); + let task = FileScanTask::builder() + .with_file_size_in_bytes(std::fs::metadata(data_path).unwrap().len()) + .with_start(0) + .with_length(0) + .with_data_file_path(data_path.to_string()) + .with_data_file_format(DataFileFormat::Parquet) + .with_schema(table_schema) + .with_project_field_ids(project_field_ids) + .with_deletes(deletes) + .with_case_sensitive(false) + .with_predicate(predicate) + .build(); + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + reader + .read(tasks) + .unwrap() + .stream() + .try_collect::>() + .await + .unwrap() + } + + fn eqd_collect_i64(batches: &[RecordBatch], col: usize) -> Vec { + batches + .iter() + .flat_map(|b| { + b.column(col) + .as_primitive::() + .iter() + .flatten() + .collect::>() + }) + .collect() + } + + // A single-column equality delete removes exactly the matching rows. + #[tokio::test] + async fn test_eq_delete_single_column_filters_matching_rows() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(2, "val", Type::Primitive(PrimitiveType::String)).into(), + ]) + .build() + .unwrap(), + ); + + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![ + eqd_field("id", DataType::Int64, 1, false), + eqd_field("val", DataType::Utf8, 2, true), + ])), + vec![ + Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5])) as ArrayRef, + Arc::new(StringArray::from(vec!["a", "b", "c", "d", "e"])) as ArrayRef, + ], + ); + + let del_path = format!("{loc}/eq-del.parquet"); + eqd_write( + &del_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![3])) as ArrayRef], + ); + + let result = eqd_read(&data_path, table_schema, vec![1, 2], vec![eqd_delete_file( + &del_path, + vec![1], + )]) + .await; + + assert_eq!(eqd_collect_i64(&result, 0), vec![1, 2, 4, 5]); + } + + // A multi-column key deletes only exact tuple matches and a row with a null in a key column + // (and a row differing in any column) is kept. + #[tokio::test] + async fn test_eq_delete_multi_column_keeps_null_and_partial_matches() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(2, "status", Type::Primitive(PrimitiveType::String)) + .into(), + ]) + .build() + .unwrap(), + ); + + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![ + eqd_field("id", DataType::Int64, 1, false), + eqd_field("status", DataType::Utf8, 2, true), + ])), + vec![ + Arc::new(Int64Array::from(vec![2, 2, 2, 3])) as ArrayRef, + Arc::new(StringArray::from(vec![ + Some("X"), + Some("Y"), + None, + Some("X"), + ])) as ArrayRef, + ], + ); + + let del_path = format!("{loc}/eq-del.parquet"); + eqd_write( + &del_path, + Arc::new(ArrowSchema::new(vec![ + eqd_field("id", DataType::Int64, 1, false), + eqd_field("status", DataType::Utf8, 2, true), + ])), + vec![ + Arc::new(Int64Array::from(vec![2])) as ArrayRef, + Arc::new(StringArray::from(vec![Some("X")])) as ArrayRef, + ], + ); + + let result = eqd_read(&data_path, table_schema, vec![1, 2], vec![eqd_delete_file( + &del_path, + vec![1, 2], + )]) + .await; + + let mut pairs = Vec::new(); + for b in &result { + let ids: Vec = b + .column(0) + .as_primitive::() + .iter() + .flatten() + .collect(); + let statuses: Vec> = b + .column(1) + .as_string::() + .iter() + .map(|o| o.map(|s| s.to_string())) + .collect(); + for (id, status) in ids.into_iter().zip(statuses) { + pairs.push((id, status)); + } + } + + assert_eq!(pairs, vec![ + (2, Some("Y".to_string())), + (2, None), + (3, Some("X".to_string())), + ]); + } + + // The data file stored `id` as int32 but the table type is long, so it must be promoted + // to the table type. + #[tokio::test] + async fn test_eq_delete_promotes_data_type_before_probe() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int32, + 1, + false, + )])), + vec![Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef], + ); + + let del_path = format!("{loc}/eq-del.parquet"); + eqd_write( + &del_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![3])) as ArrayRef], + ); + + let result = eqd_read(&data_path, table_schema, vec![1], vec![eqd_delete_file( + &del_path, + vec![1], + )]) + .await; + + assert_eq!(eqd_collect_i64(&result, 0), vec![1, 2]); + } + + // Two delete files that key on different columns each apply independently: a row is deleted + // if it matches either. + #[tokio::test] + async fn test_eq_delete_distinct_layouts_apply_independently() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(2, "id2", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(3, "val", Type::Primitive(PrimitiveType::String)).into(), + ]) + .build() + .unwrap(), + ); + + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![ + eqd_field("id", DataType::Int64, 1, false), + eqd_field("id2", DataType::Int64, 2, false), + eqd_field("val", DataType::Utf8, 3, true), + ])), + vec![ + Arc::new(Int64Array::from(vec![1, 2, 3])) as ArrayRef, + Arc::new(Int64Array::from(vec![10, 20, 30])) as ArrayRef, + Arc::new(StringArray::from(vec!["a", "b", "c"])) as ArrayRef, + ], + ); + + let del_a = format!("{loc}/eq-del-a.parquet"); + eqd_write( + &del_a, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![1])) as ArrayRef], + ); + let del_b = format!("{loc}/eq-del-b.parquet"); + eqd_write( + &del_b, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id2", + DataType::Int64, + 2, + false, + )])), + vec![Arc::new(Int64Array::from(vec![20])) as ArrayRef], + ); + + let result = eqd_read(&data_path, table_schema, vec![1, 2, 3], vec![ + eqd_delete_file(&del_a, vec![1]), + eqd_delete_file(&del_b, vec![2]), + ]) + .await; + + assert_eq!(eqd_collect_i64(&result, 0), vec![3]); + } + + // A scan predicate and an equality delete are pushed as two separate `ArrowPredicate`s + // (they used to be ANDed into one bound `Predicate`), so the result must be the + // intersection of both. + #[tokio::test] + async fn test_eq_delete_with_scan_predicate_intersects() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5, 6])) as ArrayRef], + ); + + let del_path = format!("{loc}/eq-del.parquet"); + eqd_write( + &del_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![2, 5])) as ArrayRef], + ); + + for row_selection_enabled in [false, true] { + let predicate = Reference::new("id") + .greater_than_or_equal_to(Datum::long(3)) + .bind(table_schema.clone(), false) + .unwrap(); + + let result = eqd_read_with( + &data_path, + table_schema.clone(), + vec![1], + vec![eqd_delete_file(&del_path, vec![1])], + Some(predicate), + row_selection_enabled, + ) + .await; + + assert_eq!( + eqd_collect_i64(&result, 0), + vec![3, 4, 6], + "row_selection_enabled={row_selection_enabled}" + ); + } + } + + #[tokio::test] + async fn test_pos_and_eq_delete_with_scan_predicate() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5, 6])) as ArrayRef], + ); + + // Positional delete removes row 2 (id 3), equality delete removes id 5, and the + // predicate keeps id >= 2. Each removes something the others do not. + let pos_del_path = format!("{loc}/pos-del.parquet"); + { + let pos_schema = crate::arrow::delete_filter::tests::create_pos_del_schema(); + let batch = RecordBatch::try_new(pos_schema.clone(), vec![ + Arc::new(StringArray::from(vec![data_path.as_str()])) as ArrayRef, + Arc::new(Int64Array::from(vec![2i64])) as ArrayRef, + ]) + .unwrap(); + let file = File::create(&pos_del_path).unwrap(); + let props = WriterProperties::builder() + .set_compression(Compression::SNAPPY) + .build(); + let mut writer = ArrowWriter::try_new(file, pos_schema, Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + } + + let eq_del_path = format!("{loc}/eq-del.parquet"); + eqd_write( + &eq_del_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![5])) as ArrayRef], + ); + + let pos_del = FileScanTaskDeleteFile::builder() + .with_file_path(pos_del_path.clone()) + .with_file_size_in_bytes(std::fs::metadata(&pos_del_path).unwrap().len()) + .with_file_type(DataContentType::PositionDeletes) + .with_partition_spec_id(0) + .build(); + + for row_selection_enabled in [false, true] { + let predicate = Reference::new("id") + .greater_than_or_equal_to(Datum::long(2)) + .bind(table_schema.clone(), false) + .unwrap(); + + let result = eqd_read_with( + &data_path, + table_schema.clone(), + vec![1], + vec![pos_del.clone(), eqd_delete_file(&eq_del_path, vec![1])], + Some(predicate), + row_selection_enabled, + ) + .await; + + assert_eq!( + eqd_collect_i64(&result, 0), + vec![2, 4, 6], + "row_selection_enabled={row_selection_enabled}" + ); + } + } + + // An equality-delete key column absent from the data file (added by later schema + // evolution). The probe reads it as null for every row, so nothing matches and the + // delete file removes no rows -- it does NOT resolve `initial_default`, which the spec + // asks for via normal projection rules. Pinned here so the divergence is visible and a + // later fix has something to change. + #[tokio::test] + async fn test_eq_delete_on_column_absent_from_data_file() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + // `added` exists in the table schema but not in the data file below. + let required_with_default = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(2, "added", Type::Primitive(PrimitiveType::Long)) + .with_initial_default(Literal::long(7)) + .into(), + ]) + .build() + .unwrap(), + ); + let optional_no_default = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(2, "added", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![1, 2, 3])) as ArrayRef], + ); + + // Keyed on `added` with the value that IS the initial default: if the probe ever + // resolves defaults, every row matches and this expectation flips to empty. + let del_path = format!("{loc}/eq-del.parquet"); + eqd_write( + &del_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "added", + DataType::Int64, + 2, + false, + )])), + vec![Arc::new(Int64Array::from(vec![7])) as ArrayRef], + ); + + for (label, schema) in [ + ("required with initial_default", required_with_default), + ("optional without default", optional_no_default), + ] { + let result = eqd_read(&data_path, schema, vec![1], vec![eqd_delete_file( + &del_path, + vec![2], + )]) + .await; + + assert_eq!( + eqd_collect_i64(&result, 0), + vec![1, 2, 3], + "{label}: absent key column probes as null, so nothing is deleted" + ); + } + } + + #[tokio::test] + async fn test_eq_delete_files_with_different_physical_types_merge() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![1, 2, 3, 4])) as ArrayRef], + ); + + let del_i64 = format!("{loc}/eq-del-i64.parquet"); + eqd_write( + &del_i64, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![2])) as ArrayRef], + ); + + // Written as int32: an older file from before the column was widened to long. + let del_i32 = format!("{loc}/eq-del-i32.parquet"); + eqd_write( + &del_i32, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int32, + 1, + false, + )])), + vec![Arc::new(Int32Array::from(vec![3])) as ArrayRef], + ); + + let result = eqd_read(&data_path, table_schema, vec![1], vec![ + eqd_delete_file(&del_i64, vec![1]), + eqd_delete_file(&del_i32, vec![1]), + ]) + .await; + + // Both keys apply, so the two files really did end up in one group. + assert_eq!(eqd_collect_i64(&result, 0), vec![1, 4]); + } + + // A decimal equality-delete key whose precision differs from the table's. Precision + // widening is legal Iceberg schema evolution and goes through a different `Datum::to` + // arm than the integer promotion already covered. + #[tokio::test] + async fn test_eq_delete_decimal_precision_widening() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required( + 2, + "amount", + Type::Primitive(PrimitiveType::Decimal { + precision: 18, + scale: 2, + }), + ) + .into(), + ]) + .build() + .unwrap(), + ); + + // Data file written at the narrower precision it had when it was created. + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![ + eqd_field("id", DataType::Int64, 1, false), + eqd_field("amount", DataType::Decimal128(9, 2), 2, false), + ])), + vec![ + Arc::new(Int64Array::from(vec![1, 2, 3])) as ArrayRef, + Arc::new( + Decimal128Array::from(vec![100i128, 250, 375]) + .with_precision_and_scale(9, 2) + .unwrap(), + ) as ArrayRef, + ], + ); + + let del_path = format!("{loc}/eq-del.parquet"); + eqd_write( + &del_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "amount", + DataType::Decimal128(18, 2), + 2, + false, + )])), + vec![Arc::new( + Decimal128Array::from(vec![250i128]) + .with_precision_and_scale(18, 2) + .unwrap(), + ) as ArrayRef], + ); + + let result = eqd_read(&data_path, table_schema, vec![1, 2], vec![eqd_delete_file( + &del_path, + vec![2], + )]) + .await; + + assert_eq!(eqd_collect_i64(&result, 0), vec![1, 3]); + } + + // Double delete columns are out of spec. Nothing rejects them today, so this + // pins what actually happens: `PrimitiveLiteral` stores doubles as + // `OrderedFloat`, whose `Hash`/`Eq` canonicalise NaN and signed zero. + // A NaN key therefore matches NaN data, and a -0.0 key also removes 0.0. + #[tokio::test] + async fn test_eq_delete_double_column_canonicalises_nan_and_signed_zero() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(2, "d", Type::Primitive(PrimitiveType::Double)).into(), + ]) + .build() + .unwrap(), + ); + + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![ + eqd_field("id", DataType::Int64, 1, false), + eqd_field("d", DataType::Float64, 2, false), + ])), + vec![ + Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5])) as ArrayRef, + Arc::new(Float64Array::from(vec![1.5, f64::NAN, 0.0, -0.0, 2.5])) as ArrayRef, + ], + ); + + // A NaN key matches the NaN row (IEEE-754 says NaN != NaN; OrderedFloat says + // otherwise), and a -0.0 key removes both signed zeros. + let del_path = format!("{loc}/eq-del.parquet"); + eqd_write( + &del_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "d", + DataType::Float64, + 2, + false, + )])), + vec![Arc::new(Float64Array::from(vec![f64::NAN, -0.0])) as ArrayRef], + ); + + let result = eqd_read(&data_path, table_schema, vec![1, 2], vec![eqd_delete_file( + &del_path, + vec![2], + )]) + .await; + + assert_eq!(eqd_collect_i64(&result, 0), vec![1, 5]); + } + + #[tokio::test] + async fn test_empty_eq_delete_file_deletes_nothing() { + let tmp = TempDir::new().unwrap(); + let loc = tmp.path().to_str().unwrap(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + + let data_path = format!("{loc}/data.parquet"); + eqd_write( + &data_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(vec![1, 2, 3])) as ArrayRef], + ); + + let del_path = format!("{loc}/eq-del-empty.parquet"); + eqd_write( + &del_path, + Arc::new(ArrowSchema::new(vec![eqd_field( + "id", + DataType::Int64, + 1, + false, + )])), + vec![Arc::new(Int64Array::from(Vec::::new())) as ArrayRef], + ); + + let result = eqd_read(&data_path, table_schema, vec![1], vec![eqd_delete_file( + &del_path, + vec![1], + )]) + .await; + + assert_eq!(eqd_collect_i64(&result, 0), vec![1, 2, 3]); + } }