feat(schema): complete UpdateSchema evolution semantics

crates/catalog/loader/tests/schema_update_suite.rs

@@ -282,3 +282,69 @@ async fn test_catalog_schema_update_persisted_after_reload(
 
     Ok(())
 }
+
+// Common behavior: compatible updates, an explicitly allowed nullability change, renames, and
+// moves are committed atomically and preserve field IDs.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_schema_evolution_operations(#[case] kind: CatalogKind) -> Result<()> {
+    let Some(harness) = load_catalog(kind).await else {
+        return Ok(());
+    };
+    let catalog = harness.catalog;
+    let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+        "catalog_schema_evolution_operations",
+        harness.label
+    ));
+
+    cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+    catalog.create_namespace(&namespace, HashMap::new()).await?;
+    let table_name = normalize_test_name_with_parts!(
+        "catalog_schema_evolution_operations",
+        harness.label,
+        "table"
+    );
+    let table_ident = TableIdent::new(namespace.clone(), table_name.clone());
+    let table = catalog
+        .create_table(
+            &namespace,
+            TableCreation::builder()
+                .name(table_name)
+                .schema(base_schema())
+                .build(),
+        )
+        .await?;
+
+    let tx = Transaction::new(&table);
+    let tx = tx
+        .update_schema()
+        .allow_incompatible_changes()
+        .rename_column("foo", "payload")
+        .update_column_doc("foo", Some("renamed payload".to_string()))
+        .update_column_type("bar", PrimitiveType::Long)
+        .require_column("baz")
+        .move_column_first("baz")
+        .set_identifier_fields(["bar"])
+        .apply(tx)?;
+    tx.commit(catalog.as_ref()).await?;
+
+    let reloaded = catalog.load_table(&table_ident).await?;
+    let schema = reloaded.metadata().current_schema();
+    let payload = schema.field_by_name("payload").unwrap();
+    assert_eq!(payload.id, 1);
+    assert_eq!(payload.doc.as_deref(), Some("renamed payload"));
+    assert_eq!(
+        schema.field_by_name("bar").unwrap().field_type.as_ref(),
+        &Type::Primitive(PrimitiveType::Long)
+    );
+    assert!(schema.field_by_name("baz").unwrap().required);
+    assert_eq!(schema.as_struct().fields()[0].id, 3);
+    assert_eq!(schema.identifier_field_ids().collect::<Vec<_>>(), vec![2]);
+
+    Ok(())
+}

crates/iceberg/public-api.txt

@@ -3236,16 +3236,33 @@ pub fn iceberg::transaction::Transaction::fast_append(&self) -> iceberg::transac
 pub fn iceberg::transaction::Transaction::new(table: &iceberg::table::Table) -> Self
 pub fn iceberg::transaction::Transaction::replace_sort_order(&self) -> iceberg::transaction::sort_order::ReplaceSortOrderAction
 pub fn iceberg::transaction::Transaction::update_location(&self) -> iceberg::transaction::update_location::UpdateLocationAction
-pub fn iceberg::transaction::Transaction::update_schema(&self) -> iceberg::transaction::update_schema::UpdateSchemaAction
+pub fn iceberg::transaction::Transaction::update_schema(&self) -> iceberg::transaction::UpdateSchemaAction
 pub fn iceberg::transaction::Transaction::update_statistics(&self) -> iceberg::transaction::update_statistics::UpdateStatisticsAction
 pub fn iceberg::transaction::Transaction::update_table_properties(&self) -> iceberg::transaction::update_properties::UpdatePropertiesAction
 pub fn iceberg::transaction::Transaction::upgrade_table_version(&self) -> iceberg::transaction::upgrade_format_version::UpgradeFormatVersionAction
 impl core::clone::Clone for iceberg::transaction::Transaction
 pub fn iceberg::transaction::Transaction::clone(&self) -> iceberg::transaction::Transaction
+pub struct iceberg::transaction::UpdateSchemaAction
+impl iceberg::transaction::UpdateSchemaAction
+pub fn iceberg::transaction::UpdateSchemaAction::add_column(self, add_column: iceberg::transaction::AddColumn) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::allow_incompatible_changes(self) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::case_sensitive(self, case_sensitive: bool) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::delete_column(self, name: impl alloc::string::ToString) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::make_column_optional(self, name: impl alloc::string::ToString) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::move_column_after(self, name: impl alloc::string::ToString, after_name: impl alloc::string::ToString) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::move_column_before(self, name: impl alloc::string::ToString, before_name: impl alloc::string::ToString) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::move_column_first(self, name: impl alloc::string::ToString) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::rename_column(self, name: impl alloc::string::ToString, new_name: impl alloc::string::ToString) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::require_column(self, name: impl alloc::string::ToString) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::set_identifier_fields<I, S>(self, names: I) -> Self where I: core::iter::traits::collect::IntoIterator<Item = S>, S: alloc::string::ToString
+pub fn iceberg::transaction::UpdateSchemaAction::union_by_name(self, new_schema: iceberg::spec::Schema) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::update_column_default(self, name: impl alloc::string::ToString, default: core::option::Option<iceberg::spec::Literal>) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::update_column_doc(self, name: impl alloc::string::ToString, doc: core::option::Option<alloc::string::String>) -> Self
+pub fn iceberg::transaction::UpdateSchemaAction::update_column_type(self, name: impl alloc::string::ToString, new_type: iceberg::spec::PrimitiveType) -> Self
 pub trait iceberg::transaction::ApplyTransactionAction
 pub fn iceberg::transaction::ApplyTransactionAction::apply(self, tx: iceberg::transaction::Transaction) -> iceberg::Result<iceberg::transaction::Transaction>
 impl<T: TransactionAction + 'static> iceberg::transaction::ApplyTransactionAction for T
-pub fn T::apply(self, tx: iceberg::transaction::Transaction) -> iceberg::Result<iceberg::transaction::Transaction> where Self: core::marker::Sized
+pub fn T::apply(self, tx: iceberg::transaction::Transaction) -> core::result::Result<iceberg::transaction::Transaction, iceberg::Error>
 pub mod iceberg::transform
 pub trait iceberg::transform::TransformFunction: core::marker::Send + core::marker::Sync + core::fmt::Debug
 pub fn iceberg::transform::TransformFunction::transform(&self, input: arrow_array::array::ArrayRef) -> iceberg::Result<arrow_array::array::ArrayRef>

crates/iceberg/src/transaction/mod.rs

@@ -67,7 +67,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use backon::{BackoffBuilder, ExponentialBackoff, ExponentialBuilder, RetryableWithContext};
-pub use update_schema::AddColumn;
+pub use update_schema::{AddColumn, UpdateSchemaAction};
 
 use crate::error::Result;
 use crate::spec::TableProperties;
@@ -78,7 +78,6 @@ use crate::transaction::expire_snapshots::ExpireSnapshotsAction;
 use crate::transaction::sort_order::ReplaceSortOrderAction;
 use crate::transaction::update_location::UpdateLocationAction;
 use crate::transaction::update_properties::UpdatePropertiesAction;
-use crate::transaction::update_schema::UpdateSchemaAction;
 use crate::transaction::update_statistics::UpdateStatisticsAction;
 use crate::transaction::upgrade_format_version::UpgradeFormatVersionAction;
 use crate::{Catalog, TableCommit, TableRequirement, TableUpdate};