korsakowii commented on code in PR #4023:
URL: https://github.com/apache/iceberg-python/pull/4023#discussion_r4316268732
##########
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
+
+
[email protected](
+ ("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}])
+
+
[email protected]("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}])
Review Comment:
Non-blocking: could these two cases be added to the regression coverage?
The first sets `commit.retry.total-timeout-ms=0` while allowing one retry,
so it deterministically covers the timeout terminal path rather than
retry-count exhaustion.
The second starts from a nonempty table, stages a delete that matches
nothing, then appends a matching row. It exercises the no-snapshot delete
producer and checks that the original snapshot's manifests remain available.
Both use the existing helpers and assert a single catalog commit call during
the lost response, snapshot counts, manifest availability, and exact final
rows. The second also checks parent linkage. The initial seed commit is outside
the injected failure.
Proposed tests:
```python
def
test_lost_response_after_retry_timeout_keeps_the_landed_snapshot(catalog:
Catalog) -> None:
"""Check for a landed commit when the timeout expires before the retry
budget."""
import pyarrow as pa
catalog.create_namespace("default")
table = catalog.create_table(
"default.timeout_lost",
schema=_test_schema(),
properties={
TableProperties.COMMIT_NUM_RETRIES: "1",
TableProperties.COMMIT_TOTAL_RETRY_TIME_MS: "0",
},
)
with _commit_outcomes(catalog, "lost") as commit:
table.append(pa.table({"x": [1]}))
commit.assert_called_once()
_assert_one_readable_snapshot(catalog, "default.timeout_lost", [{"x":
1}])
table = catalog.load_table("default.timeout_lost")
snapshot = table.current_snapshot()
assert snapshot is not None
manifests = snapshot.manifests(table.io)
assert len(manifests) == 1
assert table.io.new_input(manifests[0].manifest_path).exists()
def
test_lost_response_after_noop_delete_on_nonempty_table_keeps_the_landed_snapshot(catalog:
Catalog) -> None:
"""Recognize a landed append when a preceding delete produced no
snapshot."""
import pyarrow as pa
catalog.create_namespace("default")
table = catalog.create_table(
"default.nonempty_noop_delete_lost",
schema=_test_schema(),
properties={
TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "0",
TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "0",
},
)
table.append(pa.table({"x": [1]}))
initial_snapshot = table.current_snapshot()
assert initial_snapshot is not None
tx = table.transaction()
with pytest.warns(UserWarning, match="Delete operation did not match any
records"):
tx.delete("x > 45")
tx.append(pa.table({"x": [50]}))
with _commit_outcomes(catalog, "lost") as commit:
tx.commit_transaction()
commit.assert_called_once()
table = catalog.load_table("default.nonempty_noop_delete_lost")
snapshots = table.snapshots()
assert len(snapshots) == 2
snapshot = table.current_snapshot()
assert snapshot is not None
assert snapshot.parent_snapshot_id == initial_snapshot.snapshot_id
for committed_snapshot in snapshots:
assert table.io.new_input(committed_snapshot.manifest_list).exists()
manifests = committed_snapshot.manifests(table.io)
assert manifests
for manifest in manifests:
assert table.io.new_input(manifest.manifest_path).exists()
assert table.scan().to_arrow().sort_by("x").to_pylist() == [{"x": 1},
{"x": 50}]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
