From 1940c51020dcd748eb39ece8e1d0e9160141d638 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Wed, 9 Sep 2026 09:57:18 -0400 Subject: [PATCH 1/2] address followup comments from #3035 --- crates/iceberg/public-api.txt | 2 + .../src/arrow/caching_delete_file_loader.rs | 176 +-------------- crates/iceberg/src/arrow/delete_filter.rs | 12 +- crates/iceberg/src/arrow/reader/pipeline.rs | 3 +- .../src/arrow/reader/positional_deletes.rs | 15 +- crates/iceberg/src/delete_file_index.rs | 141 ++++++------ crates/iceberg/src/scan/mod.rs | 3 +- crates/iceberg/src/scan/task.rs | 213 +++++++++++++++++- 8 files changed, 316 insertions(+), 249 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index ca55804966..7bb351ad12 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1325,6 +1325,8 @@ impl core::clone::Clone for iceberg::scan::FileScanTaskDeleteFile pub fn iceberg::scan::FileScanTaskDeleteFile::clone(&self) -> iceberg::scan::FileScanTaskDeleteFile impl core::cmp::PartialEq for iceberg::scan::FileScanTaskDeleteFile pub fn iceberg::scan::FileScanTaskDeleteFile::eq(&self, other: &iceberg::scan::FileScanTaskDeleteFile) -> bool +impl core::convert::From for iceberg::Result +pub fn iceberg::Result::from(task: iceberg::scan::FileScanTaskDeleteFile) -> Self impl core::fmt::Debug for iceberg::scan::FileScanTaskDeleteFile pub fn iceberg::scan::FileScanTaskDeleteFile::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTaskDeleteFile diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 905fa86d83..bd2807f785 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -35,9 +35,9 @@ use crate::io::FileIO; use crate::runtime::Runtime; use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile}; use crate::spec::{ - DataContentType, DataFileFormat, Datum, ListType, MapType, NestedField, NestedFieldRef, - PartnerAccessor, PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor, StructType, Type, - VariantType, visit_schema_with_partner, + DataContentType, Datum, ListType, MapType, NestedField, NestedFieldRef, PartnerAccessor, + PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor, StructType, Type, VariantType, + visit_schema_with_partner, }; use crate::{Error, ErrorKind, Result}; @@ -266,7 +266,7 @@ impl CachingDeleteFileLoader { DataContentType::PositionDeletes => { // A V3 deletion vector arrives as a PositionDeletes entry whose deletes live in // a Puffin blob, not in a positional-delete parquet file. - if task.file_format == DataFileFormat::Puffin { + if task.is_deletion_vector() { return Self::load_deletion_vector(task, basic_delete_file_loader).await; } @@ -330,79 +330,6 @@ impl CachingDeleteFileLoader { } } - /// Validates a deletion-vector task and returns what the read needs as typed values: - /// `(start, len, referenced data file path, expected cardinality)`. - /// - /// The spec requires `referenced_data_file`, `content_offset` and `content_size_in_bytes` on - /// a deletion vector, and a deletion vector is always built from a manifest entry, so it - /// always carries `record_count`. A missing one is a manifest-entry inconsistency rather - /// than an I/O failure. - /// - /// Equality and ordinary position deletes have no equivalent validation in this loader: a - /// malformed equality/position delete file fails loudly when the Parquet reader can't open - /// it. A deletion vector's coordinates instead drive a raw byte-range read with no format - /// to fail against, so a bad coordinate would otherwise decode silently into the wrong (or - /// no) deletes, per the same corrupted-blob concern Iceberg-Java validates in - /// `BitmapPositionDeleteIndex.deserializeBitmap`. - fn validate_deletion_vector_task( - task: &FileScanTaskDeleteFile, - ) -> Result<(u64, u64, String, u64)> { - let content_offset = task.content_offset.ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} is missing content_offset", - task.file_path - ), - ) - })?; - let content_size = task.content_size_in_bytes.ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} is missing content_size_in_bytes", - task.file_path - ), - ) - })?; - let data_file_path = task.referenced_data_file.clone().ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} is missing referenced_data_file", - task.file_path - ), - ) - })?; - let record_count = task.record_count.ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!("deletion vector {} is missing record_count", task.file_path), - ) - })?; - - let start = u64::try_from(content_offset).map_err(|_| { - Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} has negative content_offset {content_offset}", - task.file_path - ), - ) - })?; - let len = u64::try_from(content_size).map_err(|_| { - Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} has negative content_size_in_bytes {content_size}", - task.file_path - ), - ) - })?; - - Ok((start, len, data_file_path, record_count)) - } - /// Validates a decoded deletion vector's cardinality against the manifest entry's /// `record_count`, mirroring Iceberg-Java's `BitmapPositionDeleteIndex.deserializeBitmap`. fn validate_deletion_vector_cardinality( @@ -436,7 +363,8 @@ impl CachingDeleteFileLoader { task: &FileScanTaskDeleteFile, basic_delete_file_loader: BasicDeleteFileLoader, ) -> Result { - let (start, len, data_file_path, record_count) = Self::validate_deletion_vector_task(task)?; + let (start, len, data_file_path, record_count) = task.deletion_vector_coordinates()?; + let data_file_path = data_file_path.to_string(); let input_file = basic_delete_file_loader .file_io() @@ -875,7 +803,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, DataFileFormat, Schema}; use crate::test_utils::encode_dv_blob; #[tokio::test] @@ -1426,7 +1354,8 @@ mod tests { .with_file_type(DataContentType::PositionDeletes) .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) - .build(); + .build() + .unwrap(); let eq_del = FileScanTaskDeleteFile::builder() .with_file_path(eq_delete_path.clone()) @@ -1435,7 +1364,8 @@ mod tests { .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) .with_equality_ids(Some(vec![2, 3])) // Only use field IDs that exist in both schemas - .build(); + .build() + .unwrap(); let file_scan_task = FileScanTask::builder() .with_file_size_in_bytes(0) @@ -1585,6 +1515,7 @@ mod tests { .with_record_count(Some(record_count)) .with_key_metadata(key_metadata) .build() + .unwrap() } #[tokio::test] @@ -1731,89 +1662,6 @@ mod tests { assert!(err.message().contains("expected 2 from record_count")); } - // A well-formed deletion-vector task, for tests that then clear or corrupt one field. - fn valid_dv_task() -> FileScanTaskDeleteFile { - dv_task( - "deletes.puffin".to_string(), - 100, - "data.parquet".to_string(), - 4, - 40, - 2, - None, - ) - } - - #[test] - fn test_validate_deletion_vector_task_rejects_missing_content_offset() { - let mut task = valid_dv_task(); - task.content_offset = None; - - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); - assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!(err.message().contains("missing content_offset")); - } - - #[test] - fn test_validate_deletion_vector_task_rejects_missing_content_size() { - let mut task = valid_dv_task(); - task.content_size_in_bytes = None; - - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); - assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!(err.message().contains("missing content_size_in_bytes")); - } - - #[test] - fn test_validate_deletion_vector_task_rejects_missing_referenced_data_file() { - let mut task = valid_dv_task(); - task.referenced_data_file = None; - - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); - assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!(err.message().contains("missing referenced_data_file")); - } - - #[test] - fn test_validate_deletion_vector_task_rejects_missing_record_count() { - let mut task = valid_dv_task(); - task.record_count = None; - - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); - assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!(err.message().contains("missing record_count")); - } - - #[test] - fn test_validate_deletion_vector_task_rejects_negative_content_offset() { - let mut task = valid_dv_task(); - task.content_offset = Some(-1); - - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); - assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!(err.message().contains("negative content_offset")); - } - - #[test] - fn test_validate_deletion_vector_task_rejects_negative_content_size() { - let mut task = valid_dv_task(); - task.content_size_in_bytes = Some(-1); - - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); - assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!(err.message().contains("negative content_size_in_bytes")); - } - - #[test] - fn test_validate_deletion_vector_task_accepts_valid_coordinates() { - let (start, len, data_file_path, record_count) = - CachingDeleteFileLoader::validate_deletion_vector_task(&valid_dv_task()).unwrap(); - assert_eq!(start, 4); - assert_eq!(len, 40); - assert_eq!(data_file_path, "data.parquet"); - assert_eq!(record_count, 2); - } - #[test] fn test_validate_deletion_vector_cardinality_accepts_matching_count() { let mut dv = DeleteVector::default(); diff --git a/crates/iceberg/src/arrow/delete_filter.rs b/crates/iceberg/src/arrow/delete_filter.rs index cef81afda8..f676cb1dfa 100644 --- a/crates/iceberg/src/arrow/delete_filter.rs +++ b/crates/iceberg/src/arrow/delete_filter.rs @@ -443,7 +443,8 @@ pub(crate) mod tests { .with_file_type(DataContentType::PositionDeletes) .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) - .build(); + .build() + .unwrap(); let pos_del_2 = FileScanTaskDeleteFile::builder() .with_file_path(format!( @@ -461,7 +462,8 @@ pub(crate) mod tests { .with_file_type(DataContentType::PositionDeletes) .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) - .build(); + .build() + .unwrap(); let pos_del_3 = FileScanTaskDeleteFile::builder() .with_file_path(format!( @@ -479,7 +481,8 @@ pub(crate) mod tests { .with_file_type(DataContentType::PositionDeletes) .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) - .build(); + .build() + .unwrap(); let file_scan_tasks = vec![ FileScanTask::builder() @@ -556,7 +559,8 @@ pub(crate) mod tests { .with_file_type(DataContentType::EqualityDeletes) .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) - .build(), + .build() + .unwrap(), ]) .with_case_sensitive(true) .build() diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs index 5354c72043..a74fd313f9 100644 --- a/crates/iceberg/src/arrow/reader/pipeline.rs +++ b/crates/iceberg/src/arrow/reader/pipeline.rs @@ -2414,7 +2414,8 @@ mod tests { .with_file_size_in_bytes(std::fs::metadata(&del_path).unwrap().len()) .with_file_type(DataContentType::PositionDeletes) .with_partition_spec_id(0) - .build(); + .build() + .unwrap(); let task = row_id_task_with_options(data_path, Some(100), 0, 0, vec![delete]); let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build(); diff --git a/crates/iceberg/src/arrow/reader/positional_deletes.rs b/crates/iceberg/src/arrow/reader/positional_deletes.rs index 2915596da7..149202e499 100644 --- a/crates/iceberg/src/arrow/reader/positional_deletes.rs +++ b/crates/iceberg/src/arrow/reader/positional_deletes.rs @@ -452,7 +452,8 @@ mod tests { .with_file_type(DataContentType::PositionDeletes) .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) - .build(), + .build() + .unwrap(), ]) .with_case_sensitive(false) .build() @@ -672,7 +673,8 @@ mod tests { .with_file_type(DataContentType::PositionDeletes) .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) - .build(), + .build() + .unwrap(), ]) .with_case_sensitive(false) .build() @@ -886,7 +888,8 @@ mod tests { .with_file_type(DataContentType::PositionDeletes) .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) - .build(), + .build() + .unwrap(), ]) .with_case_sensitive(false) .build() @@ -1005,7 +1008,8 @@ mod tests { .with_content_offset(Some(content_offset)) .with_content_size_in_bytes(Some(content_size)) .with_record_count(Some(2)) - .build(), + .build() + .unwrap(), ]) .with_case_sensitive(false) .build(); @@ -1102,7 +1106,8 @@ mod tests { .with_content_offset(Some(0)) .with_content_size_in_bytes(Some(blob.len() as i64)) .with_record_count(Some(5)) - .build(), + .build() + .unwrap(), ]) .with_case_sensitive(false) .build(); diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 2e749cab1f..4014680999 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -56,10 +56,10 @@ struct PopulatedDeleteFileIndex { global_equality_deletes: Vec>, eq_deletes_by_partition: HashMap>>, pos_deletes_by_partition: HashMap>>, - pos_deletes_by_path: HashMap>>, + pos_deletes_by_referenced_data_file: HashMap>>, // V3 deletion vectors, keyed by the data file they apply to (referenced_data_file). At most // one exists per data file per snapshot, and when one applies it supersedes any position - // delete files for that data file, partition-scoped or path-scoped alike. + // delete files for that data file, whether indexed by partition or by referenced data file. dvs_by_referenced_data_file: HashMap>, } @@ -136,8 +136,8 @@ impl DeleteFileIndex { /// Gets all the delete files that apply to the specified data file. /// /// Fails if building the index found a spec violation, such as multiple deletion vectors - /// referencing the same data file, or if a matched deletion vector's sequence number - /// violates the spec relative to `seq_num`. + /// referencing the same data file, or if a matched deletion vector violates the spec + /// relative to `seq_num` or is missing a field the spec requires of it. pub(crate) async fn get_deletes_for_data_file( &self, data_file: &DataFile, @@ -196,7 +196,7 @@ impl PopulatedDeleteFileIndex { HashMap::default(); let mut pos_deletes_by_partition: HashMap>> = HashMap::default(); - let mut pos_deletes_by_path: HashMap>> = + let mut pos_deletes_by_referenced_data_file: HashMap>> = HashMap::default(); let mut dvs_by_referenced_data_file: HashMap> = HashMap::default(); @@ -214,10 +214,9 @@ impl PopulatedDeleteFileIndex { // A deletion vector is a position delete stored as a Puffin blob. The file // format is what distinguishes it from a position delete parquet file. if data_file.file_format() == DataFileFormat::Puffin { - // The spec requires referenced_data_file, content_offset and - // content_size_in_bytes on a deletion vector, so a missing one is a - // malformed manifest entry, not an ordinary position delete to fall back - // on. + // referenced_data_file is what the index keys a deletion vector by, so + // its absence is fatal here rather than deferred to the scan task's own + // validation of the remaining spec-required fields. let Some(path) = data_file.referenced_data_file() else { return Err(Error::new( ErrorKind::DataInvalid, @@ -228,18 +227,6 @@ impl PopulatedDeleteFileIndex { )); }; - if data_file.content_offset().is_none() - || data_file.content_size_in_bytes().is_none() - { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} is missing content_offset or content_size_in_bytes", - arc_ctx.manifest_entry.file_path() - ), - )); - } - if let Some(existing) = dvs_by_referenced_data_file.insert(path.clone(), arc_ctx) { @@ -257,7 +244,10 @@ impl PopulatedDeleteFileIndex { } if let Some(path) = referenced_data_file(data_file) { - pos_deletes_by_path.entry(path).or_default().push(arc_ctx); + pos_deletes_by_referenced_data_file + .entry(path) + .or_default() + .push(arc_ctx); } else { pos_deletes_by_partition .entry(partition.clone()) @@ -284,7 +274,7 @@ impl PopulatedDeleteFileIndex { global_equality_deletes, eq_deletes_by_partition, pos_deletes_by_partition, - pos_deletes_by_path, + pos_deletes_by_referenced_data_file, dvs_by_referenced_data_file, }) } @@ -295,7 +285,9 @@ impl PopulatedDeleteFileIndex { /// with the data file's: a data file's path is permanently tied to one partition, and the /// spec guarantees a DV is only ever written at or after the sequence number of the data /// file it applies to, so either violation means the delete manifest is inconsistent, not - /// that the DV simply doesn't apply. + /// that the DV simply doesn't apply. Also fails if a matched delete file cannot be turned + /// into a scan task, which for a deletion vector means its manifest entry is missing a + /// spec-required Puffin coordinate. fn get_deletes_for_data_file( &self, data_file: &DataFile, @@ -303,34 +295,34 @@ impl PopulatedDeleteFileIndex { ) -> Result> { let mut results = vec![]; - self.global_equality_deletes - .iter() + // filter that returns true if the provided delete file's sequence number is **greater than** `seq_num` + let global_eq_deletes = self.global_equality_deletes.iter().filter(|&delete| { + seq_num + .map(|seq_num| delete.manifest_entry.sequence_number() > Some(seq_num)) + .unwrap_or_else(|| true) + }); + for delete in global_eq_deletes { + results.push(delete.as_ref().try_into()?); + } + + if let Some(deletes) = self.eq_deletes_by_partition.get(data_file.partition()) { // filter that returns true if the provided delete file's sequence number is **greater than** `seq_num` - .filter(|&delete| { + let deletes = deletes.iter().filter(|&delete| { seq_num .map(|seq_num| delete.manifest_entry.sequence_number() > Some(seq_num)) .unwrap_or_else(|| true) - }) - .for_each(|delete| results.push(delete.as_ref().into())); - - if let Some(deletes) = self.eq_deletes_by_partition.get(data_file.partition()) { - deletes - .iter() - // filter that returns true if the provided delete file's sequence number is **greater than** `seq_num` - .filter(|&delete| { - seq_num - .map(|seq_num| delete.manifest_entry.sequence_number() > Some(seq_num)) - .unwrap_or_else(|| true) - && data_file.partition_spec_id == delete.partition_spec_id - }) - .for_each(|delete| results.push(delete.as_ref().into())); + && data_file.partition_spec_id == delete.partition_spec_id + }); + for delete in deletes { + results.push(delete.as_ref().try_into()?); + } } // A deletion vector supersedes all position delete files for its data file, per the spec: // "readers ignore any position delete files that would otherwise match it, because the DV // subsumes them". An exact path match on referenced_data_file is sufficient proof of - // applicability, the same as for pos_deletes_by_path below, so this is checked before - // (and instead of) pos_deletes_by_partition and pos_deletes_by_path. + // applicability, the same as for the position deletes below, so this is checked before + // (and instead of) either position delete map. if let Some(dv) = self.dvs_by_referenced_data_file.get(data_file.file_path()) { let dv_data_file = dv.manifest_entry.data_file(); // A file path belongs to exactly one partition for its lifetime, so an exact path @@ -365,36 +357,39 @@ impl PopulatedDeleteFileIndex { )); } } - results.push(dv.as_ref().into()); + results.push(dv.as_ref().try_into()?); return Ok(results); } if let Some(deletes) = self.pos_deletes_by_partition.get(data_file.partition()) { - deletes - .iter() - // filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num` - .filter(|&delete| { - seq_num - .map(|seq_num| delete.manifest_entry.sequence_number() >= Some(seq_num)) - .unwrap_or_else(|| true) - && data_file.partition_spec_id == delete.partition_spec_id - }) - .for_each(|delete| results.push(delete.as_ref().into())); + // filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num` + let deletes = deletes.iter().filter(|&delete| { + seq_num + .map(|seq_num| delete.manifest_entry.sequence_number() >= Some(seq_num)) + .unwrap_or_else(|| true) + && data_file.partition_spec_id == delete.partition_spec_id + }); + for delete in deletes { + results.push(delete.as_ref().try_into()?); + } } // Position deletes indexed by the exact path of the data file they reference. // An exact path match is sufficient proof that the delete applies, so no // partition spec id check is performed. - if let Some(deletes) = self.pos_deletes_by_path.get(data_file.file_path()) { - deletes - .iter() - // filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num` - .filter(|&delete| { - seq_num - .map(|seq_num| delete.manifest_entry.sequence_number() >= Some(seq_num)) - .unwrap_or(true) - }) - .for_each(|delete| results.push(delete.as_ref().into())); + if let Some(deletes) = self + .pos_deletes_by_referenced_data_file + .get(data_file.file_path()) + { + // filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num` + let deletes = deletes.iter().filter(|&delete| { + seq_num + .map(|seq_num| delete.manifest_entry.sequence_number() >= Some(seq_num)) + .unwrap_or(true) + }); + for delete in deletes { + results.push(delete.as_ref().try_into()?); + } } Ok(results) @@ -977,7 +972,7 @@ mod tests { partition_spec_id: 0, }; - let task: FileScanTaskDeleteFile = (&ctx).into(); + let task: FileScanTaskDeleteFile = (&ctx).try_into().unwrap(); assert_eq!(task.file_type, DataContentType::PositionDeletes); assert_eq!(task.content_offset, Some(4)); assert_eq!(task.content_size_in_bytes, Some(40)); @@ -1201,12 +1196,16 @@ mod tests { #[test] fn test_deletion_vector_missing_coordinates_is_rejected() { + let data_file = build_unpartitioned_data_file(); + + // Indexing only needs referenced_data_file, so a missing Puffin coordinate surfaces when + // the matched entry is converted into a scan task, not when the index is built. let malformed_dv = DataFileBuilder::default() .file_path("deletes.puffin".to_string()) .file_format(DataFileFormat::Puffin) .content(DataContentType::PositionDeletes) .record_count(1) - .referenced_data_file(Some("data.parquet".to_string())) + .referenced_data_file(Some(data_file.file_path().to_string())) .content_size_in_bytes(Some(40)) .partition(Struct::empty()) .partition_spec_id(0) @@ -1219,12 +1218,12 @@ mod tests { partition_spec_id: 0, }]; - let err = PopulatedDeleteFileIndex::new(contexts).unwrap_err(); + let index = PopulatedDeleteFileIndex::new(contexts).unwrap(); + let err = index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!( - err.message() - .contains("missing content_offset or content_size_in_bytes") - ); + assert!(err.message().contains("missing content_offset")); } #[test] diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index db70806473..61ba145ae5 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -2767,7 +2767,8 @@ pub mod tests { .with_content_size_in_bytes(Some(34)) .with_record_count(Some(5)) .with_key_metadata(Some(vec![4, 5, 6].into_boxed_slice())) - .build(), + .build() + .unwrap(), ]) .with_partition(Some(Struct::from_iter([Some(Literal::long(42))]))) .with_partition_spec(Some(partition_spec)) diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index f1a799fcc6..e4fedd7f90 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -273,8 +273,10 @@ pub(crate) struct DeleteFileContext { pub(crate) partition_spec_id: i32, } -impl From<&DeleteFileContext> for FileScanTaskDeleteFile { - fn from(ctx: &DeleteFileContext) -> Self { +impl TryFrom<&DeleteFileContext> for FileScanTaskDeleteFile { + type Error = Error; + + fn try_from(ctx: &DeleteFileContext) -> Result { FileScanTaskDeleteFile::builder() .with_file_path(ctx.manifest_entry.file_path().to_string()) .with_file_size_in_bytes(ctx.manifest_entry.file_size_in_bytes()) @@ -299,7 +301,10 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile { /// A task to scan part of file. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, TypedBuilder)] -#[builder(field_defaults(setter(prefix = "with_")))] +#[builder( + field_defaults(setter(prefix = "with_")), + build_method(into = Result) +)] pub struct FileScanTaskDeleteFile { /// The delete file path pub file_path: String, @@ -364,6 +369,96 @@ pub struct FileScanTaskDeleteFile { pub key_metadata: Option>, } +impl FileScanTaskDeleteFile { + /// Whether this delete file is a V3 deletion vector rather than a position delete file. + pub(crate) fn is_deletion_vector(&self) -> bool { + self.file_type == DataContentType::PositionDeletes + && self.file_format == DataFileFormat::Puffin + } + + /// The deletion vector blob's byte range within its Puffin file, the data file whose rows it + /// deletes, and the cardinality the manifest entry claims for it. + /// + /// The spec requires `referenced_data_file`, `content_offset` and `content_size_in_bytes` on + /// a deletion vector, and one is always built from a manifest entry, so `record_count` is + /// always present too. A missing or negative field is a malformed manifest entry rather than + /// an I/O failure: unlike a Parquet delete file, whose reader fails loudly on garbage, these + /// coordinates drive a raw byte-range read with no format to fail against, so a bad one + /// would otherwise decode silently into the wrong (or no) deletes, per the same + /// corrupted-blob concern Iceberg-Java validates in + /// `BitmapPositionDeleteIndex.deserializeBitmap`. + pub(crate) fn deletion_vector_coordinates(&self) -> Result<(u64, u64, &str, u64)> { + let content_offset = self.content_offset.ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} is missing content_offset", + self.file_path + ), + ) + })?; + let content_size = self.content_size_in_bytes.ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} is missing content_size_in_bytes", + self.file_path + ), + ) + })?; + let data_file_path = self.referenced_data_file.as_deref().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} is missing referenced_data_file", + self.file_path + ), + ) + })?; + let record_count = self.record_count.ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("deletion vector {} is missing record_count", self.file_path), + ) + })?; + + let start = u64::try_from(content_offset).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} has negative content_offset {content_offset}", + self.file_path + ), + ) + })?; + let len = u64::try_from(content_size).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} has negative content_size_in_bytes {content_size}", + self.file_path + ), + ) + })?; + + Ok((start, len, data_file_path, record_count)) + } + + fn validate(&self) -> Result<()> { + if self.is_deletion_vector() { + self.deletion_vector_coordinates()?; + } + Ok(()) + } +} + +impl From for Result { + fn from(task: FileScanTaskDeleteFile) -> Self { + task.validate()?; + Ok(task) + } +} + mod _serde { use std::sync::Arc; @@ -696,4 +791,116 @@ mod tests { assert_eq!(err.kind(), ErrorKind::DataInvalid); } + + fn build_deletion_vector( + referenced_data_file: Option, + content_offset: Option, + content_size_in_bytes: Option, + record_count: Option, + ) -> Result { + FileScanTaskDeleteFile::builder() + .with_file_path("deletes.puffin".to_string()) + .with_file_size_in_bytes(100) + .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Puffin) + .with_partition_spec_id(0) + .with_referenced_data_file(referenced_data_file) + .with_content_offset(content_offset) + .with_content_size_in_bytes(content_size_in_bytes) + .with_record_count(record_count) + .build() + } + + // A well-formed deletion vector, for tests that then clear or corrupt one field. + fn valid_deletion_vector() -> Result { + build_deletion_vector(Some("data.parquet".to_string()), Some(4), Some(40), Some(2)) + } + + #[test] + fn test_delete_file_builder_accepts_valid_deletion_vector() { + let task = valid_deletion_vector().unwrap(); + + assert!(task.is_deletion_vector()); + let (start, len, data_file_path, record_count) = + task.deletion_vector_coordinates().unwrap(); + assert_eq!(start, 4); + assert_eq!(len, 40); + assert_eq!(data_file_path, "data.parquet"); + assert_eq!(record_count, 2); + } + + #[test] + fn test_delete_file_builder_rejects_deletion_vector_missing_content_offset() { + let err = build_deletion_vector(Some("data.parquet".to_string()), None, Some(40), Some(2)) + .unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("missing content_offset")); + } + + #[test] + fn test_delete_file_builder_rejects_deletion_vector_missing_content_size() { + let err = build_deletion_vector(Some("data.parquet".to_string()), Some(4), None, Some(2)) + .unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("missing content_size_in_bytes")); + } + + #[test] + fn test_delete_file_builder_rejects_deletion_vector_missing_referenced_data_file() { + let err = build_deletion_vector(None, Some(4), Some(40), Some(2)).unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("missing referenced_data_file")); + } + + #[test] + fn test_delete_file_builder_rejects_deletion_vector_missing_record_count() { + let err = build_deletion_vector(Some("data.parquet".to_string()), Some(4), Some(40), None) + .unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("missing record_count")); + } + + #[test] + fn test_delete_file_builder_rejects_deletion_vector_negative_content_offset() { + let err = build_deletion_vector( + Some("data.parquet".to_string()), + Some(-1), + Some(40), + Some(2), + ) + .unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("negative content_offset")); + } + + #[test] + fn test_delete_file_builder_rejects_deletion_vector_negative_content_size() { + let err = + build_deletion_vector(Some("data.parquet".to_string()), Some(4), Some(-1), Some(2)) + .unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("negative content_size_in_bytes")); + } + + // A position delete file carries none of the deletion vector fields, so validation must not + // demand them of it. + #[test] + fn test_delete_file_builder_accepts_position_delete_without_deletion_vector_fields() { + let task = FileScanTaskDeleteFile::builder() + .with_file_path("pos-deletes.parquet".to_string()) + .with_file_size_in_bytes(100) + .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) + .with_partition_spec_id(0) + .build() + .unwrap(); + + assert!(!task.is_deletion_vector()); + } } From b738363f93672bf711df7de67b96519185a48245 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Wed, 9 Sep 2026 10:10:32 -0400 Subject: [PATCH 2/2] fix comment --- crates/iceberg/src/delete_file_index.rs | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 4014680999..9dedac8c23 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -318,11 +318,12 @@ impl PopulatedDeleteFileIndex { } } - // A deletion vector supersedes all position delete files for its data file, per the spec: - // "readers ignore any position delete files that would otherwise match it, because the DV - // subsumes them". An exact path match on referenced_data_file is sufficient proof of - // applicability, the same as for the position deletes below, so this is checked before - // (and instead of) either position delete map. + // A deletion vector supersedes all position delete files for its data file: a DV must + // replace every position delete file previously written for it, so that "readers can + // safely ignore matching position delete files" (spec, Deletion Vectors). An exact path + // match on referenced_data_file is sufficient proof of applicability, the same as for the + // position deletes below, so this is checked before (and instead of) either position + // delete map. if let Some(dv) = self.dvs_by_referenced_data_file.get(data_file.file_path()) { let dv_data_file = dv.manifest_entry.data_file(); // A file path belongs to exactly one partition for its lifetime, so an exact path