From d3ab1b2a0b98d3dd9ad83a3429d0f872be41bfb6 Mon Sep 17 00:00:00 2001 From: Daniel Szilagyi Date: Fri, 25 Sep 2026 12:05:06 +0200 Subject: [PATCH 1/3] Check whether a failed commit landed before cleaning up its files --- pyiceberg/table/__init__.py | 62 +++++++++++------ tests/table/test_commit_retry.py | 116 +++++++++++++++++++++++++++++++ 2 files changed, 157 insertions(+), 21 deletions(-) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 2c5c26800c..fabba8080c 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -33,7 +33,7 @@ from pydantic import Field -from pyiceberg.exceptions import CommitFailedException, ValidationException +from pyiceberg.exceptions import CommitFailedException, CommitStateUnknownException, ValidationException from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, And, BooleanExpression, EqualTo, IsNull, Or, Reference from pyiceberg.expressions.visitors import ( ResidualEvaluator, @@ -66,6 +66,7 @@ from pyiceberg.table.update import ( AddPartitionSpecUpdate, AddSchemaUpdate, + AddSnapshotUpdate, AddSortOrderUpdate, AssertCreate, AssertRefSnapshotId, @@ -1123,7 +1124,13 @@ def commit_transaction(self) -> Table: try: try: + # Snapshot ids sent to the catalog so far. A commit applies all of an attempt's + # snapshots atomically, so finding any of them means that attempt landed. + sent_snapshot_ids: set[int] = set() for attempt in range(num_retries + 1): + sent_snapshot_ids.update( + update.snapshot.snapshot_id for update in self._updates if isinstance(update, AddSnapshotUpdate) + ) try: self._table._do_commit( # pylint: disable=W0212 updates=self._updates, @@ -1131,33 +1138,32 @@ def commit_transaction(self) -> Table: ) self._cleanup_uncommitted_manifests() break - except CommitFailedException: + except CommitFailedException as e: elapsed_ms = (time.monotonic() - start_time) * 1000 - if attempt == num_retries or not self._snapshot_producers or elapsed_ms >= total_timeout_ms: - raise - - wait = min(min_wait_ms * (2**attempt), max_wait_ms) - jitter = random.uniform(0, 0.1 * wait) - logger.warning( - "Commit failed due to a concurrent update, retrying (%s/%s) in %s ms", - attempt + 1, - num_retries, - round(wait + jitter), + last_attempt = ( + attempt == num_retries or not self._snapshot_producers or elapsed_ms >= total_timeout_ms ) - time.sleep((wait + jitter) / 1000.0) - self._table.refresh() - if all( - self._table.metadata.snapshot_by_id(producer._snapshot_id) is not None - for producer in self._snapshot_producers - ): + if not last_attempt: + wait = min(min_wait_ms * (2**attempt), max_wait_ms) + jitter = random.uniform(0, 0.1 * wait) + logger.warning( + "Commit failed due to a concurrent update, retrying (%s/%s) in %s ms", + attempt + 1, + num_retries, + round(wait + jitter), + ) + time.sleep((wait + jitter) / 1000.0) + + if sent_snapshot_ids and self._attempt_landed(sent_snapshot_ids, e): # A previous attempt actually landed even though it was reported as # failed (for example a lost response that the transport layer retried). - # The snapshot id is stable across attempts, so finding it in the - # refreshed metadata means the commit is already applied. Stop here - # instead of committing the same data again. + # Stop here instead of committing the same data again, and never clean + # up the files the landed snapshot references. self._cleanup_uncommitted_manifests() break + if last_attempt: + raise self._rebuild_snapshot_updates() except (CommitFailedException, ValidationException): # These exceptions guarantee the commit did not land, so it is safe to delete the @@ -1203,6 +1209,20 @@ def commit_transaction(self) -> Table: return self._table + def _attempt_landed(self, snapshot_ids: set[int], commit_error: CommitFailedException) -> bool: + """Refresh the table and return whether any of the given snapshots is in its metadata. + + Raises CommitStateUnknownException if the refresh fails. That skips the cleanup of this + transaction's files, since a snapshot that may have landed could reference them. + """ + try: + self._table.refresh() + except Exception as refresh_error: + raise CommitStateUnknownException( + f"Commit failed ({commit_error}); could not refresh the table to check whether it landed: {refresh_error}" + ) from commit_error + return any(self._table.metadata.snapshot_by_id(snapshot_id) is not None for snapshot_id in snapshot_ids) + def _cleanup_uncommitted_manifests(self) -> None: """Clean up manifests from failed retry attempts after a successful commit.""" for producer in self._snapshot_producers: diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index ab95457e23..d4688d8257 100644 --- a/tests/table/test_commit_retry.py +++ b/tests/table/test_commit_retry.py @@ -1290,3 +1290,119 @@ def test_negative_wait_properties_do_not_mask_commit_failure(catalog: Catalog) - result = catalog.load_table("default.negative_wait_test").scan().to_arrow() assert len(result) == 6 + + +def _commit_outcomes(catalog: Catalog, *outcomes: str) -> Any: + """Patch commit_table to fail its first calls with CommitFailedException, then commit normally. + + A "conflict" outcome fails without committing. A "lost" outcome commits, then fails as if the + response was lost. + """ + real_commit = catalog.commit_table + remaining = list(outcomes) + + def commit(*args: Any, **kwargs: Any) -> Any: + outcome = remaining.pop(0) if remaining else "ok" + if outcome == "conflict": + raise CommitFailedException("concurrent update") + result = real_commit(*args, **kwargs) + if outcome == "lost": + raise CommitFailedException("response lost after the commit landed") + return result + + return patch.object(catalog, "commit_table", side_effect=commit) + + +def _assert_one_readable_snapshot(catalog: Catalog, identifier: str, expected: list[dict[str, Any]]) -> None: + table = catalog.load_table(identifier) + assert len(table.snapshots()) == 1 + snapshot = table.current_snapshot() + assert snapshot is not None + assert table.io.new_input(snapshot.manifest_list).exists() + assert table.scan().to_arrow().to_pylist() == expected + + +@pytest.mark.parametrize( + ("num_retries", "outcomes"), + [ + pytest.param("0", ["lost"], id="no-retries"), + pytest.param("1", ["conflict", "lost"], id="retries-exhausted"), + ], +) +def test_lost_response_on_the_last_attempt_keeps_the_landed_snapshot( + catalog: Catalog, num_retries: str, outcomes: list[str] +) -> None: + """A lost response on the last attempt must be checked for a landed snapshot before cleaning up. + + With no attempt left, the landed check of the retry loop is never reached. Cleaning up + unconditionally deletes the manifest list of the snapshot the catalog just committed. + """ + import pyarrow as pa + + catalog.create_namespace("default") + catalog.create_table( + "default.last_attempt_lost", + schema=_test_schema(), + properties={ + TableProperties.COMMIT_NUM_RETRIES: num_retries, + TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1", + TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2", + }, + ) + table = catalog.load_table("default.last_attempt_lost") + + with _commit_outcomes(catalog, *outcomes): + table.append(pa.table({"x": [1]})) + + _assert_one_readable_snapshot(catalog, "default.last_attempt_lost", [{"x": 1}]) + + +@pytest.mark.filterwarnings("ignore:Delete operation did not match any records") +def test_lost_response_for_an_overwrite_of_an_empty_table_keeps_the_landed_snapshot(catalog: Catalog) -> None: + """The landed check must use the snapshots that were committed, not every producer. + + An overwrite of an empty table registers a delete producer that adds no snapshot. Requiring a + snapshot from every producer never finds the landed append, so the retry rebuilds the updates. + The rebuilt delete then sees its own landed append as a conflicting commit, and the resulting + ValidationException cleans up the landed snapshot's files. + """ + import pyarrow as pa + + catalog.create_namespace("default") + catalog.create_table( + "default.empty_overwrite_lost", + schema=_test_schema(), + properties={ + TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1", + TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2", + }, + ) + table = catalog.load_table("default.empty_overwrite_lost") + + with _commit_outcomes(catalog, "lost"): + table.overwrite(pa.table({"x": [1]})) + + _assert_one_readable_snapshot(catalog, "default.empty_overwrite_lost", [{"x": 1}]) + + +def test_lost_response_with_a_failed_refresh_keeps_the_files(catalog: Catalog) -> None: + """If the landed check cannot refresh the table, the outcome is unknown and nothing is deleted.""" + import pyarrow as pa + + catalog.create_namespace("default") + catalog.create_table( + "default.lost_refresh_failed", + schema=_test_schema(), + properties={TableProperties.COMMIT_NUM_RETRIES: "0"}, + ) + table = catalog.load_table("default.lost_refresh_failed") + + with ( + _commit_outcomes(catalog, "lost"), + patch.object(table, "refresh", side_effect=ConnectionError("catalog unreachable")), + ): + with pytest.raises(CommitStateUnknownException) as exc_info: + table.append(pa.table({"x": [1]})) + assert isinstance(exc_info.value.__cause__, CommitFailedException) + + _assert_one_readable_snapshot(catalog, "default.lost_refresh_failed", [{"x": 1}]) From 7848c5f0097c95b51c63cd88c04dbe963ae5c6a3 Mon Sep 17 00:00:00 2001 From: Daniel Szilagyi Date: Fri, 25 Sep 2026 12:21:59 +0200 Subject: [PATCH 2/3] Refresh before rebuilding a retry that sent no snapshots A transaction whose producers staged no snapshot skipped the refresh, so the rebuild validated against stale metadata and missed concurrent conflicts. Co-Authored-By: Claude Opus 5.5 (1M context) --- pyiceberg/table/__init__.py | 19 ++++++++++------- tests/table/test_commit_retry.py | 35 ++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 7 deletions(-) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index fabba8080c..5486202022 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -1155,13 +1155,18 @@ def commit_transaction(self) -> Table: ) time.sleep((wait + jitter) / 1000.0) - if sent_snapshot_ids and self._attempt_landed(sent_snapshot_ids, e): - # A previous attempt actually landed even though it was reported as - # failed (for example a lost response that the transport layer retried). - # Stop here instead of committing the same data again, and never clean - # up the files the landed snapshot references. - self._cleanup_uncommitted_manifests() - break + if sent_snapshot_ids: + if self._attempt_landed(sent_snapshot_ids, e): + # A previous attempt actually landed even though it was reported as + # failed (for example a lost response that the transport layer retried). + # Stop here instead of committing the same data again, and never clean + # up the files the landed snapshot references. + self._cleanup_uncommitted_manifests() + break + elif not last_attempt: + # No snapshot was sent, so nothing can have landed, but the rebuild still + # needs fresh metadata to validate against concurrent commits. + self._table.refresh() if last_attempt: raise self._rebuild_snapshot_updates() diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index d4688d8257..93545e5823 100644 --- a/tests/table/test_commit_retry.py +++ b/tests/table/test_commit_retry.py @@ -1385,6 +1385,41 @@ def test_lost_response_for_an_overwrite_of_an_empty_table_keeps_the_landed_snaps _assert_one_readable_snapshot(catalog, "default.empty_overwrite_lost", [{"x": 1}]) +def test_retry_without_staged_snapshots_validates_against_refreshed_metadata(catalog: Catalog) -> None: + """A retry must refresh the table even when no attempt sent a snapshot. + + A property update with a delete that matched nothing has producers but adds no snapshot. Skipping + the refresh rebuilds the delete against stale metadata, which misses a concurrent append matching + the delete predicate. + """ + import pyarrow as pa + + catalog.create_namespace("default") + catalog.create_table( + "default.no_snapshot_retry", + schema=_test_schema(), + properties={ + TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1", + TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2", + }, + ) + + tx = catalog.load_table("default.no_snapshot_retry").transaction() + tx.set_properties({"key": "value"}) + with pytest.warns(UserWarning): # the delete matches nothing at staging time + tx.delete("x > 45") + + # A row matching the delete predicate lands concurrently. + catalog.load_table("default.no_snapshot_retry").append(pa.table({"x": [50]})) + + with _commit_outcomes(catalog, "conflict"), pytest.raises(ValidationException): + tx.commit_transaction() + + table = catalog.load_table("default.no_snapshot_retry") + assert "key" not in table.properties + assert table.scan().to_arrow()["x"].to_pylist() == [50] + + def test_lost_response_with_a_failed_refresh_keeps_the_files(catalog: Catalog) -> None: """If the landed check cannot refresh the table, the outcome is unknown and nothing is deleted.""" import pyarrow as pa From 8153d7878bb5659f87a065d6889100ef92672f0b Mon Sep 17 00:00:00 2001 From: Daniel Szilagyi Date: Fri, 25 Sep 2026 12:32:54 +0200 Subject: [PATCH 3/3] Simplify commit retry loop and tests Run the post-commit cleanup once after the loop, flatten the landed check, drop a redundant local import, and trim test setup and docstrings. Co-Authored-By: Claude Opus 5.5 (1M context) --- pyiceberg/table/__init__.py | 29 +++++++----------- tests/table/test_commit_retry.py | 50 ++++++++++++-------------------- 2 files changed, 28 insertions(+), 51 deletions(-) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 5486202022..7d9b332759 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -1124,8 +1124,7 @@ def commit_transaction(self) -> Table: try: try: - # Snapshot ids sent to the catalog so far. A commit applies all of an attempt's - # snapshots atomically, so finding any of them means that attempt landed. + # Each attempt is atomic, so finding any snapshot it sent proves it landed. sent_snapshot_ids: set[int] = set() for attempt in range(num_retries + 1): sent_snapshot_ids.update( @@ -1136,7 +1135,6 @@ def commit_transaction(self) -> Table: updates=self._updates, requirements=self._requirements, ) - self._cleanup_uncommitted_manifests() break except CommitFailedException as e: elapsed_ms = (time.monotonic() - start_time) * 1000 @@ -1155,21 +1153,16 @@ def commit_transaction(self) -> Table: ) time.sleep((wait + jitter) / 1000.0) - if sent_snapshot_ids: - if self._attempt_landed(sent_snapshot_ids, e): - # A previous attempt actually landed even though it was reported as - # failed (for example a lost response that the transport layer retried). - # Stop here instead of committing the same data again, and never clean - # up the files the landed snapshot references. - self._cleanup_uncommitted_manifests() - break - elif not last_attempt: - # No snapshot was sent, so nothing can have landed, but the rebuild still - # needs fresh metadata to validate against concurrent commits. - self._table.refresh() + if sent_snapshot_ids and self._attempt_landed(sent_snapshot_ids, e): + # A lost response can report failure after the commit landed. + break if last_attempt: raise + if not sent_snapshot_ids: + # Retries without snapshot updates still need fresh metadata for validation. + self._table.refresh() self._rebuild_snapshot_updates() + self._cleanup_uncommitted_manifests() except (CommitFailedException, ValidationException): # These exceptions guarantee the commit did not land, so it is safe to delete the # files written for it. Any other exception (unknown outcome, or a commit that already @@ -1215,10 +1208,9 @@ def commit_transaction(self) -> Table: return self._table def _attempt_landed(self, snapshot_ids: set[int], commit_error: CommitFailedException) -> bool: - """Refresh the table and return whether any of the given snapshots is in its metadata. + """Refresh the table and check for a landed snapshot. - Raises CommitStateUnknownException if the refresh fails. That skips the cleanup of this - transaction's files, since a snapshot that may have landed could reference them. + Raise CommitStateUnknownException if refresh fails, preserving potentially committed files. """ try: self._table.refresh() @@ -1235,7 +1227,6 @@ def _cleanup_uncommitted_manifests(self) -> None: def _rebuild_snapshot_updates(self) -> None: """Rebuild snapshot updates for retry by re-executing registered producers.""" - from pyiceberg.table.update import AddSnapshotUpdate, AssertRefSnapshotId, SetSnapshotRefUpdate from pyiceberg.table.update.snapshot import CommitWindow self._updates = tuple(u for u in self._updates if not isinstance(u, (AddSnapshotUpdate, SetSnapshotRefUpdate))) diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index 93545e5823..ecfd2b023c 100644 --- a/tests/table/test_commit_retry.py +++ b/tests/table/test_commit_retry.py @@ -1293,16 +1293,12 @@ def test_negative_wait_properties_do_not_mask_commit_failure(catalog: Catalog) - def _commit_outcomes(catalog: Catalog, *outcomes: str) -> Any: - """Patch commit_table to fail its first calls with CommitFailedException, then commit normally. - - A "conflict" outcome fails without committing. A "lost" outcome commits, then fails as if the - response was lost. - """ + """Inject CommitFailedException before ("conflict") or after ("lost") a commit.""" real_commit = catalog.commit_table - remaining = list(outcomes) + remaining = iter(outcomes) def commit(*args: Any, **kwargs: Any) -> Any: - outcome = remaining.pop(0) if remaining else "ok" + outcome = next(remaining, "ok") if outcome == "conflict": raise CommitFailedException("concurrent update") result = real_commit(*args, **kwargs) @@ -1332,15 +1328,11 @@ def _assert_one_readable_snapshot(catalog: Catalog, identifier: str, expected: l def test_lost_response_on_the_last_attempt_keeps_the_landed_snapshot( catalog: Catalog, num_retries: str, outcomes: list[str] ) -> None: - """A lost response on the last attempt must be checked for a landed snapshot before cleaning up. - - With no attempt left, the landed check of the retry loop is never reached. Cleaning up - unconditionally deletes the manifest list of the snapshot the catalog just committed. - """ + """Check for a landed snapshot before cleanup, even when retries are exhausted.""" import pyarrow as pa catalog.create_namespace("default") - catalog.create_table( + table = catalog.create_table( "default.last_attempt_lost", schema=_test_schema(), properties={ @@ -1349,7 +1341,6 @@ def test_lost_response_on_the_last_attempt_keeps_the_landed_snapshot( TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2", }, ) - table = catalog.load_table("default.last_attempt_lost") with _commit_outcomes(catalog, *outcomes): table.append(pa.table({"x": [1]})) @@ -1359,17 +1350,15 @@ def test_lost_response_on_the_last_attempt_keeps_the_landed_snapshot( @pytest.mark.filterwarnings("ignore:Delete operation did not match any records") def test_lost_response_for_an_overwrite_of_an_empty_table_keeps_the_landed_snapshot(catalog: Catalog) -> None: - """The landed check must use the snapshots that were committed, not every producer. + """Recognize a landed overwrite even when its delete producer adds no snapshot. - An overwrite of an empty table registers a delete producer that adds no snapshot. Requiring a - snapshot from every producer never finds the landed append, so the retry rebuilds the updates. - The rebuilt delete then sees its own landed append as a conflicting commit, and the resulting - ValidationException cleans up the landed snapshot's files. + Requiring a snapshot from every producer misses the landed append, so the rebuilt delete sees + that append as a conflict, and the ValidationException cleanup deletes the landed snapshot's files. """ import pyarrow as pa catalog.create_namespace("default") - catalog.create_table( + table = catalog.create_table( "default.empty_overwrite_lost", schema=_test_schema(), properties={ @@ -1377,7 +1366,6 @@ def test_lost_response_for_an_overwrite_of_an_empty_table_keeps_the_landed_snaps TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2", }, ) - table = catalog.load_table("default.empty_overwrite_lost") with _commit_outcomes(catalog, "lost"): table.overwrite(pa.table({"x": [1]})) @@ -1386,16 +1374,15 @@ def test_lost_response_for_an_overwrite_of_an_empty_table_keeps_the_landed_snaps def test_retry_without_staged_snapshots_validates_against_refreshed_metadata(catalog: Catalog) -> None: - """A retry must refresh the table even when no attempt sent a snapshot. + """Refresh before retrying a property update combined with a delete that matched nothing. - A property update with a delete that matched nothing has producers but adds no snapshot. Skipping - the refresh rebuilds the delete against stale metadata, which misses a concurrent append matching - the delete predicate. + No snapshot is sent, so without a refresh the delete is rebuilt against stale metadata and misses + a concurrent append matching its predicate. """ import pyarrow as pa catalog.create_namespace("default") - catalog.create_table( + table = catalog.create_table( "default.no_snapshot_retry", schema=_test_schema(), properties={ @@ -1404,7 +1391,7 @@ def test_retry_without_staged_snapshots_validates_against_refreshed_metadata(cat }, ) - tx = catalog.load_table("default.no_snapshot_retry").transaction() + tx = table.transaction() tx.set_properties({"key": "value"}) with pytest.warns(UserWarning): # the delete matches nothing at staging time tx.delete("x > 45") @@ -1421,23 +1408,22 @@ def test_retry_without_staged_snapshots_validates_against_refreshed_metadata(cat def test_lost_response_with_a_failed_refresh_keeps_the_files(catalog: Catalog) -> None: - """If the landed check cannot refresh the table, the outcome is unknown and nothing is deleted.""" + """Preserve committed files when the outcome cannot be verified.""" import pyarrow as pa catalog.create_namespace("default") - catalog.create_table( + table = catalog.create_table( "default.lost_refresh_failed", schema=_test_schema(), properties={TableProperties.COMMIT_NUM_RETRIES: "0"}, ) - table = catalog.load_table("default.lost_refresh_failed") with ( _commit_outcomes(catalog, "lost"), patch.object(table, "refresh", side_effect=ConnectionError("catalog unreachable")), + pytest.raises(CommitStateUnknownException) as exc_info, ): - with pytest.raises(CommitStateUnknownException) as exc_info: - table.append(pa.table({"x": [1]})) + table.append(pa.table({"x": [1]})) assert isinstance(exc_info.value.__cause__, CommitFailedException) _assert_one_readable_snapshot(catalog, "default.lost_refresh_failed", [{"x": 1}])