From 06eca0e814663ad754ee30dc028185f98d250eec Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Thu, 9 Jul 2026 17:56:43 -0400 Subject: [PATCH 1/6] feat(scan): carry deletion-vector coordinates on FileScanTaskDeleteFile Add referenced_data_file, content_offset, and content_size_in_bytes to FileScanTaskDeleteFile, populated from the delete file's manifest entry. These locate a deletion-vector blob and scope it to its data file, which the delete loader needs to read and apply V3 deletion vectors. Refs #2792. --- crates/iceberg/src/arrow/reader/row_filter.rs | 3 ++ crates/iceberg/src/delete_file_index.rs | 35 +++++++++++++++++++ crates/iceberg/src/scan/task.rs | 18 ++++++++++ 3 files changed, 56 insertions(+) diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index a94c159d1d..a4477431c7 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1241,6 +1241,9 @@ 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, key_metadata: None, }], partition: None, diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index e58bb671b6..0a7f74d5bf 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -416,6 +416,41 @@ mod tests { assert!(actual_paths_to_apply_for_different_spec.is_empty()); } + #[test] + fn 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 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.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 faeac51be9..97efd0adf1 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -174,6 +174,9 @@ 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_key_metadata( ctx.manifest_entry .data_file @@ -205,6 +208,21 @@ 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. + #[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. + #[builder(default)] + pub content_offset: Option, + + /// For a deletion vector, the length in bytes of the blob within its Puffin file. Set + /// whenever `content_offset` is. + #[builder(default)] + pub content_size_in_bytes: Option, + /// Key metadata for encrypted delete files (Parquet Modular Encryption). /// When present, the reader uses this to build `FileDecryptionProperties`. /// From fb84ffca38c14359401235e4ab6570be0cd31fd4 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 14:21:16 -0400 Subject: [PATCH 2/6] fix after merging in main --- crates/iceberg/src/arrow/delete_file_loader.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index efdc2632cf..af470cf561 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -237,6 +237,9 @@ 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, }; let scan_metrics = ScanMetrics::new(); @@ -311,6 +314,9 @@ 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, }; let scan_metrics = ScanMetrics::new(); From 4e54829d1c81855e3e3596bc9123d11e798034ac Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 14:41:56 -0400 Subject: [PATCH 3/6] fix public-api.txt --- crates/iceberg/public-api.txt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index e6de7ab4d8..26c7a60d27 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1287,12 +1287,15 @@ 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::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 @@ -1301,7 +1304,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 From e17dfa3ad188c6a3bc8f2258ccb08485cc62cc92 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 15:59:45 -0400 Subject: [PATCH 4/6] fix test name --- crates/iceberg/src/delete_file_index.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 0a7f74d5bf..fa1ceaaa31 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -417,7 +417,7 @@ mod tests { } #[test] - fn deletion_vector_context_carries_coordinates() { + 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 must survive the conversion into a FileScanTaskDeleteFile so the loader From ad2cb1b310c3b3ea260cf99f9377de7fcabb1d65 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 18 Aug 2026 15:24:40 -0400 Subject: [PATCH 5/6] update --- crates/iceberg/src/scan/task.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 3a72538af0..50b7399cdb 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -244,16 +244,22 @@ pub struct FileScanTaskDeleteFile { /// 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. Set /// whenever `content_offset` is. + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] pub content_size_in_bytes: Option, From 004449e611a8ee20846fe37891408e12b35663ff Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Wed, 26 Aug 2026 17:29:18 -0400 Subject: [PATCH 6/6] Address PR feedback --- crates/iceberg/public-api.txt | 3 ++- crates/iceberg/src/arrow/delete_file_loader.rs | 2 ++ crates/iceberg/src/arrow/reader/row_filter.rs | 1 + crates/iceberg/src/delete_file_index.rs | 5 +++-- crates/iceberg/src/scan/task.rs | 12 ++++++++++-- 5 files changed, 18 insertions(+), 5 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 2dacd5801e..1c3250b7dd 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1318,6 +1318,7 @@ 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 @@ -1327,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 af470cf561..6a232cea0d 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -240,6 +240,7 @@ mod tests { referenced_data_file: None, content_offset: None, content_size_in_bytes: None, + record_count: None, }; let scan_metrics = ScanMetrics::new(); @@ -317,6 +318,7 @@ mod tests { 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 6a33e90fd7..8c7b00aef3 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1249,6 +1249,7 @@ mod tests { 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 09f7b3c25b..bab589385f 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -759,8 +759,8 @@ mod tests { 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 must survive the conversion into a FileScanTaskDeleteFile so the loader - // can find and apply the blob. + // 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) @@ -784,6 +784,7 @@ mod tests { 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") diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 50b7399cdb..3b25bc3134 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -211,6 +211,7 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile { .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 @@ -256,13 +257,20 @@ pub struct FileScanTaskDeleteFile { #[builder(default)] pub content_offset: Option, - /// For a deletion vector, the length in bytes of the blob within its Puffin file. Set - /// whenever `content_offset` is. + /// 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`. ///