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_conflicts(&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,54 @@ 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.
+    ///
+    /// This validation prevents introducing new field names that match existing partition field names
+    /// that don't correspond to any field in the current schema. Field names that already exist in the
+    /// current schema are allowed since partition fields are typically derived from schema fields.
+    ///
+    /// # Errors
+    /// - Schema field name conflicts with existing partition field name that doesn't correspond to any current schema field.
+    fn check_schema_partition_name_conflicts(&self, schema: &Schema) -> Result<()> {
+        let existing_partition_field_names: HashSet<&str> = self
+            .metadata
+            .partition_specs
+            .values()
+            .flat_map(|spec| spec.fields().iter().map(|field| field.name.as_str()))
+            .collect();
+
+        let current_schema_field_names: HashSet<&str> = self
+            .metadata
+            .schemas
+            .get(&self.metadata.current_schema_id)
+            .map(|current_schema| {
+                current_schema
+                    .as_struct()
+                    .fields()
+                    .iter()
+                    .map(|field| field.name.as_str())
+                    .collect()
+            })
+            .unwrap_or_default();
+
+        if let Some(conflicting_field) = schema.as_struct().fields().iter().find(|field| {
+            existing_partition_field_names.contains(field.name.as_str())
+                && !current_schema_field_names.contains(field.name.as_str())
+        }) {
+            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.",
+                    conflicting_field.name
+                ),
+            ));
+        }
+
+        Ok(())
     }
 
     /// Add a partition spec to the table metadata.
@@ -1859,6 +1909,7 @@ mod tests {
 
         let build_result = builder
             .add_schema(added_schema.clone())
+            .unwrap()
             .set_current_schema(1)
             .unwrap()
             .build()
@@ -2496,4 +2547,110 @@ 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());
+    }
 }