feat(schema): Add update_schema action to enable table schema updates (#2120)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #2119.

## What changes are included in this PR?

<!--
Provide a summary of the modifications in this PR. List the main changes
such as new features, bug fixes, refactoring, or any other updates.
-->
This PR creates an `UpdateSchema` which implements the
`TransactionAction` allowing users of the crate to add new fields to or
delete fields from an iceberg table by updating the table schema. It
also checks that the added fields are either optional or have a default
value to avoid data corruption downstream.

## Are these changes tested?

<!--
Specify what test covers (unit test, integration test, etc.).

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
- [X] Unit tests for `UpdateSchema`.
- [X] Adds one integration tests which tests that we can add data,
update the table and can still read the data, ensuring backwards
compatiblity and one integration test for deleting a schema.

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: tomighita

crates/catalog/loader/tests/schema_update_suite.rs

@@ -0,0 +1,284 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Common schema-update behavior across catalogs.
+//!
+//! These tests assume Docker containers are started externally via `make docker-up`.
+
+mod common;
+
+use std::collections::HashMap;
+
+use common::{CatalogKind, cleanup_namespace_dyn, load_catalog};
+use iceberg::spec::{NestedField, PrimitiveType, Schema, StructType, Type};
+use iceberg::transaction::{AddColumn, ApplyTransactionAction, Transaction};
+use iceberg::{ErrorKind, NamespaceIdent, Result, TableCreation, TableIdent};
+use iceberg_test_utils::normalize_test_name_with_parts;
+use rstest::rstest;
+
+fn base_schema() -> Schema {
+    Schema::builder()
+        .with_fields(vec![
+            NestedField::optional(1, "foo", Type::Primitive(PrimitiveType::String)).into(),
+            NestedField::required(2, "bar", Type::Primitive(PrimitiveType::Int)).into(),
+            NestedField::optional(3, "baz", Type::Primitive(PrimitiveType::Boolean)).into(),
+        ])
+        .with_identifier_field_ids(vec![2])
+        .build()
+        .unwrap()
+}
+
+// Common behavior: adding a top-level column appends it to the schema.
+// HMS is excluded because update_table is not yet supported.
+#[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_add_column(#[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_add_column",
+        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_add_column", harness.label, "table");
+    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()
+        .add_column(AddColumn::optional(
+            "a",
+            Type::Primitive(PrimitiveType::Int),
+        ))
+        .apply(tx)?;
+    let updated = tx.commit(catalog.as_ref()).await?;
+
+    let schema = updated.metadata().current_schema();
+    let field_a = schema.field_by_name("a").expect("field 'a' should exist");
+    assert_eq!(field_a.id, 4);
+    assert_eq!(*field_a.field_type, Type::Primitive(PrimitiveType::Int));
+
+    Ok(())
+}
+
+// Common behavior: adding a nested struct column then a sub-field within it,
+// and deleting another top-level column, all persist correctly.
+// HMS is excluded because update_table is not yet supported.
+#[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_add_nested_and_delete_column(#[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_add_nested_and_delete_column",
+        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_add_nested_and_delete_column",
+        harness.label,
+        "table"
+    );
+    let table = catalog
+        .create_table(
+            &namespace,
+            TableCreation::builder()
+                .name(table_name)
+                .schema(base_schema())
+                .build(),
+        )
+        .await?;
+
+    // First transaction: add a nested struct column.
+    let tx = Transaction::new(&table);
+    let tx = tx
+        .update_schema()
+        .add_column(AddColumn::optional(
+            "info",
+            Type::Struct(StructType::new(vec![
+                NestedField::optional(0, "city", Type::Primitive(PrimitiveType::String)).into(),
+            ])),
+        ))
+        .apply(tx)?;
+    let table = tx.commit(catalog.as_ref()).await?;
+
+    // Second transaction: add a sub-field to the nested struct and delete a top-level column.
+    let tx = Transaction::new(&table);
+    let tx = tx
+        .update_schema()
+        .add_column(
+            AddColumn::builder()
+                .name("zip")
+                .field_type(Type::Primitive(PrimitiveType::String))
+                .parent("info")
+                .build(),
+        )
+        .delete_column("baz")
+        .apply(tx)?;
+    let table = tx.commit(catalog.as_ref()).await?;
+
+    let schema = table.metadata().current_schema();
+    assert!(schema.field_by_name("info").is_some());
+    assert!(schema.field_by_name("info.city").is_some());
+    assert!(schema.field_by_name("info.zip").is_some());
+    assert!(schema.field_by_name("baz").is_none());
+
+    Ok(())
+}
+
+// Common behavior: deleting an identifier field or a nonexistent field must fail.
+// HMS is excluded because update_table is not yet supported.
+#[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_delete_invalid_column_errors(#[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_delete_invalid_column_errors",
+        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_delete_invalid_column_errors",
+        harness.label,
+        "table"
+    );
+    let table = catalog
+        .create_table(
+            &namespace,
+            TableCreation::builder()
+                .name(table_name)
+                .schema(base_schema())
+                .build(),
+        )
+        .await?;
+
+    // Deleting an identifier field must fail.
+    let tx = Transaction::new(&table);
+    let tx = tx.update_schema().delete_column("bar").apply(tx)?;
+    let err = tx.commit(catalog.as_ref()).await.unwrap_err();
+    assert_eq!(err.kind(), ErrorKind::PreconditionFailed);
+
+    // Deleting a nonexistent field must fail.
+    let tx = Transaction::new(&table);
+    let tx = tx.update_schema().delete_column("nonexistent").apply(tx)?;
+    let err = tx.commit(catalog.as_ref()).await.unwrap_err();
+    assert_eq!(err.kind(), ErrorKind::PreconditionFailed);
+
+    Ok(())
+}
+
+// Common behavior: loading a table after a schema update reflects the new schema.
+// HMS is excluded because update_table is not yet supported.
+#[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_update_persisted_after_reload(
+    #[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_update_persisted_after_reload",
+        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_update_persisted_after_reload",
+        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()
+        .add_column(AddColumn::optional(
+            "new_field",
+            Type::Primitive(PrimitiveType::Long),
+        ))
+        .apply(tx)?;
+    tx.commit(catalog.as_ref()).await?;
+
+    let reloaded = catalog.load_table(&table_ident).await?;
+    assert!(
+        reloaded
+            .metadata()
+            .current_schema()
+            .field_by_name("new_field")
+            .is_some()
+    );
+
+    Ok(())
+}

crates/iceberg/src/spec/schema/mod.rs

@@ -51,6 +51,8 @@ pub type SchemaId = i32;
 pub type SchemaRef = Arc<Schema>;
 /// Default schema id.
 pub const DEFAULT_SCHEMA_ID: SchemaId = 0;
+/// Delimiter for schema name, which denotes a nested struct.
+pub const SCHEMA_NAME_DELIMITER: &str = ".";
 
 /// Defines schema in iceberg.
 #[derive(Debug, Serialize, Deserialize, Clone)]

crates/iceberg/src/transaction/mod.rs

@@ -58,13 +58,15 @@ mod snapshot;
 mod sort_order;
 mod update_location;
 mod update_properties;
+mod update_schema;
 mod update_statistics;
 mod upgrade_format_version;
 
 use std::sync::Arc;
 use std::time::Duration;
 
 use backon::{BackoffBuilder, ExponentialBackoff, ExponentialBuilder, RetryableWithContext};
+pub use update_schema::AddColumn;
 
 use crate::error::Result;
 use crate::spec::TableProperties;
@@ -74,6 +76,7 @@ use crate::transaction::append::FastAppendAction;
 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};
@@ -136,6 +139,11 @@ impl Transaction {
         UpdatePropertiesAction::new()
     }
 
+    /// Creates an update schema action.
+    pub fn update_schema(&self) -> UpdateSchemaAction {
+        UpdateSchemaAction::new()
+    }
+
     /// Creates a fast append action.
     pub fn fast_append(&self) -> FastAppendAction {
         FastAppendAction::new()