This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3431-86e837f90d16074d67218db05ffc2629443dafd5 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 1c05452b53f65d4d37f6399c550ca0d2413ded0c Author: Dhruv Arya <[email protected]> AuthorDate: Thu Oct 22 13:49:50 2026 +0000 fix(spec): accept partition specs produced by spec evolution (#3431) * fix(spec): accept partition specs produced by spec evolution Partition spec evolution produces specs that the spec builders rejected: - A source column can have more than one partition field. Adding hour(ts) to a spec that has day(ts) is allowed, and format version 1 tables keep each removed field as a `void` placeholder. Binding an unbound spec (`UnboundPartitionSpec::bind`, `PartitionSpecBuilder::new_from_unbound`, and so `TableMetadataBuilder::add_partition_spec`) and loading the legacy v1 `partition-spec` field ran the authoring-time redundancy check and rejected these specs. Both now skip it, as Java does. `PartitionSpecBuilder::add_unbound_field` and `UnboundPartitionSpecBuilder::add_partition_field` still reject redundant fields. - Removing identity(category) from a format version 1 spec leaves a void field named `category`. Both the spec builder and the table metadata builder allowed only identity fields to share the name of their source column. Void fields may now do the same, matching Java's `alwaysNull`. Co-authored-by: Isaac <[email protected]> * fix(spec): address review feedback on evolved spec binding Share the identity-or-void name rule through `Transform::may_share_name_with_source` and name void in its error. Check names before redundancy in `add_unbound_field`, document the authoring-vs-bind split, and pin it with tests, including a v2 day+hour default spec that survives later commits. Co-authored-by: Isaac <[email protected]> --------- Co-authored-by: Dhruv Arya <[email protected]> Co-authored-by: Isaac <[email protected]> Co-authored-by: Kevin Liu <[email protected]> --- crates/iceberg/src/spec/partition.rs | 168 +++++++++++++++++++++- crates/iceberg/src/spec/table_metadata.rs | 41 +++++- crates/iceberg/src/spec/table_metadata_builder.rs | 102 +++++++++++-- crates/iceberg/src/spec/transform.rs | 8 ++ 4 files changed, 300 insertions(+), 19 deletions(-) diff --git a/crates/iceberg/src/spec/partition.rs b/crates/iceberg/src/spec/partition.rs index fdb5a4a7b..3e639de00 100644 --- a/crates/iceberg/src/spec/partition.rs +++ b/crates/iceberg/src/spec/partition.rs @@ -437,6 +437,9 @@ impl UnboundPartitionSpec { } /// Bind this unbound partition spec to a schema. + /// + /// Unlike [`PartitionSpecBuilder::add_unbound_field`], this accepts redundant fields, such + /// as two time granularities of one column, because spec evolution produces them. pub fn bind(self, schema: impl Into<SchemaRef>) -> Result<PartitionSpec> { PartitionSpecBuilder::new_from_unbound(self, schema)?.build() } @@ -578,7 +581,7 @@ impl PartitionSpecBuilder { Self::new(schema).with_spec_id(unbound.spec_id.unwrap_or(DEFAULT_PARTITION_SPEC_ID)); for field in unbound.fields { - builder = builder.add_unbound_field(field)?; + builder = builder.add_field_allowing_redundancy(field)?; } Ok(builder) } @@ -630,9 +633,20 @@ impl PartitionSpecBuilder { /// /// If partition field id is set, it is used as the field id. /// Otherwise, a new `field_id` is assigned. - pub fn add_unbound_field(mut self, field: UnboundPartitionField) -> Result<Self> { + /// + /// Rejects a field that is redundant with an existing one, such as a second time + /// granularity of the same column. [`UnboundPartitionSpec::bind`] accepts such fields. + pub fn add_unbound_field(self, field: UnboundPartitionField) -> Result<Self> { self.check_name_set_and_unique(&field.name)?; self.check_for_redundant_partitions(&field.source_ids, &field.transform)?; + self.add_field_allowing_redundancy(field) + } + + /// Same as [`Self::add_unbound_field`] without the redundancy check, as Java's + /// `PartitionSpec.Builder.add` does. Used to bind persisted specs, where evolution can leave + /// several time granularities or void fields for one column. + fn add_field_allowing_redundancy(mut self, field: UnboundPartitionField) -> Result<Self> { + self.check_name_set_and_unique(&field.name)?; Self::check_name_does_not_collide_with_schema(&field, &self.schema)?; Self::check_transform_compatibility(&field, &self.schema)?; if let Some(partition_field_id) = field.field_id { @@ -709,19 +723,19 @@ impl PartitionSpecBuilder { /// Ensure that the partition name is unique among columns in the schema. /// Duplicate names are allowed if: /// 1. The column is sourced from the column with the same name. - /// 2. AND the transformation is identity + /// 2. AND [`Transform::may_share_name_with_source`] holds for the transform. fn check_name_does_not_collide_with_schema( field: &UnboundPartitionField, schema: &Schema, ) -> Result<()> { match schema.field_by_name(field.name.as_str()) { Some(schema_collision) => { - if field.transform == Transform::Identity { + if field.transform.may_share_name_with_source() { if schema_collision.id == field.source_id()? { Ok(()) } else { Err(invalid_data!( - "Cannot create identity partition sourced from different field in schema. Field name '{}' has id `{}` in schema but partition source id is `{}`", + "Cannot create identity or void partition sourced from different field in schema. Field name '{}' has id `{}` in schema but partition source id is `{}`", field.name, schema_collision.id, field.source_id()? @@ -729,7 +743,7 @@ impl PartitionSpecBuilder { } } else { Err(invalid_data!( - "Cannot create partition with name: '{}' that conflicts with schema field and is not an identity transform.", + "Cannot create partition with name: '{}' that conflicts with schema field and is not an identity or void transform.", field.name )) } @@ -1564,6 +1578,148 @@ mod tests { assert!(err.message().contains("redundant partition")); } + fn ts_schema() -> Schema { + Schema::builder() + .with_fields(vec![ + NestedField::required(1, "ts", Type::Primitive(PrimitiveType::Timestamptz)).into(), + NestedField::required(2, "id", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap() + } + + fn ts_field(field_id: i32, name: &str, transform: Transform) -> UnboundPartitionField { + UnboundPartitionField { + source_ids: vec![1], + field_id: Some(field_id), + name: name.to_string(), + transform, + } + } + + #[test] + fn test_add_unbound_field_rejects_second_time_granularity() { + let err = PartitionSpec::builder(ts_schema()) + .add_unbound_field(ts_field(1000, "ts_day", Transform::Day)) + .unwrap() + .add_unbound_field(ts_field(1001, "ts_hour", Transform::Hour)) + .unwrap_err(); + assert!(err.message().contains("redundant partition")); + } + + #[test] + fn test_bind_accepts_evolved_spec_fields() { + let fields = vec![ + ts_field(1000, "ts_day", Transform::Day), + ts_field(1001, "ts_hour", Transform::Hour), + ts_field(1002, "ts", Transform::Void), + ts_field(1003, "ts_null", Transform::Void), + ]; + let unbound = UnboundPartitionSpec { + spec_id: Some(1), + fields: fields.clone(), + }; + + let spec = unbound.bind(ts_schema()).unwrap(); + + let expected = PartitionSpec { + spec_id: 1, + fields: fields + .into_iter() + .map(|f| PartitionField { + source_id: 1, + field_id: f.field_id.unwrap(), + name: f.name, + transform: f.transform, + }) + .collect(), + }; + assert_eq!(spec, expected); + } + + #[test] + fn test_bind_rejects_void_field_named_after_other_column() { + let unbound = UnboundPartitionSpec { + spec_id: Some(1), + fields: vec![ts_field(1000, "id", Transform::Void)], + }; + + let err = unbound.bind(ts_schema()).unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("Field name 'id'")); + } + + #[test] + fn test_bind_accepts_identical_fields_with_different_names() { + let bucket_field = |field_id: i32, name: &str| UnboundPartitionField { + source_ids: vec![2], + field_id: Some(field_id), + name: name.to_string(), + transform: Transform::Bucket(16), + }; + let unbound = UnboundPartitionSpec { + spec_id: Some(1), + fields: vec![ + bucket_field(1000, "id_bucket"), + bucket_field(1001, "id_bucket_2"), + ], + }; + + let spec = unbound.bind(ts_schema()).unwrap(); + + let expected = PartitionSpec { + spec_id: 1, + fields: vec![ + PartitionField { + source_id: 2, + field_id: 1000, + name: "id_bucket".to_string(), + transform: Transform::Bucket(16), + }, + PartitionField { + source_id: 2, + field_id: 1001, + name: "id_bucket_2".to_string(), + transform: Transform::Bucket(16), + }, + ], + }; + assert_eq!(spec, expected); + } + + #[test] + fn test_bind_rejects_repeated_partition_name() { + let unbound = UnboundPartitionSpec { + spec_id: Some(1), + fields: vec![ + ts_field(1000, "ts_part", Transform::Day), + ts_field(1001, "ts_part", Transform::Hour), + ], + }; + + let err = unbound.bind(ts_schema()).unwrap_err(); + + assert!( + err.message() + .contains("Cannot use partition name more than once") + ); + } + + #[test] + fn test_add_unbound_field_checks_name_before_redundancy() { + let err = PartitionSpec::builder(ts_schema()) + .add_unbound_field(ts_field(1000, "ts_part", Transform::Day)) + .unwrap() + .add_unbound_field(ts_field(1001, "ts_part", Transform::Hour)) + .unwrap_err(); + + assert!( + err.message() + .contains("Cannot use partition name more than once") + ); + } + #[test] fn test_builder_incompatible_transforms_disallowed() { let schema = Schema::builder() diff --git a/crates/iceberg/src/spec/table_metadata.rs b/crates/iceberg/src/spec/table_metadata.rs index 83468602c..99df566f6 100644 --- a/crates/iceberg/src/spec/table_metadata.rs +++ b/crates/iceberg/src/spec/table_metadata.rs @@ -785,7 +785,7 @@ pub(super) mod _serde { use crate::spec::{ EncryptedKey, INITIAL_ROW_ID, PartitionField, PartitionSpec, PartitionSpecRef, PartitionStatisticsFile, Schema, SchemaRef, Snapshot, SnapshotReference, SnapshotRetention, - SortOrder, StatisticsFile, + SortOrder, StatisticsFile, UnboundPartitionSpec, }; use crate::{Error, ErrorKind}; @@ -1220,10 +1220,14 @@ pub(super) mod _serde { .collect::<HashMap<_, _>>() } else if let Some(partition_spec) = value.partition_spec { // Option 2: Fall back to 'partition_spec' - let spec = PartitionSpec::builder(current_schema.clone()) - .with_spec_id(DEFAULT_PARTITION_SPEC_ID) - .add_unbound_fields(partition_spec.into_iter().map(|f| f.into_unbound()))? - .build()?; + let spec = UnboundPartitionSpec { + spec_id: Some(DEFAULT_PARTITION_SPEC_ID), + fields: partition_spec + .into_iter() + .map(|f| f.into_unbound()) + .collect(), + } + .bind(current_schema.clone())?; HashMap::from_iter(vec![(DEFAULT_PARTITION_SPEC_ID, Arc::new(spec))]) } else { @@ -3317,6 +3321,33 @@ mod tests { check_table_metadata_serde(&metadata, expected); } + #[test] + fn test_table_metadata_v1_legacy_partition_spec_with_void_fields() { + let mut metadata: serde_json::Value = serde_json::from_str( + &fs::read_to_string("testdata/table_metadata/TableMetadataV1Valid.json").unwrap(), + ) + .unwrap(); + // What a format version 1 spec keeps after identity(x) was removed twice. + metadata["partition-spec"] = serde_json::json!([ + {"name": "x", "transform": "void", "source-id": 1, "field-id": 1000}, + {"name": "x_null", "transform": "void", "source-id": 1, "field-id": 1001}, + ]); + + let metadata: TableMetadata = serde_json::from_value(metadata).unwrap(); + + let fields = metadata.default_partition_spec().fields(); + assert_eq!( + fields + .iter() + .map(|f| (f.field_id, f.name.as_str(), f.transform)) + .collect::<Vec<_>>(), + vec![ + (1000, "x", Transform::Void), + (1001, "x_null", Transform::Void) + ] + ); + } + #[test] fn test_empty_snapshot_id_is_normalized_to_none() { let metadata = diff --git a/crates/iceberg/src/spec/table_metadata_builder.rs b/crates/iceberg/src/spec/table_metadata_builder.rs index 44deb3bfe..a8ef84a18 100644 --- a/crates/iceberg/src/spec/table_metadata_builder.rs +++ b/crates/iceberg/src/spec/table_metadata_builder.rs @@ -726,12 +726,13 @@ impl TableMetadataBuilder { /// Validate partition field names against schema field names across all historical schemas. /// /// Due to Iceberg's multi-version property, partition fields can share names with schema fields - /// if they meet specific requirements (identity transform + matching source field ID). + /// if they meet specific requirements (identity or void transform + matching source field ID). /// This validation enforces those rules across all historical schema versions. /// /// # Errors - /// - Partition field name conflicts with schema field name but doesn't use identity transform. - /// - Partition field uses identity transform but references wrong source field ID. + /// - Partition field name conflicts with schema field name but doesn't use identity or void + /// transform. + /// - Partition field uses identity or void transform but references wrong source field ID. fn validate_partition_field_names(&self, unbound_spec: &UnboundPartitionSpec) -> Result<()> { if self.metadata.schemas.is_empty() { return Ok(()); @@ -750,19 +751,18 @@ impl TableMetadataBuilder { // If name exists in schemas, validate against current schema rules if let Some(schema_field) = current_schema.field_by_name(partition_field.name()) { - let is_identity_transform = partition_field.transform() == Transform::Identity; let has_matching_source_id = schema_field.id == partition_field.source_id()?; - if !is_identity_transform { + if !partition_field.transform().may_share_name_with_source() { return Err(invalid_data!( - "Cannot create partition with name '{}' that conflicts with schema field and is not an identity transform.", + "Cannot create partition with name '{}' that conflicts with schema field and is not an identity or void transform.", partition_field.name() )); } if !has_matching_source_id { return Err(invalid_data!( - "Cannot create identity partition sourced from different field in schema. \ + "Cannot create identity or void partition sourced from different field in schema. \ Field name '{}' has id `{}` in schema but partition source id is `{}`", partition_field.name(), schema_field.id, @@ -2471,6 +2471,92 @@ mod tests { )); } + #[test] + fn test_add_partition_spec_accepts_void_fields_of_removed_fields() { + let builder = builder_without_changes(FormatVersion::V1); + let void_field = |field_id: i32, name: &str| { + UnboundPartitionField::builder() + .source_ids(vec![2]) + .field_id(field_id) + .name(name) + .transform(Transform::Void) + .build() + .unwrap() + }; + let added_spec = UnboundPartitionSpec { + spec_id: None, + fields: vec![void_field(1000, "y"), void_field(1001, "y_null")], + }; + + let metadata = builder + .add_default_partition_spec(added_spec.clone()) + .unwrap() + .build() + .unwrap() + .metadata; + + let expected = added_spec.with_spec_id(1).bind(schema()).unwrap(); + assert_eq!(metadata.default_partition_spec().as_ref(), &expected); + assert_eq!(metadata.last_partition_id, 1001); + } + + #[test] + fn test_add_partition_spec_accepts_two_time_granularities_of_one_column() { + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "ts", Type::Primitive(PrimitiveType::Timestamptz)).into(), + ]) + .build() + .unwrap(); + let time_field = |field_id: i32, name: &str, transform: Transform| { + UnboundPartitionField::builder() + .source_ids(vec![1]) + .field_id(field_id) + .name(name) + .transform(transform) + .build() + .unwrap() + }; + let added_spec = UnboundPartitionSpec { + spec_id: None, + fields: vec![ + time_field(1000, "ts_day", Transform::Day), + time_field(1001, "ts_hour", Transform::Hour), + ], + }; + let metadata = TableMetadataBuilder::new( + schema.clone(), + PartitionSpec::unpartition_spec(), + SortOrder::unsorted_order(), + TEST_LOCATION.to_string(), + FormatVersion::V2, + HashMap::new(), + ) + .unwrap() + .build() + .unwrap() + .metadata; + + let metadata = metadata + .into_builder(None) + .add_default_partition_spec(added_spec.clone()) + .unwrap() + .build() + .unwrap() + .metadata; + // Every later commit re-validates the default spec. + let metadata = metadata + .into_builder(None) + .set_properties(HashMap::from([("key".to_string(), "value".to_string())])) + .unwrap() + .build() + .unwrap() + .metadata; + + let expected = added_spec.with_spec_id(1).bind(schema).unwrap(); + assert_eq!(metadata.default_partition_spec().as_ref(), &expected); + } + #[test] fn test_expire_metadata_log() { let builder = builder_without_changes(FormatVersion::V2); @@ -2957,7 +3043,7 @@ mod tests { assert!(error_message.contains( "Cannot create partition with name 'existing_field' that conflicts with schema field" )); - assert!(error_message.contains("and is not an identity transform")); + assert!(error_message.contains("and is not an identity or void transform")); } #[test] diff --git a/crates/iceberg/src/spec/transform.rs b/crates/iceberg/src/spec/transform.rs index 0d2b61e73..f52202b78 100644 --- a/crates/iceberg/src/spec/transform.rs +++ b/crates/iceberg/src/spec/transform.rs @@ -404,6 +404,14 @@ impl Transform { ) } + /// Whether a partition field with this transform may be named after its source column. + /// + /// Void is allowed because removing an identity field from a format version 1 spec leaves + /// a void field that keeps the identity field's name. + pub(crate) fn may_share_name_with_source(&self) -> bool { + matches!(self, Transform::Identity | Transform::Void) + } + /// Return the unique transform name to check if similar transforms for the same source field /// are added multiple times in partition spec builder. pub fn dedup_name(&self) -> String {
