From 583a78e215f3a7b101aaadf05f95b0155f9dc19c Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Fri, 28 Aug 2026 16:24:58 +0800 Subject: [PATCH 1/5] feat(scan): make FileScanTask serializable Replace the placeholder serde errors with a lossless representation for partition literals and use the existing serde implementations for the remaining scan context fields. Expand round-trip coverage to include a fully populated task and every primitive partition literal. [apache/iceberg-rust#3089](https://github.com/apache/iceberg-rust/issues/3089) --- crates/iceberg/src/scan/mod.rs | 72 +++++++++-- crates/iceberg/src/scan/task.rs | 206 +++++++++++++++++++++++++++----- 2 files changed, 239 insertions(+), 39 deletions(-) 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..2c3fd39a7a 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -18,35 +18,148 @@ 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, + PrimitiveLiteral, Schema, SchemaRef, Struct, StructType, }; /// 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", - )) -} +mod partition_serde { + use std::result::Result as StdResult; + + use serde::ser::Error as _; + use serde::{Deserialize, Deserializer, Serialize, Serializer}; + + use super::{Literal, PrimitiveLiteral, Struct}; + + /// A self-describing representation for the physical values in a partition struct. + /// + /// The logical types are carried separately by the task's schema and partition spec. Keeping + /// this representation physical avoids duplicating that context while still distinguishing + /// values such as `Int` and `Long`. Floats and 128-bit integers use byte representations so + /// that every value can round-trip through JSON, including NaNs and infinities. + #[derive(Serialize, Deserialize)] + #[serde(tag = "type", content = "value", rename_all = "kebab-case")] + enum SerializablePrimitiveLiteral { + Boolean(bool), + Int(i32), + Long(i64), + Float([u8; 4]), + Double([u8; 8]), + String(String), + Binary(Vec), + Int128([u8; 16]), + UInt128([u8; 16]), + AboveMax, + BelowMin, + } + + impl TryFrom<&Literal> for SerializablePrimitiveLiteral { + type Error = &'static str; + + fn try_from(value: &Literal) -> StdResult { + match value { + Literal::Primitive(PrimitiveLiteral::Boolean(value)) => Ok(Self::Boolean(*value)), + Literal::Primitive(PrimitiveLiteral::Int(value)) => Ok(Self::Int(*value)), + Literal::Primitive(PrimitiveLiteral::Long(value)) => Ok(Self::Long(*value)), + Literal::Primitive(PrimitiveLiteral::Float(value)) => { + Ok(Self::Float(value.to_bits().to_be_bytes())) + } + Literal::Primitive(PrimitiveLiteral::Double(value)) => { + Ok(Self::Double(value.to_bits().to_be_bytes())) + } + Literal::Primitive(PrimitiveLiteral::String(value)) => { + Ok(Self::String(value.clone())) + } + Literal::Primitive(PrimitiveLiteral::Binary(value)) => { + Ok(Self::Binary(value.clone())) + } + Literal::Primitive(PrimitiveLiteral::Int128(value)) => { + Ok(Self::Int128(value.to_be_bytes())) + } + Literal::Primitive(PrimitiveLiteral::UInt128(value)) => { + Ok(Self::UInt128(value.to_be_bytes())) + } + Literal::Primitive(PrimitiveLiteral::AboveMax) => Ok(Self::AboveMax), + Literal::Primitive(PrimitiveLiteral::BelowMin) => Ok(Self::BelowMin), + Literal::Struct(_) | Literal::List(_) | Literal::Map(_) => { + Err("partition structs can contain only primitive literal values") + } + } + } + } + + impl From for Literal { + fn from(value: SerializablePrimitiveLiteral) -> Self { + let value = match value { + SerializablePrimitiveLiteral::Boolean(value) => PrimitiveLiteral::Boolean(value), + SerializablePrimitiveLiteral::Int(value) => PrimitiveLiteral::Int(value), + SerializablePrimitiveLiteral::Long(value) => PrimitiveLiteral::Long(value), + SerializablePrimitiveLiteral::Float(value) => { + PrimitiveLiteral::Float(f32::from_bits(u32::from_be_bytes(value)).into()) + } + SerializablePrimitiveLiteral::Double(value) => { + PrimitiveLiteral::Double(f64::from_bits(u64::from_be_bytes(value)).into()) + } + SerializablePrimitiveLiteral::String(value) => PrimitiveLiteral::String(value), + SerializablePrimitiveLiteral::Binary(value) => PrimitiveLiteral::Binary(value), + SerializablePrimitiveLiteral::Int128(value) => { + PrimitiveLiteral::Int128(i128::from_be_bytes(value)) + } + SerializablePrimitiveLiteral::UInt128(value) => { + PrimitiveLiteral::UInt128(u128::from_be_bytes(value)) + } + SerializablePrimitiveLiteral::AboveMax => PrimitiveLiteral::AboveMax, + SerializablePrimitiveLiteral::BelowMin => PrimitiveLiteral::BelowMin, + }; + Literal::Primitive(value) + } + } -/// 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", - )) + pub(super) fn serialize( + partition: &Option, + serializer: S, + ) -> StdResult + where + S: Serializer, + { + let partition = partition + .as_ref() + .map(|partition| { + partition + .iter() + .map(|value| { + value + .map(SerializablePrimitiveLiteral::try_from) + .transpose() + }) + .collect::, _>>() + }) + .transpose() + .map_err(S::Error::custom)?; + + partition.serialize(serializer) + } + + pub(super) fn deserialize<'de, D>(deserializer: D) -> StdResult, D::Error> + where D: Deserializer<'de> { + let partition = + Option::>>::deserialize(deserializer)?; + + Ok(partition.map(|partition| { + partition + .into_iter() + .map(|value| value.map(Literal::from)) + .collect() + })) + } } /// A task to scan part of file. @@ -109,8 +222,7 @@ pub struct FileScanTask { /// 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")] + #[serde(with = "partition_serde")] #[builder(default)] pub partition: Option, @@ -119,8 +231,6 @@ pub struct FileScanTask { /// 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>, @@ -129,8 +239,6 @@ pub struct FileScanTask { /// 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 +250,8 @@ 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>, @@ -282,3 +387,50 @@ pub struct FileScanTaskDeleteFile { #[builder(default)] pub key_metadata: Option>, } + +#[cfg(test)] +mod tests { + use serde::{Deserialize, Serialize}; + + use super::*; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct PartitionWrapper { + #[serde(with = "partition_serde")] + partition: Option, + } + + #[test] + fn test_partition_serde_round_trip_all_primitive_literals() { + let partition = Struct::from_iter([ + Some(Literal::Primitive(PrimitiveLiteral::Boolean(true))), + Some(Literal::Primitive(PrimitiveLiteral::Int(i32::MIN))), + Some(Literal::Primitive(PrimitiveLiteral::Long(i64::MAX))), + Some(Literal::Primitive(PrimitiveLiteral::Float( + f32::INFINITY.into(), + ))), + Some(Literal::Primitive(PrimitiveLiteral::Double( + f64::NEG_INFINITY.into(), + ))), + Some(Literal::Primitive(PrimitiveLiteral::String( + "partition".to_string(), + ))), + Some(Literal::Primitive(PrimitiveLiteral::Binary(vec![ + 0, 1, 255, + ]))), + Some(Literal::Primitive(PrimitiveLiteral::Int128(i128::MIN))), + Some(Literal::Primitive(PrimitiveLiteral::UInt128(u128::MAX))), + Some(Literal::Primitive(PrimitiveLiteral::AboveMax)), + Some(Literal::Primitive(PrimitiveLiteral::BelowMin)), + None, + ]); + let expected = PartitionWrapper { + partition: Some(partition), + }; + + let serialized = serde_json::to_string(&expected).unwrap(); + let actual = serde_json::from_str(&serialized).unwrap(); + + assert_eq!(expected, actual); + } +} From a1b2fce7da58bb90c95e6a6db0377863c29b842b Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Fri, 28 Aug 2026 17:01:51 +0800 Subject: [PATCH 2/5] refactor(spec): serialize Literal directly Move the self-describing serde representation into Literal so scan task partitions use the shared values-layer implementation. Cover primitive and nested literal round trips and remove the task-local helper requested in review. --- crates/iceberg/src/scan/task.rs | 183 +----------------- crates/iceberg/src/spec/values/literal.rs | 128 ++++++++++++ crates/iceberg/src/spec/values/map.rs | 4 + .../iceberg/src/spec/values/struct_value.rs | 18 ++ crates/iceberg/src/spec/values/tests.rs | 36 ++++ 5 files changed, 188 insertions(+), 181 deletions(-) diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 2c3fd39a7a..2a93ecd69b 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -24,144 +24,13 @@ use typed_builder::TypedBuilder; use crate::Result; use crate::expr::BoundPredicate; use crate::spec::{ - DataContentType, DataFileFormat, Literal, ManifestEntryRef, NameMapping, PartitionSpec, - PrimitiveLiteral, Schema, SchemaRef, Struct, StructType, + DataContentType, DataFileFormat, ManifestEntryRef, NameMapping, PartitionSpec, Schema, + SchemaRef, Struct, StructType, }; /// A stream of [`FileScanTask`]. pub type FileScanTaskStream = BoxStream<'static, Result>; -mod partition_serde { - use std::result::Result as StdResult; - - use serde::ser::Error as _; - use serde::{Deserialize, Deserializer, Serialize, Serializer}; - - use super::{Literal, PrimitiveLiteral, Struct}; - - /// A self-describing representation for the physical values in a partition struct. - /// - /// The logical types are carried separately by the task's schema and partition spec. Keeping - /// this representation physical avoids duplicating that context while still distinguishing - /// values such as `Int` and `Long`. Floats and 128-bit integers use byte representations so - /// that every value can round-trip through JSON, including NaNs and infinities. - #[derive(Serialize, Deserialize)] - #[serde(tag = "type", content = "value", rename_all = "kebab-case")] - enum SerializablePrimitiveLiteral { - Boolean(bool), - Int(i32), - Long(i64), - Float([u8; 4]), - Double([u8; 8]), - String(String), - Binary(Vec), - Int128([u8; 16]), - UInt128([u8; 16]), - AboveMax, - BelowMin, - } - - impl TryFrom<&Literal> for SerializablePrimitiveLiteral { - type Error = &'static str; - - fn try_from(value: &Literal) -> StdResult { - match value { - Literal::Primitive(PrimitiveLiteral::Boolean(value)) => Ok(Self::Boolean(*value)), - Literal::Primitive(PrimitiveLiteral::Int(value)) => Ok(Self::Int(*value)), - Literal::Primitive(PrimitiveLiteral::Long(value)) => Ok(Self::Long(*value)), - Literal::Primitive(PrimitiveLiteral::Float(value)) => { - Ok(Self::Float(value.to_bits().to_be_bytes())) - } - Literal::Primitive(PrimitiveLiteral::Double(value)) => { - Ok(Self::Double(value.to_bits().to_be_bytes())) - } - Literal::Primitive(PrimitiveLiteral::String(value)) => { - Ok(Self::String(value.clone())) - } - Literal::Primitive(PrimitiveLiteral::Binary(value)) => { - Ok(Self::Binary(value.clone())) - } - Literal::Primitive(PrimitiveLiteral::Int128(value)) => { - Ok(Self::Int128(value.to_be_bytes())) - } - Literal::Primitive(PrimitiveLiteral::UInt128(value)) => { - Ok(Self::UInt128(value.to_be_bytes())) - } - Literal::Primitive(PrimitiveLiteral::AboveMax) => Ok(Self::AboveMax), - Literal::Primitive(PrimitiveLiteral::BelowMin) => Ok(Self::BelowMin), - Literal::Struct(_) | Literal::List(_) | Literal::Map(_) => { - Err("partition structs can contain only primitive literal values") - } - } - } - } - - impl From for Literal { - fn from(value: SerializablePrimitiveLiteral) -> Self { - let value = match value { - SerializablePrimitiveLiteral::Boolean(value) => PrimitiveLiteral::Boolean(value), - SerializablePrimitiveLiteral::Int(value) => PrimitiveLiteral::Int(value), - SerializablePrimitiveLiteral::Long(value) => PrimitiveLiteral::Long(value), - SerializablePrimitiveLiteral::Float(value) => { - PrimitiveLiteral::Float(f32::from_bits(u32::from_be_bytes(value)).into()) - } - SerializablePrimitiveLiteral::Double(value) => { - PrimitiveLiteral::Double(f64::from_bits(u64::from_be_bytes(value)).into()) - } - SerializablePrimitiveLiteral::String(value) => PrimitiveLiteral::String(value), - SerializablePrimitiveLiteral::Binary(value) => PrimitiveLiteral::Binary(value), - SerializablePrimitiveLiteral::Int128(value) => { - PrimitiveLiteral::Int128(i128::from_be_bytes(value)) - } - SerializablePrimitiveLiteral::UInt128(value) => { - PrimitiveLiteral::UInt128(u128::from_be_bytes(value)) - } - SerializablePrimitiveLiteral::AboveMax => PrimitiveLiteral::AboveMax, - SerializablePrimitiveLiteral::BelowMin => PrimitiveLiteral::BelowMin, - }; - Literal::Primitive(value) - } - } - - pub(super) fn serialize( - partition: &Option, - serializer: S, - ) -> StdResult - where - S: Serializer, - { - let partition = partition - .as_ref() - .map(|partition| { - partition - .iter() - .map(|value| { - value - .map(SerializablePrimitiveLiteral::try_from) - .transpose() - }) - .collect::, _>>() - }) - .transpose() - .map_err(S::Error::custom)?; - - partition.serialize(serializer) - } - - pub(super) fn deserialize<'de, D>(deserializer: D) -> StdResult, D::Error> - where D: Deserializer<'de> { - let partition = - Option::>>::deserialize(deserializer)?; - - Ok(partition.map(|partition| { - partition - .into_iter() - .map(|value| value.map(Literal::from)) - .collect() - })) - } -} - /// A task to scan part of file. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, TypedBuilder)] #[builder(field_defaults(setter(prefix = "with_")))] @@ -222,7 +91,6 @@ pub struct FileScanTask { /// Per the Iceberg spec, only identity-transformed partition fields should use constants. #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] - #[serde(with = "partition_serde")] #[builder(default)] pub partition: Option, @@ -387,50 +255,3 @@ pub struct FileScanTaskDeleteFile { #[builder(default)] pub key_metadata: Option>, } - -#[cfg(test)] -mod tests { - use serde::{Deserialize, Serialize}; - - use super::*; - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - struct PartitionWrapper { - #[serde(with = "partition_serde")] - partition: Option, - } - - #[test] - fn test_partition_serde_round_trip_all_primitive_literals() { - let partition = Struct::from_iter([ - Some(Literal::Primitive(PrimitiveLiteral::Boolean(true))), - Some(Literal::Primitive(PrimitiveLiteral::Int(i32::MIN))), - Some(Literal::Primitive(PrimitiveLiteral::Long(i64::MAX))), - Some(Literal::Primitive(PrimitiveLiteral::Float( - f32::INFINITY.into(), - ))), - Some(Literal::Primitive(PrimitiveLiteral::Double( - f64::NEG_INFINITY.into(), - ))), - Some(Literal::Primitive(PrimitiveLiteral::String( - "partition".to_string(), - ))), - Some(Literal::Primitive(PrimitiveLiteral::Binary(vec![ - 0, 1, 255, - ]))), - Some(Literal::Primitive(PrimitiveLiteral::Int128(i128::MIN))), - Some(Literal::Primitive(PrimitiveLiteral::UInt128(u128::MAX))), - Some(Literal::Primitive(PrimitiveLiteral::AboveMax)), - Some(Literal::Primitive(PrimitiveLiteral::BelowMin)), - None, - ]); - let expected = PartitionWrapper { - partition: Some(partition), - }; - - let serialized = serde_json::to_string(&expected).unwrap(); - let actual = serde_json::from_str(&serialized).unwrap(); - - assert_eq!(expected, actual); - } -} diff --git a/crates/iceberg/src/spec/values/literal.rs b/crates/iceberg/src/spec/values/literal.rs index 5296eff2a2..0db3a0e3a2 100644 --- a/crates/iceberg/src/spec/values/literal.rs +++ b/crates/iceberg/src/spec/values/literal.rs @@ -22,6 +22,8 @@ use std::str::FromStr; use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, TimeZone, Utc}; use ordered_float::OrderedFloat; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde_derive::{Deserialize as DeserializeDerive, Serialize as SerializeDerive}; use serde_json::{Map as JsonMap, Number, Value as JsonValue}; use uuid::Uuid; @@ -55,6 +57,132 @@ pub enum Literal { Map(Map), } +#[derive(SerializeDerive, DeserializeDerive)] +#[serde(tag = "type", content = "value", rename_all = "kebab-case")] +enum SerializableLiteral { + Boolean(bool), + Int(i32), + Long(i64), + Float([u8; 4]), + Double([u8; 8]), + String(String), + Binary(Vec), + Int128([u8; 16]), + UInt128([u8; 16]), + AboveMax, + BelowMin, + Struct(Vec>), + List(Vec>), + Map(Vec<(SerializableLiteral, Option)>), +} + +impl From<&Literal> for SerializableLiteral { + fn from(value: &Literal) -> Self { + match value { + Literal::Primitive(PrimitiveLiteral::Boolean(value)) => Self::Boolean(*value), + Literal::Primitive(PrimitiveLiteral::Int(value)) => Self::Int(*value), + Literal::Primitive(PrimitiveLiteral::Long(value)) => Self::Long(*value), + Literal::Primitive(PrimitiveLiteral::Float(value)) => { + Self::Float(value.to_bits().to_be_bytes()) + } + Literal::Primitive(PrimitiveLiteral::Double(value)) => { + Self::Double(value.to_bits().to_be_bytes()) + } + Literal::Primitive(PrimitiveLiteral::String(value)) => Self::String(value.clone()), + Literal::Primitive(PrimitiveLiteral::Binary(value)) => Self::Binary(value.clone()), + Literal::Primitive(PrimitiveLiteral::Int128(value)) => { + Self::Int128(value.to_be_bytes()) + } + Literal::Primitive(PrimitiveLiteral::UInt128(value)) => { + Self::UInt128(value.to_be_bytes()) + } + Literal::Primitive(PrimitiveLiteral::AboveMax) => Self::AboveMax, + Literal::Primitive(PrimitiveLiteral::BelowMin) => Self::BelowMin, + Literal::Struct(value) => Self::Struct( + value + .iter() + .map(|value| value.map(SerializableLiteral::from)) + .collect(), + ), + Literal::List(value) => Self::List( + value + .iter() + .map(|value| value.as_ref().map(SerializableLiteral::from)) + .collect(), + ), + Literal::Map(value) => Self::Map( + value + .iter() + .map(|(key, value)| { + ( + SerializableLiteral::from(key), + value.map(SerializableLiteral::from), + ) + }) + .collect(), + ), + } + } +} + +impl From for Literal { + fn from(value: SerializableLiteral) -> Self { + match value { + SerializableLiteral::Boolean(value) => Self::bool(value), + SerializableLiteral::Int(value) => Self::int(value), + SerializableLiteral::Long(value) => Self::long(value), + SerializableLiteral::Float(value) => { + Self::float(f32::from_bits(u32::from_be_bytes(value))) + } + SerializableLiteral::Double(value) => { + Self::double(f64::from_bits(u64::from_be_bytes(value))) + } + SerializableLiteral::String(value) => Self::string(value), + SerializableLiteral::Binary(value) => Self::binary(value), + SerializableLiteral::Int128(value) => { + Self::Primitive(PrimitiveLiteral::Int128(i128::from_be_bytes(value))) + } + SerializableLiteral::UInt128(value) => { + Self::Primitive(PrimitiveLiteral::UInt128(u128::from_be_bytes(value))) + } + SerializableLiteral::AboveMax => Self::Primitive(PrimitiveLiteral::AboveMax), + SerializableLiteral::BelowMin => Self::Primitive(PrimitiveLiteral::BelowMin), + SerializableLiteral::Struct(value) => Self::Struct( + value + .into_iter() + .map(|value| value.map(Literal::from)) + .collect(), + ), + SerializableLiteral::List(value) => Self::List( + value + .into_iter() + .map(|value| value.map(Literal::from)) + .collect(), + ), + SerializableLiteral::Map(value) => Self::Map( + value + .into_iter() + .map(|(key, value)| (Literal::from(key), value.map(Literal::from))) + .collect(), + ), + } + } +} + +impl Serialize for Literal { + fn serialize(&self, serializer: S) -> std::result::Result + where S: Serializer { + SerializableLiteral::from(self).serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for Literal { + fn deserialize(deserializer: D) -> std::result::Result + where D: Deserializer<'de> { + Ok(SerializableLiteral::deserialize(deserializer)?.into()) + } +} + impl Literal { /// Creates a boolean value. /// diff --git a/crates/iceberg/src/spec/values/map.rs b/crates/iceberg/src/spec/values/map.rs index e0f75205f0..49021dd2a6 100644 --- a/crates/iceberg/src/spec/values/map.rs +++ b/crates/iceberg/src/spec/values/map.rs @@ -67,6 +67,10 @@ impl Map { self.pair.is_empty() } + pub(crate) fn iter(&self) -> impl ExactSizeIterator)> { + self.pair.iter().map(|(key, value)| (key, value.as_ref())) + } + /// Inserts a key-value pair into the map. /// If the map did not have this key present, None is returned. /// If the map did have this key present, the value is updated, and the old value is returned. diff --git a/crates/iceberg/src/spec/values/struct_value.rs b/crates/iceberg/src/spec/values/struct_value.rs index 2b44961edf..04f05a05b7 100644 --- a/crates/iceberg/src/spec/values/struct_value.rs +++ b/crates/iceberg/src/spec/values/struct_value.rs @@ -19,6 +19,8 @@ use std::ops::Index; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; + use super::Literal; /// The partition struct stores the tuple of partition values for each file. @@ -30,6 +32,22 @@ pub struct Struct { fields: Vec>, } +impl Serialize for Struct { + fn serialize(&self, serializer: S) -> Result + where S: Serializer { + self.fields.serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for Struct { + fn deserialize(deserializer: D) -> Result + where D: Deserializer<'de> { + Ok(Self { + fields: Vec::>::deserialize(deserializer)?, + }) + } +} + impl Struct { /// Create a empty struct. pub fn empty() -> Self { diff --git a/crates/iceberg/src/spec/values/tests.rs b/crates/iceberg/src/spec/values/tests.rs index 5ee4ac8d02..00ee3d7c91 100644 --- a/crates/iceberg/src/spec/values/tests.rs +++ b/crates/iceberg/src/spec/values/tests.rs @@ -65,6 +65,42 @@ fn check_avro_bytes_serde(input: Vec, expected_datum: Datum, expected_type: } } +#[test] +fn test_literal_serde_round_trip() { + let expected = Literal::Struct(Struct::from_iter([ + Some(Literal::Primitive(PrimitiveLiteral::Boolean(true))), + Some(Literal::Primitive(PrimitiveLiteral::Int(i32::MIN))), + Some(Literal::Primitive(PrimitiveLiteral::Long(i64::MAX))), + Some(Literal::Primitive(PrimitiveLiteral::Float( + f32::INFINITY.into(), + ))), + Some(Literal::Primitive(PrimitiveLiteral::Double( + f64::NEG_INFINITY.into(), + ))), + Some(Literal::Primitive(PrimitiveLiteral::String( + "literal".to_string(), + ))), + Some(Literal::Primitive(PrimitiveLiteral::Binary(vec![ + 0, 1, 255, + ]))), + Some(Literal::Primitive(PrimitiveLiteral::Int128(i128::MIN))), + Some(Literal::Primitive(PrimitiveLiteral::UInt128(u128::MAX))), + Some(Literal::Primitive(PrimitiveLiteral::AboveMax)), + Some(Literal::Primitive(PrimitiveLiteral::BelowMin)), + Some(Literal::List(vec![Some(Literal::long(42)), None])), + Some(Literal::Map(Map::from([ + (Literal::string("key"), Some(Literal::int(7))), + (Literal::long(8), None), + ]))), + None, + ])); + + let serialized = serde_json::to_string(&expected).unwrap(); + let actual = serde_json::from_str(&serialized).unwrap(); + + assert_eq!(expected, actual); +} + fn check_convert_with_avro(expected_literal: Literal, expected_type: &Type) { let fields = vec![NestedField::required(1, "col", expected_type.clone()).into()]; let schema = Schema::builder() From acd1c571f2f7efc466cb92a67558b7264a07fe39 Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Fri, 28 Aug 2026 17:21:20 +0800 Subject: [PATCH 3/5] chore: update public API snapshot Record the new Serialize and Deserialize implementations for Literal and Struct. --- crates/iceberg/public-api.txt | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 19f6c26968..b0bcafe7c7 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1534,6 +1534,10 @@ pub fn iceberg::spec::Literal::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> c impl core::hash::Hash for iceberg::spec::Literal pub fn iceberg::spec::Literal::hash<__H: core::hash::Hasher>(&self, state: &mut __H) impl core::marker::StructuralPartialEq for iceberg::spec::Literal +impl serde_core::ser::Serialize for iceberg::spec::Literal +pub fn iceberg::spec::Literal::serialize(&self, serializer: S) -> core::result::Result<::Ok, ::Error> where S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::Literal +pub fn iceberg::spec::Literal::deserialize(deserializer: D) -> core::result::Result::Error> where D: serde_core::de::Deserializer<'de> pub enum iceberg::spec::ManifestContentType pub iceberg::spec::ManifestContentType::Data = 0 pub iceberg::spec::ManifestContentType::Deletes = 1 @@ -2646,6 +2650,10 @@ impl core::marker::StructuralPartialEq for iceberg::spec::Struct impl core::ops::index::Index for iceberg::spec::Struct pub type iceberg::spec::Struct::Output = core::option::Option pub fn iceberg::spec::Struct::index(&self, idx: usize) -> &Self::Output +impl serde_core::ser::Serialize for iceberg::spec::Struct +pub fn iceberg::spec::Struct::serialize(&self, serializer: S) -> core::result::Result<::Ok, ::Error> where S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::Struct +pub fn iceberg::spec::Struct::deserialize(deserializer: D) -> core::result::Result::Error> where D: serde_core::de::Deserializer<'de> pub struct iceberg::spec::StructType impl iceberg::spec::StructType pub fn iceberg::spec::StructType::field_by_id(&self, id: i32) -> core::option::Option<&iceberg::spec::NestedFieldRef> From 986c051189c24a391433066eea31abec6b193b66 Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Fri, 28 Aug 2026 17:34:08 +0800 Subject: [PATCH 4/5] refactor(scan): use typed FileScanTask serde adapter Serialize partition data through the existing RawLiteral implementation using the task schema and partition spec. Remove the direct Literal and Struct serde implementation and keep the public API unchanged. --- crates/iceberg/public-api.txt | 8 - crates/iceberg/src/scan/task.rs | 171 ++++++++++++++++-- crates/iceberg/src/spec/values/literal.rs | 128 ------------- crates/iceberg/src/spec/values/map.rs | 4 - .../iceberg/src/spec/values/struct_value.rs | 18 -- crates/iceberg/src/spec/values/tests.rs | 36 ---- 6 files changed, 154 insertions(+), 211 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index b0bcafe7c7..19f6c26968 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1534,10 +1534,6 @@ pub fn iceberg::spec::Literal::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> c impl core::hash::Hash for iceberg::spec::Literal pub fn iceberg::spec::Literal::hash<__H: core::hash::Hasher>(&self, state: &mut __H) impl core::marker::StructuralPartialEq for iceberg::spec::Literal -impl serde_core::ser::Serialize for iceberg::spec::Literal -pub fn iceberg::spec::Literal::serialize(&self, serializer: S) -> core::result::Result<::Ok, ::Error> where S: serde_core::ser::Serializer -impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::Literal -pub fn iceberg::spec::Literal::deserialize(deserializer: D) -> core::result::Result::Error> where D: serde_core::de::Deserializer<'de> pub enum iceberg::spec::ManifestContentType pub iceberg::spec::ManifestContentType::Data = 0 pub iceberg::spec::ManifestContentType::Deletes = 1 @@ -2650,10 +2646,6 @@ impl core::marker::StructuralPartialEq for iceberg::spec::Struct impl core::ops::index::Index for iceberg::spec::Struct pub type iceberg::spec::Struct::Output = core::option::Option pub fn iceberg::spec::Struct::index(&self, idx: usize) -> &Self::Output -impl serde_core::ser::Serialize for iceberg::spec::Struct -pub fn iceberg::spec::Struct::serialize(&self, serializer: S) -> core::result::Result<::Ok, ::Error> where S: serde_core::ser::Serializer -impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::Struct -pub fn iceberg::spec::Struct::deserialize(deserializer: D) -> core::result::Result::Error> where D: serde_core::de::Deserializer<'de> pub struct iceberg::spec::StructType impl iceberg::spec::StructType pub fn iceberg::spec::StructType::field_by_id(&self, id: i32) -> core::option::Option<&iceberg::spec::NestedFieldRef> diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 2a93ecd69b..37bf0d149a 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -24,15 +24,147 @@ 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>; +mod _serde { + use serde_derive::{Deserialize as DeserializeDerive, Serialize as SerializeDerive}; + + use super::*; + use crate::{Error, ErrorKind}; + + #[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 TryFrom<&FileScanTask> for FileScanTaskSerde { + type Error = Error; + + fn try_from(value: &FileScanTask) -> Result { + let partition = value + .partition + .as_ref() + .map(|partition| { + RawLiteral::try_from( + Literal::Struct(partition.clone()), + &partition_type(&value.schema, value.partition_spec.as_deref())?, + ) + }) + .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.clone(), + data_file_format: value.data_file_format, + schema: value.schema.clone(), + project_field_ids: value.project_field_ids.clone(), + predicate: value.predicate.clone(), + deletes: value.deletes.clone(), + partition, + partition_spec: value.partition_spec.clone(), + name_mapping: value.name_mapping.clone(), + unified_partition_type: value.unified_partition_type.clone(), + case_sensitive: value.case_sensitive, + key_metadata: value.key_metadata.clone(), + }) + } + } + + 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.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, + }) + } + } +} + /// A task to scan part of file. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, TypedBuilder)] +#[derive(Debug, Clone, PartialEq, TypedBuilder)] #[builder(field_defaults(setter(prefix = "with_")))] pub struct FileScanTask { /// The total size of the data file in bytes, from the manifest entry. @@ -53,7 +185,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, @@ -63,7 +194,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, @@ -78,7 +208,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, @@ -89,24 +218,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")] #[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")] #[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")] #[builder(default)] pub name_mapping: Option>, @@ -118,8 +241,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(default)] - #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] pub unified_partition_type: Option>, @@ -131,15 +252,31 @@ 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>, } +impl Serialize for FileScanTask { + fn serialize<__S>(&self, __serializer: __S) -> std::result::Result<__S::Ok, __S::Error> + where __S: serde::Serializer { + _serde::FileScanTaskSerde::try_from(self) + .map_err(serde::ser::Error::custom)? + .serialize(__serializer) + } +} + +impl<'de> Deserialize<'de> for FileScanTask { + fn deserialize<__D>(__deserializer: __D) -> std::result::Result + where __D: serde::Deserializer<'de> { + _serde::FileScanTaskSerde::deserialize(__deserializer)? + .try_into() + .map_err(serde::de::Error::custom) + } +} + impl FileScanTask { /// Returns the data file path of this file scan task. pub fn data_file_path(&self) -> &str { diff --git a/crates/iceberg/src/spec/values/literal.rs b/crates/iceberg/src/spec/values/literal.rs index 0db3a0e3a2..5296eff2a2 100644 --- a/crates/iceberg/src/spec/values/literal.rs +++ b/crates/iceberg/src/spec/values/literal.rs @@ -22,8 +22,6 @@ use std::str::FromStr; use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, TimeZone, Utc}; use ordered_float::OrderedFloat; -use serde::{Deserialize, Deserializer, Serialize, Serializer}; -use serde_derive::{Deserialize as DeserializeDerive, Serialize as SerializeDerive}; use serde_json::{Map as JsonMap, Number, Value as JsonValue}; use uuid::Uuid; @@ -57,132 +55,6 @@ pub enum Literal { Map(Map), } -#[derive(SerializeDerive, DeserializeDerive)] -#[serde(tag = "type", content = "value", rename_all = "kebab-case")] -enum SerializableLiteral { - Boolean(bool), - Int(i32), - Long(i64), - Float([u8; 4]), - Double([u8; 8]), - String(String), - Binary(Vec), - Int128([u8; 16]), - UInt128([u8; 16]), - AboveMax, - BelowMin, - Struct(Vec>), - List(Vec>), - Map(Vec<(SerializableLiteral, Option)>), -} - -impl From<&Literal> for SerializableLiteral { - fn from(value: &Literal) -> Self { - match value { - Literal::Primitive(PrimitiveLiteral::Boolean(value)) => Self::Boolean(*value), - Literal::Primitive(PrimitiveLiteral::Int(value)) => Self::Int(*value), - Literal::Primitive(PrimitiveLiteral::Long(value)) => Self::Long(*value), - Literal::Primitive(PrimitiveLiteral::Float(value)) => { - Self::Float(value.to_bits().to_be_bytes()) - } - Literal::Primitive(PrimitiveLiteral::Double(value)) => { - Self::Double(value.to_bits().to_be_bytes()) - } - Literal::Primitive(PrimitiveLiteral::String(value)) => Self::String(value.clone()), - Literal::Primitive(PrimitiveLiteral::Binary(value)) => Self::Binary(value.clone()), - Literal::Primitive(PrimitiveLiteral::Int128(value)) => { - Self::Int128(value.to_be_bytes()) - } - Literal::Primitive(PrimitiveLiteral::UInt128(value)) => { - Self::UInt128(value.to_be_bytes()) - } - Literal::Primitive(PrimitiveLiteral::AboveMax) => Self::AboveMax, - Literal::Primitive(PrimitiveLiteral::BelowMin) => Self::BelowMin, - Literal::Struct(value) => Self::Struct( - value - .iter() - .map(|value| value.map(SerializableLiteral::from)) - .collect(), - ), - Literal::List(value) => Self::List( - value - .iter() - .map(|value| value.as_ref().map(SerializableLiteral::from)) - .collect(), - ), - Literal::Map(value) => Self::Map( - value - .iter() - .map(|(key, value)| { - ( - SerializableLiteral::from(key), - value.map(SerializableLiteral::from), - ) - }) - .collect(), - ), - } - } -} - -impl From for Literal { - fn from(value: SerializableLiteral) -> Self { - match value { - SerializableLiteral::Boolean(value) => Self::bool(value), - SerializableLiteral::Int(value) => Self::int(value), - SerializableLiteral::Long(value) => Self::long(value), - SerializableLiteral::Float(value) => { - Self::float(f32::from_bits(u32::from_be_bytes(value))) - } - SerializableLiteral::Double(value) => { - Self::double(f64::from_bits(u64::from_be_bytes(value))) - } - SerializableLiteral::String(value) => Self::string(value), - SerializableLiteral::Binary(value) => Self::binary(value), - SerializableLiteral::Int128(value) => { - Self::Primitive(PrimitiveLiteral::Int128(i128::from_be_bytes(value))) - } - SerializableLiteral::UInt128(value) => { - Self::Primitive(PrimitiveLiteral::UInt128(u128::from_be_bytes(value))) - } - SerializableLiteral::AboveMax => Self::Primitive(PrimitiveLiteral::AboveMax), - SerializableLiteral::BelowMin => Self::Primitive(PrimitiveLiteral::BelowMin), - SerializableLiteral::Struct(value) => Self::Struct( - value - .into_iter() - .map(|value| value.map(Literal::from)) - .collect(), - ), - SerializableLiteral::List(value) => Self::List( - value - .into_iter() - .map(|value| value.map(Literal::from)) - .collect(), - ), - SerializableLiteral::Map(value) => Self::Map( - value - .into_iter() - .map(|(key, value)| (Literal::from(key), value.map(Literal::from))) - .collect(), - ), - } - } -} - -impl Serialize for Literal { - fn serialize(&self, serializer: S) -> std::result::Result - where S: Serializer { - SerializableLiteral::from(self).serialize(serializer) - } -} - -impl<'de> Deserialize<'de> for Literal { - fn deserialize(deserializer: D) -> std::result::Result - where D: Deserializer<'de> { - Ok(SerializableLiteral::deserialize(deserializer)?.into()) - } -} - impl Literal { /// Creates a boolean value. /// diff --git a/crates/iceberg/src/spec/values/map.rs b/crates/iceberg/src/spec/values/map.rs index 49021dd2a6..e0f75205f0 100644 --- a/crates/iceberg/src/spec/values/map.rs +++ b/crates/iceberg/src/spec/values/map.rs @@ -67,10 +67,6 @@ impl Map { self.pair.is_empty() } - pub(crate) fn iter(&self) -> impl ExactSizeIterator)> { - self.pair.iter().map(|(key, value)| (key, value.as_ref())) - } - /// Inserts a key-value pair into the map. /// If the map did not have this key present, None is returned. /// If the map did have this key present, the value is updated, and the old value is returned. diff --git a/crates/iceberg/src/spec/values/struct_value.rs b/crates/iceberg/src/spec/values/struct_value.rs index 04f05a05b7..2b44961edf 100644 --- a/crates/iceberg/src/spec/values/struct_value.rs +++ b/crates/iceberg/src/spec/values/struct_value.rs @@ -19,8 +19,6 @@ use std::ops::Index; -use serde::{Deserialize, Deserializer, Serialize, Serializer}; - use super::Literal; /// The partition struct stores the tuple of partition values for each file. @@ -32,22 +30,6 @@ pub struct Struct { fields: Vec>, } -impl Serialize for Struct { - fn serialize(&self, serializer: S) -> Result - where S: Serializer { - self.fields.serialize(serializer) - } -} - -impl<'de> Deserialize<'de> for Struct { - fn deserialize(deserializer: D) -> Result - where D: Deserializer<'de> { - Ok(Self { - fields: Vec::>::deserialize(deserializer)?, - }) - } -} - impl Struct { /// Create a empty struct. pub fn empty() -> Self { diff --git a/crates/iceberg/src/spec/values/tests.rs b/crates/iceberg/src/spec/values/tests.rs index 00ee3d7c91..5ee4ac8d02 100644 --- a/crates/iceberg/src/spec/values/tests.rs +++ b/crates/iceberg/src/spec/values/tests.rs @@ -65,42 +65,6 @@ fn check_avro_bytes_serde(input: Vec, expected_datum: Datum, expected_type: } } -#[test] -fn test_literal_serde_round_trip() { - let expected = Literal::Struct(Struct::from_iter([ - Some(Literal::Primitive(PrimitiveLiteral::Boolean(true))), - Some(Literal::Primitive(PrimitiveLiteral::Int(i32::MIN))), - Some(Literal::Primitive(PrimitiveLiteral::Long(i64::MAX))), - Some(Literal::Primitive(PrimitiveLiteral::Float( - f32::INFINITY.into(), - ))), - Some(Literal::Primitive(PrimitiveLiteral::Double( - f64::NEG_INFINITY.into(), - ))), - Some(Literal::Primitive(PrimitiveLiteral::String( - "literal".to_string(), - ))), - Some(Literal::Primitive(PrimitiveLiteral::Binary(vec![ - 0, 1, 255, - ]))), - Some(Literal::Primitive(PrimitiveLiteral::Int128(i128::MIN))), - Some(Literal::Primitive(PrimitiveLiteral::UInt128(u128::MAX))), - Some(Literal::Primitive(PrimitiveLiteral::AboveMax)), - Some(Literal::Primitive(PrimitiveLiteral::BelowMin)), - Some(Literal::List(vec![Some(Literal::long(42)), None])), - Some(Literal::Map(Map::from([ - (Literal::string("key"), Some(Literal::int(7))), - (Literal::long(8), None), - ]))), - None, - ])); - - let serialized = serde_json::to_string(&expected).unwrap(); - let actual = serde_json::from_str(&serialized).unwrap(); - - assert_eq!(expected, actual); -} - fn check_convert_with_avro(expected_literal: Literal, expected_type: &Type) { let fields = vec![NestedField::required(1, "col", expected_type.clone()).into()]; let schema = Schema::builder() From 9d23600937c6e4af9be57be5316633f31629fb11 Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Sat, 29 Aug 2026 10:26:09 +0800 Subject: [PATCH 5/5] refactor(scan): simplify FileScanTask serde adapter --- crates/iceberg/src/scan/task.rs | 303 ++++++++++++++++---------------- 1 file changed, 152 insertions(+), 151 deletions(-) diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 37bf0d149a..29132bb9ec 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -31,140 +31,12 @@ use crate::spec::{ /// A stream of [`FileScanTask`]. pub type FileScanTaskStream = BoxStream<'static, Result>; -mod _serde { - use serde_derive::{Deserialize as DeserializeDerive, Serialize as SerializeDerive}; - - use super::*; - use crate::{Error, ErrorKind}; - - #[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 TryFrom<&FileScanTask> for FileScanTaskSerde { - type Error = Error; - - fn try_from(value: &FileScanTask) -> Result { - let partition = value - .partition - .as_ref() - .map(|partition| { - RawLiteral::try_from( - Literal::Struct(partition.clone()), - &partition_type(&value.schema, value.partition_spec.as_deref())?, - ) - }) - .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.clone(), - data_file_format: value.data_file_format, - schema: value.schema.clone(), - project_field_ids: value.project_field_ids.clone(), - predicate: value.predicate.clone(), - deletes: value.deletes.clone(), - partition, - partition_spec: value.partition_spec.clone(), - name_mapping: value.name_mapping.clone(), - unified_partition_type: value.unified_partition_type.clone(), - case_sensitive: value.case_sensitive, - key_metadata: value.key_metadata.clone(), - }) - } - } - - 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.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, - }) - } - } -} - /// A task to scan part of file. -#[derive(Debug, Clone, PartialEq, TypedBuilder)] +#[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. @@ -259,24 +131,6 @@ pub struct FileScanTask { pub key_metadata: Option>, } -impl Serialize for FileScanTask { - fn serialize<__S>(&self, __serializer: __S) -> std::result::Result<__S::Ok, __S::Error> - where __S: serde::Serializer { - _serde::FileScanTaskSerde::try_from(self) - .map_err(serde::ser::Error::custom)? - .serialize(__serializer) - } -} - -impl<'de> Deserialize<'de> for FileScanTask { - fn deserialize<__D>(__deserializer: __D) -> std::result::Result - where __D: serde::Deserializer<'de> { - _serde::FileScanTaskSerde::deserialize(__deserializer)? - .try_into() - .map_err(serde::de::Error::custom) - } -} - impl FileScanTask { /// Returns the data file path of this file scan task. pub fn data_file_path(&self) -> &str { @@ -392,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, + }) + } + } +}