diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 2c5c26800c..7d9b332759 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,42 +1124,45 @@ def commit_transaction(self) -> Table: try: try: + # 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( + update.snapshot.snapshot_id for update in self._updates if isinstance(update, AddSnapshotUpdate) + ) try: self._table._do_commit( # pylint: disable=W0212 updates=self._updates, requirements=self._requirements, ) - 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 - ): - # 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. - self._cleanup_uncommitted_manifests() + + 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 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 @@ -1203,6 +1207,19 @@ def commit_transaction(self) -> Table: return self._table + def _attempt_landed(self, snapshot_ids: set[int], commit_error: CommitFailedException) -> bool: + """Refresh the table and check for a landed snapshot. + + Raise CommitStateUnknownException if refresh fails, preserving potentially committed files. + """ + 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: @@ -1210,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 ab95457e23..ecfd2b023c 100644 --- a/tests/table/test_commit_retry.py +++ b/tests/table/test_commit_retry.py @@ -1290,3 +1290,140 @@ 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: + """Inject CommitFailedException before ("conflict") or after ("lost") a commit.""" + real_commit = catalog.commit_table + remaining = iter(outcomes) + + def commit(*args: Any, **kwargs: Any) -> Any: + outcome = next(remaining, "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: + """Check for a landed snapshot before cleanup, even when retries are exhausted.""" + import pyarrow as pa + + catalog.create_namespace("default") + table = 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", + }, + ) + + 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: + """Recognize a landed overwrite even when its delete producer adds no snapshot. + + 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") + table = 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", + }, + ) + + with _commit_outcomes(catalog, "lost"): + table.overwrite(pa.table({"x": [1]})) + + _assert_one_readable_snapshot(catalog, "default.empty_overwrite_lost", [{"x": 1}]) + + +def test_retry_without_staged_snapshots_validates_against_refreshed_metadata(catalog: Catalog) -> None: + """Refresh before retrying a property update combined with a delete that matched nothing. + + 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") + table = 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 = table.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: + """Preserve committed files when the outcome cannot be verified.""" + import pyarrow as pa + + catalog.create_namespace("default") + table = catalog.create_table( + "default.lost_refresh_failed", + schema=_test_schema(), + properties={TableProperties.COMMIT_NUM_RETRIES: "0"}, + ) + + with ( + _commit_outcomes(catalog, "lost"), + patch.object(table, "refresh", side_effect=ConnectionError("catalog unreachable")), + 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}])