This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3145-49b91d1c9508758f116bdc6947d2f2cba9f9980b in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit b0d8baaf966813f82363279db46d6688bf455d24 Author: hylin <[email protected]> AuthorDate: Fri Oct 9 12:36:50 2026 +0000 fix(scan): validate deletion vector delete tasks (#3145) * docs: design delete task validation * fix(scan): validate delete task construction * refactor(scan): use validated delete task builder * refactor(scan): centralize deletion vector validation * style(scan): format validated delete task changes * fix(scan): satisfy public API and license checks * test(scan): adapt merged reader fixtures * refactor(scan): encapsulate delete file metadata * fix(scan): propagate invalid delete task errors * fix(scan): unwrap the delete-file builder in the scan test `FileScanTaskDeleteFile::builder().build()` returns `Result` since this branch made it validate deletion-vector metadata, and this call site was missed, so `cargo check --all-targets` failed with E0308 once the branch was rebased onto a main that also compiles the lib test target. The task built here is a complete equality delete, so validation cannot fail; `.unwrap()` matches how the other tests in this file build tasks. * refactor(scan): make delete-file byte offsets unsigned Addresses @blackmwk's review. content_offset and content_size_in_bytes are byte quantities, so Option<u64> makes a negative unrepresentable instead of something every layer re-checks. That collapses the three findings into one change: - 'Why not change it to u64?' / 'Ditto': the getters and fields are u64. - 'This duplicates a lot.': the duplication existed only to build the 'deletion vector' vs 'delete file' prefix for two near-identical negative-value errors. Both are gone, and the remaining presence checks collapse to one lookup plus one error site. - 'I think this should also apply to non deletion vector?': it now does, and unconditionally. The i64 -> u64 conversion happens once in TryFrom<&DeleteFileContext>, where the manifest value enters, so every delete-file kind is covered rather than only Puffin position deletes. Also addresses @mbutrovich's deserialization concern from the same review: FileScanTaskDeleteFile derives plain Deserialize, so a scan plan rebuilt on a worker never passes through build(). With u64 fields, serde itself rejects a negative offset, so a hand-written or corrupted plan fails before any loader code runs -- stronger than the runtime check it replaces. Downstream, validate_deletion_vector_task loses its two u64::try_from conversions; the values are already u64. Tests: the negative cases moved from the builder to the manifest boundary (test_delete_context_rejects_negative_*), including the non-DV case, and the two loader tests now assert serde refuses the value. cargo test -p iceberg --lib: 1683 passed, 0 failed. cargo fmt --all -- --check and cargo clippy -p iceberg --all-targets -D warnings: clean. public-api.txt updated for the two getters. Signed-off-by: linhongyu510 <[email protected]> * refactor(scan): validate delete files on deserialize and enforce size pairing Address review follow-ups on the delete-file validation design: - Route FileScanTaskDeleteFile deserialization through a FileScanTaskDeleteFileSerde mirror that builds via builder().build(), so a deletion vector rebuilt from a serialized scan plan (planner->worker path) passes validate() like a manifest- built task. A malformed plan now fails at deserialize time instead of reaching a loader expect/ok_or_else. - Enforce the spec pairing rule for every delete-file kind: content_size_in_bytes is required whenever content_offset is present, not only for deletion vectors. - Replace two builder-only duplicate tests with one driving the missing-coordinate check through get_deletes_for_data_file, where the manifest entry is converted. - Reword PR-history-laden comments to describe the code as it is now. cargo test -p iceberg --lib: 1683 passed, 0 failed. * docs(scan): clarify deletion vector loader extraction --------- Signed-off-by: linhongyu510 <[email protected]> Co-authored-by: linhongyu510 <[email protected]> Co-authored-by: linhongyu510 <mailto:[email protected]> --- crates/iceberg/public-api.txt | 25 +- .../src/arrow/caching_delete_file_loader.rs | 207 ++++----- crates/iceberg/src/arrow/delete_file_loader.rs | 55 ++- crates/iceberg/src/arrow/delete_filter.rs | 18 +- crates/iceberg/src/arrow/reader/pipeline.rs | 5 +- .../iceberg/src/arrow/reader/positional_deletes.rs | 31 +- crates/iceberg/src/arrow/reader/row_filter.rs | 23 +- crates/iceberg/src/cow_rewrite/plan.rs | 2 +- crates/iceberg/src/delete_file_index.rs | 154 +++---- crates/iceberg/src/scan/mod.rs | 5 +- crates/iceberg/src/scan/task.rs | 468 ++++++++++++++++++++- 11 files changed, 707 insertions(+), 286 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index bb4d350e5..bd5539534 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1367,21 +1367,24 @@ pub fn iceberg::scan::FileScanTask::serialize<S>(&self, serializer: S) -> core:: impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTask pub fn iceberg::scan::FileScanTask::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::scan::FileScanTaskDeleteFile -pub iceberg::scan::FileScanTaskDeleteFile::content_offset: core::option::Option<i64> -pub iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes: core::option::Option<i64> -pub iceberg::scan::FileScanTaskDeleteFile::equality_ids: core::option::Option<alloc::vec::Vec<i32>> -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<alloc::boxed::Box<[u8]>> -pub iceberg::scan::FileScanTaskDeleteFile::partition_spec_id: i32 -pub iceberg::scan::FileScanTaskDeleteFile::record_count: core::option::Option<u64> -pub iceberg::scan::FileScanTaskDeleteFile::referenced_data_file: core::option::Option<alloc::string::String> +impl iceberg::scan::FileScanTaskDeleteFile +pub fn iceberg::scan::FileScanTaskDeleteFile::content_offset(&self) -> core::option::Option<u64> +pub fn iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes(&self) -> core::option::Option<u64> +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<u64> +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 pub fn iceberg::scan::FileScanTaskDeleteFile::eq(&self, other: &iceberg::scan::FileScanTaskDeleteFile) -> bool +impl core::convert::From<iceberg::scan::FileScanTaskDeleteFile> for iceberg::Result<iceberg::scan::FileScanTaskDeleteFile> +pub fn iceberg::Result<iceberg::scan::FileScanTaskDeleteFile>::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 e1cb6de15..5582e4397 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -263,15 +263,15 @@ impl CachingDeleteFileLoader { del_filter: DeleteFilter, schema: SchemaRef, ) -> Result<DeleteFileContext> { - 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. @@ -281,12 +281,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?, }), @@ -294,22 +294,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, @@ -331,59 +331,47 @@ impl CachingDeleteFileLoader { } } - /// Validates a deletion-vector task and returns what the read needs as typed values: + /// Extracts a deletion-vector task's metadata as the values needed by the read: /// `(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`. + /// Both builder construction and deserialization validate the task's required fields. + /// The fallible extraction here converts the optional fields into values for the + /// byte-range read, preserving `DataInvalid` errors instead of unwrapping them. fn validate_deletion_vector_task( task: &FileScanTaskDeleteFile, ) -> Result<(u64, u64, String, u64)> { - let content_offset = task.content_offset.ok_or_else(|| { + let content_offset = task.content_offset().ok_or_else(|| { invalid_data!( "deletion vector {} is missing content_offset", - task.file_path + task.file_path() ) })?; - let content_size = task.content_size_in_bytes.ok_or_else(|| { + let content_size = task.content_size_in_bytes().ok_or_else(|| { invalid_data!( "deletion vector {} is missing content_size_in_bytes", - task.file_path - ) - })?; - let data_file_path = task.referenced_data_file.clone().ok_or_else(|| { - invalid_data!( - "deletion vector {} is missing referenced_data_file", - task.file_path - ) - })?; - let record_count = task.record_count.ok_or_else(|| { - invalid_data!("deletion vector {} is missing record_count", task.file_path) - })?; - - let start = u64::try_from(content_offset).map_err(|_| { - invalid_data!( - "deletion vector {} has negative content_offset {content_offset}", - task.file_path + task.file_path() ) })?; - let len = u64::try_from(content_size).map_err(|_| { + let data_file_path = task + .referenced_data_file() + .map(ToOwned::to_owned) + .ok_or_else(|| { + invalid_data!( + "deletion vector {} is missing referenced_data_file", + task.file_path() + ) + })?; + let record_count = task.record_count().ok_or_else(|| { invalid_data!( - "deletion vector {} has negative content_size_in_bytes {content_size}", - task.file_path + "deletion vector {} is missing record_count", + task.file_path() ) })?; - Ok((start, len, data_file_path, record_count)) + // content_offset / content_size_in_bytes are u64 on the task, and the + // manifest -> task conversion already rejected a negative, so there is + // nothing left to convert or re-check here. + Ok((content_offset, content_size, data_file_path, record_count)) } /// Validates a decoded deletion vector's cardinality against the manifest entry's @@ -420,8 +408,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) @@ -437,7 +425,7 @@ impl CachingDeleteFileLoader { data_file_path, blob, record_count, - dv_path: task.file_path.clone(), + dv_path: task.file_path().to_string(), }) } @@ -849,9 +837,29 @@ 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_deserialized_deletion_vector_rejects_missing_fields() { + // A deletion vector that omits its required coordinates fails validation + // during deserialization (the try_from mirror routes through `build()`), + // before the loader ever sees the task. + let err = serde_json::from_value::<FileScanTaskDeleteFile>(serde_json::json!({ + "file_path": "dv.puffin", + "file_size_in_bytes": 100, + "file_type": "PositionDeletes", + "file_format": "Puffin", + "partition_spec_id": 0 + })) + .unwrap_err(); + + assert!( + err.to_string().contains("missing referenced_data_file"), + "expected a missing-field validation error, got `{err}`" + ); + } + #[tokio::test] async fn test_delete_file_loader_parse_equality_deletes() { let tmp_dir = TempDir::new().unwrap(); @@ -1400,7 +1408,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()) @@ -1409,7 +1418,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) @@ -1542,8 +1552,8 @@ mod tests { dv_path: String, file_size: u64, data_file_path: String, - content_offset: i64, - content_size: i64, + content_offset: u64, + content_size: u64, record_count: u64, key_metadata: Option<Box<[u8]>>, ) -> FileScanTaskDeleteFile { @@ -1559,6 +1569,7 @@ mod tests { .with_record_count(Some(record_count)) .with_key_metadata(key_metadata) .build() + .expect("deletion vector task should be valid") } #[tokio::test] @@ -1571,8 +1582,8 @@ mod tests { // Embed the blob in a Puffin-like file behind leading bytes so content_offset is // non-zero, then let the loader read it back by range. - let content_offset = 12i64; - let content_size = blob.len() as i64; + let content_offset = 12u64; + let content_size = blob.len() as u64; let mut file_bytes = vec![0u8; content_offset as usize]; file_bytes.extend_from_slice(&blob); file_bytes.extend_from_slice(&[0u8; 8]); @@ -1623,7 +1634,7 @@ mod tests { .with_aad_prefix(b"test-aad-prefix!"); let blob = encode_dv_blob([2u64, 4]); - let plaintext_size = blob.len() as i64; + let plaintext_size = blob.len() as u64; let dv_path = format!("{table_location}/deletes.puffin"); let output = EncryptedOutputFile::new(file_io.new_output(&dv_path).unwrap(), key_metadata); let file_metadata = output.write(Bytes::from(blob)).await.unwrap(); @@ -1683,7 +1694,7 @@ mod tests { std::fs::metadata(&dv_path).unwrap().len(), data_file_path, 0, - blob.len() as i64, + blob.len() as u64, 2, None, ); @@ -1708,7 +1719,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(), @@ -1721,64 +1733,31 @@ mod tests { ) } + // The coordinate fields are unsigned, so a negative byte offset is rejected during + // deserialization: a hand-written or corrupted scan plan fails before any loader + // code runs. #[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); + fn test_deserializing_negative_content_offset_is_rejected() { + let mut task = serde_json::to_value(valid_dv_task()).unwrap(); + task["content_offset"] = serde_json::json!(-1); - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); - assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!(err.message().contains("negative content_offset")); + let err = serde_json::from_value::<FileScanTaskDeleteFile>(task).unwrap_err(); + assert!( + err.to_string().contains("invalid value"), + "expected an out-of-range integer error, got `{err}`" + ); } #[test] - fn test_validate_deletion_vector_task_rejects_negative_content_size() { - let mut task = valid_dv_task(); - task.content_size_in_bytes = Some(-1); + fn test_deserializing_negative_content_size_is_rejected() { + let mut task = serde_json::to_value(valid_dv_task()).unwrap(); + task["content_size_in_bytes"] = serde_json::json!(-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")); + let err = serde_json::from_value::<FileScanTaskDeleteFile>(task).unwrap_err(); + assert!( + err.to_string().contains("invalid value"), + "expected an out-of-range integer error, got `{err}`" + ); } #[test] diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index 5852cf1ac..ec61710a4 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<ArrowRecordBatchStream> { 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(), }; @@ -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/delete_filter.rs b/crates/iceberg/src/arrow/delete_filter.rs index 6f13be71d..29843295d 100644 --- a/crates/iceberg/src/arrow/delete_filter.rs +++ b/crates/iceberg/src/arrow/delete_filter.rs @@ -208,14 +208,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() ), )); }; @@ -279,7 +279,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)] @@ -448,7 +448,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!( @@ -466,7 +467,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!( @@ -484,7 +486,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() @@ -561,7 +564,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 11fa200ee..18ba42d3b 100644 --- a/crates/iceberg/src/arrow/reader/pipeline.rs +++ b/crates/iceberg/src/arrow/reader/pipeline.rs @@ -2486,11 +2486,12 @@ 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) .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 2915596da..a261bd536 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() @@ -975,8 +978,8 @@ mod tests { // Deletion vector deleting positions 1 and 3 (ids 2 and 4), serialized as a // deletion-vector-v1 blob and embedded in a Puffin-like file at a non-zero offset. let blob = encode_dv_blob([1u64, 3]); - let content_offset = 12i64; - let content_size = blob.len() as i64; + let content_offset = 12u64; + let content_size = blob.len() as u64; let mut dv_file_bytes = vec![0u8; content_offset as usize]; dv_file_bytes.extend_from_slice(&blob); let dv_path = format!("{table_location}/deletes.puffin"); @@ -1005,12 +1008,14 @@ 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![task])) as FileScanTaskStream; + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; let result = reader .read(tasks) .unwrap() @@ -1100,14 +1105,16 @@ mod tests { .with_partition_spec_id(0) .with_referenced_data_file(Some(data_file_path.clone())) .with_content_offset(Some(0)) - .with_content_size_in_bytes(Some(blob.len() as i64)) + .with_content_size_in_bytes(Some(blob.len() as u64)) .with_record_count(Some(5)) - .build(), + .build() + .unwrap(), ]) .with_case_sensitive(false) - .build(); + .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<Vec<RecordBatch>, _> = reader.read(tasks).unwrap().stream().try_collect().await; diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index 34b739113..5d92f6bc7 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1235,19 +1235,16 @@ 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(); diff --git a/crates/iceberg/src/cow_rewrite/plan.rs b/crates/iceberg/src/cow_rewrite/plan.rs index b96aea635..8e96e9ab4 100644 --- a/crates/iceberg/src/cow_rewrite/plan.rs +++ b/crates/iceberg/src/cow_rewrite/plan.rs @@ -151,7 +151,7 @@ mod tests { for file in files { assert_eq!(file.scan_task.deletes().len(), 1); assert_eq!( - file.scan_task.deletes()[0].file_path, + file.scan_task.deletes()[0].file_path(), format!("{}/del.parquet", fixture.table_location) ); } diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 692ab9315..8bae43941 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -215,25 +215,14 @@ 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(invalid_data!( + // referenced_data_file is the map key the index needs; the offset/size/ + // record_count invariants are enforced centrally when the task is built. + let path = data_file.referenced_data_file().ok_or_else(|| { + invalid_data!( "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(invalid_data!( - "deletion vector {} is missing content_offset or content_size_in_bytes", - arc_ctx.manifest_entry.file_path() - )); - } + data_file.file_path() + ) + })?; if let Some(existing) = dvs_by_referenced_data_file.insert(path.clone(), arc_ctx) @@ -295,27 +284,25 @@ impl PopulatedDeleteFileIndex { ) -> Result<Vec<FileScanTaskDeleteFile>> { 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: @@ -351,12 +338,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| { @@ -365,14 +352,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| { @@ -380,7 +369,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) @@ -442,7 +433,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_4: Vec<String> = delete_files_to_apply_for_seq_4 .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert_eq!( @@ -456,7 +447,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_5: Vec<String> = 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, @@ -469,7 +460,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_6: Vec<String> = 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, @@ -486,7 +477,7 @@ mod tests { let actual_paths_to_apply_for_partitioned_file: Vec<String> = 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, @@ -541,7 +532,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_4: Vec<String> = delete_files_to_apply_for_seq_4 .into_iter() - .map(|file| file.file_path) + .map(|file| file.file_path().to_string()) .collect(); assert_eq!( @@ -555,7 +546,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_5: Vec<String> = 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, @@ -568,7 +559,7 @@ mod tests { .unwrap(); let actual_paths_to_apply_for_seq_6: Vec<String> = 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, @@ -584,7 +575,7 @@ mod tests { let actual_paths_to_apply_for_different_partition: Vec<String> = 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()); @@ -596,7 +587,7 @@ mod tests { let actual_paths_to_apply_for_different_spec: Vec<String> = 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()); } @@ -617,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. @@ -785,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(); @@ -822,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(); @@ -861,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 @@ -964,13 +958,13 @@ mod tests { partition_spec_id: 0, }; - 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)); + 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)); + 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") ); } @@ -1001,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 @@ -1116,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 @@ -1151,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(); @@ -1162,7 +1156,7 @@ mod tests { } #[test] - fn test_deletion_vector_missing_referenced_data_file_is_rejected() { + 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) @@ -1175,43 +1169,59 @@ mod tests { .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_missing_coordinates_is_rejected() { + fn test_deletion_vector_index_rejects_missing_coordinates() { + // The missing-coordinate check fires in `get_deletes_for_data_file`, where the + // manifest entry is converted into a task. `PopulatedDeleteFileIndex::new` only + // needs `referenced_data_file` as its map key, so a DV that carries it but omits + // `content_offset` survives construction and is rejected when the data file is + // resolved against it. + let data_file = DataFileBuilder::default() + .file_path("target-data.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .content(DataContentType::Data) + .record_count(100) + .partition(Struct::empty()) + .partition_spec_id(0) + .file_size_in_bytes(100) + .build() + .unwrap(); + 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("target-data.parquet".to_string())) .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(); + let delete_file_index = PopulatedDeleteFileIndex::new(contexts).unwrap(); + let err = delete_file_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 ccdb5c7d5..0c241125d 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -1881,7 +1881,8 @@ 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)) @@ -2905,7 +2906,7 @@ 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 97ca7e518..45940b0df 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -21,13 +21,13 @@ use futures::stream::BoxStream; use serde::{Deserialize, Serialize}; use typed_builder::TypedBuilder; -use crate::Result; use crate::error::invalid_data; use crate::expr::BoundPredicate; use crate::spec::{ DataContentType, DataFileFormat, ManifestEntryRef, NameMapping, PartitionSpec, Schema, SchemaRef, SortOrderRef, Struct, StructType, }; +use crate::{Error, Result}; /// A stream of [`FileScanTask`]. pub type FileScanTaskStream = BoxStream<'static, Result<FileScanTask>>; @@ -309,8 +309,24 @@ 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<Self> { + // The manifest stores these as i64. Convert here so a negative byte offset is + // rejected once at the boundary, for every delete-file kind rather than only + // deletion vectors, instead of being carried inward and re-checked per consumer. + let file_path = ctx.manifest_entry.file_path(); + let to_offset = |value: Option<i64>, field: &str| -> Result<Option<u64>> { + value + .map(|value| { + u64::try_from(value).map_err(|_| { + invalid_data!("delete file {file_path} has negative {field} {value}") + }) + }) + .transpose() + }; + FileScanTaskDeleteFile::builder() .with_file_path(ctx.manifest_entry.file_path().to_string()) .with_file_size_in_bytes(ctx.manifest_entry.file_size_in_bytes()) @@ -319,8 +335,14 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile { .with_partition_spec_id(ctx.partition_spec_id) .with_equality_ids(ctx.manifest_entry.data_file.equality_ids.clone()) .with_referenced_data_file(ctx.manifest_entry.data_file.referenced_data_file.clone()) - .with_content_offset(ctx.manifest_entry.data_file.content_offset) - .with_content_size_in_bytes(ctx.manifest_entry.data_file.content_size_in_bytes) + .with_content_offset(to_offset( + ctx.manifest_entry.data_file.content_offset, + "content_offset", + )?) + .with_content_size_in_bytes(to_offset( + ctx.manifest_entry.data_file.content_size_in_bytes, + "content_size_in_bytes", + )?) .with_record_count(Some(ctx.manifest_entry.record_count())) .with_key_metadata( ctx.manifest_entry @@ -335,55 +357,62 @@ 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_")))] +#[serde(try_from = "crate::scan::task::_serde::FileScanTaskDeleteFileSerde")] +#[builder( + field_defaults(setter(prefix = "with_")), + build_method(into = Result<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<Vec<i32>>, + equality_ids: Option<Vec<i32>>, /// 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<String>, + referenced_data_file: Option<String>, /// 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. + /// deletion vectors, where it locates the blob for direct access. Unsigned because it is a + /// byte offset: a negative value is rejected once, when the task is built from a manifest + /// entry, rather than re-checked by every consumer. #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] - pub content_offset: Option<i64>, + content_offset: Option<u64>, /// 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. + /// Unsigned for the same reason as `content_offset`. #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] - pub content_size_in_bytes: Option<i64>, + content_size_in_bytes: Option<u64>, /// 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<u64>, + record_count: Option<u64>, /// 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 @@ -397,7 +426,7 @@ pub struct FileScanTaskDeleteFile { #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] - pub key_metadata: Option<Box<[u8]>>, + key_metadata: Option<Box<[u8]>>, } mod _serde { @@ -409,8 +438,8 @@ mod _serde { use crate::error::invalid_data; use crate::expr::BoundPredicate; use crate::spec::{ - DataFileFormat, Literal, NameMapping, PartitionSpec, RawLiteral, SchemaRef, SortOrderRef, - StructType, Type, + DataContentType, DataFileFormat, Literal, NameMapping, PartitionSpec, RawLiteral, + SchemaRef, SortOrderRef, StructType, Type, }; use crate::{Error, Result}; @@ -579,13 +608,168 @@ mod _serde { .build() } } + + #[derive(Deserialize)] + pub(super) struct FileScanTaskDeleteFileSerde { + file_path: String, + file_size_in_bytes: u64, + file_type: DataContentType, + file_format: DataFileFormat, + partition_spec_id: i32, + equality_ids: Option<Vec<i32>>, + #[serde(default)] + referenced_data_file: Option<String>, + #[serde(default)] + content_offset: Option<u64>, + #[serde(default)] + content_size_in_bytes: Option<u64>, + #[serde(default)] + record_count: Option<u64>, + #[serde(default)] + key_metadata: Option<Box<[u8]>>, + } + + impl TryFrom<FileScanTaskDeleteFileSerde> for FileScanTaskDeleteFile { + type Error = Error; + + fn try_from(value: FileScanTaskDeleteFileSerde) -> Result<Self> { + Self::builder() + .with_file_path(value.file_path) + .with_file_size_in_bytes(value.file_size_in_bytes) + .with_file_type(value.file_type) + .with_file_format(value.file_format) + .with_partition_spec_id(value.partition_spec_id) + .with_equality_ids(value.equality_ids) + .with_referenced_data_file(value.referenced_data_file) + .with_content_offset(value.content_offset) + .with_content_size_in_bytes(value.content_size_in_bytes) + .with_record_count(value.record_count) + .with_key_metadata(value.key_metadata) + .build() + } + } +} + +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<u64> { + self.content_offset + } + + /// Returns the deletion vector blob size in bytes. + pub fn content_size_in_bytes(&self) -> Option<u64> { + self.content_size_in_bytes + } + + /// Returns the number of records in the delete file. + pub fn record_count(&self) -> Option<u64> { + 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 + } + + fn validate(&self) -> Result<()> { + // Negative offsets/sizes are unrepresentable: both fields are u64, and the + // conversion from the manifest entry's i64 rejects a negative there. + + // Spec: `content_size_in_bytes` is required whenever `content_offset` is + // present, for every delete-file kind (not only deletion vectors). + if self.content_offset.is_some() && self.content_size_in_bytes.is_none() { + return Err(invalid_data!( + "delete file {} is missing content_size_in_bytes while content_offset is present", + self.file_path + )); + } + + // Presence of the deletion-vector coordinates is required only for + // deletion vectors; other delete files legitimately carry none of these. + if !self.is_deletion_vector() { + return Ok(()); + } + + let missing = if self.referenced_data_file.is_none() { + Some("referenced_data_file") + } else if self.content_offset.is_none() { + Some("content_offset") + } else if self.content_size_in_bytes.is_none() { + Some("content_size_in_bytes") + } else if self.record_count.is_none() { + Some("record_count") + } else { + None + }; + + if let Some(field) = missing { + return Err(invalid_data!( + "deletion vector {} is missing {field}", + self.file_path + )); + } + + Ok(()) + } +} + +impl From<FileScanTaskDeleteFile> for Result<FileScanTaskDeleteFile> { + fn from(task: FileScanTaskDeleteFile) -> Self { + task.validate()?; + Ok(task) + } } #[cfg(test)] mod tests { + use std::collections::HashMap; + use super::*; use crate::ErrorKind; - use crate::spec::{Literal, NestedField, PrimitiveType, Transform, Type}; + use crate::spec::{ + DataFileBuilder, Literal, ManifestEntry, ManifestStatus, NestedField, PrimitiveType, + Transform, Type, + }; fn build_file_scan_task( schema: SchemaRef, @@ -630,6 +814,34 @@ mod tests { (schema, partition_spec) } + fn build_delete_file_task( + file_type: DataContentType, + file_format: DataFileFormat, + ) -> Result<FileScanTaskDeleteFile> { + 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 assert_delete_file_builder_error( + result: Result<FileScanTaskDeleteFile>, + 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. @@ -739,4 +951,218 @@ mod tests { assert_eq!(err.kind(), ErrorKind::DataInvalid); } + + #[test] + fn test_delete_file_builder_accepts_valid_deletion_vector() { + 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] + 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(), + "delete file dv.puffin is missing content_size_in_bytes while content_offset is present", + ); + } + + #[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_non_dv_offset_without_size() { + // Spec pairing rule applies to every delete-file kind: a Parquet + // position-delete file that carries a content_offset must also carry a + // content_size_in_bytes, even though it is not a deletion vector. + 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(7)) + .build(), + "delete file position-deletes.parquet is missing content_size_in_bytes while content_offset is present", + ); + } + + fn delete_manifest_entry( + file_path: &str, + file_format: DataFileFormat, + content_offset: Option<i64>, + content_size_in_bytes: Option<i64>, + ) -> DeleteFileContext { + let mut data_file = DataFileBuilder::default() + .content(DataContentType::PositionDeletes) + .file_path(file_path.to_string()) + .file_format(file_format) + .partition(Struct::empty()) + .record_count(3) + .file_size_in_bytes(100) + .column_sizes(HashMap::new()) + .value_counts(HashMap::new()) + .null_value_counts(HashMap::new()) + .partition_spec_id(0) + .referenced_data_file(Some("data.parquet".to_string())) + .build() + .unwrap(); + data_file.content_offset = content_offset; + data_file.content_size_in_bytes = content_size_in_bytes; + + DeleteFileContext { + manifest_entry: Arc::new( + ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file(data_file) + .build(), + ), + partition_spec_id: 0, + } + } + + fn assert_delete_context_error(ctx: &DeleteFileContext, expected_message: &str) { + match FileScanTaskDeleteFile::try_from(ctx) { + Ok(task) => panic!( + "expected conversion to fail with `{expected_message}`, but got Ok({task:?})" + ), + Err(error) => { + assert_eq!(error.kind(), ErrorKind::DataInvalid); + assert!( + error.message().contains(expected_message), + "expected `{expected_message}`, got `{}`", + error.message() + ); + } + } + } + + /// The manifest entry supplies `content_offset` / `content_size_in_bytes` as + /// `i64`; the conversion into a task rejects a negative value as it enters, + /// since the task fields are `u64`. + #[test] + fn test_delete_context_rejects_negative_dv_offset() { + assert_delete_context_error( + &delete_manifest_entry("dv.puffin", DataFileFormat::Puffin, Some(-1), Some(11)), + "delete file dv.puffin has negative content_offset -1", + ); + } + + #[test] + fn test_delete_context_rejects_negative_dv_size() { + assert_delete_context_error( + &delete_manifest_entry("dv.puffin", DataFileFormat::Puffin, Some(7), Some(-1)), + "delete file dv.puffin has negative content_size_in_bytes -1", + ); + } + + /// The negative-coordinate check applies to every delete-file kind, including + /// a Parquet position-delete file rather than only deletion vectors. + #[test] + fn test_delete_context_rejects_negative_coordinates_for_non_dv() { + assert_delete_context_error( + &delete_manifest_entry( + "position-deletes.parquet", + DataFileFormat::Parquet, + Some(-1), + None, + ), + "delete file position-deletes.parquet has negative content_offset -1", + ); + assert_delete_context_error( + &delete_manifest_entry( + "position-deletes.parquet", + DataFileFormat::Parquet, + None, + Some(-1), + ), + "delete file position-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(); + build_delete_file_task(DataContentType::EqualityDeletes, DataFileFormat::Parquet).unwrap(); + } }
