Commit cleanup deletes a landed snapshot's manifests when the last attempt's response is lost

[dancsi]

Apache Iceberg version: 0.12.0 (also reproduced on main at ebbc0ba)

Please describe the bug 🐞

When commit_table raises CommitFailedException on the last allowed attempt, Transaction.commit_transaction calls _clean_all_uncommitted() on every producer without first checking whether that attempt landed. If the catalog applied the commit and only the response was lost, the snapshot is live, but its manifest list and manifests have been deleted. The table can no longer be scanned.

The retry loop does check for a landed snapshot, but only before the next attempt (snapshot_by_id(producer._snapshot_id) after refresh()). When there is no next attempt, the check is skipped:

• commit.retry.num-retries=0: every lost response.
• Default retries: a lost response on the final attempt, or once commit.retry.total-timeout-ms has elapsed.

A lost response is realistic for REST catalogs: a timeout, or a proxy or transport retry that gets a 409 for a request that already applied.

Reproduction

import tempfile

import pyarrow as pa
from pyiceberg.catalog.sql import SqlCatalog
from pyiceberg.exceptions import CommitFailedException
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField

warehouse = tempfile.mkdtemp()
catalog = SqlCatalog("default", uri=f"sqlite:///{warehouse}/catalog.db", warehouse=f"file://{warehouse}")
catalog.create_namespace("default")
table = catalog.create_table(
    "default.t",
    Schema(NestedField(1, "a", LongType(), required=False)),
    properties={"commit.retry.num-retries": "0"},
)

# The catalog applies the commit, but the client sees a failure.
commit_table = catalog.commit_table
def commit_table_lose_response(*args):
    commit_table(*args)
    raise CommitFailedException("simulated lost response")
catalog.commit_table = commit_table_lose_response

try:
    table.append(pa.table({"a": pa.array([1, 2, 3], pa.int64())}))
except CommitFailedException as e:
    print(f"append raised: {e}")

table = catalog.load_table("default.t")
snapshot = table.current_snapshot()
print("manifest list exists:", table.io.new_input(snapshot.manifest_list).exists())
table.scan().to_arrow()

Output:

append raised: simulated lost response
manifest list exists: False
FileNotFoundError: [Errno 2] Failed to open local file '.../metadata/snap-...avro'

Suggested fix

Before _clean_all_uncommitted() on CommitFailedException, refresh and run the same landed check the retry loop uses. If the snapshot is present, treat the commit as successful. If the refresh itself fails, the outcome is unknown, so raise without deleting anything, the same way other exceptions are handled today. ValidationException raised before the attempt reached the catalog can keep cleaning up unconditionally.

Related: #4022. The landed check this fix would reuse gives the wrong answer for transactions with a producer that writes no snapshot.

Willingness to contribute

• I would be willing to contribute a fix for this bug with guidance from the Iceberg community

I have a fix with regression tests for both this and #4022: it keys the landed check on the AddSnapshotUpdate snapshot ids that were sent, and runs it after every CommitFailedException, including on the last attempt. I'm happy to open a PR.