This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new b5cc6a1c1 fix(spec): Skip the partition name check once the source
column is dropped (#3420)
b5cc6a1c1 is described below
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]
