fix(schema): check if new schema field names conflict with existing partition field names (#1609)

- Add a validation step in add_schema() method to prevent conflicts

Author: fvaleye

crates/iceberg/src/catalog/mod.rs

@@ -561,7 +561,7 @@ impl TableUpdate {
     pub fn apply(self, builder: TableMetadataBuilder) -> Result<TableMetadataBuilder> {
         match self {
             TableUpdate::AssignUuid { uuid } => Ok(builder.assign_uuid(uuid)),
-            TableUpdate::AddSchema { schema, .. } => Ok(builder.add_schema(schema)),
+            TableUpdate::AddSchema { schema, .. } => Ok(builder.add_schema(schema)?),
             TableUpdate::SetCurrentSchema { schema_id } => builder.set_current_schema(schema_id),
             TableUpdate::AddSpec { spec } => builder.add_partition_spec(spec),
             TableUpdate::SetDefaultSpec { spec_id } => builder.set_default_partition_spec(spec_id),

crates/iceberg/src/spec/table_metadata_builder.rs

@@ -589,7 +589,10 @@ 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) -> Self {
+    pub fn add_schema(mut self, schema: Schema) -> Result<Self> {
+        // Validate that new schema fields don't conflict with existing partition field names
+        self.check_schema_partition_name_conflict(&schema)?;
+
         let new_schema_id = self.reuse_or_create_new_schema_id(&schema);
         let schema_found = self.metadata.schemas.contains_key(&new_schema_id);
 
@@ -601,7 +604,7 @@ impl TableMetadataBuilder {
                 self.last_added_schema_id = Some(new_schema_id);
             }
 
-            return self;
+            return Ok(self);
         }
 
         // New schemas might contain only old columns. In this case last_column_id should not be
@@ -623,7 +626,7 @@ impl TableMetadataBuilder {
 
         self.last_added_schema_id = Some(new_schema_id);
 
-        self
+        Ok(self)
     }
 
     /// Set the current schema id.
@@ -679,7 +682,41 @@ impl TableMetadataBuilder {
 
     /// Add a schema and set it as the current schema.
     pub fn add_current_schema(self, schema: Schema) -> Result<Self> {
-        self.add_schema(schema).set_current_schema(Self::LAST_ADDED)
+        self.add_schema(schema)?
+            .set_current_schema(Self::LAST_ADDED)
+    }
+
+    /// Check if schema field names conflict with existing partition field names.
+    ///
+    /// # Errors
+    /// - Schema field name conflicts with existing partition field name.  
+    fn check_schema_partition_name_conflict(&self, schema: &Schema) -> Result<()> {
+        let current_schema = match self.get_current_schema() {
+            Ok(schema) => schema,
+            Err(_) => return Ok(()), // No current schema, no conflicts possible
+        };
+
+        for field_name in schema.field_id_to_name_map().values() {
+            let conflicts_with_partition_field =
+                self.metadata.partition_specs.values().any(|spec| {
+                    spec.fields()
+                        .iter()
+                        .any(|partition_field| &partition_field.name == field_name)
+                });
+
+            if conflicts_with_partition_field && current_schema.field_by_name(field_name).is_none()
+            {
+                return Err(Error::new(
+                    ErrorKind::DataInvalid,
+                    format!(
+                        "Cannot add schema field '{}' because it conflicts with existing partition field name. Schema evolution cannot introduce field names that match existing partition field names.",
+                        field_name
+                    ),
+                ));
+            }
+        }
+
+        Ok(())
     }
 
     /// Add a partition spec to the table metadata.
@@ -694,6 +731,7 @@ impl TableMetadataBuilder {
     /// - The partition spec has non-sequential field ids and the table format version is 1.
     pub fn add_partition_spec(mut self, unbound_spec: UnboundPartitionSpec) -> Result<Self> {
         let schema = self.get_current_schema()?.clone();
+
         let spec = PartitionSpecBuilder::new_from_unbound(unbound_spec.clone(), schema)?
             .with_last_assigned_field_id(self.metadata.last_partition_id)
             .build()?;
@@ -1859,6 +1897,7 @@ mod tests {
 
         let build_result = builder
             .add_schema(added_schema.clone())
+            .unwrap()
             .set_current_schema(1)
             .unwrap()
             .build()
@@ -2496,4 +2535,208 @@ mod tests {
             };
         assert_eq!(remove_schema_ids, &[0]);
     }
+
+    #[test]
+    fn test_schema_evolution_now_correctly_validates_partition_field_name_conflicts() {
+        let initial_schema = Schema::builder()
+            .with_fields(vec![
+                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
+            ])
+            .build()
+            .unwrap();
+
+        let partition_spec_with_bucket = UnboundPartitionSpec::builder()
+            .with_spec_id(0)
+            .add_partition_field(1, "bucket_data", Transform::Bucket(16))
+            .unwrap()
+            .build();
+
+        let metadata = TableMetadataBuilder::new(
+            initial_schema,
+            partition_spec_with_bucket,
+            SortOrder::unsorted_order(),
+            TEST_LOCATION.to_string(),
+            FormatVersion::V2,
+            HashMap::new(),
+        )
+        .unwrap()
+        .build()
+        .unwrap()
+        .metadata;
+
+        let partition_field_names: Vec<String> = metadata
+            .default_partition_spec()
+            .fields()
+            .iter()
+            .map(|f| f.name.clone())
+            .collect();
+        assert!(partition_field_names.contains(&"bucket_data".to_string()));
+
+        let evolved_schema = Schema::builder()
+            .with_fields(vec![
+                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
+                // Adding a schema field with the same name as an existing partition field
+                NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
+            ])
+            .build()
+            .unwrap();
+
+        let builder = metadata.into_builder(Some(
+            "s3://bucket/test/location/metadata/metadata1.json".to_string(),
+        ));
+
+        // Try to add the evolved schema - this should now fail immediately with a clear error
+        let result = builder.add_current_schema(evolved_schema);
+
+        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"));
+    }
+
+    #[test]
+    fn test_schema_evolution_should_validate_on_schema_add_not_metadata_build() {
+        let initial_schema = Schema::builder()
+            .with_fields(vec![
+                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
+            ])
+            .build()
+            .unwrap();
+
+        let partition_spec = UnboundPartitionSpec::builder()
+            .with_spec_id(0)
+            .add_partition_field(1, "partition_col", Transform::Bucket(16))
+            .unwrap()
+            .build();
+
+        let metadata = TableMetadataBuilder::new(
+            initial_schema,
+            partition_spec,
+            SortOrder::unsorted_order(),
+            TEST_LOCATION.to_string(),
+            FormatVersion::V2,
+            HashMap::new(),
+        )
+        .unwrap()
+        .build()
+        .unwrap()
+        .metadata;
+
+        let non_conflicting_schema = Schema::builder()
+            .with_fields(vec![
+                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
+                NestedField::required(2, "new_field", Type::Primitive(PrimitiveType::Int)).into(),
+            ])
+            .build()
+            .unwrap();
+
+        // This should succeed since there's no name conflict
+        let result = metadata
+            .clone()
+            .into_builder(Some("test_location".to_string()))
+            .add_current_schema(non_conflicting_schema)
+            .unwrap()
+            .build();
+
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_partition_spec_evolution_validates_schema_field_name_conflicts() {
+        let initial_schema = Schema::builder()
+            .with_fields(vec![
+                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
+                NestedField::required(2, "existing_field", Type::Primitive(PrimitiveType::Int))
+                    .into(),
+            ])
+            .build()
+            .unwrap();
+
+        let partition_spec = UnboundPartitionSpec::builder()
+            .with_spec_id(0)
+            .add_partition_field(1, "data_bucket", Transform::Bucket(16))
+            .unwrap()
+            .build();
+
+        let metadata = TableMetadataBuilder::new(
+            initial_schema,
+            partition_spec,
+            SortOrder::unsorted_order(),
+            TEST_LOCATION.to_string(),
+            FormatVersion::V2,
+            HashMap::new(),
+        )
+        .unwrap()
+        .build()
+        .unwrap()
+        .metadata;
+
+        let builder = metadata.into_builder(Some(
+            "s3://bucket/test/location/metadata/metadata1.json".to_string(),
+        ));
+
+        let conflicting_partition_spec = UnboundPartitionSpec::builder()
+            .with_spec_id(1)
+            .add_partition_field(1, "existing_field", Transform::Bucket(8))
+            .unwrap()
+            .build();
+
+        let result = builder.add_partition_spec(conflicting_partition_spec);
+
+        assert!(result.is_err());
+        let error = result.unwrap_err();
+        let error_message = error.message();
+        // The error comes from existing PartitionSpecBuilder validation
+        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"));
+    }
+
+    #[test]
+    fn test_partition_spec_evolution_allows_non_conflicting_names() {
+        let initial_schema = Schema::builder()
+            .with_fields(vec![
+                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
+                NestedField::required(2, "existing_field", Type::Primitive(PrimitiveType::Int))
+                    .into(),
+            ])
+            .build()
+            .unwrap();
+
+        let partition_spec = UnboundPartitionSpec::builder()
+            .with_spec_id(0)
+            .add_partition_field(1, "data_bucket", Transform::Bucket(16))
+            .unwrap()
+            .build();
+
+        let metadata = TableMetadataBuilder::new(
+            initial_schema,
+            partition_spec,
+            SortOrder::unsorted_order(),
+            TEST_LOCATION.to_string(),
+            FormatVersion::V2,
+            HashMap::new(),
+        )
+        .unwrap()
+        .build()
+        .unwrap()
+        .metadata;
+
+        let builder = metadata.into_builder(Some(
+            "s3://bucket/test/location/metadata/metadata1.json".to_string(),
+        ));
+
+        // Try to add a partition spec with a field name that does NOT conflict with existing schema fields
+        let non_conflicting_partition_spec = UnboundPartitionSpec::builder()
+            .with_spec_id(1)
+            .add_partition_field(2, "new_partition_field", Transform::Bucket(8))
+            .unwrap()
+            .build();
+
+        let result = builder.add_partition_spec(non_conflicting_partition_spec);
+
+        assert!(result.is_ok());
+    }
 }