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-3420-340b6fa0d1e44fbcfdac68174298519e35819e47 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit b5cc6a1c179eafd3bc3159f1f537b8928f293c15 Author: Xander <[email protected]> AuthorDate: Tue Oct 27 14:17:57 2026 +0000 fix(spec): Skip the partition name check once the source column is dropped (#3420) * fix(spec): skip the partition name check once the source column is dropped `validate_schema_field_names` rejected any new schema field whose name matched a partition field name in any spec, unless some schema already had that name. Iceberg keeps specs forever, so a name used by a spec whose source column has since been dropped stayed reserved even though the spec is only retained to interpret old metadata. Iceberg Java runs the same check, in `checkAndAddPartitionName`, as `setCurrentSchema` rebuilds every spec. It skips a field whose source column is absent from the new schema, and otherwise allows the name only for the source column itself: boolean sourceFieldExists = schema.findField(sourceColumnId) != null; if (sourceFieldExists) { Preconditions.checkArgument( schemaField == null || schemaField.fieldId() == sourceColumnId, ...); } Adopt that rule. It keeps every spec in scope, so nothing that was rejected before is accepted now unless its source column is gone, and it drops the `name_exists_in_any_schema` escape hatch, which was a Rust-only approximation of the identity-partition case. The check now reads partition fields rather than schema names, so its result no longer depends on whether the schema or the spec is added first. `partition_name_exists` loses its only caller and is removed. Note a related divergence this does not address: `build()` rebinds the default spec with `bind()`, which requires the source column to exist, so dropping the source column of the *current* spec still fails. Java uses `buildUnchecked()` there and tolerates it. * maybe this? * update error message --- crates/iceberg/src/spec/table_metadata.rs | 86 ------- crates/iceberg/src/spec/table_metadata_builder.rs | 290 +++++++++++++++++++--- 2 files changed, 262 insertions(+), 114 deletions(-) diff --git a/crates/iceberg/src/spec/table_metadata.rs b/crates/iceberg/src/spec/table_metadata.rs index 99df566f6..63199742f 100644 --- a/crates/iceberg/src/spec/table_metadata.rs +++ b/crates/iceberg/src/spec/table_metadata.rs @@ -153,14 +153,6 @@ impl TableMetadata { TableMetadataBuilder::new_from_metadata(self, current_file_location) } - /// Check if a partition field name exists in any partition spec. - #[inline] - pub(crate) fn partition_name_exists(&self, name: &str) -> bool { - self.partition_specs - .values() - .any(|spec| spec.fields().iter().any(|pf| pf.name == name)) - } - /// Check if a field name exists in any schema. #[inline] pub(crate) fn name_exists_in_any_schema(&self, name: &str) -> bool { @@ -3830,82 +3822,6 @@ mod tests { assert_eq!(read_metadata, compressed_metadata); } - #[test] - fn test_partition_name_exists() { - let schema = Schema::builder() - .with_fields(vec![ - NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(), - NestedField::required(2, "partition_col", Type::Primitive(PrimitiveType::Int)) - .into(), - ]) - .build() - .unwrap(); - - let spec1 = PartitionSpec::builder(schema.clone()) - .with_spec_id(1) - .add_partition_field("data", "data_partition", Transform::Identity) - .unwrap() - .build() - .unwrap(); - - let spec2 = PartitionSpec::builder(schema.clone()) - .with_spec_id(2) - .add_partition_field("partition_col", "partition_bucket", Transform::Bucket(16)) - .unwrap() - .build() - .unwrap(); - - // Build metadata with these specs - let metadata = TableMetadataBuilder::new( - schema, - spec1.clone().into_unbound(), - SortOrder::unsorted_order(), - "s3://test/location".to_string(), - FormatVersion::V2, - HashMap::new(), - ) - .unwrap() - .add_partition_spec(spec2.into_unbound()) - .unwrap() - .build() - .unwrap() - .metadata; - - assert!(metadata.partition_name_exists("data_partition")); - assert!(metadata.partition_name_exists("partition_bucket")); - - assert!(!metadata.partition_name_exists("nonexistent_field")); - assert!(!metadata.partition_name_exists("data")); // schema field name, not partition field name - assert!(!metadata.partition_name_exists("")); - } - - #[test] - fn test_partition_name_exists_empty_specs() { - // Create metadata with no partition specs (unpartitioned table) - let schema = Schema::builder() - .with_fields(vec![ - NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(), - ]) - .build() - .unwrap(); - - let metadata = TableMetadataBuilder::new( - schema, - PartitionSpec::unpartition_spec().into_unbound(), - SortOrder::unsorted_order(), - "s3://test/location".to_string(), - FormatVersion::V2, - HashMap::new(), - ) - .unwrap() - .build() - .unwrap() - .metadata; - - assert!(!metadata.partition_name_exists("any_field")); - assert!(!metadata.partition_name_exists("data")); - } - #[test] fn test_name_exists_in_any_schema() { // Create multiple schemas with different fields @@ -4034,8 +3950,6 @@ mod tests { .unwrap() .metadata; - assert!(!final_metadata.partition_name_exists("nonexistent_partition")); // unpartitioned table - assert!(final_metadata.name_exists_in_any_schema("id")); // exists in both schemas assert!(final_metadata.name_exists_in_any_schema("name")); // exists in both schemas assert!(final_metadata.name_exists_in_any_schema("deprecated_field")); // exists in both schemas diff --git a/crates/iceberg/src/spec/table_metadata_builder.rs b/crates/iceberg/src/spec/table_metadata_builder.rs index a8ef84a18..6d78f77d4 100644 --- a/crates/iceberg/src/spec/table_metadata_builder.rs +++ b/crates/iceberg/src/spec/table_metadata_builder.rs @@ -606,9 +606,6 @@ impl TableMetadataBuilder { /// Important: Use this method with caution. The builder does not check /// if the added schema is compatible with the current schema. pub fn add_schema(mut self, schema: Schema) -> Result<Self> { - // Validate that new schema fields don't conflict with existing partition field names - self.validate_schema_field_names(&schema)?; - let new_schema_id = self.reuse_or_create_new_schema_id(&schema); let schema_found = self.metadata.schemas.contains_key(&new_schema_id); @@ -652,6 +649,7 @@ impl TableMetadataBuilder { /// Errors: /// - provided `schema_id` is -1 but no schema has been added via `add_schema`. /// - No schema with the provided `schema_id` exists. + /// - A field of the schema takes the name of a partition field it is not the source of. pub fn set_current_schema(mut self, mut schema_id: i32) -> Result<Self> { if schema_id == Self::LAST_ADDED { schema_id = self.last_added_schema_id.ok_or_else(|| { @@ -666,9 +664,10 @@ impl TableMetadataBuilder { return Ok(self); } - let _schema = self.metadata.schemas.get(&schema_id).ok_or_else(|| { + let schema = self.metadata.schemas.get(&schema_id).ok_or_else(|| { invalid_data!("Cannot set current schema to unknown schema with id: '{schema_id}'") })?; + self.validate_schema_field_names(schema)?; // Old partition specs and sort-orders should be preserved even if they are not compatible with the new schema, // so that older metadata can still be interpreted. @@ -695,28 +694,37 @@ impl TableMetadataBuilder { .set_current_schema(Self::LAST_ADDED) } - /// Validate schema field names against partition field names across all historical schemas. + /// Validate schema field names against the partition field names of every spec. /// - /// Due to Iceberg's multi-version property, this check ignores existing schema fields - /// that match partition names (schema evolution allows re-adding previously removed fields). - /// Only NEW field names that conflict with partition names are rejected. + /// A partition field whose source column is absent from `schema` is skipped: its spec is + /// historical, kept only so old metadata can be interpreted, so the name is free to reuse. + /// Otherwise the name may only be taken by the source column itself, which is the identity + /// partition case that [`Self::validate_partition_field_names`] allows in the other direction. /// /// # Errors - /// - Schema field name conflicts with partition field name but doesn't exist in any historical schema. + /// - A schema field takes the name of a partition field it is not the source of. fn validate_schema_field_names(&self, schema: &Schema) -> Result<()> { - if self.metadata.schemas.is_empty() { - return Ok(()); - } + for (spec_id, spec) in &self.metadata.partition_specs { + for partition_field in spec.fields() { + if schema.field_by_id(partition_field.source_id).is_none() { + continue; + } - for field_name in schema.field_id_to_name_map().values() { - let has_partition_conflict = self.metadata.partition_name_exists(field_name); - let is_new_field = !self.metadata.name_exists_in_any_schema(field_name); + let Some(schema_field) = schema.field_by_name(&partition_field.name) else { + continue; + }; - if has_partition_conflict && is_new_field { - return Err(invalid_data!( - "Cannot add schema field '{field_name}' because it conflicts with existing partition field name. \ - Schema evolution cannot introduce field names that match existing partition field names." - )); + if schema_field.id != partition_field.source_id { + return Err(invalid_data!( + "Cannot set current schema {}: field '{}' with id {} conflicts with a \ + partition field of that name in spec {} sourced from field id {}", + schema.schema_id(), + partition_field.name, + schema_field.id, + spec_id, + partition_field.source_id + )); + } } } @@ -2922,8 +2930,8 @@ mod tests { assert!(result.is_err()); let error = result.unwrap_err(); let error_message = error.message(); - assert!(error_message.contains("Cannot add schema field 'bucket_data' because it conflicts with existing partition field name")); - assert!(error_message.contains("Schema evolution cannot introduce field names that match existing partition field names")); + assert!(error_message.contains("Cannot set current schema")); + assert!(error_message.contains("sourced from field id 1")); } #[test] @@ -3142,11 +3150,12 @@ mod tests { let builder2 = metadata.into_builder(Some("test_location".to_string())); let result2 = builder2.add_current_schema(conflicting_schema); - // This should fail because "bucket_data" conflicts with partition field name - // and doesn't exist in any historical schema + // This should fail because "bucket_data" takes the name of a partition field whose + // source column `data` is still in the schema assert!(result2.is_err()); let error = result2.unwrap_err(); - assert!(error.message().contains("Cannot add schema field 'bucket_data' because it conflicts with existing partition field name")); + assert!(error.message().contains("Cannot set current schema")); + assert!(error.message().contains("sourced from field id 1")); } #[test] @@ -3207,6 +3216,230 @@ mod tests { assert!(result.is_ok()); } + /// A v2 table of `data: string` partitioned by `bucket(data, 16)` as `bucket_data`. + fn bucket_partitioned_metadata() -> TableMetadata { + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(), + ]) + .build() + .unwrap(); + let spec = UnboundPartitionSpec::builder() + .with_spec_id(0) + .add_partition_field( + UnboundPartitionField::builder() + .source_ids(vec![1]) + .name("bucket_data") + .transform(Transform::Bucket(16)) + .build() + .unwrap(), + ) + .unwrap() + .build(); + + TableMetadataBuilder::new( + schema, + spec, + SortOrder::unsorted_order(), + TEST_LOCATION.to_string(), + FormatVersion::V2, + HashMap::new(), + ) + .unwrap() + .build() + .unwrap() + .metadata + } + + /// Replaces the default spec with an unpartitioned one. Spec 0 is retained but superseded. + fn unpartition(metadata: TableMetadata, file: &str) -> TableMetadata { + let metadata = metadata + .into_builder(Some(file.to_string())) + .add_default_partition_spec(UnboundPartitionSpec::builder().build()) + .unwrap() + .build() + .unwrap() + .metadata; + assert_ne!(metadata.default_partition_spec().spec_id(), 0); + assert!(metadata.partition_specs.contains_key(&0)); + metadata + } + + #[test] + fn test_schema_evolution_allows_a_superseded_partition_field_name_once_its_source_is_dropped() { + // `data`, the source of spec 0's `bucket_data`, is gone, so spec 0 can only be historical + // and its names are free. Iceberg Java skips its name check for exactly this case. + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(); + + let metadata = unpartition( + bucket_partitioned_metadata(), + "s3://bucket/t/metadata/m1.json", + ) + .into_builder(Some("s3://bucket/t/metadata/m2.json".to_string())) + .add_current_schema(schema.clone()) + .unwrap() + .build() + .unwrap() + .metadata; + + assert_eq!(metadata.current_schema().as_struct(), schema.as_struct()); + assert!(metadata.current_schema().field_by_name("data").is_none()); + // Spec 0 is kept so the snapshots written under it stay interpretable. + assert_eq!( + metadata.partition_spec_by_id(0).unwrap().fields()[0].name, + "bucket_data" + ); + } + + #[test] + fn test_schema_evolution_rejects_a_partition_field_name_while_its_source_remains() { + // `data` is still here, so `bucket_data` would be ambiguous. Iceberg Java rejects this + // whether or not spec 0 is still the default, because it rebuilds every spec. + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(), + NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(); + + let error = bucket_partitioned_metadata() + .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string())) + .add_current_schema(schema.clone()) + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::DataInvalid); + assert!( + error.message().contains("sourced from field id 1"), + "{error}" + ); + + let error = unpartition( + bucket_partitioned_metadata(), + "s3://bucket/t/metadata/m1.json", + ) + .into_builder(Some("s3://bucket/t/metadata/m2.json".to_string())) + .add_current_schema(schema) + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::DataInvalid); + assert!( + error.message().contains("sourced from field id 1"), + "{error}" + ); + } + + #[test] + fn test_schema_evolution_partition_name_check_is_order_independent() { + // Dropping the source column and dropping the partitioning in one set of changes must + // behave the same either way round. + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(); + let unpartitioned = UnboundPartitionSpec::builder().build(); + + let schema_first = bucket_partitioned_metadata() + .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string())) + .add_current_schema(schema.clone()) + .and_then(|b| b.add_default_partition_spec(unpartitioned.clone())) + .and_then(|b| b.build()); + let spec_first = bucket_partitioned_metadata() + .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string())) + .add_default_partition_spec(unpartitioned) + .and_then(|b| b.add_current_schema(schema)) + .and_then(|b| b.build()); + + assert!(schema_first.is_ok(), "{:?}", schema_first.err()); + assert!(spec_first.is_ok(), "{:?}", spec_first.err()); + } + + #[test] + fn test_set_current_schema_rejects_a_partition_field_name_from_a_spec_added_after_the_schema() { + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(), + ]) + .build() + .unwrap(); + let conflicting_schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(), + NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(); + let spec = UnboundPartitionSpec::builder() + .add_partition_field( + UnboundPartitionField::builder() + .source_ids(vec![1]) + .name("bucket_data") + .transform(Transform::Bucket(16)) + .build() + .unwrap(), + ) + .unwrap() + .build(); + + let error = TableMetadataBuilder::new( + schema, + UnboundPartitionSpec::builder().build(), + SortOrder::unsorted_order(), + TEST_LOCATION.to_string(), + FormatVersion::V2, + HashMap::new(), + ) + .unwrap() + .add_schema(conflicting_schema) + .and_then(|b| b.add_partition_spec(spec)) + .and_then(|b| b.set_current_schema(TableMetadataBuilder::LAST_ADDED)) + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::DataInvalid); + assert!( + error.message().contains("sourced from field id 1"), + "{error}" + ); + } + + #[test] + fn test_schema_with_a_conflicting_partition_field_name_can_be_added_but_not_made_current() { + let conflicting_schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(), + NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(); + + let metadata = bucket_partitioned_metadata() + .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string())) + .add_schema(conflicting_schema.clone()) + .unwrap() + .build() + .unwrap() + .metadata; + assert_eq!(metadata.current_schema_id(), 0); + assert_eq!( + metadata.schema_by_id(1).unwrap().as_struct(), + conflicting_schema.as_struct() + ); + + let error = metadata + .into_builder(Some("s3://bucket/t/metadata/m2.json".to_string())) + .set_current_schema(1) + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::DataInvalid); + assert!( + error.message().contains("sourced from field id 1"), + "{error}" + ); + } + #[test] fn test_schema_evolution_prevents_new_field_conflicting_with_partition_field() { // Create initial schema WITHOUT the conflicting field @@ -3256,11 +3489,12 @@ mod tests { let builder = metadata.into_builder(Some("test_location".to_string())); let result = builder.add_current_schema(conflicting_schema); - // This should fail because "bucket_data" conflicts with partition field name - // and doesn't exist in any historical schema + // This should fail because "bucket_data" takes the name of a partition field whose + // source column `data` is still in the schema assert!(result.is_err()); let error = result.unwrap_err(); - assert!(error.message().contains("Cannot add schema field 'bucket_data' because it conflicts with existing partition field name")); + assert!(error.message().contains("Cannot set current schema")); + assert!(error.message().contains("sourced from field id 1")); } #[test]
