diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 7abbf2dddf..0ed5fd095c 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/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] diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 3b25bc3134..29132bb9ec 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/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>; -/// 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(_: &T, _: S) -> std::result::Result -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 -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, @@ -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, @@ -96,7 +80,6 @@ pub struct FileScanTask { /// The field ids to project. pub project_field_ids: Vec, /// The predicate to filter. - #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] pub predicate: Option, @@ -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, /// 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>, /// 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>, @@ -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>, @@ -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>, } @@ -282,3 +246,150 @@ pub struct FileScanTaskDeleteFile { #[builder(default)] pub key_metadata: Option>, } + +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); + + impl Serialize for PartitionSerde { + fn serialize(&self, serializer: S) -> std::result::Result + where S: serde::Serializer { + self.0 + .as_ref() + .map_err(serde::ser::Error::custom)? + .serialize(serializer) + } + } + + impl<'de> Deserialize<'de> for PartitionSerde { + fn deserialize(deserializer: D) -> std::result::Result + 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, + #[serde(skip_serializing_if = "Option::is_none")] + first_row_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + data_sequence_number: Option, + data_file_path: String, + data_file_format: DataFileFormat, + schema: SchemaRef, + project_field_ids: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + predicate: Option, + deletes: Vec, + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + partition: Option, + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + partition_spec: Option>, + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + name_mapping: Option>, + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + unified_partition_type: Option>, + case_sensitive: bool, + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + key_metadata: Option>, + } + + fn partition_type(schema: &Schema, partition_spec: Option<&PartitionSpec>) -> Result { + 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 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 for FileScanTask { + type Error = Error; + + fn try_from(value: FileScanTaskSerde) -> Result { + 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, + }) + } + } +}