feat(scan): [2/N] carry deletion-vector coordinates on FileScanTaskDeleteFile

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<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_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 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

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();

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,

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)
     }

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<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>,
+
+    /// 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<i64>,
+
+    /// 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<i64>,
+
+    /// 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>,
+
     /// Key metadata for encrypted delete files (Parquet Modular Encryption).
     /// When present, the reader uses this to build `FileDecryptionProperties`.
     ///