From 0dcc16e3807ecd85a16d339a089de03cf6f11ae5 Mon Sep 17 00:00:00 2001
From: KK <37928252+dragonknightk@users.noreply.github.com>
Date: Fri, 30 Oct 2026 23:20:42 +0800
Subject: [PATCH] Start writes to a missing branch from main
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Closes issue 4195

# Rationale for this change

Writing to a branch that does not exist creates it, but the new snapshot gets no parent, so the branch starts empty instead of from the state of the table.

```python
tbl.append(pa.table({"x": [1, 2]}))
tbl.append(pa.table({"x": [3]}), branch="b")   # b is [3] instead of [1, 2, 3], with no parent
```

`overwrite` does the same, and so does an append that is retried after the branch was removed concurrently. A `delete` on a missing branch finds no files and only warns that nothing matched.

Java starts from main. `SnapshotProducer.apply()` takes the parent from `SnapshotUtil.latestSnapshot(base, targetBranch)`, which returns `metadata.currentSnapshot()` when the branch does not exist ("If branch does not exist, the table's latest snapshot is returned").

### What changed

The parent of a write is the head of the target branch, and looking that up by name returns nothing when the branch does not exist. Five places depend on that lookup:

| Where | Now |
| --- | --- |
| `_SnapshotProducer._current_branch_head_id` | Falls back to the current snapshot. Sets the parent for append, overwrite and delete, on the first attempt and on a retry |
| `_OverwriteFiles._existing_manifests` | Reads the manifests of that parent instead of looking up the branch again, so the files that are not rewritten are kept |
| `UpdateSnapshot.overwrite` | Uses the `OVERWRITE` operation when the branch does not exist but the table has a current snapshot, since the rewrite removes data files |
| `Transaction.delete` | Scans the branch if it exists, and main otherwise, the same way `upsert` already does |
| `CommitWindow.resolve` | Uses the same fallback for the end of the validation window, so a retried `delete` or `overwrite` validates the snapshots added to main since it started |

### Left out

- A table without snapshots: writing to a missing branch still raises `CommitFailedException: Requirement failed: Table has no snapshots and can only be written to the main BRANCH.`, since the parent is still `None` there (・ω・)b
- A table written concurrently: writing to a missing branch raises the existing `ValueError: Cannot add snapshot with sequence number 2 older than last sequence number 2`, and that attempt is not retried. The same happens when the branch exists, so it is left out of this PR

## Are these changes tested?

Yes, against the memory and SQL catalogs:

| Test | Checks |
| --- | --- |
| `test_append_to_missing_branch_starts_from_main` (`tests/catalog/test_catalog_behaviors.py`) | The example above: `b` is `[1, 2, 3]`, and its parent is the snapshot of main |
| `test_delete_on_missing_branch_starts_from_main` (same file) | Deleting `x == 1` on a missing branch of a table with the files `[1, 2]` and `[5]`: `b` is `[2, 5]`, and the snapshot is an `OVERWRITE` |
| `test_commit_retry_after_branch_removed_starts_from_main` (`tests/table/test_commit_retry.py`) | The branch is removed while another writer appends to it: the retried append is `[1, 2, 3, 4]`, with main as its parent |
| `test_delete_retried_after_branch_removed_validates_against_main` (same file) | The branch is removed and a matching row is appended to main: the retried `delete` raises a `ValidationException` for that row |

Reverting any one of the five changes makes at least one test fail. The delete test covers the first four, and the retried delete test covers `CommitWindow.resolve`.

## Are there any user-facing changes?

Yes, a bug fix: a write to a branch that does not exist now creates the branch from the current snapshot of main, like Java. Before, the branch only contained the rows of that write (｀・ω・´)

## 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/__init__.py             |  2 +-
 pyiceberg/table/update/snapshot.py      | 20 +++++++---
 tests/catalog/test_catalog_behaviors.py | 34 ++++++++++++++++
 tests/table/test_commit_retry.py        | 53 +++++++++++++++++++++++++
 4 files changed, 102 insertions(+), 7 deletions(-)

diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py
index 8d79472af5..03fcf5fd00 100644
--- a/pyiceberg/table/__init__.py
+++ b/pyiceberg/table/__init__.py
@@ -794,7 +794,7 @@ def delete(
             preserve_row_filter = _expression_to_complementary_pyarrow(bound_delete_filter, self.table_metadata.schema())
 
             file_scan = self._scan(row_filter=delete_filter, case_sensitive=case_sensitive)
-            if branch is not None:
+            if branch in self.table_metadata.refs:
                 file_scan = file_scan.use_ref(branch)
             files = file_scan.plan_files()
 
diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py
index 8feebb9901..764653e92f 100644
--- a/pyiceberg/table/update/snapshot.py
+++ b/pyiceberg/table/update/snapshot.py
@@ -108,7 +108,8 @@ class CommitWindow:
     """Tracks the commit range to validate against during retry.
 
     base: The snapshot when the operation began (exclusive lower bound, fixed across retries).
-    head: The branch HEAD snapshot after metadata refresh (inclusive upper bound).
+    head: The branch HEAD snapshot after metadata refresh, or the current snapshot if the branch does not exist
+          (inclusive upper bound).
     """
 
     base: Snapshot | None
@@ -117,7 +118,7 @@ class CommitWindow:
     @classmethod
     def resolve(cls, metadata: TableMetadata, base_id: int | None, branch: str | None) -> CommitWindow:
         """Resolve a CommitWindow from metadata, starting snapshot ID, and target branch."""
-        head = metadata.snapshot_by_name(branch)
+        head = metadata.snapshot_by_name(branch) or metadata.current_snapshot()
         base = metadata.snapshot_by_id(base_id) if base_id is not None else None
         if base_id is not None and base is None:
             raise ValidationException(f"Cannot find starting snapshot {base_id}")
@@ -197,8 +198,9 @@ def _validate_target_branch(self, branch: str | None) -> str | None:
         return branch
 
     def _current_branch_head_id(self) -> int | None:
-        """Return the snapshot id at the head of the target branch, or None if the branch has no snapshot."""
-        snapshot = self._transaction.table_metadata.snapshot_by_name(self._target_branch)
+        """Return the snapshot id at the head of the target branch, or the current snapshot id if the branch does not exist."""
+        table_metadata = self._transaction.table_metadata
+        snapshot = table_metadata.snapshot_by_name(self._target_branch) or table_metadata.current_snapshot()
         return snapshot.snapshot_id if snapshot else None
 
     def append_data_file(self, data_file: DataFile) -> _SnapshotProducer[U]:
@@ -782,7 +784,9 @@ def _existing_manifests(self) -> list[ManifestFile]:
         existing_files = []
 
         manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)
-        if snapshot := self._transaction.table_metadata.snapshot_by_name(name=self._target_branch):
+        if self._parent_snapshot_id is not None and (
+            snapshot := self._transaction.table_metadata.snapshot_by_id(self._parent_snapshot_id)
+        ):
             for manifest_file in snapshot.manifests(io=self._io):
                 # Manifest does not contain rows that match the files to delete partitions
                 if not manifest_evaluators[manifest_file.partition_spec_id](manifest_file):
@@ -922,7 +926,11 @@ def overwrite(self, commit_uuid: uuid.UUID | None = None) -> _OverwriteFiles:
         return _OverwriteFiles(
             commit_uuid=commit_uuid,
             operation=Operation.OVERWRITE
-            if self._transaction.table_metadata.snapshot_by_name(name=self._branch) is not None
+            if (
+                self._transaction.table_metadata.snapshot_by_name(name=self._branch)
+                or self._transaction.table_metadata.current_snapshot()
+            )
+            is not None
             else Operation.APPEND,
             transaction=self._transaction,
             io=self._io,
diff --git a/tests/catalog/test_catalog_behaviors.py b/tests/catalog/test_catalog_behaviors.py
index 1148ca8527..fa79292fa1 100644
--- a/tests/catalog/test_catalog_behaviors.py
+++ b/tests/catalog/test_catalog_behaviors.py
@@ -1409,3 +1409,37 @@ def test_record_batch_reader_schema_mismatch_writes_no_files(catalog: Catalog) -
     # _append_snapshot_producer opens.
     assert tbl.metadata.current_snapshot() is None
     assert len(tbl.scan().to_arrow()) == 0
+
+
+def test_append_to_missing_branch_starts_from_main(catalog: Catalog) -> None:
+    catalog.create_namespace("default")
+    identifier = f"default.append_to_missing_branch_{catalog.name}"
+    tbl = catalog.create_table(identifier=identifier, schema=pa.schema([("x", pa.int64())]))
+    tbl.append(pa.table({"x": [1, 2]}))
+    main_snapshot_id = tbl.metadata.current_snapshot_id
+
+    tbl.append(pa.table({"x": [3]}), branch="b")
+
+    # Like Java's SnapshotUtil.latestSnapshot, a branch that does not exist starts from the current snapshot of main
+    branch_snapshot = tbl.snapshot_by_name("b")
+    assert branch_snapshot is not None
+    assert branch_snapshot.parent_snapshot_id == main_snapshot_id
+    assert sorted(tbl.scan().use_ref("b").to_arrow()["x"].to_pylist()) == [1, 2, 3]
+
+
+def test_delete_on_missing_branch_starts_from_main(catalog: Catalog) -> None:
+    catalog.create_namespace("default")
+    identifier = f"default.delete_on_missing_branch_{catalog.name}"
+    tbl = catalog.create_table(identifier=identifier, schema=pa.schema([("x", pa.int64())]))
+    tbl.append(pa.table({"x": [1, 2]}))
+    tbl.append(pa.table({"x": [5]}))
+
+    # x == 1 matches only part of the file [1, 2], so that file is rewritten to [2] and the file [5] is kept
+    tbl.delete("x == 1", branch="b")
+
+    branch_snapshot = tbl.snapshot_by_name("b")
+    assert branch_snapshot is not None
+    assert sorted(tbl.scan().use_ref("b").to_arrow()["x"].to_pylist()) == [2, 5]
+    # The rewrite removes a data file of main, so it is an overwrite and not an append
+    assert branch_snapshot.summary is not None
+    assert branch_snapshot.summary.operation == Operation.OVERWRITE
diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py
index ab95457e23..2dc8d7d457 100644
--- a/tests/table/test_commit_retry.py
+++ b/tests/table/test_commit_retry.py
@@ -957,6 +957,59 @@ def test_commit_retry_delete_on_non_main_branch(catalog: Catalog) -> None:
     assert sorted(result.column("x").to_pylist()) == [2, 3, 4, 5, 6, 7, 8, 9]
 
 
+def test_commit_retry_after_branch_removed_starts_from_main(catalog: Catalog) -> None:
+    """Verify that a retried append to a branch removed concurrently starts from the current snapshot of main."""
+    catalog.create_namespace("default")
+    catalog.create_table("default.branch_removed_retry_test", schema=_test_schema())
+
+    import pyarrow as pa
+
+    tbl = catalog.load_table("default.branch_removed_retry_test")
+    tbl.append(pa.table({"x": [1, 2]}))
+    assert tbl.metadata.current_snapshot_id is not None
+    tbl.manage_snapshots().create_branch(snapshot_id=tbl.metadata.current_snapshot_id, branch_name="test-branch").commit()
+    tbl.append(pa.table({"x": [3]}))
+
+    tbl1 = catalog.load_table("default.branch_removed_retry_test")
+    tbl2 = catalog.load_table("default.branch_removed_retry_test")
+
+    tbl1.manage_snapshots().remove_branch("test-branch").commit()
+
+    # tbl2 still sees test-branch, so the first attempt fails and the retry finds the branch missing
+    tbl2.append(pa.table({"x": [4]}), branch="test-branch")
+
+    refreshed = catalog.load_table("default.branch_removed_retry_test")
+    branch_snapshot = refreshed.snapshot_by_name("test-branch")
+    assert branch_snapshot is not None
+    # The parent is the current snapshot of main ([1, 2] + [3]), not the removed head of test-branch ([1, 2])
+    assert branch_snapshot.parent_snapshot_id == refreshed.metadata.current_snapshot_id
+    assert sorted(refreshed.scan(snapshot_id=branch_snapshot.snapshot_id).to_arrow().column("x").to_pylist()) == [1, 2, 3, 4]
+
+
+def test_delete_retried_after_branch_removed_validates_against_main(catalog: Catalog) -> None:
+    """Verify that a delete retried on main after its branch was removed concurrently validates the changes on main."""
+    catalog.create_namespace("default")
+    catalog.create_table("default.branch_removed_delete_test", schema=_test_schema())
+
+    import pyarrow as pa
+
+    tbl = catalog.load_table("default.branch_removed_delete_test")
+    tbl.append(pa.table({"x": [1, 2]}))
+    assert tbl.metadata.current_snapshot_id is not None
+    tbl.manage_snapshots().create_branch(snapshot_id=tbl.metadata.current_snapshot_id, branch_name="test-branch").commit()
+
+    tbl1 = catalog.load_table("default.branch_removed_delete_test")
+    tbl2 = catalog.load_table("default.branch_removed_delete_test")
+
+    tbl1.manage_snapshots().remove_branch("test-branch").commit()
+    # A concurrent append to main adds a row that matches the delete filter
+    catalog.load_table("default.branch_removed_delete_test").append(pa.table({"x": [1]}))
+
+    # tbl2 still sees test-branch, so the retry starts from main, and the default serializable isolation rejects the new row
+    with pytest.raises(ValidationException, match="Added data files were found matching the filter"):
+        tbl2.delete("x <= 2", branch="test-branch")
+
+
 def test_negative_num_retries_still_commits(catalog: Catalog) -> None:
     """A negative commit.retry.num-retries is clamped to 0, so one attempt still runs (matches Java).