fix: skip serializing file scan task partition

.github/workflows/ci.yml

@@ -21,7 +21,10 @@ on:
   push:
     branches:
       - main
+      - dev_rebase_main_20251111
   pull_request:
+    branches:
+      - dev_rebase_main_20251111
     paths:
       - '**'                              # Include all files and directories in the repository by default.
       - '!.github/ISSUE_TEMPLATE/**'      # Exclude files and directories that don't impact tests or code like templates, metadata, and documentation.

Cargo.toml

@@ -47,6 +47,7 @@ arrow-array = "56.2"
 arrow-buffer = "56.2"
 arrow-cast = "56.2"
 arrow-ord = "56.2"
+arrow-row = "56.2"
 arrow-schema = "56.2"
 arrow-select = "56.2"
 arrow-string = "56.2"

crates/iceberg/Cargo.toml

@@ -30,14 +30,15 @@ repository = { workspace = true }
 
 [features]
 default = ["storage-memory", "storage-fs", "storage-s3", "tokio"]
-storage-all = ["storage-memory", "storage-fs", "storage-s3", "storage-gcs"]
+storage-all = ["storage-memory", "storage-fs", "storage-s3", "storage-gcs", "storage-azblob"]
 
 storage-azdls = ["opendal/services-azdls"]
 storage-fs = ["opendal/services-fs"]
 storage-gcs = ["opendal/services-gcs"]
 storage-memory = ["opendal/services-memory"]
 storage-oss = ["opendal/services-oss"]
 storage-s3 = ["opendal/services-s3", "reqsign"]
+storage-azblob = ["opendal/services-azblob"]
 
 smol = ["dep:smol"]
 tokio = ["tokio/rt-multi-thread"]
@@ -51,6 +52,7 @@ arrow-array = { workspace = true }
 arrow-buffer = { workspace = true }
 arrow-cast = { workspace = true }
 arrow-ord = { workspace = true }
+arrow-row = { workspace = true }
 arrow-schema = { workspace = true }
 arrow-select = { workspace = true }
 arrow-string = { workspace = true }
@@ -89,6 +91,7 @@ smol = { workspace = true, optional = true }
 strum = { workspace = true, features = ["derive"] }
 thrift = { workspace = true }
 tokio = { workspace = true, optional = false, features = ["sync"] }
+tracing = { workspace = true }
 typed-builder = { workspace = true }
 url = { workspace = true }
 uuid = { workspace = true }

crates/iceberg/src/arrow/caching_delete_file_loader.rs

@@ -30,7 +30,7 @@ use crate::delete_vector::DeleteVector;
 use crate::expr::Predicate::AlwaysTrue;
 use crate::expr::{Predicate, Reference};
 use crate::io::FileIO;
-use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile};
+use crate::scan::{ArrowRecordBatchStream, FileScanTask};
 use crate::spec::{
     DataContentType, Datum, ListType, MapType, NestedField, NestedFieldRef, PartnerAccessor,
     PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor, StructType, Type,
@@ -138,7 +138,7 @@ impl CachingDeleteFileLoader {
     /// ```
     pub(crate) fn load_deletes(
         &self,
-        delete_file_entries: &[FileScanTaskDeleteFile],
+        delete_file_entries: &[Arc<FileScanTask>],
         schema: SchemaRef,
     ) -> Receiver<Result<DeleteFilter>> {
         let (tx, rx) = channel();
@@ -204,32 +204,32 @@ impl CachingDeleteFileLoader {
     }
 
     async fn load_file_for_task(
-        task: &FileScanTaskDeleteFile,
+        task: &FileScanTask,
         basic_delete_file_loader: BasicDeleteFileLoader,
         del_filter: DeleteFilter,
         schema: SchemaRef,
     ) -> Result<DeleteFileContext> {
-        match task.file_type {
+        match task.data_file_content {
             DataContentType::PositionDeletes => Ok(DeleteFileContext::PosDels(
                 basic_delete_file_loader
-                    .parquet_to_batch_stream(&task.file_path)
+                    .parquet_to_batch_stream(task.data_file_path())
                     .await?,
             )),
 
             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.data_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.data_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 evolved_stream = BasicDeleteFileLoader::evolve_schema(
                     basic_delete_file_loader
-                        .parquet_to_batch_stream(&task.file_path)
+                        .parquet_to_batch_stream(task.data_file_path())
                         .await?,
                     schema,
                     &equality_ids_vec,
@@ -552,6 +552,7 @@ mod tests {
 
     use super::*;
     use crate::arrow::delete_filter::tests::setup;
+    use crate::scan::FileScanTaskDeleteFile;
 
     #[tokio::test]
     async fn test_delete_file_loader_parse_equality_deletes() {
@@ -871,6 +872,44 @@ mod tests {
             equality_ids: Some(vec![2, 3]), // Only use field IDs that exist in both schemas
         };
 
+        let pos_del_scan_task = FileScanTask {
+            start: 0,
+            length: 0,
+            record_count: None,
+            data_file_path: pos_del.file_path,
+            data_file_content: DataContentType::PositionDeletes,
+            data_file_format: DataFileFormat::Parquet,
+            schema: data_file_schema.clone(),
+            project_field_ids: vec![],
+            predicate: None,
+            deletes: vec![],
+            sequence_number: 0,
+            equality_ids: None,
+            file_size_in_bytes: 0,
+            partition: None,
+            partition_spec: None,
+            name_mapping: None,
+        };
+
+        let eq_del_scan_task = FileScanTask {
+            start: 0,
+            length: 0,
+            record_count: None,
+            data_file_path: eq_del.file_path.clone(),
+            data_file_content: DataContentType::EqualityDeletes,
+            data_file_format: DataFileFormat::Parquet,
+            schema: data_file_schema.clone(),
+            project_field_ids: vec![],
+            predicate: None,
+            deletes: vec![],
+            sequence_number: 0,
+            equality_ids: eq_del.equality_ids.clone(),
+            file_size_in_bytes: 0,
+            partition: None,
+            partition_spec: None,
+            name_mapping: None,
+        };
+
         let file_scan_task = FileScanTask {
             start: 0,
             length: 0,
@@ -880,7 +919,14 @@ mod tests {
             schema: data_file_schema.clone(),
             project_field_ids: vec![2, 3],
             predicate: None,
-            deletes: vec![pos_del, eq_del],
+            deletes: vec![pos_del_scan_task.into(), eq_del_scan_task.into()],
+            sequence_number: 0,
+            data_file_content: DataContentType::Data,
+            equality_ids: None,
+            file_size_in_bytes: 0,
+            partition: None,
+            partition_spec: None,
+            name_mapping: None,
         };
 
         // Load the deletes - should handle both types without error

crates/iceberg/src/arrow/delete_file_loader.rs

@@ -20,7 +20,7 @@ use std::sync::Arc;
 use futures::{StreamExt, TryStreamExt};
 
 use crate::arrow::ArrowReader;
-use crate::arrow::record_batch_transformer::RecordBatchTransformer;
+use crate::arrow::record_batch_transformer::RecordBatchTransformerBuilder;
 use crate::io::FileIO;
 use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile};
 use crate::spec::{Schema, SchemaRef};
@@ -82,7 +82,7 @@ impl BasicDeleteFileLoader {
         equality_ids: &[i32],
     ) -> Result<ArrowRecordBatchStream> {
         let mut record_batch_transformer =
-            RecordBatchTransformer::build(target_schema.clone(), equality_ids);
+            RecordBatchTransformerBuilder::new(target_schema.clone(), equality_ids).build();
 
         let record_batch_stream = record_batch_stream.map(move |record_batch| {
             record_batch.and_then(|record_batch| {
@@ -120,6 +120,7 @@ mod tests {
 
     use super::*;
     use crate::arrow::delete_filter::tests::setup;
+    use crate::spec::DataContentType;
 
     #[tokio::test]
     async fn test_basic_delete_file_loader_read_delete_file() {
@@ -134,11 +135,15 @@ mod tests {
 
         let file_scan_tasks = setup(table_location);
 
+        let delete_task = FileScanTaskDeleteFile {
+            file_path: file_scan_tasks[0].deletes[0].data_file_path.clone(),
+            file_type: DataContentType::PositionDeletes,
+            partition_spec_id: 0,
+            equality_ids: None,
+        };
+
         let result = delete_file_loader
-            .read_delete_file(
-                &file_scan_tasks[0].deletes[0],
-                file_scan_tasks[0].schema_ref(),
-            )
+            .read_delete_file(&delete_task, file_scan_tasks[0].schema_ref())
             .await
             .unwrap();
 

crates/iceberg/src/arrow/delete_filter.rs

@@ -24,7 +24,7 @@ use tokio::sync::oneshot::Receiver;
 use crate::delete_vector::DeleteVector;
 use crate::expr::Predicate::AlwaysTrue;
 use crate::expr::{Bind, BoundPredicate, Predicate};
-use crate::scan::{FileScanTask, FileScanTaskDeleteFile};
+use crate::scan::FileScanTask;
 use crate::spec::DataContentType;
 use crate::{Error, ErrorKind, Result};
 
@@ -122,14 +122,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.data_file_path())
                 .await
             else {
                 return Err(Error::new(
                     ErrorKind::Unexpected,
                     format!(
                         "Missing predicate for equality delete file '{}'",
-                        delete.file_path
+                        delete.data_file_path()
                     ),
                 ));
             };
@@ -192,8 +192,8 @@ impl DeleteFilter {
     }
 }
 
-pub(crate) fn is_equality_delete(f: &FileScanTaskDeleteFile) -> bool {
-    matches!(f.file_type, DataContentType::EqualityDeletes)
+pub(crate) fn is_equality_delete(f: &FileScanTask) -> bool {
+    matches!(f.data_file_content, DataContentType::EqualityDeletes)
 }
 
 #[cfg(test)]
@@ -309,25 +309,24 @@ pub(crate) mod tests {
             writer.close().unwrap();
         }
 
-        let pos_del_1 = FileScanTaskDeleteFile {
-            file_path: format!("{}/pos-del-1.parquet", table_location.to_str().unwrap()),
-            file_type: DataContentType::PositionDeletes,
-            partition_spec_id: 0,
-            equality_ids: None,
-        };
-
-        let pos_del_2 = FileScanTaskDeleteFile {
-            file_path: format!("{}/pos-del-2.parquet", table_location.to_str().unwrap()),
-            file_type: DataContentType::PositionDeletes,
-            partition_spec_id: 0,
-            equality_ids: None,
-        };
-
-        let pos_del_3 = FileScanTaskDeleteFile {
-            file_path: format!("{}/pos-del-3.parquet", table_location.to_str().unwrap()),
-            file_type: DataContentType::PositionDeletes,
-            partition_spec_id: 0,
+        // Helper to build a positional delete task with minimal fields
+        let make_pos_del_task = |n: u8| FileScanTask {
+            start: 0,
+            length: 0,
+            record_count: None,
+            data_file_path: format!("{}/pos-del-{}.parquet", table_location.to_str().unwrap(), n),
+            data_file_content: DataContentType::PositionDeletes,
+            data_file_format: DataFileFormat::Parquet,
+            schema: data_file_schema.clone(),
+            project_field_ids: vec![],
+            predicate: None,
+            deletes: vec![],
+            sequence_number: 0,
             equality_ids: None,
+            file_size_in_bytes: 0,
+            partition: None,
+            partition_spec: None,
+            name_mapping: None,
         };
 
         let file_scan_tasks = vec![
@@ -336,22 +335,36 @@ pub(crate) mod tests {
                 length: 0,
                 record_count: None,
                 data_file_path: format!("{}/1.parquet", table_location.to_str().unwrap()),
+                data_file_content: DataContentType::Data,
                 data_file_format: DataFileFormat::Parquet,
                 schema: data_file_schema.clone(),
                 project_field_ids: vec![],
                 predicate: None,
-                deletes: vec![pos_del_1, pos_del_2.clone()],
+                deletes: vec![make_pos_del_task(1).into(), make_pos_del_task(2).into()],
+                sequence_number: 0,
+                equality_ids: None,
+                file_size_in_bytes: 0,
+                partition: None,
+                partition_spec: None,
+                name_mapping: None,
             },
             FileScanTask {
                 start: 0,
                 length: 0,
                 record_count: None,
                 data_file_path: format!("{}/2.parquet", table_location.to_str().unwrap()),
+                data_file_content: DataContentType::Data,
                 data_file_format: DataFileFormat::Parquet,
                 schema: data_file_schema.clone(),
                 project_field_ids: vec![],
                 predicate: None,
-                deletes: vec![pos_del_3],
+                deletes: vec![make_pos_del_task(3).into()],
+                sequence_number: 0,
+                equality_ids: None,
+                file_size_in_bytes: 0,
+                partition: None,
+                partition_spec: None,
+                name_mapping: None,
             },
         ];