From faf92455882a9d2335f9e694023b4e96b2662693 Mon Sep 17 00:00:00 2001 From: linhongyu510 Date: Fri, 4 Sep 2026 01:53:17 +0800 Subject: [PATCH 1/9] docs: design delete task validation --- ...9-04-file-scan-delete-validation-design.md | 78 +++++++++++++++++++ 1 file changed, 78 insertions(+) create mode 100644 docs/superpowers/specs/2026-09-04-file-scan-delete-validation-design.md diff --git a/docs/superpowers/specs/2026-09-04-file-scan-delete-validation-design.md b/docs/superpowers/specs/2026-09-04-file-scan-delete-validation-design.md new file mode 100644 index 0000000000..e74cfe6b7e --- /dev/null +++ b/docs/superpowers/specs/2026-09-04-file-scan-delete-validation-design.md @@ -0,0 +1,78 @@ +# FileScanTaskDeleteFile Validation Design + +## Context + +Apache Iceberg Rust issue #3135 asks for deletion-vector validation to happen when a +`FileScanTaskDeleteFile` is built. The current implementation validates required +deletion-vector metadata in the delete-file index and caching loader, after an invalid +task can already exist and cross planning boundaries. + +## Goals + +- Make every builder-created deletion-vector task structurally valid. +- Preserve the behavior of equality-delete and ordinary position-delete tasks. +- Keep decoded bitmap cardinality validation in the loader because it depends on I/O. +- Return `DataInvalid` errors with the delete-file path and invalid field. + +## Builder Contract + +`FileScanTaskDeleteFile` will use the same typed-builder conversion pattern as +`FileScanTask`: `build()` returns `Result`, and conversion calls +a private `validate()` method. + +A task is a deletion vector exactly when: + +```text +file_type == PositionDeletes && file_format == Puffin +``` + +Deletion-vector tasks must contain: + +- `referenced_data_file` +- `content_offset` +- `content_size_in_bytes` +- `record_count` + +`content_offset` and `content_size_in_bytes` must be non-negative so they can safely +be converted to the unsigned values required by range reads. Non-deletion-vector tasks +will not acquire new restrictions. + +## Call-Site Migration + +Builder call sites will propagate `Result` where possible and use `expect` or `unwrap` +only in tests that intentionally construct valid fixtures. Direct struct literals will +remain unchanged unless making fields private is required by the builder invariant; field +visibility changes are outside this issue. + +The loader will stop repeating checks that the builder now guarantees. It will continue +to validate decoded bitmap cardinality against `record_count`, because that check can only +happen after reading the Puffin blob. + +The delete-file index will stop repeating required-field checks after task construction +becomes the invariant boundary. Duplicate deletion vectors for one data file remain an +index-level validation because they involve multiple tasks. + +## Error Handling + +All builder validation failures use `ErrorKind::DataInvalid`. Messages identify the +deletion-vector path and the missing or negative field. Validation runs before any I/O, +so malformed manifest metadata fails during scan planning rather than during execution. + +## Testing + +TDD coverage will first demonstrate failures on current `main`, then verify: + +- a complete deletion-vector task builds successfully; +- each required field is rejected when absent; +- negative offset and size are rejected; +- ordinary Parquet position-delete tasks remain valid without DV metadata; +- equality-delete tasks remain valid without DV metadata. + +Verification will include the focused task tests, the `iceberg` library test suite, +`cargo fmt --all -- --check`, targeted Clippy with warnings denied, and +`git diff --check`. + +## Scope + +This change does not alter Puffin decoding, bitmap cardinality rules, duplicate-DV +detection, transaction semantics, or public deletion-vector formats. From 897c7b036ee2b4c1109e4e3ca119921838387898 Mon Sep 17 00:00:00 2001 From: linhongyu510 Date: Fri, 4 Sep 2026 10:09:10 +0800 Subject: [PATCH 2/9] fix(scan): validate delete task construction --- .../src/arrow/caching_delete_file_loader.rs | 7 +- crates/iceberg/src/arrow/delete_filter.rs | 12 +- crates/iceberg/src/arrow/reader/pipeline.rs | 4 +- .../src/arrow/reader/positional_deletes.rs | 21 +- crates/iceberg/src/scan/task.rs | 244 +++++++++++++++++- 5 files changed, 273 insertions(+), 15 deletions(-) diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 905fa86d83..e7d0e9beab 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -1426,7 +1426,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 +1436,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 +1587,7 @@ mod tests { .with_record_count(Some(record_count)) .with_key_metadata(key_metadata) .build() + .expect("deletion vector task should be valid") } #[tokio::test] 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 894e51d76d..14b93d294e 100644 --- a/crates/iceberg/src/arrow/reader/pipeline.rs +++ b/crates/iceberg/src/arrow/reader/pipeline.rs @@ -2412,8 +2412,10 @@ mod tests { .with_file_path(del_path.clone()) .with_file_size_in_bytes(std::fs::metadata(&del_path).unwrap().len()) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) .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 f0c7bf7f95..5dcba26d60 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,10 +1008,12 @@ 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(); + .build() + .unwrap(); let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; let result = reader @@ -1102,10 +1107,12 @@ 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(); + .build() + .unwrap(); let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; let result: Result, _> = diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 3aff0f4ae6..9b02bd3c8c 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -333,12 +333,16 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile { .map(Box::from), ) .build() + .expect("delete file context should build a valid 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, @@ -403,6 +407,87 @@ pub struct FileScanTaskDeleteFile { pub key_metadata: Option>, } +impl FileScanTaskDeleteFile { + /// Returns whether this delete file is a V3 deletion vector stored in Puffin. + pub fn is_deletion_vector(&self) -> bool { + self.file_type == DataContentType::PositionDeletes + && self.file_format == DataFileFormat::Puffin + } + + fn validate(&self) -> Result<()> { + if !self.is_deletion_vector() { + return Ok(()); + } + + if self.referenced_data_file.is_none() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} is missing referenced_data_file", + self.file_path + ), + )); + } + + match self.content_offset { + None => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("deletion vector {} is missing content_offset", self.file_path), + )); + } + Some(offset) if offset < 0 => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} has negative content_offset {}", + self.file_path, offset + ), + )); + } + Some(_) => {} + } + + match self.content_size_in_bytes { + None => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} is missing content_size_in_bytes", + self.file_path + ), + )); + } + Some(size) if size < 0 => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} has negative content_size_in_bytes {}", + self.file_path, size + ), + )); + } + Some(_) => {} + } + + if self.record_count.is_none() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("deletion vector {} is missing record_count", self.file_path), + )); + } + + Ok(()) + } +} + +impl From for Result { + fn from(task: FileScanTaskDeleteFile) -> Self { + task.validate()?; + Ok(task) + } +} + #[cfg(test)] mod tests { use super::*; @@ -452,6 +537,48 @@ mod tests { (schema, partition_spec) } + fn build_delete_file_task( + file_type: DataContentType, + file_format: DataFileFormat, + ) -> Result { + FileScanTaskDeleteFile::builder() + .with_file_path("delete-file".to_string()) + .with_file_size_in_bytes(100) + .with_file_type(file_type) + .with_file_format(file_format) + .with_partition_spec_id(0) + .build() + } + + fn build_deletion_vector_task() -> Result { + FileScanTaskDeleteFile::builder() + .with_file_path("dv.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(Some("data.parquet".to_string())) + .with_content_offset(Some(7)) + .with_content_size_in_bytes(Some(11)) + .with_record_count(Some(3)) + .build() + } + + fn assert_delete_file_builder_error( + result: Result, + expected_message: &str, + ) { + match result { + Ok(task) => panic!( + "expected delete file builder to fail with `{expected_message}`, but got Ok({task:?})" + ), + Err(err) => { + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert_eq!(err.message(), expected_message); + } + } + } + #[test] fn test_file_scan_task_builder_rejects_non_empty_partition_without_spec() { // Regression test for https://github.com/apache/iceberg-rust/issues/3130. @@ -564,4 +691,119 @@ mod tests { assert_eq!(err.kind(), ErrorKind::DataInvalid); } + + #[test] + fn test_delete_file_builder_accepts_valid_deletion_vector() { + build_deletion_vector_task().unwrap(); + } + + #[test] + fn test_delete_file_builder_rejects_dv_missing_referenced_data_file() { + assert_delete_file_builder_error( + FileScanTaskDeleteFile::builder() + .with_file_path("dv.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_content_offset(Some(7)) + .with_content_size_in_bytes(Some(11)) + .with_record_count(Some(3)) + .build(), + "deletion vector dv.puffin is missing referenced_data_file", + ); + } + + #[test] + fn test_delete_file_builder_rejects_dv_missing_content_offset() { + assert_delete_file_builder_error( + FileScanTaskDeleteFile::builder() + .with_file_path("dv.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(Some("data.parquet".to_string())) + .with_content_size_in_bytes(Some(11)) + .with_record_count(Some(3)) + .build(), + "deletion vector dv.puffin is missing content_offset", + ); + } + + #[test] + fn test_delete_file_builder_rejects_dv_missing_content_size() { + assert_delete_file_builder_error( + FileScanTaskDeleteFile::builder() + .with_file_path("dv.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(Some("data.parquet".to_string())) + .with_content_offset(Some(7)) + .with_record_count(Some(3)) + .build(), + "deletion vector dv.puffin is missing content_size_in_bytes", + ); + } + + #[test] + fn test_delete_file_builder_rejects_dv_missing_record_count() { + assert_delete_file_builder_error( + FileScanTaskDeleteFile::builder() + .with_file_path("dv.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(Some("data.parquet".to_string())) + .with_content_offset(Some(7)) + .with_content_size_in_bytes(Some(11)) + .build(), + "deletion vector dv.puffin is missing record_count", + ); + } + + #[test] + fn test_delete_file_builder_rejects_negative_dv_offset() { + assert_delete_file_builder_error( + FileScanTaskDeleteFile::builder() + .with_file_path("dv.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(Some("data.parquet".to_string())) + .with_content_offset(Some(-1)) + .with_content_size_in_bytes(Some(11)) + .with_record_count(Some(3)) + .build(), + "deletion vector dv.puffin has negative content_offset -1", + ); + } + + #[test] + fn test_delete_file_builder_rejects_negative_dv_size() { + assert_delete_file_builder_error( + FileScanTaskDeleteFile::builder() + .with_file_path("dv.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(Some("data.parquet".to_string())) + .with_content_offset(Some(7)) + .with_content_size_in_bytes(Some(-1)) + .with_record_count(Some(3)) + .build(), + "deletion vector dv.puffin has negative content_size_in_bytes -1", + ); + } + + #[test] + fn test_delete_file_builder_accepts_non_dv_delete_without_dv_fields() { + build_delete_file_task(DataContentType::PositionDeletes, DataFileFormat::Parquet).unwrap(); + build_delete_file_task(DataContentType::EqualityDeletes, DataFileFormat::Parquet).unwrap(); + } } From e08db93834bbbd8805a57d76227f427aaa1704ee Mon Sep 17 00:00:00 2001 From: linhongyu510 Date: Fri, 4 Sep 2026 10:13:17 +0800 Subject: [PATCH 3/9] refactor(scan): use validated delete task builder --- .../iceberg/src/arrow/delete_file_loader.rs | 45 ++++++++----------- crates/iceberg/src/arrow/reader/row_filter.rs | 21 ++++----- 2 files changed, 27 insertions(+), 39 deletions(-) diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index 5852cf1acd..f994bbe7a8 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -230,19 +230,15 @@ mod tests { .unwrap(), ); - let task = FileScanTaskDeleteFile { - file_path: del_path.clone(), - file_size_in_bytes: std::fs::metadata(&del_path).unwrap().len(), - file_type: DataContentType::PositionDeletes, - file_format: DataFileFormat::Parquet, - partition_spec_id: 0, - equality_ids: None, - key_metadata: Some(Box::from(key_metadata.as_ref())), - referenced_data_file: None, - content_offset: None, - content_size_in_bytes: None, - record_count: None, - }; + let task = FileScanTaskDeleteFile::builder() + .with_file_path(del_path.clone()) + .with_file_size_in_bytes(std::fs::metadata(&del_path).unwrap().len()) + .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) + .with_partition_spec_id(0) + .with_key_metadata(Some(Box::from(key_metadata.as_ref()))) + .build() + .unwrap(); let scan_metrics = ScanMetrics::new(); let delete_file_loader = BasicDeleteFileLoader::new(file_io, scan_metrics); @@ -309,19 +305,16 @@ mod tests { .unwrap(), ); - let task = FileScanTaskDeleteFile { - file_path: del_path.clone(), - file_size_in_bytes: std::fs::metadata(&del_path).unwrap().len(), - file_type: DataContentType::EqualityDeletes, - file_format: DataFileFormat::Parquet, - partition_spec_id: 0, - equality_ids: Some(vec![1]), - key_metadata: Some(Box::from(key_metadata.as_ref())), - referenced_data_file: None, - content_offset: None, - content_size_in_bytes: None, - record_count: None, - }; + let task = FileScanTaskDeleteFile::builder() + .with_file_path(del_path.clone()) + .with_file_size_in_bytes(std::fs::metadata(&del_path).unwrap().len()) + .with_file_type(DataContentType::EqualityDeletes) + .with_file_format(DataFileFormat::Parquet) + .with_partition_spec_id(0) + .with_equality_ids(Some(vec![1])) + .with_key_metadata(Some(Box::from(key_metadata.as_ref()))) + .build() + .unwrap(); let scan_metrics = ScanMetrics::new(); let delete_file_loader = BasicDeleteFileLoader::new(file_io, scan_metrics); diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index 389231c461..fcef7f65e2 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1234,19 +1234,14 @@ mod tests { .with_schema(iceberg_schema.clone()) .with_project_field_ids(vec![1, 2]) .with_predicate(Some(predicate_sub2)) - .with_deletes(vec![FileScanTaskDeleteFile { - file_path: pos_del_path.clone(), - file_type: DataContentType::PositionDeletes, - file_format: DataFileFormat::Parquet, - partition_spec_id: 0, - equality_ids: None, - file_size_in_bytes: std::fs::metadata(&pos_del_path).unwrap().len(), - referenced_data_file: None, - content_offset: None, - content_size_in_bytes: None, - record_count: None, - key_metadata: None, - }]) + .with_deletes(vec![FileScanTaskDeleteFile::builder() + .with_file_path(pos_del_path.clone()) + .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) + .with_partition_spec_id(0) + .with_file_size_in_bytes(std::fs::metadata(&pos_del_path).unwrap().len()) + .build() + .unwrap()]) .with_case_sensitive(false) .build() .unwrap(); From 3ab4e683cce94b1692fd83b6d61a8efa4ccbebb0 Mon Sep 17 00:00:00 2001 From: linhongyu510 Date: Fri, 4 Sep 2026 10:17:43 +0800 Subject: [PATCH 4/9] refactor(scan): centralize deletion vector validation --- .../src/arrow/caching_delete_file_loader.rs | 96 ++++--------------- crates/iceberg/src/delete_file_index.rs | 91 +++++------------- 2 files changed, 44 insertions(+), 143 deletions(-) diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index e7d0e9beab..46a818d75d 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -333,10 +333,9 @@ 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. + /// The builder guarantees that a deletion vector already carries + /// `referenced_data_file`, `content_offset`, `content_size_in_bytes`, and `record_count`. + /// This helper keeps only the conversions still needed for the read. /// /// 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 @@ -347,39 +346,19 @@ impl CachingDeleteFileLoader { 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 content_offset = task + .content_offset + .expect("validated deletion vector must have content_offset"); + let content_size = task + .content_size_in_bytes + .expect("validated deletion vector must have content_size_in_bytes"); + let data_file_path = task + .referenced_data_file + .clone() + .expect("validated deletion vector must have referenced_data_file"); + let record_count = task + .record_count + .expect("validated deletion vector must have record_count"); let start = u64::try_from(content_offset).map_err(|_| { Error::new( @@ -1734,7 +1713,8 @@ 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. + // A well-formed deletion-vector task, for tests that then corrupt one field after builder + // validation. fn valid_dv_task() -> FileScanTaskDeleteFile { dv_task( "deletes.puffin".to_string(), @@ -1747,46 +1727,6 @@ mod tests { ) } - #[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(); diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 2e749cab1f..d3cd205dec 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -214,31 +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. - let Some(path) = data_file.referenced_data_file() else { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} is missing referenced_data_file", - arc_ctx.manifest_entry.file_path() - ), - )); - }; - - 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() - ), - )); - } + let path = data_file + .referenced_data_file() + .expect("validated deletion vector must have referenced_data_file"); if let Some(existing) = dvs_by_referenced_data_file.insert(path.clone(), arc_ctx) @@ -1175,56 +1153,39 @@ mod tests { } #[test] - fn test_deletion_vector_missing_referenced_data_file_is_rejected() { - let malformed_dv = DataFileBuilder::default() - .file_path("deletes.puffin".to_string()) - .file_format(DataFileFormat::Puffin) - .content(DataContentType::PositionDeletes) - .record_count(1) - .content_offset(Some(4)) - .content_size_in_bytes(Some(40)) - .partition(Struct::empty()) - .partition_spec_id(0) - .file_size_in_bytes(60) + fn test_deletion_vector_builder_rejects_missing_referenced_data_file() { + let err = FileScanTaskDeleteFile::builder() + .with_file_path("deletes.puffin".to_string()) + .with_file_size_in_bytes(60) + .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Puffin) + .with_partition_spec_id(0) + .with_content_offset(Some(4)) + .with_content_size_in_bytes(Some(40)) + .with_record_count(Some(1)) .build() - .unwrap(); - - let contexts = vec![DeleteFileContext { - manifest_entry: build_added_manifest_entry(5, &malformed_dv).into(), - partition_spec_id: 0, - }]; + .unwrap_err(); - let err = PopulatedDeleteFileIndex::new(contexts).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); assert!(err.message().contains("missing referenced_data_file")); } #[test] - fn test_deletion_vector_missing_coordinates_is_rejected() { - 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())) - .content_size_in_bytes(Some(40)) - .partition(Struct::empty()) - .partition_spec_id(0) - .file_size_in_bytes(60) + fn test_deletion_vector_builder_rejects_missing_coordinates() { + let err = FileScanTaskDeleteFile::builder() + .with_file_path("deletes.puffin".to_string()) + .with_file_size_in_bytes(60) + .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Puffin) + .with_partition_spec_id(0) + .with_referenced_data_file(Some("data.parquet".to_string())) + .with_content_size_in_bytes(Some(40)) + .with_record_count(Some(1)) .build() - .unwrap(); - - let contexts = vec![DeleteFileContext { - manifest_entry: build_added_manifest_entry(5, &malformed_dv).into(), - partition_spec_id: 0, - }]; + .unwrap_err(); - let err = PopulatedDeleteFileIndex::new(contexts).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] From c5c986fb260521c81293a18b0835756fe6c64b7f Mon Sep 17 00:00:00 2001 From: linhongyu510 Date: Fri, 4 Sep 2026 10:19:40 +0800 Subject: [PATCH 5/9] style(scan): format validated delete task changes --- crates/iceberg/src/arrow/reader/row_filter.rs | 18 ++++++++++-------- crates/iceberg/src/scan/task.rs | 5 ++++- 2 files changed, 14 insertions(+), 9 deletions(-) diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index fcef7f65e2..26d42c801b 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1234,14 +1234,16 @@ mod tests { .with_schema(iceberg_schema.clone()) .with_project_field_ids(vec![1, 2]) .with_predicate(Some(predicate_sub2)) - .with_deletes(vec![FileScanTaskDeleteFile::builder() - .with_file_path(pos_del_path.clone()) - .with_file_type(DataContentType::PositionDeletes) - .with_file_format(DataFileFormat::Parquet) - .with_partition_spec_id(0) - .with_file_size_in_bytes(std::fs::metadata(&pos_del_path).unwrap().len()) - .build() - .unwrap()]) + .with_deletes(vec![ + FileScanTaskDeleteFile::builder() + .with_file_path(pos_del_path.clone()) + .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) + .with_partition_spec_id(0) + .with_file_size_in_bytes(std::fs::metadata(&pos_del_path).unwrap().len()) + .build() + .unwrap(), + ]) .with_case_sensitive(false) .build() .unwrap(); diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 9b02bd3c8c..323867010b 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -433,7 +433,10 @@ impl FileScanTaskDeleteFile { None => { return Err(Error::new( ErrorKind::DataInvalid, - format!("deletion vector {} is missing content_offset", self.file_path), + format!( + "deletion vector {} is missing content_offset", + self.file_path + ), )); } Some(offset) if offset < 0 => { From 8ff9d122a58269f8158f60e9854f1452933230fe Mon Sep 17 00:00:00 2001 From: linhongyu510 Date: Fri, 4 Sep 2026 12:27:15 +0800 Subject: [PATCH 6/9] fix(scan): satisfy public API and license checks --- crates/iceberg/public-api.txt | 2 + crates/iceberg/src/scan/task.rs | 3 +- ...9-04-file-scan-delete-validation-design.md | 78 ------------------- 3 files changed, 3 insertions(+), 80 deletions(-) delete mode 100644 docs/superpowers/specs/2026-09-04-file-scan-delete-validation-design.md diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 57edfd6421..71743d2840 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/scan/task.rs b/crates/iceberg/src/scan/task.rs index 323867010b..83a29f7e05 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -408,8 +408,7 @@ pub struct FileScanTaskDeleteFile { } impl FileScanTaskDeleteFile { - /// Returns whether this delete file is a V3 deletion vector stored in Puffin. - pub fn is_deletion_vector(&self) -> bool { + fn is_deletion_vector(&self) -> bool { self.file_type == DataContentType::PositionDeletes && self.file_format == DataFileFormat::Puffin } diff --git a/docs/superpowers/specs/2026-09-04-file-scan-delete-validation-design.md b/docs/superpowers/specs/2026-09-04-file-scan-delete-validation-design.md deleted file mode 100644 index e74cfe6b7e..0000000000 --- a/docs/superpowers/specs/2026-09-04-file-scan-delete-validation-design.md +++ /dev/null @@ -1,78 +0,0 @@ -# FileScanTaskDeleteFile Validation Design - -## Context - -Apache Iceberg Rust issue #3135 asks for deletion-vector validation to happen when a -`FileScanTaskDeleteFile` is built. The current implementation validates required -deletion-vector metadata in the delete-file index and caching loader, after an invalid -task can already exist and cross planning boundaries. - -## Goals - -- Make every builder-created deletion-vector task structurally valid. -- Preserve the behavior of equality-delete and ordinary position-delete tasks. -- Keep decoded bitmap cardinality validation in the loader because it depends on I/O. -- Return `DataInvalid` errors with the delete-file path and invalid field. - -## Builder Contract - -`FileScanTaskDeleteFile` will use the same typed-builder conversion pattern as -`FileScanTask`: `build()` returns `Result`, and conversion calls -a private `validate()` method. - -A task is a deletion vector exactly when: - -```text -file_type == PositionDeletes && file_format == Puffin -``` - -Deletion-vector tasks must contain: - -- `referenced_data_file` -- `content_offset` -- `content_size_in_bytes` -- `record_count` - -`content_offset` and `content_size_in_bytes` must be non-negative so they can safely -be converted to the unsigned values required by range reads. Non-deletion-vector tasks -will not acquire new restrictions. - -## Call-Site Migration - -Builder call sites will propagate `Result` where possible and use `expect` or `unwrap` -only in tests that intentionally construct valid fixtures. Direct struct literals will -remain unchanged unless making fields private is required by the builder invariant; field -visibility changes are outside this issue. - -The loader will stop repeating checks that the builder now guarantees. It will continue -to validate decoded bitmap cardinality against `record_count`, because that check can only -happen after reading the Puffin blob. - -The delete-file index will stop repeating required-field checks after task construction -becomes the invariant boundary. Duplicate deletion vectors for one data file remain an -index-level validation because they involve multiple tasks. - -## Error Handling - -All builder validation failures use `ErrorKind::DataInvalid`. Messages identify the -deletion-vector path and the missing or negative field. Validation runs before any I/O, -so malformed manifest metadata fails during scan planning rather than during execution. - -## Testing - -TDD coverage will first demonstrate failures on current `main`, then verify: - -- a complete deletion-vector task builds successfully; -- each required field is rejected when absent; -- negative offset and size are rejected; -- ordinary Parquet position-delete tasks remain valid without DV metadata; -- equality-delete tasks remain valid without DV metadata. - -Verification will include the focused task tests, the `iceberg` library test suite, -`cargo fmt --all -- --check`, targeted Clippy with warnings denied, and -`git diff --check`. - -## Scope - -This change does not alter Puffin decoding, bitmap cardinality rules, duplicate-DV -detection, transaction semantics, or public deletion-vector formats. From e499648328e7b7003115d9b54e6b4ae22d20c30a Mon Sep 17 00:00:00 2001 From: linhongyu510 Date: Fri, 4 Sep 2026 12:33:52 +0800 Subject: [PATCH 7/9] test(scan): adapt merged reader fixtures --- crates/iceberg/src/arrow/reader/pipeline.rs | 1 - crates/iceberg/src/arrow/reader/positional_deletes.rs | 4 ++-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs index 354b78476c..14b93d294e 100644 --- a/crates/iceberg/src/arrow/reader/pipeline.rs +++ b/crates/iceberg/src/arrow/reader/pipeline.rs @@ -2410,7 +2410,6 @@ mod tests { let delete = FileScanTaskDeleteFile::builder() .with_file_path(del_path.clone()) - .with_file_format(DataFileFormat::Parquet) .with_file_size_in_bytes(std::fs::metadata(&del_path).unwrap().len()) .with_file_type(DataContentType::PositionDeletes) .with_file_format(DataFileFormat::Parquet) diff --git a/crates/iceberg/src/arrow/reader/positional_deletes.rs b/crates/iceberg/src/arrow/reader/positional_deletes.rs index 96ddfb4b61..5dcba26d60 100644 --- a/crates/iceberg/src/arrow/reader/positional_deletes.rs +++ b/crates/iceberg/src/arrow/reader/positional_deletes.rs @@ -1015,7 +1015,7 @@ mod tests { .build() .unwrap(); - let tasks = Box::pin(futures::stream::iter(vec![task])) as FileScanTaskStream; + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; let result = reader .read(tasks) .unwrap() @@ -1114,7 +1114,7 @@ mod tests { .build() .unwrap(); - let tasks = Box::pin(futures::stream::iter(vec![task])) as FileScanTaskStream; + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; let result: Result, _> = reader.read(tasks).unwrap().stream().try_collect().await; From 0228cd4f99cf3c4fcb508a7b3961b150366ca4be Mon Sep 17 00:00:00 2001 From: linhongyu510 Date: Sat, 5 Sep 2026 06:28:37 +0800 Subject: [PATCH 8/9] refactor(scan): encapsulate delete file metadata --- crates/iceberg/public-api.txt | 23 ++-- .../src/arrow/caching_delete_file_loader.rs | 56 +++++---- .../iceberg/src/arrow/delete_file_loader.rs | 10 +- crates/iceberg/src/arrow/delete_filter.rs | 6 +- crates/iceberg/src/delete_file_index.rs | 45 +++---- crates/iceberg/src/scan/mod.rs | 2 +- crates/iceberg/src/scan/task.rs | 117 ++++++++++++++---- 7 files changed, 165 insertions(+), 94 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 71743d2840..3f8c6a3ccc 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1310,17 +1310,18 @@ pub fn iceberg::scan::FileScanTask::serialize<__S>(&self, __serializer: __S) -> impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTask pub fn iceberg::scan::FileScanTask::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::scan::FileScanTaskDeleteFile -pub iceberg::scan::FileScanTaskDeleteFile::content_offset: core::option::Option -pub iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes: core::option::Option -pub iceberg::scan::FileScanTaskDeleteFile::equality_ids: core::option::Option> -pub iceberg::scan::FileScanTaskDeleteFile::file_format: iceberg::spec::DataFileFormat -pub iceberg::scan::FileScanTaskDeleteFile::file_path: alloc::string::String -pub iceberg::scan::FileScanTaskDeleteFile::file_size_in_bytes: u64 -pub iceberg::scan::FileScanTaskDeleteFile::file_type: iceberg::spec::DataContentType -pub iceberg::scan::FileScanTaskDeleteFile::key_metadata: core::option::Option> -pub iceberg::scan::FileScanTaskDeleteFile::partition_spec_id: i32 -pub iceberg::scan::FileScanTaskDeleteFile::record_count: core::option::Option -pub iceberg::scan::FileScanTaskDeleteFile::referenced_data_file: core::option::Option +impl iceberg::scan::FileScanTaskDeleteFile +pub fn iceberg::scan::FileScanTaskDeleteFile::content_offset(&self) -> core::option::Option +pub fn iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes(&self) -> core::option::Option +pub fn iceberg::scan::FileScanTaskDeleteFile::equality_ids(&self) -> core::option::Option<&[i32]> +pub fn iceberg::scan::FileScanTaskDeleteFile::file_format(&self) -> iceberg::spec::DataFileFormat +pub fn iceberg::scan::FileScanTaskDeleteFile::file_path(&self) -> &str +pub fn iceberg::scan::FileScanTaskDeleteFile::file_size_in_bytes(&self) -> u64 +pub fn iceberg::scan::FileScanTaskDeleteFile::file_type(&self) -> iceberg::spec::DataContentType +pub fn iceberg::scan::FileScanTaskDeleteFile::key_metadata(&self) -> core::option::Option<&[u8]> +pub fn iceberg::scan::FileScanTaskDeleteFile::partition_spec_id(&self) -> i32 +pub fn iceberg::scan::FileScanTaskDeleteFile::record_count(&self) -> core::option::Option +pub fn iceberg::scan::FileScanTaskDeleteFile::referenced_data_file(&self) -> core::option::Option<&str> 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 diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 46a818d75d..f836d7d5ed 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -262,15 +262,15 @@ impl CachingDeleteFileLoader { del_filter: DeleteFilter, schema: SchemaRef, ) -> Result { - match task.file_type { + match task.file_type() { 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.file_format() == DataFileFormat::Puffin { return Self::load_deletion_vector(task, basic_delete_file_loader).await; } - match del_filter.try_start_pos_del_load(&task.file_path) { + match del_filter.try_start_pos_del_load(task.file_path()) { PosDelLoadAction::AlreadyLoaded => Ok(DeleteFileContext::ExistingPosDel), PosDelLoadAction::WaitFor(notified) => { // Positional deletes are accessed synchronously by ArrowReader. @@ -280,12 +280,12 @@ impl CachingDeleteFileLoader { Ok(DeleteFileContext::ExistingPosDel) } PosDelLoadAction::Load => Ok(DeleteFileContext::PosDels { - file_path: task.file_path.clone(), + file_path: task.file_path().to_string(), stream: basic_delete_file_loader .parquet_to_batch_stream( - &task.file_path, - task.file_size_in_bytes, - task.key_metadata.as_deref(), + task.file_path(), + task.file_size_in_bytes(), + task.key_metadata(), ) .await?, }), @@ -293,22 +293,22 @@ impl CachingDeleteFileLoader { } DataContentType::EqualityDeletes => { - let Some(notify) = del_filter.try_start_eq_del_load(&task.file_path) else { + let Some(notify) = del_filter.try_start_eq_del_load(task.file_path()) else { return Ok(DeleteFileContext::ExistingEqDel); }; let (sender, receiver) = channel(); - del_filter.insert_equality_delete(&task.file_path, receiver); + del_filter.insert_equality_delete(task.file_path(), receiver); // Per the Iceberg spec, evolve schema for equality deletes but only for the // equality_ids columns, not all table columns. - let equality_ids_vec = task.equality_ids.clone().unwrap(); + let equality_ids_vec = task.equality_ids().unwrap().to_vec(); let evolved_stream = BasicDeleteFileLoader::evolve_schema( basic_delete_file_loader .parquet_to_batch_stream( - &task.file_path, - task.file_size_in_bytes, - task.key_metadata.as_deref(), + task.file_path(), + task.file_size_in_bytes(), + task.key_metadata(), ) .await?, schema, @@ -347,17 +347,17 @@ impl CachingDeleteFileLoader { task: &FileScanTaskDeleteFile, ) -> Result<(u64, u64, String, u64)> { let content_offset = task - .content_offset + .content_offset() .expect("validated deletion vector must have content_offset"); let content_size = task - .content_size_in_bytes + .content_size_in_bytes() .expect("validated deletion vector must have content_size_in_bytes"); let data_file_path = task - .referenced_data_file - .clone() + .referenced_data_file() + .map(ToOwned::to_owned) .expect("validated deletion vector must have referenced_data_file"); let record_count = task - .record_count + .record_count() .expect("validated deletion vector must have record_count"); let start = u64::try_from(content_offset).map_err(|_| { @@ -365,7 +365,7 @@ impl CachingDeleteFileLoader { ErrorKind::DataInvalid, format!( "deletion vector {} has negative content_offset {content_offset}", - task.file_path + task.file_path() ), ) })?; @@ -374,7 +374,7 @@ impl CachingDeleteFileLoader { ErrorKind::DataInvalid, format!( "deletion vector {} has negative content_size_in_bytes {content_size}", - task.file_path + task.file_path() ), ) })?; @@ -419,8 +419,8 @@ impl CachingDeleteFileLoader { let input_file = basic_delete_file_loader .file_io() - .new_input(&task.file_path)?; - let blob = match task.key_metadata.as_deref() { + .new_input(task.file_path())?; + let blob = match task.key_metadata() { Some(key_metadata) => { let key_metadata = StandardKeyMetadata::decode(key_metadata)?; EncryptedInputFile::new(input_file, key_metadata) @@ -436,7 +436,7 @@ impl CachingDeleteFileLoader { data_file_path, blob, record_count, - dv_path: task.file_path.clone(), + dv_path: task.file_path().to_string(), }) } @@ -1729,8 +1729,9 @@ mod tests { #[test] fn test_validate_deletion_vector_task_rejects_negative_content_offset() { - let mut task = valid_dv_task(); - task.content_offset = Some(-1); + let mut task = serde_json::to_value(valid_dv_task()).unwrap(); + task["content_offset"] = serde_json::json!(-1); + let task = serde_json::from_value(task).unwrap(); let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); @@ -1739,8 +1740,9 @@ mod tests { #[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 mut task = serde_json::to_value(valid_dv_task()).unwrap(); + task["content_size_in_bytes"] = serde_json::json!(-1); + let task = serde_json::from_value(task).unwrap(); let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index f994bbe7a8..ec61710a42 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -124,16 +124,16 @@ impl DeleteFileLoader for BasicDeleteFileLoader { ) -> Result { let raw_batch_stream = self .parquet_to_batch_stream( - &task.file_path, - task.file_size_in_bytes, - task.key_metadata.as_deref(), + task.file_path(), + task.file_size_in_bytes(), + task.key_metadata(), ) .await?; // For equality deletes, only evolve the equality_ids columns. // For positional deletes (equality_ids is None), use all field IDs. - let field_ids = match &task.equality_ids { - Some(ids) => ids.clone(), + let field_ids = match task.equality_ids() { + Some(ids) => ids.to_vec(), None => schema.field_id_to_name_map().keys().cloned().collect(), }; diff --git a/crates/iceberg/src/arrow/delete_filter.rs b/crates/iceberg/src/arrow/delete_filter.rs index f676cb1dfa..baf9b821cf 100644 --- a/crates/iceberg/src/arrow/delete_filter.rs +++ b/crates/iceberg/src/arrow/delete_filter.rs @@ -203,14 +203,14 @@ impl DeleteFilter { } let Some(predicate) = self - .get_equality_delete_predicate_for_delete_file_path(&delete.file_path) + .get_equality_delete_predicate_for_delete_file_path(delete.file_path()) .await else { return Err(Error::new( ErrorKind::Unexpected, format!( "Missing predicate for equality delete file '{}'", - delete.file_path + delete.file_path() ), )); }; @@ -274,7 +274,7 @@ impl DeleteFilter { } pub(crate) fn is_equality_delete(f: &FileScanTaskDeleteFile) -> bool { - matches!(f.file_type, DataContentType::EqualityDeletes) + matches!(f.file_type(), DataContentType::EqualityDeletes) } #[cfg(test)] diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index d3cd205dec..6b2a0b9159 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -433,7 +433,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_4: Vec = delete_files_to_apply_for_seq_4 .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert_eq!( @@ -447,7 +447,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_5: Vec = delete_files_to_apply_for_seq_5 .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert_eq!( actual_paths_to_apply_for_seq_5, @@ -460,7 +460,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_6: Vec = delete_files_to_apply_for_seq_6 .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert_eq!( actual_paths_to_apply_for_seq_6, @@ -477,7 +477,7 @@ mod tests { let actual_paths_to_apply_for_partitioned_file: Vec = delete_files_to_apply_for_partitioned_file .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert_eq!( actual_paths_to_apply_for_partitioned_file, @@ -532,7 +532,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_4: Vec = delete_files_to_apply_for_seq_4 .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert_eq!( @@ -546,7 +546,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_5: Vec = delete_files_to_apply_for_seq_5 .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert_eq!( actual_paths_to_apply_for_seq_5, @@ -559,7 +559,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_6: Vec = delete_files_to_apply_for_seq_6 .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert_eq!( actual_paths_to_apply_for_seq_6, @@ -575,7 +575,7 @@ mod tests { let actual_paths_to_apply_for_different_partition: Vec = delete_files_to_apply_for_different_partition .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert!(actual_paths_to_apply_for_different_partition.is_empty()); @@ -587,7 +587,7 @@ mod tests { let actual_paths_to_apply_for_different_spec: Vec = delete_files_to_apply_for_different_spec .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert!(actual_paths_to_apply_for_different_spec.is_empty()); } @@ -608,7 +608,7 @@ mod tests { .get_deletes_for_data_file(&data_file_a, Some(0)) .unwrap(); assert_eq!(deletes_for_a.len(), 1); - assert_eq!(deletes_for_a[0].file_path, pos_delete.file_path()); + assert_eq!(deletes_for_a[0].file_path(), pos_delete.file_path()); // The delete references data file A, so it must not apply to data file B // even though B shares A's partition. @@ -776,7 +776,7 @@ mod tests { .get_deletes_for_data_file(&data_file, Some(0)) .unwrap() .into_iter() - .map(|delete| delete.file_path) + .map(|delete| delete.file_path().to_string()) .collect(); actual_paths.sort(); @@ -813,7 +813,7 @@ mod tests { .get_deletes_for_data_file(&data_file, Some(0)) .unwrap() .into_iter() - .map(|delete| delete.file_path) + .map(|delete| delete.file_path().to_string()) .collect(); actual_paths.sort(); @@ -852,7 +852,10 @@ mod tests { .get_deletes_for_data_file(&data_file, Some(0)) .unwrap(); assert_eq!(deletes_for_referenced.len(), 1); - assert_eq!(deletes_for_referenced[0].file_path, pos_delete.file_path()); + assert_eq!( + deletes_for_referenced[0].file_path(), + pos_delete.file_path() + ); assert!( index @@ -956,12 +959,12 @@ mod tests { }; let task: FileScanTaskDeleteFile = (&ctx).into(); - assert_eq!(task.file_type, DataContentType::PositionDeletes); - assert_eq!(task.content_offset, Some(4)); - assert_eq!(task.content_size_in_bytes, Some(40)); - assert_eq!(task.record_count, Some(3)); + assert_eq!(task.file_type(), DataContentType::PositionDeletes); + assert_eq!(task.content_offset(), Some(4)); + assert_eq!(task.content_size_in_bytes(), Some(40)); + assert_eq!(task.record_count(), Some(3)); assert_eq!( - task.referenced_data_file.as_deref(), + task.referenced_data_file(), Some("s3://bucket/data/part-0.parquet") ); } @@ -992,7 +995,7 @@ mod tests { .get_deletes_for_data_file(&data_file, Some(0)) .unwrap() .into_iter() - .map(|f| f.file_path) + .map(|f| f.file_path().to_string()) .collect(); // Only the deletion vector applies; the partition-scoped position delete file, which @@ -1107,7 +1110,7 @@ mod tests { .get_deletes_for_data_file(&data_file, Some(0)) .unwrap() .into_iter() - .map(|f| f.file_path) + .map(|f| f.file_path().to_string()) .collect(); // Only the deletion vector applies; the path-scoped position delete file, which would @@ -1142,7 +1145,7 @@ mod tests { .get_deletes_for_data_file(&data_file, Some(0)) .unwrap() .into_iter() - .map(|f| f.file_path) + .map(|f| f.file_path().to_string()) .collect(); applied.sort(); diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 2f0a876096..63c84fb296 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -3613,7 +3613,7 @@ pub mod tests { "positional delete file should be planned into the task" ); assert_eq!( - tasks[0].deletes()[0].file_type, + tasks[0].deletes()[0].file_type(), DataContentType::PositionDeletes ); diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 83a29f7e05..0d68319f9c 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -345,52 +345,52 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile { )] pub struct FileScanTaskDeleteFile { /// The delete file path - pub file_path: String, + file_path: String, /// The total size of the delete file in bytes, from the manifest entry. - pub file_size_in_bytes: u64, + file_size_in_bytes: u64, /// delete file type - pub file_type: DataContentType, + file_type: DataContentType, /// The delete file's format, from the manifest entry. A `PositionDeletes` entry written as /// `Puffin` is a V3 deletion vector; one written as `Parquet` is a position delete file. - pub file_format: DataFileFormat, + file_format: DataFileFormat, /// partition id - pub partition_spec_id: i32, + partition_spec_id: i32, /// equality ids for equality deletes (null for anything other than equality-deletes) #[builder(default)] - pub equality_ids: Option>, + equality_ids: Option>, /// For a deletion vector, the location of the data file whose rows it deletes. Required for /// deletion vectors, and may also be set on a position delete file scoped to one data file. #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] - pub referenced_data_file: Option, + referenced_data_file: Option, /// For a deletion vector, the offset of the blob within its Puffin file. Set only for /// deletion vectors, where it locates the blob for direct access. #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] - pub content_offset: Option, + content_offset: Option, /// For a deletion vector, the length in bytes of the blob within its Puffin file. /// Required together with `content_offset`; both are absent for non-DV delete files. #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] - pub content_size_in_bytes: Option, + content_size_in_bytes: Option, /// The number of records in the delete file, from the manifest entry; for a deletion vector, /// the cardinality of its bitmap. `None` only for a task not built from a manifest entry. #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] - pub record_count: Option, + record_count: Option, /// Key metadata for an encrypted delete file. When present, the reader uses this to /// decrypt the file: for a Parquet equality or position delete file, this builds @@ -404,10 +404,65 @@ pub struct FileScanTaskDeleteFile { #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] - pub key_metadata: Option>, + key_metadata: Option>, } impl FileScanTaskDeleteFile { + /// Returns the delete file path. + pub fn file_path(&self) -> &str { + &self.file_path + } + + /// Returns the total size of the delete file in bytes. + pub fn file_size_in_bytes(&self) -> u64 { + self.file_size_in_bytes + } + + /// Returns the delete file content type. + pub fn file_type(&self) -> DataContentType { + self.file_type + } + + /// Returns the delete file format. + pub fn file_format(&self) -> DataFileFormat { + self.file_format + } + + /// Returns the partition spec id. + pub fn partition_spec_id(&self) -> i32 { + self.partition_spec_id + } + + /// Returns the equality field ids for an equality delete file. + pub fn equality_ids(&self) -> Option<&[i32]> { + self.equality_ids.as_deref() + } + + /// Returns the referenced data file path. + pub fn referenced_data_file(&self) -> Option<&str> { + self.referenced_data_file.as_deref() + } + + /// Returns the deletion vector blob offset. + pub fn content_offset(&self) -> Option { + self.content_offset + } + + /// Returns the deletion vector blob size in bytes. + pub fn content_size_in_bytes(&self) -> Option { + self.content_size_in_bytes + } + + /// Returns the number of records in the delete file. + pub fn record_count(&self) -> Option { + self.record_count + } + + /// Returns the key metadata for the encrypted delete file. + pub fn key_metadata(&self) -> Option<&[u8]> { + self.key_metadata.as_deref() + } + fn is_deletion_vector(&self) -> bool { self.file_type == DataContentType::PositionDeletes && self.file_format == DataFileFormat::Puffin @@ -552,20 +607,6 @@ mod tests { .build() } - fn build_deletion_vector_task() -> Result { - FileScanTaskDeleteFile::builder() - .with_file_path("dv.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(Some("data.parquet".to_string())) - .with_content_offset(Some(7)) - .with_content_size_in_bytes(Some(11)) - .with_record_count(Some(3)) - .build() - } - fn assert_delete_file_builder_error( result: Result, expected_message: &str, @@ -696,7 +737,31 @@ mod tests { #[test] fn test_delete_file_builder_accepts_valid_deletion_vector() { - build_deletion_vector_task().unwrap(); + let task = FileScanTaskDeleteFile::builder() + .with_file_path("dv.puffin".to_string()) + .with_file_size_in_bytes(100) + .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Puffin) + .with_partition_spec_id(7) + .with_referenced_data_file(Some("data.parquet".to_string())) + .with_content_offset(Some(11)) + .with_content_size_in_bytes(Some(13)) + .with_record_count(Some(3)) + .with_key_metadata(Some(vec![17, 19].into_boxed_slice())) + .build() + .unwrap(); + + assert_eq!(task.file_path(), "dv.puffin"); + assert_eq!(task.file_size_in_bytes(), 100); + assert_eq!(task.file_type(), DataContentType::PositionDeletes); + assert_eq!(task.file_format(), DataFileFormat::Puffin); + assert_eq!(task.partition_spec_id(), 7); + assert_eq!(task.equality_ids(), None); + assert_eq!(task.referenced_data_file(), Some("data.parquet")); + assert_eq!(task.content_offset(), Some(11)); + assert_eq!(task.content_size_in_bytes(), Some(13)); + assert_eq!(task.record_count(), Some(3)); + assert_eq!(task.key_metadata(), Some([17, 19].as_slice())); } #[test] From a27d15970531cbb31dffbf2482c4016637745d75 Mon Sep 17 00:00:00 2001 From: linhongyu510 Date: Fri, 11 Sep 2026 06:15:38 +0800 Subject: [PATCH 9/9] fix(scan): propagate invalid delete task errors --- .../src/arrow/caching_delete_file_loader.rs | 73 +++++++++-- crates/iceberg/src/delete_file_index.rs | 85 ++++++++---- crates/iceberg/src/scan/task.rs | 124 ++++++++++++------ 3 files changed, 199 insertions(+), 83 deletions(-) diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index f836d7d5ed..efaaefc39c 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -333,9 +333,9 @@ 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 builder guarantees that a deletion vector already carries - /// `referenced_data_file`, `content_offset`, `content_size_in_bytes`, and `record_count`. - /// This helper keeps only the conversions still needed for the read. + /// Builder-created tasks are validated up front, but deserialized scan plans can bypass the + /// builder. Keep the required-field checks here so malformed plans return `DataInvalid` + /// instead of panicking. /// /// 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 @@ -346,19 +346,45 @@ impl CachingDeleteFileLoader { fn validate_deletion_vector_task( task: &FileScanTaskDeleteFile, ) -> Result<(u64, u64, String, u64)> { - let content_offset = task - .content_offset() - .expect("validated deletion vector must have content_offset"); - let content_size = task - .content_size_in_bytes() - .expect("validated deletion vector must have content_size_in_bytes"); + 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() .map(ToOwned::to_owned) - .expect("validated deletion vector must have referenced_data_file"); - let record_count = task - .record_count() - .expect("validated deletion vector must have record_count"); + .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( @@ -854,9 +880,28 @@ 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; + #[test] + fn test_validate_deserialized_deletion_vector_rejects_missing_fields() { + let task: FileScanTaskDeleteFile = serde_json::from_value(serde_json::json!({ + "file_path": "dv.puffin", + "file_size_in_bytes": 100, + "file_type": "PositionDeletes", + "file_format": "Puffin", + "partition_spec_id": 0 + })) + .unwrap(); + assert_eq!(task.file_type(), DataContentType::PositionDeletes); + assert_eq!(task.file_format(), DataFileFormat::Puffin); + + let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("missing content_offset")); + } + #[tokio::test] async fn test_delete_file_loader_parse_equality_deletes() { let tmp_dir = TempDir::new().unwrap(); diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 6b2a0b9159..fc942f8438 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -214,9 +214,15 @@ 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 { - let path = data_file - .referenced_data_file() - .expect("validated deletion vector must have referenced_data_file"); + let path = data_file.referenced_data_file().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} is missing referenced_data_file", + data_file.file_path() + ), + ) + })?; if let Some(existing) = dvs_by_referenced_data_file.insert(path.clone(), arc_ctx) @@ -281,27 +287,25 @@ impl PopulatedDeleteFileIndex { ) -> Result> { let mut results = vec![]; - self.global_equality_deletes - .iter() + for delete in self.global_equality_deletes.iter().filter(|&delete| { // 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) - }) - .for_each(|delete| results.push(delete.as_ref().into())); + seq_num + .map(|seq_num| delete.manifest_entry.sequence_number() > Some(seq_num)) + .unwrap_or_else(|| true) + }) { + results.push(FileScanTaskDeleteFile::try_from(delete.as_ref())?); + } if let Some(deletes) = self.eq_deletes_by_partition.get(data_file.partition()) { - deletes - .iter() + for delete in deletes.iter().filter(|&delete| { // 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())); + 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 + }) { + results.push(FileScanTaskDeleteFile::try_from(delete.as_ref())?); + } } // A deletion vector supersedes all position delete files for its data file, per the spec: @@ -343,12 +347,12 @@ impl PopulatedDeleteFileIndex { )); } } - results.push(dv.as_ref().into()); + results.push(FileScanTaskDeleteFile::try_from(dv.as_ref())?); return Ok(results); } if let Some(deletes) = self.pos_deletes_by_partition.get(data_file.partition()) { - deletes + for delete in deletes .iter() // filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num` .filter(|&delete| { @@ -357,14 +361,16 @@ impl PopulatedDeleteFileIndex { .unwrap_or_else(|| true) && data_file.partition_spec_id == delete.partition_spec_id }) - .for_each(|delete| results.push(delete.as_ref().into())); + { + results.push(FileScanTaskDeleteFile::try_from(delete.as_ref())?); + } } // 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 + for delete in deletes .iter() // filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num` .filter(|&delete| { @@ -372,7 +378,9 @@ impl PopulatedDeleteFileIndex { .map(|seq_num| delete.manifest_entry.sequence_number() >= Some(seq_num)) .unwrap_or(true) }) - .for_each(|delete| results.push(delete.as_ref().into())); + { + results.push(FileScanTaskDeleteFile::try_from(delete.as_ref())?); + } } Ok(results) @@ -958,7 +966,7 @@ mod tests { partition_spec_id: 0, }; - let task: FileScanTaskDeleteFile = (&ctx).into(); + let task = FileScanTaskDeleteFile::try_from(&ctx).unwrap(); assert_eq!(task.file_type(), DataContentType::PositionDeletes); assert_eq!(task.content_offset(), Some(4)); assert_eq!(task.content_size_in_bytes(), Some(40)); @@ -1173,6 +1181,31 @@ mod tests { assert!(err.message().contains("missing referenced_data_file")); } + #[test] + fn test_deletion_vector_index_rejects_missing_referenced_data_file() { + let malformed_dv = DataFileBuilder::default() + .file_path("deletes.puffin".to_string()) + .file_format(DataFileFormat::Puffin) + .content(DataContentType::PositionDeletes) + .record_count(1) + .content_offset(Some(4)) + .content_size_in_bytes(Some(40)) + .partition(Struct::empty()) + .partition_spec_id(0) + .file_size_in_bytes(60) + .build() + .unwrap(); + let contexts = vec![DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &malformed_dv).into(), + partition_spec_id: 0, + }]; + + let err = PopulatedDeleteFileIndex::new(contexts).unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("missing referenced_data_file")); + } + #[test] fn test_deletion_vector_builder_rejects_missing_coordinates() { let err = FileScanTaskDeleteFile::builder() diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 0d68319f9c..8dc84238ef 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -312,8 +312,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()) @@ -333,7 +335,6 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile { .map(Box::from), ) .build() - .expect("delete file context should build a valid FileScanTaskDeleteFile") } } @@ -469,6 +470,40 @@ impl FileScanTaskDeleteFile { } fn validate(&self) -> Result<()> { + if let Some(offset) = self.content_offset + && offset < 0 + { + let kind = if self.is_deletion_vector() { + "deletion vector" + } else { + "delete file" + }; + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "{kind} {} has negative content_offset {}", + self.file_path, offset + ), + )); + } + + if let Some(size) = self.content_size_in_bytes + && size < 0 + { + let kind = if self.is_deletion_vector() { + "deletion vector" + } else { + "delete file" + }; + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "{kind} {} has negative content_size_in_bytes {}", + self.file_path, size + ), + )); + } + if !self.is_deletion_vector() { return Ok(()); } @@ -483,48 +518,24 @@ impl FileScanTaskDeleteFile { )); } - match self.content_offset { - None => { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} is missing content_offset", - self.file_path - ), - )); - } - Some(offset) if offset < 0 => { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} has negative content_offset {}", - self.file_path, offset - ), - )); - } - Some(_) => {} + if self.content_offset.is_none() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} is missing content_offset", + self.file_path + ), + )); } - match self.content_size_in_bytes { - None => { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} is missing content_size_in_bytes", - self.file_path - ), - )); - } - Some(size) if size < 0 => { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "deletion vector {} has negative content_size_in_bytes {}", - self.file_path, size - ), - )); - } - Some(_) => {} + if self.content_size_in_bytes.is_none() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} is missing content_size_in_bytes", + self.file_path + ), + )); } if self.record_count.is_none() { @@ -868,6 +879,33 @@ mod tests { ); } + #[test] + fn test_delete_file_builder_rejects_negative_coordinates_for_non_dv() { + assert_delete_file_builder_error( + FileScanTaskDeleteFile::builder() + .with_file_path("position-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) + .with_content_offset(Some(-1)) + .build(), + "delete file position-deletes.parquet has negative content_offset -1", + ); + + assert_delete_file_builder_error( + FileScanTaskDeleteFile::builder() + .with_file_path("equality-deletes.parquet".to_string()) + .with_file_size_in_bytes(100) + .with_file_type(DataContentType::EqualityDeletes) + .with_file_format(DataFileFormat::Parquet) + .with_partition_spec_id(0) + .with_content_size_in_bytes(Some(-1)) + .build(), + "delete file equality-deletes.parquet has negative content_size_in_bytes -1", + ); + } + #[test] fn test_delete_file_builder_accepts_non_dv_delete_without_dv_fields() { build_delete_file_task(DataContentType::PositionDeletes, DataFileFormat::Parquet).unwrap();