feat(spec): read partition and sort fields with multi-argument transforms (source-ids)

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<alloc::vec::Vec<i32>>
 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<alloc::vec::Vec<i32>>
 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

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<Vec<i32>>,
     /// 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<i32>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        source_ids: Option<Vec<i32>>,
+        field_id: i32,
+        name: String,
+        transform: Option<String>,
+    }
+
+    impl TryFrom<PartitionFieldSerde> for PartitionField {
+        type Error = Error;
+
+        fn try_from(value: PartitionFieldSerde) -> Result<Self, Error> {
+            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<PartitionField> 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<PartitionSpec>;
 /// 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::<PartitionField>(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::<PartitionField>(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),

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<Vec<i32>>,
     /// 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<i32>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        source_ids: Option<Vec<i32>>,
+        transform: Option<String>,
+        direction: SortDirection,
+        null_order: NullOrder,
+    }
+
+    impl TryFrom<SortFieldSerde> for SortField {
+        type Error = Error;
+
+        fn try_from(value: SortFieldSerde) -> Result<Self, Error> {
+            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<SortField> 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::<SortField>(spec).unwrap_err();
+        assert!(
+            err.to_string()
+                .contains("Transform is required for a multi-argument field")
+        );
+    }
+
     #[test]
     fn test_sort_order() {
         let spec = r#"