feat(scan): make FileScanTask serializable

crates/iceberg/src/scan/mod.rs

@@ -662,13 +662,13 @@ pub mod tests {
         RESERVED_COL_NAME_POS, RESERVED_COL_NAME_SPEC_ID, RESERVED_FIELD_ID_DELETE_FILE_PATH,
         RESERVED_FIELD_ID_DELETE_FILE_POS, RESERVED_FIELD_ID_POS,
     };
-    use crate::scan::FileScanTask;
+    use crate::scan::{FileScanTask, FileScanTaskDeleteFile};
     use crate::spec::{
         DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, DataFileBuilder, DataFileFormat, Datum,
         FormatVersion, Literal, MAIN_BRANCH, ManifestEntry, ManifestListWriter, ManifestStatus,
-        ManifestWriterBuilder, NestedField, Operation, PartitionSpec, PrimitiveType, Schema,
-        Snapshot, Struct, StructType, Summary, TableMetadata, TableMetadataBuilder, Type,
-        UnboundPartitionSpec,
+        ManifestWriterBuilder, MappedField, NameMapping, NestedField, Operation, PartitionSpec,
+        PrimitiveType, Schema, Snapshot, Struct, StructType, Summary, TableMetadata,
+        TableMetadataBuilder, Transform, Type, UnboundPartitionSpec,
     };
     use crate::table::Table;
     use crate::test_utils::test_runtime;
@@ -2650,18 +2650,12 @@ pub mod tests {
 
     #[test]
     fn test_file_scan_task_serialize_deserialize() {
+        // Regression test for https://github.com/apache/iceberg-rust/issues/3089.
         let test_fn = |task: FileScanTask| {
             let serialized = serde_json::to_string(&task).unwrap();
             let deserialized: FileScanTask = serde_json::from_str(&serialized).unwrap();
 
-            assert_eq!(task.data_file_path, deserialized.data_file_path);
-            assert_eq!(task.start, deserialized.start);
-            assert_eq!(task.length, deserialized.length);
-            assert_eq!(task.project_field_ids, deserialized.project_field_ids);
-            assert_eq!(task.predicate, deserialized.predicate);
-            assert_eq!(task.schema, deserialized.schema);
-            assert_eq!(task.first_row_id, deserialized.first_row_id);
-            assert_eq!(task.data_sequence_number, deserialized.data_sequence_number);
+            assert_eq!(task, deserialized);
         };
 
         // without predicate
@@ -2703,6 +2697,60 @@ pub mod tests {
             .with_case_sensitive(false)
             .build();
         test_fn(task);
+
+        // with every optional scan context field populated
+        let schema = Arc::new(
+            Schema::builder()
+                .with_fields(vec![Arc::new(NestedField::required(
+                    1,
+                    "x",
+                    Type::Primitive(PrimitiveType::Long),
+                ))])
+                .build()
+                .unwrap(),
+        );
+        let partition_spec = Arc::new(
+            PartitionSpec::builder(schema.clone())
+                .add_partition_field("x", "x", Transform::Identity)
+                .unwrap()
+                .build()
+                .unwrap(),
+        );
+        let unified_partition_type = Arc::new(partition_spec.partition_type(&schema).unwrap());
+        let task = FileScanTask::builder()
+            .with_data_file_path("data_file_path".to_string())
+            .with_file_size_in_bytes(123)
+            .with_start(10)
+            .with_length(100)
+            .with_project_field_ids(vec![1])
+            .with_schema(schema)
+            .with_data_file_format(DataFileFormat::Parquet)
+            .with_deletes(vec![
+                FileScanTaskDeleteFile::builder()
+                    .with_file_path("delete_file_path".to_string())
+                    .with_file_size_in_bytes(23)
+                    .with_file_type(DataContentType::EqualityDeletes)
+                    .with_partition_spec_id(0)
+                    .with_equality_ids(Some(vec![1]))
+                    .with_referenced_data_file(Some("data_file_path".to_string()))
+                    .with_content_offset(Some(12))
+                    .with_content_size_in_bytes(Some(34))
+                    .with_record_count(Some(5))
+                    .with_key_metadata(Some(vec![4, 5, 6].into_boxed_slice()))
+                    .build(),
+            ])
+            .with_partition(Some(Struct::from_iter([Some(Literal::long(42))])))
+            .with_partition_spec(Some(partition_spec))
+            .with_name_mapping(Some(Arc::new(NameMapping::new(vec![MappedField::new(
+                Some(1),
+                vec!["x".to_string()],
+                vec![],
+            )]))))
+            .with_unified_partition_type(Some(unified_partition_type))
+            .with_case_sensitive(true)
+            .with_key_metadata(Some(vec![1, 2, 3].into_boxed_slice()))
+            .build();
+        test_fn(task);
     }
 
     #[tokio::test]

crates/iceberg/src/scan/task.rs

@@ -18,39 +18,25 @@
 use std::sync::Arc;
 
 use futures::stream::BoxStream;
-use serde::{Deserialize, Serialize, Serializer};
+use serde::{Deserialize, Serialize};
 use typed_builder::TypedBuilder;
 
 use crate::Result;
 use crate::expr::BoundPredicate;
 use crate::spec::{
-    DataContentType, DataFileFormat, ManifestEntryRef, NameMapping, PartitionSpec, Schema,
-    SchemaRef, Struct, StructType,
+    DataContentType, DataFileFormat, Literal, ManifestEntryRef, NameMapping, PartitionSpec,
+    RawLiteral, Schema, SchemaRef, Struct, StructType, Type,
 };
 
 /// A stream of [`FileScanTask`].
 pub type FileScanTaskStream = BoxStream<'static, Result<FileScanTask>>;
 
-/// Serialization helper that always returns NotImplementedError.
-/// Used for fields that should not be serialized but we want to be explicit about it.
-fn serialize_not_implemented<S, T>(_: &T, _: S) -> std::result::Result<S::Ok, S::Error>
-where S: Serializer {
-    Err(serde::ser::Error::custom(
-        "Serialization not implemented for this field",
-    ))
-}
-
-/// Deserialization helper that always returns NotImplementedError.
-/// Used for fields that should not be deserialized but we want to be explicit about it.
-fn deserialize_not_implemented<'de, D, T>(_: D) -> std::result::Result<T, D::Error>
-where D: serde::Deserializer<'de> {
-    Err(serde::de::Error::custom(
-        "Deserialization not implemented for this field",
-    ))
-}
-
 /// A task to scan part of file.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, TypedBuilder)]
+#[serde(
+    into = "crate::scan::task::_serde::FileScanTaskSerde",
+    try_from = "crate::scan::task::_serde::FileScanTaskSerde"
+)]
 #[builder(field_defaults(setter(prefix = "with_")))]
 pub struct FileScanTask {
     /// The total size of the data file in bytes, from the manifest entry.
@@ -71,7 +57,6 @@ pub struct FileScanTask {
     ///
     /// Used to derive the `_row_id` metadata column: for a row without an
     /// explicit `_row_id`, it is this value plus the row's ordinal position.
-    #[serde(skip_serializing_if = "Option::is_none")]
     #[builder(default)]
     pub first_row_id: Option<i64>,
 
@@ -81,7 +66,6 @@ pub struct FileScanTask {
     /// manifest that lacks one.
     ///
     /// Used to derive the `_last_updated_sequence_number` metadata column.
-    #[serde(skip_serializing_if = "Option::is_none")]
     #[builder(default)]
     pub data_sequence_number: Option<i64>,
 
@@ -96,7 +80,6 @@ pub struct FileScanTask {
     /// The field ids to project.
     pub project_field_ids: Vec<i32>,
     /// The predicate to filter.
-    #[serde(skip_serializing_if = "Option::is_none")]
     #[builder(default)]
     pub predicate: Option<BoundPredicate>,
 
@@ -107,30 +90,18 @@ pub struct FileScanTask {
     /// Partition data from the manifest entry, used to identify which columns can use
     /// constant values from partition metadata vs. reading from the data file.
     /// Per the Iceberg spec, only identity-transformed partition fields should use constants.
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[serde(serialize_with = "serialize_not_implemented")]
-    #[serde(deserialize_with = "deserialize_not_implemented")]
     #[builder(default)]
     pub partition: Option<Struct>,
 
     /// The partition spec for this file, used to distinguish identity transforms
     /// (which use partition metadata constants) from non-identity transforms like
     /// bucket/truncate (which must read source columns from the data file).
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[serde(serialize_with = "serialize_not_implemented")]
-    #[serde(deserialize_with = "deserialize_not_implemented")]
     #[builder(default)]
     pub partition_spec: Option<Arc<PartitionSpec>>,
 
     /// Name mapping from table metadata (property: schema.name-mapping.default),
     /// used to resolve field IDs from column names when Parquet files lack field IDs
     /// or have field ID conflicts.
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[serde(serialize_with = "serialize_not_implemented")]
-    #[serde(deserialize_with = "deserialize_not_implemented")]
     #[builder(default)]
     pub name_mapping: Option<Arc<NameMapping>>,
 
@@ -142,11 +113,6 @@ pub struct FileScanTask {
     /// This is a table-level value (same for all tasks in a scan), stored per-task
     /// so that readers are self-contained without needing back-pointers to table
     /// metadata. The cost is one Arc clone per task.
-    /// Serde: not yet implemented (same pattern as partition, partition_spec, name_mapping).
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[serde(serialize_with = "serialize_not_implemented")]
-    #[serde(deserialize_with = "deserialize_not_implemented")]
     #[builder(default)]
     pub unified_partition_type: Option<Arc<StructType>>,
 
@@ -158,11 +124,9 @@ pub struct FileScanTask {
     ///
     /// Note on the trust boundary: for the standard encryption scheme this
     /// carries `StandardKeyMetadata`, whose payload is the *plaintext* DEK.
-    /// Because `FileScanTask` derives `Serialize`, that plaintext DEK is part
+    /// Because `FileScanTask` implements [`Serialize`], that plaintext DEK is part
     /// of the serialized scan plan should these tasks ever be serialized and sent
     /// over the network.
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
     #[builder(default)]
     pub key_metadata: Option<Box<[u8]>>,
 }
@@ -282,3 +246,150 @@ pub struct FileScanTaskDeleteFile {
     #[builder(default)]
     pub key_metadata: Option<Box<[u8]>>,
 }
+
+mod _serde {
+    use serde_derive::{Deserialize as DeserializeDerive, Serialize as SerializeDerive};
+
+    use super::*;
+    use crate::{Error, ErrorKind};
+
+    // Container-level `into` conversion is infallible. Keep a failed typed conversion here so
+    // serialization can return that error instead of panicking or changing the wire format.
+    struct PartitionSerde(Result<RawLiteral>);
+
+    impl Serialize for PartitionSerde {
+        fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+        where S: serde::Serializer {
+            self.0
+                .as_ref()
+                .map_err(serde::ser::Error::custom)?
+                .serialize(serializer)
+        }
+    }
+
+    impl<'de> Deserialize<'de> for PartitionSerde {
+        fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+        where D: serde::Deserializer<'de> {
+            Ok(Self(Ok(RawLiteral::deserialize(deserializer)?)))
+        }
+    }
+
+    #[derive(SerializeDerive, DeserializeDerive)]
+    pub(super) struct FileScanTaskSerde {
+        file_size_in_bytes: u64,
+        start: u64,
+        length: u64,
+        record_count: Option<u64>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        first_row_id: Option<i64>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        data_sequence_number: Option<i64>,
+        data_file_path: String,
+        data_file_format: DataFileFormat,
+        schema: SchemaRef,
+        project_field_ids: Vec<i32>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        predicate: Option<BoundPredicate>,
+        deletes: Vec<FileScanTaskDeleteFile>,
+        #[serde(default)]
+        #[serde(skip_serializing_if = "Option::is_none")]
+        partition: Option<PartitionSerde>,
+        #[serde(default)]
+        #[serde(skip_serializing_if = "Option::is_none")]
+        partition_spec: Option<Arc<PartitionSpec>>,
+        #[serde(default)]
+        #[serde(skip_serializing_if = "Option::is_none")]
+        name_mapping: Option<Arc<NameMapping>>,
+        #[serde(default)]
+        #[serde(skip_serializing_if = "Option::is_none")]
+        unified_partition_type: Option<Arc<StructType>>,
+        case_sensitive: bool,
+        #[serde(default)]
+        #[serde(skip_serializing_if = "Option::is_none")]
+        key_metadata: Option<Box<[u8]>>,
+    }
+
+    fn partition_type(schema: &Schema, partition_spec: Option<&PartitionSpec>) -> Result<Type> {
+        let partition_spec = partition_spec.ok_or_else(|| {
+            Error::new(
+                ErrorKind::DataInvalid,
+                "FileScanTask partition requires a partition spec",
+            )
+        })?;
+        Ok(Type::Struct(partition_spec.partition_type(schema)?))
+    }
+
+    impl From<FileScanTask> for FileScanTaskSerde {
+        fn from(value: FileScanTask) -> Self {
+            let partition = value.partition.map(|partition| {
+                PartitionSerde(
+                    partition_type(&value.schema, value.partition_spec.as_deref())
+                        .and_then(|ty| RawLiteral::try_from(Literal::Struct(partition), &ty)),
+                )
+            });
+
+            Self {
+                file_size_in_bytes: value.file_size_in_bytes,
+                start: value.start,
+                length: value.length,
+                record_count: value.record_count,
+                first_row_id: value.first_row_id,
+                data_sequence_number: value.data_sequence_number,
+                data_file_path: value.data_file_path,
+                data_file_format: value.data_file_format,
+                schema: value.schema,
+                project_field_ids: value.project_field_ids,
+                predicate: value.predicate,
+                deletes: value.deletes,
+                partition,
+                partition_spec: value.partition_spec,
+                name_mapping: value.name_mapping,
+                unified_partition_type: value.unified_partition_type,
+                case_sensitive: value.case_sensitive,
+                key_metadata: value.key_metadata,
+            }
+        }
+    }
+
+    impl TryFrom<FileScanTaskSerde> for FileScanTask {
+        type Error = Error;
+
+        fn try_from(value: FileScanTaskSerde) -> Result<Self> {
+            let partition = value
+                .partition
+                .map(|partition| {
+                    let partition_type =
+                        partition_type(&value.schema, value.partition_spec.as_deref())?;
+                    match partition.0?.try_into(&partition_type)? {
+                        Some(Literal::Struct(partition)) => Ok(partition),
+                        _ => Err(Error::new(
+                            ErrorKind::DataInvalid,
+                            "FileScanTask partition must be a struct",
+                        )),
+                    }
+                })
+                .transpose()?;
+
+            Ok(Self {
+                file_size_in_bytes: value.file_size_in_bytes,
+                start: value.start,
+                length: value.length,
+                record_count: value.record_count,
+                first_row_id: value.first_row_id,
+                data_sequence_number: value.data_sequence_number,
+                data_file_path: value.data_file_path,
+                data_file_format: value.data_file_format,
+                schema: value.schema,
+                project_field_ids: value.project_field_ids,
+                predicate: value.predicate,
+                deletes: value.deletes,
+                partition,
+                partition_spec: value.partition_spec,
+                name_mapping: value.name_mapping,
+                unified_partition_type: value.unified_partition_type,
+                case_sensitive: value.case_sensitive,
+                key_metadata: value.key_metadata,
+            })
+        }
+    }
+}