From 5402bdfa16e540a2496eba3ed7c33f784af52d37 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Sat, 11 Jul 2026 10:23:23 +0900 Subject: [PATCH 1/2] Read partition and sort fields with multi-argument transforms (source-ids) --- crates/iceberg/src/spec/partition.rs | 153 +++++++++++++++++- crates/iceberg/src/spec/sort.rs | 143 +++++++++++++++- crates/iceberg/src/spec/table_metadata.rs | 6 + .../src/spec/table_metadata_builder.rs | 4 + crates/iceberg/src/spec/transform.rs | 48 ++++++ .../src/physical_plan/repartition.rs | 2 + 6 files changed, 349 insertions(+), 7 deletions(-) diff --git a/crates/iceberg/src/spec/partition.rs b/crates/iceberg/src/spec/partition.rs index 43b56dcdaa..cd463e9493 100644 --- a/crates/iceberg/src/spec/partition.rs +++ b/crates/iceberg/src/spec/partition.rs @@ -34,10 +34,18 @@ pub(crate) const DEFAULT_PARTITION_SPEC_ID: i32 = 0; /// Partition fields capture the transform from table data to partition values. #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, TypedBuilder)] -#[serde(rename_all = "kebab-case")] +#[serde( + try_from = "_serde_partition_field::PartitionFieldSerde", + into = "_serde_partition_field::PartitionFieldSerde" +)] pub struct PartitionField { /// A source column id from the table’s schema pub source_id: i32, + /// Source column ids when the transform takes multiple arguments (v3 multi-argument + /// transforms). `None` for single-argument transforms, where `source_id` is used instead. + /// When set, `source_id` holds the first id so that existing consumers keep working. + #[builder(default)] + pub source_ids: Option>, /// A partition field id that is used to identify a partition field and is unique within a partition spec. /// In v2 table metadata, it is unique across all partition specs. pub field_id: i32, @@ -54,6 +62,57 @@ impl PartitionField { } } +mod _serde_partition_field { + use serde::{Deserialize, Serialize}; + + use super::PartitionField; + use crate::Error; + use crate::spec::transform::normalize_transform_sources; + + #[derive(Serialize, Deserialize)] + #[serde(rename_all = "kebab-case")] + pub(super) struct PartitionFieldSerde { + #[serde(default, skip_serializing_if = "Option::is_none")] + source_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + source_ids: Option>, + field_id: i32, + name: String, + transform: Option, + } + + impl TryFrom for PartitionField { + type Error = Error; + + fn try_from(value: PartitionFieldSerde) -> Result { + let (source_id, source_ids, transform) = + normalize_transform_sources(value.source_id, value.source_ids, value.transform)?; + Ok(PartitionField { + source_id, + source_ids, + field_id: value.field_id, + name: value.name, + transform, + }) + } + } + + impl From for PartitionFieldSerde { + fn from(value: PartitionField) -> Self { + // Per the spec, single-argument transforms write only source-id and + // multi-argument transforms write only source-ids + let multi_arg = value.source_ids.as_ref().is_some_and(|ids| ids.len() > 1); + Self { + source_id: (!multi_arg).then_some(value.source_id), + source_ids: if multi_arg { value.source_ids } else { None }, + field_id: value.field_id, + name: value.name, + transform: Some(value.transform.to_string()), + } + } + } +} + /// Reference to [`PartitionSpec`]. pub type PartitionSpecRef = Arc; /// Partition spec that defines how to produce a tuple of partition values from a record. @@ -145,6 +204,7 @@ impl PartitionSpec { for (this_field, other_field) in self.fields.iter().zip(other.fields.iter()) { if this_field.source_id != other_field.source_id + || this_field.source_ids != other_field.source_ids || this_field.name != other_field.name || this_field.transform != other_field.transform { @@ -556,6 +616,7 @@ impl PartitionSpecBuilder { bound_fields.push(PartitionField { source_id: field.source_id, + source_ids: None, field_id: partition_field_id, name: field.name, transform: field.transform, @@ -785,6 +846,95 @@ mod tests { assert_eq!(Transform::Truncate(4), partition_spec.fields[2].transform); } + #[test] + fn test_deserialize_partition_field_multi_arg() { + let spec = r#" + { + "source-ids": [1, 2], + "field-id": 1000, + "name": "multi_bucket", + "transform": "bucket[4]" + } + "#; + + let field: PartitionField = serde_json::from_str(spec).unwrap(); + + // v3 readers must read tables with multi-argument transforms, treating them as unknown + assert_eq!(Transform::Unknown, field.transform); + assert_eq!(1, field.source_id); + assert_eq!(Some(vec![1, 2]), field.source_ids); + + // the field must round-trip with source-ids only + let serialized = serde_json::to_value(&field).unwrap(); + assert_eq!( + Some(&serde_json::json!([1, 2])), + serialized.get("source-ids") + ); + assert!(serialized.get("source-id").is_none()); + } + + #[test] + fn test_deserialize_partition_field_single_element_source_ids() { + let spec = r#" + { + "source-ids": [1], + "field-id": 1000, + "name": "str_truncate", + "transform": "truncate[19]" + } + "#; + + let field: PartitionField = serde_json::from_str(spec).unwrap(); + + // a single-element source-ids is normalized onto source-id + assert_eq!(Transform::Truncate(19), field.transform); + assert_eq!(1, field.source_id); + assert_eq!(None, field.source_ids); + + let serialized = serde_json::to_value(&field).unwrap(); + assert_eq!(Some(&serde_json::json!(1)), serialized.get("source-id")); + assert!(serialized.get("source-ids").is_none()); + } + + #[test] + fn test_deserialize_partition_field_empty_source_ids_rejected() { + let spec = r#"{"source-ids": [], "field-id": 1000, "name": "m", "transform": "bucket[4]"}"#; + let err = serde_json::from_str::(spec).unwrap_err(); + assert!(err.to_string().contains("Empty source-ids is not allowed")); + } + + #[test] + fn test_deserialize_partition_field_multi_arg_requires_transform() { + let spec = r#"{"source-ids": [1, 2], "field-id": 1000, "name": "m"}"#; + let err = serde_json::from_str::(spec).unwrap_err(); + assert!( + err.to_string() + .contains("Transform is required for a multi-argument field") + ); + } + + #[test] + fn test_partition_type_with_multi_arg_field() { + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "a", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required(2, "b", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(); + + let spec: PartitionSpec = serde_json::from_str( + r#"{"spec-id": 1, "fields": [{"source-ids": [1, 2], "field-id": 1000, "name": "m", "transform": "bucket[4]"}]}"#, + ) + .unwrap(); + + let struct_type = spec.partition_type(&schema).unwrap(); + assert_eq!( + Type::Primitive(PrimitiveType::String), + *struct_type.fields()[0].field_type + ); + } + #[test] fn test_is_unpartitioned() { let schema = Schema::builder() @@ -1283,6 +1433,7 @@ mod tests { spec_id: 1, fields: vec![PartitionField { source_id: 1, + source_ids: None, field_id: 1000, name: "id_bucket[16]".to_string(), transform: Transform::Bucket(16), diff --git a/crates/iceberg/src/spec/sort.rs b/crates/iceberg/src/spec/sort.rs index 379d44cc2d..665752e90f 100644 --- a/crates/iceberg/src/spec/sort.rs +++ b/crates/iceberg/src/spec/sort.rs @@ -73,11 +73,19 @@ impl fmt::Display for NullOrder { } #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, TypedBuilder)] -#[serde(rename_all = "kebab-case")] +#[serde( + try_from = "_serde_sort_field::SortFieldSerde", + into = "_serde_sort_field::SortFieldSerde" +)] /// Entry for every column that is to be sorted pub struct SortField { /// A source column id from the table’s schema pub source_id: i32, + /// Source column ids when the transform takes multiple arguments (v3 multi-argument + /// transforms). `None` for single-argument transforms, where `source_id` is used instead. + /// When set, `source_id` holds the first id so that existing consumers keep working. + #[builder(default)] + pub source_ids: Option>, /// A transform that is used to produce values to be sorted on from the source column. pub transform: Transform, /// A sort direction, that can only be either asc or desc @@ -88,11 +96,70 @@ pub struct SortField { impl fmt::Display for SortField { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!( - f, - "SortField {{ source_id: {}, transform: {}, direction: {}, null_order: {} }}", - self.source_id, self.transform, self.direction, self.null_order - ) + if let Some(source_ids) = self.source_ids.as_ref().filter(|ids| ids.len() > 1) { + write!( + f, + "SortField {{ source_ids: {source_ids:?}, transform: {}, direction: {}, null_order: {} }}", + self.transform, self.direction, self.null_order + ) + } else { + write!( + f, + "SortField {{ source_id: {}, transform: {}, direction: {}, null_order: {} }}", + self.source_id, self.transform, self.direction, self.null_order + ) + } + } +} + +mod _serde_sort_field { + use serde::{Deserialize, Serialize}; + + use super::{NullOrder, SortDirection, SortField}; + use crate::Error; + use crate::spec::transform::normalize_transform_sources; + + #[derive(Serialize, Deserialize)] + #[serde(rename_all = "kebab-case")] + pub(super) struct SortFieldSerde { + #[serde(default, skip_serializing_if = "Option::is_none")] + source_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + source_ids: Option>, + transform: Option, + direction: SortDirection, + null_order: NullOrder, + } + + impl TryFrom for SortField { + type Error = Error; + + fn try_from(value: SortFieldSerde) -> Result { + let (source_id, source_ids, transform) = + normalize_transform_sources(value.source_id, value.source_ids, value.transform)?; + Ok(SortField { + source_id, + source_ids, + transform, + direction: value.direction, + null_order: value.null_order, + }) + } + } + + impl From for SortFieldSerde { + fn from(value: SortField) -> Self { + // Per the spec, single-argument transforms write only source-id and + // multi-argument transforms write only source-ids + let multi_arg = value.source_ids.as_ref().is_some_and(|ids| ids.len() > 1); + Self { + source_id: (!multi_arg).then_some(value.source_id), + source_ids: if multi_arg { value.source_ids } else { None }, + transform: Some(value.transform.to_string()), + direction: value.direction, + null_order: value.null_order, + } + } } } @@ -234,6 +301,70 @@ mod tests { assert_eq!(NullOrder::Last, field.null_order); } + #[test] + fn test_deserialize_sort_field_multi_arg() { + let spec = r#" + { + "transform": "bucket[4]", + "source-ids": [19, 20], + "direction": "asc", + "null-order": "nulls-first" + } + "#; + + let field: SortField = serde_json::from_str(spec).unwrap(); + + // v3 readers must read tables with multi-argument transforms, treating them as unknown + assert_eq!(Transform::Unknown, field.transform); + assert_eq!(19, field.source_id); + assert_eq!(Some(vec![19, 20]), field.source_ids); + assert_eq!( + "SortField { source_ids: [19, 20], transform: unknown, direction: ascending, null_order: first }", + field.to_string() + ); + + // the field must round-trip with source-ids only + let serialized = serde_json::to_value(&field).unwrap(); + assert_eq!( + Some(&serde_json::json!([19, 20])), + serialized.get("source-ids") + ); + assert!(serialized.get("source-id").is_none()); + } + + #[test] + fn test_deserialize_sort_field_single_element_source_ids() { + let spec = r#" + { + "transform": "bucket[4]", + "source-ids": [19], + "direction": "asc", + "null-order": "nulls-first" + } + "#; + + let field: SortField = serde_json::from_str(spec).unwrap(); + + // a single-element source-ids is normalized onto source-id + assert_eq!(Transform::Bucket(4), field.transform); + assert_eq!(19, field.source_id); + assert_eq!(None, field.source_ids); + + let serialized = serde_json::to_value(&field).unwrap(); + assert_eq!(Some(&serde_json::json!(19)), serialized.get("source-id")); + assert!(serialized.get("source-ids").is_none()); + } + + #[test] + fn test_deserialize_sort_field_multi_arg_requires_transform() { + let spec = r#"{"source-ids": [19, 20], "direction": "asc", "null-order": "nulls-first"}"#; + let err = serde_json::from_str::(spec).unwrap_err(); + assert!( + err.to_string() + .contains("Transform is required for a multi-argument field") + ); + } + #[test] fn test_sort_order() { let spec = r#" diff --git a/crates/iceberg/src/spec/table_metadata.rs b/crates/iceberg/src/spec/table_metadata.rs index 607fd98350..cc37e4ba52 100644 --- a/crates/iceberg/src/spec/table_metadata.rs +++ b/crates/iceberg/src/spec/table_metadata.rs @@ -2963,12 +2963,14 @@ mod tests { .with_order_id(3) .with_sort_field(SortField { source_id: 2, + source_ids: None, transform: Transform::Identity, direction: SortDirection::Ascending, null_order: NullOrder::First, }) .with_sort_field(SortField { source_id: 3, + source_ids: None, transform: Transform::Bucket(4), direction: SortDirection::Descending, null_order: NullOrder::Last, @@ -3060,12 +3062,14 @@ mod tests { .with_order_id(3) .with_sort_field(SortField { source_id: 2, + source_ids: None, transform: Transform::Identity, direction: SortDirection::Ascending, null_order: NullOrder::First, }) .with_sort_field(SortField { source_id: 3, + source_ids: None, transform: Transform::Bucket(4), direction: SortDirection::Descending, null_order: NullOrder::Last, @@ -3189,12 +3193,14 @@ mod tests { .with_order_id(3) .with_sort_field(SortField { source_id: 2, + source_ids: None, transform: Transform::Identity, direction: SortDirection::Ascending, null_order: NullOrder::First, }) .with_sort_field(SortField { source_id: 3, + source_ids: None, transform: Transform::Bucket(4), direction: SortDirection::Descending, null_order: NullOrder::Last, diff --git a/crates/iceberg/src/spec/table_metadata_builder.rs b/crates/iceberg/src/spec/table_metadata_builder.rs index 3191d6c13c..bcb8e6524b 100644 --- a/crates/iceberg/src/spec/table_metadata_builder.rs +++ b/crates/iceberg/src/spec/table_metadata_builder.rs @@ -1495,6 +1495,7 @@ mod tests { .with_order_id(1) .with_sort_field(SortField { source_id: 3, + source_ids: None, transform: Transform::Bucket(4), direction: SortDirection::Descending, null_order: NullOrder::First, @@ -1639,6 +1640,7 @@ mod tests { let sort_order = SortOrder::builder() .with_fields(vec![SortField { source_id: 11, + source_ids: None, transform: Transform::Identity, direction: SortDirection::Ascending, null_order: NullOrder::First, @@ -1680,6 +1682,7 @@ mod tests { let expected_sort_order = SortOrder::builder() .with_fields(vec![SortField { source_id: 1, + source_ids: None, transform: Transform::Identity, direction: SortDirection::Ascending, null_order: NullOrder::First, @@ -1988,6 +1991,7 @@ mod tests { .with_order_id(10) .with_fields(vec![SortField { source_id: 1, + source_ids: None, transform: Transform::Identity, direction: SortDirection::Ascending, null_order: NullOrder::First, diff --git a/crates/iceberg/src/spec/transform.rs b/crates/iceberg/src/spec/transform.rs index 97ab638e79..7ea27c1f24 100644 --- a/crates/iceberg/src/spec/transform.rs +++ b/crates/iceberg/src/spec/transform.rs @@ -1058,6 +1058,54 @@ impl FromStr for Transform { } } +/// Normalizes the `source-id`/`source-ids` pair read from partition and sort field JSON. +/// +/// Per the spec, single-argument transforms are written with `source-id` and multi-argument +/// transforms with `source-ids`. Multi-argument transforms cannot be evaluated yet; per the +/// spec, v3 readers must read tables with such transforms, ignoring them, so the transform is +/// mapped to [`Transform::Unknown`] while the original ids are kept for round-tripping. +/// +/// Returns the normalized `(source_id, source_ids, transform)` triple, where `source_id` is +/// the first id so that existing consumers keep working. +pub(crate) fn normalize_transform_sources( + source_id: Option, + source_ids: Option>, + transform: Option, +) -> Result<(i32, Option>, Transform)> { + let parse_transform = |transform: Option| { + transform + .ok_or_else(|| Error::new(ErrorKind::DataInvalid, "missing field `transform`"))? + .parse::() + }; + + if let Some(source_id) = source_id { + return Ok((source_id, source_ids, parse_transform(transform)?)); + } + + match source_ids { + Some(source_ids) => match source_ids.as_slice() { + [] => Err(Error::new( + ErrorKind::DataInvalid, + "Empty source-ids is not allowed", + )), + [source_id] => Ok((*source_id, None, parse_transform(transform)?)), + [first, ..] => { + if transform.is_none() { + return Err(Error::new( + ErrorKind::DataInvalid, + "Transform is required for a multi-argument field", + )); + } + Ok((*first, Some(source_ids), Transform::Unknown)) + } + }, + None => Err(Error::new( + ErrorKind::DataInvalid, + "missing field `source-id`", + )), + } +} + impl Serialize for Transform { fn serialize(&self, serializer: S) -> std::result::Result where S: Serializer { diff --git a/crates/integrations/datafusion/src/physical_plan/repartition.rs b/crates/integrations/datafusion/src/physical_plan/repartition.rs index 31a7f94a35..5f82b4ee16 100644 --- a/crates/integrations/datafusion/src/physical_plan/repartition.rs +++ b/crates/integrations/datafusion/src/physical_plan/repartition.rs @@ -356,6 +356,7 @@ mod tests { .with_order_id(1) .with_sort_field(SortField { source_id: 2, + source_ids: None, transform: Transform::Bucket(4), direction: SortDirection::Ascending, null_order: NullOrder::First, @@ -445,6 +446,7 @@ mod tests { .with_order_id(1) .with_sort_field(SortField { source_id: 2, + source_ids: None, transform: Transform::Bucket(8), direction: SortDirection::Ascending, null_order: NullOrder::First, From c68a8f5c1b9538ed0d4fd995ba25d3dae86b6a14 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Sat, 11 Jul 2026 10:40:33 +0900 Subject: [PATCH 2/2] Update public-api.txt for the new source_ids fields --- crates/iceberg/public-api.txt | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index c2013294e3..39eb3a1cb5 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -2229,6 +2229,7 @@ pub struct iceberg::spec::PartitionField pub iceberg::spec::PartitionField::field_id: i32 pub iceberg::spec::PartitionField::name: alloc::string::String pub iceberg::spec::PartitionField::source_id: i32 +pub iceberg::spec::PartitionField::source_ids: core::option::Option> pub iceberg::spec::PartitionField::transform: iceberg::spec::Transform impl iceberg::spec::PartitionField pub fn iceberg::spec::PartitionField::into_unbound(self) -> iceberg::spec::UnboundPartitionField @@ -2243,7 +2244,7 @@ impl core::fmt::Debug for iceberg::spec::PartitionField pub fn iceberg::spec::PartitionField::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl core::marker::StructuralPartialEq for iceberg::spec::PartitionField impl iceberg::spec::PartitionField -pub fn iceberg::spec::PartitionField::builder() -> PartitionFieldBuilder<((), (), (), ())> +pub fn iceberg::spec::PartitionField::builder() -> PartitionFieldBuilder<((), (), (), (), ())> impl serde_core::ser::Serialize for iceberg::spec::PartitionField pub fn iceberg::spec::PartitionField::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::PartitionField @@ -2451,6 +2452,7 @@ pub struct iceberg::spec::SortField pub iceberg::spec::SortField::direction: iceberg::spec::SortDirection pub iceberg::spec::SortField::null_order: iceberg::spec::NullOrder pub iceberg::spec::SortField::source_id: i32 +pub iceberg::spec::SortField::source_ids: core::option::Option> pub iceberg::spec::SortField::transform: iceberg::spec::Transform impl core::clone::Clone for iceberg::spec::SortField pub fn iceberg::spec::SortField::clone(&self) -> iceberg::spec::SortField @@ -2463,7 +2465,7 @@ impl core::fmt::Display for iceberg::spec::SortField pub fn iceberg::spec::SortField::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl core::marker::StructuralPartialEq for iceberg::spec::SortField impl iceberg::spec::SortField -pub fn iceberg::spec::SortField::builder() -> SortFieldBuilder<((), (), (), ())> +pub fn iceberg::spec::SortField::builder() -> SortFieldBuilder<((), (), (), (), ())> impl serde_core::ser::Serialize for iceberg::spec::SortField pub fn iceberg::spec::SortField::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::SortField