diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 9c79233aab..19f6c26968 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1310,12 +1310,16 @@ 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_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 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 @@ -1324,7 +1328,7 @@ 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 impl iceberg::scan::FileScanTaskDeleteFile -pub fn iceberg::scan::FileScanTaskDeleteFile::builder() -> FileScanTaskDeleteFileBuilder<((), (), (), (), (), ())> +pub fn iceberg::scan::FileScanTaskDeleteFile::builder() -> FileScanTaskDeleteFileBuilder<((), (), (), (), (), (), (), (), (), ())> impl serde_core::ser::Serialize for iceberg::scan::FileScanTaskDeleteFile pub fn iceberg::scan::FileScanTaskDeleteFile::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTaskDeleteFile diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index efdc2632cf..6a232cea0d 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -237,6 +237,10 @@ mod tests { 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 scan_metrics = ScanMetrics::new(); @@ -311,6 +315,10 @@ mod tests { 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 scan_metrics = ScanMetrics::new(); diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index 2e6b37ba30..8c7b00aef3 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1246,6 +1246,10 @@ mod tests { 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, }], partition: None, diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 19407467db..bab589385f 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -755,6 +755,42 @@ mod tests { .unwrap() } + #[test] + fn test_deletion_vector_context_carries_coordinates() { + // A deletion vector is a PositionDeletes entry stored as a Puffin blob, located by + // content_offset / content_size_in_bytes and scoped by referenced_data_file. Those + // three fields, and the record count its bitmap is validated against, must survive the + // conversion into a FileScanTaskDeleteFile so the loader can find and apply the blob. + let dv = DataFileBuilder::default() + .file_path("s3://bucket/data/part-0.parquet-deletes.puffin".to_string()) + .file_format(DataFileFormat::Puffin) + .content(DataContentType::PositionDeletes) + .record_count(3) + .referenced_data_file(Some("s3://bucket/data/part-0.parquet".to_string())) + .content_offset(Some(4)) + .content_size_in_bytes(Some(40)) + .partition(Struct::empty()) + .partition_spec_id(0) + .file_size_in_bytes(44) + .build() + .unwrap(); + + let ctx = DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &dv).into(), + 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)); + assert_eq!( + task.referenced_data_file.as_deref(), + Some("s3://bucket/data/part-0.parquet") + ); + } + fn build_unpartitioned_eq_delete() -> DataFile { build_partitioned_eq_delete(&Struct::empty(), 0) } diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index e69397b066..3b25bc3134 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -208,6 +208,10 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile { .with_file_type(ctx.manifest_entry.content_type()) .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_record_count(Some(ctx.manifest_entry.record_count())) .with_key_metadata( ctx.manifest_entry .data_file @@ -239,6 +243,34 @@ pub struct FileScanTaskDeleteFile { #[builder(default)] pub 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, + + /// 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, + + /// 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, + + /// 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, + /// Key metadata for encrypted delete files (Parquet Modular Encryption). /// When present, the reader uses this to build `FileDecryptionProperties`. ///