fix(scan): validate deletion vector delete tasks

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

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

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

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

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;
 

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

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