Core: Retry appends on catalogs that only expose the current snapshot

pyiceberg/table/update/snapshot.py

@@ -29,6 +29,7 @@
 
 from pyiceberg.avro.codecs import AvroCompressionCodec
 from pyiceberg.exceptions import (
+    CommitFailedException,
     NoSuchSnapshotRefError,
     NotAncestorError,
     SnapshotRefTypeError,
@@ -109,26 +110,29 @@ class CommitWindow:
 
     base: The snapshot when the operation began (exclusive lower bound, fixed across retries).
     head: The branch HEAD snapshot after metadata refresh (inclusive upper bound).
+    base_id: The starting snapshot id, also kept when the catalog no longer exposes it.
+    base_missing: True when the starting snapshot is not in the refreshed metadata, for
+        example when the catalog only exposes the current snapshot.
     """
 
     base: Snapshot | None
     head: Snapshot | None
+    base_id: int | None = None
+    base_missing: bool = False
 
     @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)
         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}")
-        return cls(base=base, head=head)
+        return cls(base=base, head=head, base_id=base_id, base_missing=base_id is not None and base is None)
 
     def is_empty(self) -> bool:
         """Return True if no concurrent commits occurred (validation can be skipped).
 
-        A None base means the operation started on a table with no snapshots. If a head
-        exists, another writer landed the first snapshot concurrently, so the window is not
-        empty and validation must run against the whole history.
+        A None base means the operation started on a table with no snapshots, or that the
+        catalog did not expose the starting snapshot. If a head exists, another writer landed
+        a snapshot concurrently, so the window is not empty.
         """
         return self.head is None or (self.base is not None and self.base.snapshot_id == self.head.snapshot_id)
 
@@ -487,6 +491,12 @@ def _validate_concurrency(self) -> None:
         if self._commit_window is None or self._commit_window.is_empty():
             return
 
+        if self._commit_window.base_missing:
+            raise CommitFailedException(
+                f"Cannot determine history between starting snapshot {self._commit_window.base_id} "
+                "and the current snapshot; the catalog does not expose the starting snapshot"
+            )
+
         catalog_head = self._commit_window.head
         starting_snapshot = self._commit_window.base
 

tests/table/test_commit_retry.py

@@ -735,8 +735,8 @@ def test_validate_concurrency_skips_when_commit_window_is_empty(catalog: Catalog
     producer._validate_concurrency()
 
 
-def test_validate_concurrency_raises_on_missing_starting_snapshot(catalog: Catalog) -> None:
-    """CommitWindow.resolve should raise when starting_snapshot_id cannot be resolved."""
+def test_validate_concurrency_marks_missing_starting_snapshot(catalog: Catalog) -> None:
+    """CommitWindow.resolve should record a starting snapshot the refreshed metadata does not expose."""
     catalog.create_namespace("default")
     schema = _test_schema()
     table = catalog.create_table("default.missing_starting_test", schema=schema)
@@ -747,8 +747,82 @@ def test_validate_concurrency_raises_on_missing_starting_snapshot(catalog: Catal
 
     from pyiceberg.table.update.snapshot import CommitWindow
 
-    with pytest.raises(ValidationException, match="Cannot find starting snapshot"):
-        CommitWindow.resolve(table.metadata, base_id=99999999, branch="main")
+    window = CommitWindow.resolve(table.metadata, base_id=99999999, branch="main")
+
+    assert window.base is None
+    assert window.base_missing
+    assert window.head is not None
+    assert not window.is_empty()
+
+
+def _hiding_starting_snapshot(rebuild: Any) -> Any:
+    def patched_rebuild(self_tx: Any) -> None:
+        metadata_cls = type(self_tx._table.metadata)
+        base_id = self_tx._snapshot_producers[0]._starting_snapshot_id
+        original_snapshot_by_id = metadata_cls.snapshot_by_id
+
+        def snapshot_by_id(self_md: Any, snapshot_id: int) -> Any:
+            return None if snapshot_id == base_id else original_snapshot_by_id(self_md, snapshot_id)
+
+        with patch.object(metadata_cls, "snapshot_by_id", snapshot_by_id):
+            rebuild(self_tx)
+
+    return patched_rebuild
+
+
+def test_append_retry_without_starting_snapshot(catalog: Catalog) -> None:
+    """An append retries on catalogs that only expose the current snapshot."""
+    catalog.create_namespace("default")
+    schema = _test_schema()
+    catalog.create_table("default.hidden_history_append_test", schema=schema)
+
+    import pyarrow as pa
+
+    import pyiceberg.table as _table_module
+
+    df = pa.table({"x": [1, 2, 3]})
+    table = catalog.load_table("default.hidden_history_append_test")
+    table.append(df)
+
+    tbl1 = catalog.load_table("default.hidden_history_append_test")
+    tbl2 = catalog.load_table("default.hidden_history_append_test")
+
+    tbl1.append(df)
+
+    RuntimeTransaction = _table_module.Transaction
+    original_rebuild = RuntimeTransaction._rebuild_snapshot_updates
+
+    with patch.object(RuntimeTransaction, "_rebuild_snapshot_updates", _hiding_starting_snapshot(original_rebuild)):
+        tbl2.append(df)
+
+    assert len(catalog.load_table("default.hidden_history_append_test").scan().to_arrow()) == 9
+
+
+def test_delete_retry_without_starting_snapshot_raises_commit_failed(catalog: Catalog) -> None:
+    """A delete cannot validate without the starting snapshot and raises CommitFailedException."""
+    catalog.create_namespace("default")
+    schema = _test_schema()
+    catalog.create_table("default.hidden_history_delete_test", schema=schema)
+
+    import pyarrow as pa
+
+    import pyiceberg.table as _table_module
+
+    df = pa.table({"x": [1, 2, 3]})
+    table = catalog.load_table("default.hidden_history_delete_test")
+    table.append(df)
+
+    tbl1 = catalog.load_table("default.hidden_history_delete_test")
+    tbl2 = catalog.load_table("default.hidden_history_delete_test")
+
+    tbl1.append(df)
+
+    RuntimeTransaction = _table_module.Transaction
+    original_rebuild = RuntimeTransaction._rebuild_snapshot_updates
+
+    with patch.object(RuntimeTransaction, "_rebuild_snapshot_updates", _hiding_starting_snapshot(original_rebuild)):
+        with pytest.raises(CommitFailedException, match="Cannot determine history"):
+            tbl2.delete("x == 1")
 
 
 def test_mixed_delete_overwrite_retries_successfully(catalog: Catalog) -> None: