feat: add column mapping support for add_column

When column mapping is enabled (mode = name or id), newly added columns
via ALTER TABLE ADD COLUMN are now assigned column mapping metadata
(unique ID and UUID-based physical name). The maxColumnId table property
is updated in the evolved metadata configuration.

Reuses the existing assign_field_column_mapping function (now pub(crate))
which handles recursive assignment for nested types.

Author: william-ch-databricks

kernel/src/actions/mod.rs

@@ -355,6 +355,14 @@ impl Metadata {
         })
     }
 
+    /// Returns a new Metadata with the configuration replaced, preserving all other fields.
+    pub(crate) fn with_configuration(self, configuration: HashMap<String, String>) -> Self {
+        Self {
+            configuration,
+            ..self
+        }
+    }
+
     #[cfg(test)]
     #[allow(clippy::too_many_arguments)]
     pub(crate) fn new_unchecked(

kernel/src/table_features/column_mapping.rs

@@ -245,7 +245,7 @@ pub(crate) fn get_column_mapping_mode_from_properties(
 /// arrays, and maps. Each field is assigned a new unique ID and physical name.
 ///
 /// Fields with pre-existing column mapping metadata (id or physicalName) are rejected
-/// to avoid conflicts. ALTER TABLE will need different handling in the future.
+/// to avoid conflicts.
 ///
 /// # Arguments
 ///
@@ -271,24 +271,27 @@ pub(crate) fn assign_column_mapping_metadata(
 /// Assigns column mapping metadata to a single field, recursively processing nested types.
 ///
 /// Rejects fields with pre-existing column mapping metadata. Otherwise, assigns a new
-/// unique ID and physical name (incrementing `max_id`).
-fn assign_field_column_mapping(field: &StructField, max_id: &mut i64) -> DeltaResult<StructField> {
+/// unique ID and physical name (incrementing `max_id`). Used during CREATE TABLE for the full
+/// schema and during ALTER TABLE ADD COLUMN for newly added fields.
+pub(crate) fn assign_field_column_mapping(
+    field: &StructField,
+    max_id: &mut i64,
+) -> DeltaResult<StructField> {
     let has_id = field
         .get_config_value(&ColumnMetadataKey::ColumnMappingId)
         .is_some();
     let has_physical_name = field
         .get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName)
         .is_some();
 
-    // For CREATE TABLE, reject any pre-existing column mapping metadata.
-    // This avoids conflicts between user-provided IDs/physical names and the ones we assign.
-    // ALTER TABLE (adding columns) will need different handling in the future.
+    // Reject fields with pre-existing column mapping metadata to avoid ID/physical-name
+    // conflicts with the ones we assign.
     // TODO: Also check for nested column IDs (`delta.columnMapping.nested.ids`) once
     // Iceberg compatibility (IcebergCompatV2+) is supported. See issue #1125.
     if has_id || has_physical_name {
         return Err(Error::generic(format!(
             "Field '{}' already has column mapping metadata. \
-             Pre-existing column mapping metadata is not supported for CREATE TABLE.",
+             Pre-existing column mapping metadata is not supported when adding columns.",
             field.name
         )));
     }

kernel/src/table_features/mod.rs

@@ -21,6 +21,21 @@ use crate::schema::derive_macro_utils::ToDataType;
 use crate::schema::DataType;
 use crate::table_properties::TableProperties;
 use crate::{DeltaResult, Error};
+use delta_kernel_derive::internal_api;
+
+#[internal_api]
+pub(crate) use column_mapping::get_any_level_column_physical_name;
+pub(crate) use column_mapping::physical_to_logical_column_name;
+#[deprecated = "Enable internal-api and use TableConfiguration instead"]
+pub use column_mapping::validate_schema_column_mapping;
+pub use column_mapping::ColumnMappingMode;
+pub(crate) use column_mapping::{
+    assign_column_mapping_metadata, assign_field_column_mapping, column_mapping_mode,
+    get_column_mapping_mode_from_properties, get_field_column_mapping_info,
+};
+pub(crate) use timestamp_ntz::{
+    schema_contains_timestamp_ntz, validate_timestamp_ntz_feature_support,
+};
 mod column_mapping;
 mod timestamp_ntz;
 

kernel/src/transaction/builder/alter_table.rs

@@ -72,7 +72,9 @@ impl AlterTableTransactionBuilder<Ready> {
     ///
     /// The field must not already exist in the schema (case-insensitive). The field must be
     /// nullable because existing data files do not contain this column and will read NULL for it.
-    /// These constraints are validated during [`build()`](AlterTableTransactionBuilder::build).
+    /// If column mapping is enabled, the builder automatically assigns a new column ID and physical
+    /// name at build time. These constraints are validated during
+    /// [`build()`](AlterTableTransactionBuilder::build).
     pub fn add_column(mut self, field: StructField) -> AlterTableTransactionBuilder<Modifying> {
         self.operations.push(SchemaOperation::AddColumn { field });
         self.transition()
@@ -115,22 +117,50 @@ impl AlterTableTransactionBuilder<Modifying> {
     ///   `timestampNtz` column without the `timestampNtz` feature)
     pub fn build(
         self,
-        _engine: &dyn Engine,
+        _engine: &dyn Engine, // used by later operations (e.g., drop_column reads clustering columns)
         committer: Box<dyn Committer>,
     ) -> DeltaResult<AlterTableTransaction> {
         let table_config = self.snapshot.table_configuration();
         table_config.ensure_operation_supported(Operation::Write)?;
 
         let schema = Arc::unwrap_or_clone(table_config.logical_schema());
+        let column_mapping_mode = table_config.column_mapping_mode();
+        let current_max_column_id = table_config
+            .metadata()
+            .configuration()
+            .get(crate::table_properties::COLUMN_MAPPING_MAX_COLUMN_ID)
+            .map(|v| {
+                v.parse::<i64>().map_err(|_| {
+                    crate::Error::generic(format!(
+                        "Invalid delta.columnMapping.maxColumnId value: '{v}'"
+                    ))
+                })
+            })
+            .transpose()?;
         let SchemaEvolutionResult {
             schema: evolved_schema,
-        } = apply_schema_operations(schema, self.operations)?;
+            new_max_column_id,
+        } = apply_schema_operations(
+            schema,
+            self.operations,
+            column_mapping_mode,
+            current_max_column_id,
+        )?;
 
-        let evolved_metadata = table_config
+        let mut evolved_metadata = table_config
             .metadata()
             .clone()
             .with_schema(evolved_schema.clone())?;
 
+        if let Some(new_id) = new_max_column_id {
+            let mut config = table_config.metadata().configuration().clone();
+            config.insert(
+                crate::table_properties::COLUMN_MAPPING_MAX_COLUMN_ID.to_string(),
+                new_id.to_string(),
+            );
+            evolved_metadata = evolved_metadata.with_configuration(config);
+        }
+
         // Validates the evolved metadata against the protocol.
         let evolved_table_config = TableConfiguration::try_new_with_schema(
             evolved_metadata,