diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 5ca3eb7a73..213f4c9a5f 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/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 diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index ab95457e23..b93510fae8 100644 --- a/tests/table/test_commit_retry.py +++ b/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: