From 359047a88538e540da39c7eac684a94439c5f7d8 Mon Sep 17 00:00:00 2001
From: KK <37928252+dragonknightk@users.noreply.github.com>
Date: Fri, 30 Oct 2026 19:50:56 +0800
Subject: [PATCH] Assert the last assigned field id when update_schema adds a
 schema
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Closes issue 4190

# Rationale for this change

`UpdateSchema._commit` only sent `AssertCurrentSchemaId`, so a stale `update_schema` could commit after another writer had moved the current schema away and back. For example:

- writer A adds a column (field id 2, schema 1), then drops it, which makes schema 0 current again
- writer B, loaded before A's changes, adds a column: its `AssertCurrentSchemaId(0)` still passes, so it reuses field id 2 and adds a second schema with id 1

Java's `UpdateRequirements` adds `AssertLastAssignedFieldId(base.lastColumnId())` when a commit adds a schema (`update(MetadataUpdate.AddSchema)`). pyiceberg already defines `AssertLastAssignedFieldId` but never sent it.

### What changed

- `UpdateSchema._commit` adds `AssertLastAssignedFieldId` with the transaction's `last_column_id` when it adds a new schema
- this follows `UpdateSpec._commit`, which adds `AssertLastAssignedPartitionId` when it adds a spec
- switching back to an existing schema adds no schema, so it only sends `AssertCurrentSchemaId`, as before

With this change, writer B fails with `CommitFailedException` and the table stays unchanged (｀・ω・´)

## Are these changes tested?

Yes. `test_concurrent_add_column_after_add_and_drop` in `tests/catalog/test_catalog_behaviors.py` runs the scenario from the issue against the memory and SQL catalogs:

- with this change, writer B fails with `last assigned field id has changed: expected 3, found 4`
- without it, the test fails with `DID NOT RAISE CommitFailedException`

## Are there any user-facing changes?

Yes, a bug fix: a stale `update_schema` that adds a schema now raises `CommitFailedException` when the table's last assigned field id has changed, instead of reusing field ids. Refresh the table and retry, as with other requirement failures (・ω・)b

## AI Disclosure

Human-led contribution. Grok Bot, working for dragonknightk, assisted with this change. All submitted content was reviewed by a human before submission.
---
 pyiceberg/table/update/schema.py        |  3 +++
 tests/catalog/test_catalog_behaviors.py | 28 +++++++++++++++++++++++++
 2 files changed, 31 insertions(+)

diff --git a/pyiceberg/table/update/schema.py b/pyiceberg/table/update/schema.py
index 828f1e877a..f04513bb12 100644
--- a/pyiceberg/table/update/schema.py
+++ b/pyiceberg/table/update/schema.py
@@ -42,6 +42,7 @@
 from pyiceberg.table.update import (
     AddSchemaUpdate,
     AssertCurrentSchemaId,
+    AssertLastAssignedFieldId,
     SetCurrentSchemaUpdate,
     SetPropertiesUpdate,
     TableRequirement,
@@ -635,6 +636,8 @@ def _commit(self) -> UpdatesAndRequirements:
         if existing_schema_id != self._schema.schema_id:
             requirements += (AssertCurrentSchemaId(current_schema_id=self._schema.schema_id),)
             if existing_schema_id is None:
+                last_column_id = self._transaction.table_metadata.last_column_id
+                requirements += (AssertLastAssignedFieldId(last_assigned_field_id=last_column_id),)
                 updates += (
                     AddSchemaUpdate(schema=new_schema),
                     SetCurrentSchemaUpdate(schema_id=-1),
diff --git a/tests/catalog/test_catalog_behaviors.py b/tests/catalog/test_catalog_behaviors.py
index 1148ca8527..9540bfb08f 100644
--- a/tests/catalog/test_catalog_behaviors.py
+++ b/tests/catalog/test_catalog_behaviors.py
@@ -622,6 +622,34 @@ def test_concurrent_commit_table(catalog: Catalog, table_schema_nested: Schema,
             update.add_column(path="c", field_type=IntegerType())
 
 
+def test_concurrent_add_column_after_add_and_drop(
+    catalog: Catalog, table_schema_simple: Schema, random_table_identifier: Identifier
+) -> None:
+    namespace = Catalog.namespace_from(random_table_identifier)
+    catalog.create_namespace(namespace)
+    table_a = catalog.create_table(random_table_identifier, table_schema_simple)
+    table_b = catalog.load_table(random_table_identifier)
+
+    with table_a.update_schema() as update:
+        update.add_column(path="b", field_type=IntegerType())
+    with table_a.update_schema() as update:
+        update.delete_column("b")
+
+    # Dropping b makes schema 0 current again, so only the last assigned field id shows the change:
+    # table_b loaded it as 3 (foo, bar, baz), adding b raised it to 4
+    with pytest.raises(
+        CommitFailedException, match="Requirement failed: last assigned field id has changed: expected 3, found 4"
+    ):
+        with table_b.update_schema() as update:
+            update.add_column(path="c", field_type=IntegerType())
+
+    # After refresh the last assigned field id is 4 (b), so c gets 5
+    table_b.refresh()
+    with table_b.update_schema() as update:
+        update.add_column(path="c", field_type=IntegerType())
+    assert table_b.schema().find_field("c").field_id == 5
+
+
 def test_delete_metadata_multiple(catalog: Catalog, table_schema_nested: Schema, random_table_identifier: Identifier) -> None:
     namespace = Catalog.namespace_from(random_table_identifier)
     catalog.create_namespace(namespace)