Start writes to a missing branch from main

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()
 

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,

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

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).