Check whether a failed commit landed before cleaning up its files

[dancsi]

Closes #4021
Closes #4022

Rationale for this change

When commit_table raises CommitFailedException after the catalog has already applied the commit (a lost response), Transaction.commit_transaction can delete the landed snapshot's manifest list and manifests, which leaves the table unreadable:

• Commit cleanup deletes a landed snapshot's manifests when the last attempt's response is lost #4021: on the last attempt (including commit.retry.num-retries=0, or after the total timeout), the cleanup runs without checking whether the attempt landed.
• Retry "already landed" check fails for an overwrite of an empty table, then deletes the landed snapshot's manifests #4022: the landed check requires a snapshot from every producer. A producer that adds no snapshot, such as the delete in an overwrite of an empty table, makes it always false. The retry then rebuilds the updates, the rebuilt delete sees its own landed append as a conflict, and the resulting ValidationException cleans up.

We ran into both bugs in Polars' sink_iceberg, which commits through pyiceberg, and work around them in pola-rs/polars#29511. The workaround wraps the catalog's commit_table, reloads the table after a failed commit, and returns a success response when a snapshot it sent is present, so pyiceberg never reaches the cleanup. This PR moves that check into commit_transaction, and Polars can drop the wrapper once it's released.

This change:

• bases the landed check on the AddSnapshotUpdate snapshot ids that were actually sent to the catalog. Each attempt is atomic, so finding any of them proves an attempt landed;
• runs the check after every CommitFailedException, including on the last attempt, before any cleanup;
• raises CommitStateUnknownException (chained to the original error) if the refresh for the check fails. That exception skips the cleanup, so files a landed snapshot may reference are never deleted when the outcome can't be verified;
• still refreshes before rebuilding a retry that sent no snapshots, so its validation sees current metadata;
• runs the post-commit cleanup of superseded attempts' manifests once, after the loop.

Are these changes tested?

Yes, in tests/table/test_commit_retry.py, parametrized over the existing catalog fixture:

• a lost response on the last attempt, with no retries and with retries exhausted;
• a lost response for an overwrite of an empty table;
• a lost response whose landed check can't refresh the table.

Each asserts that exactly one snapshot exists, that its manifest list is still on disk and that the table scans. All of them fail on main.

Are there any user-facing changes?

A lost response on the final attempt now returns successfully when the commit landed, instead of raising CommitFailedException. If the landed check itself can't reach the catalog, CommitStateUnknownException is raised instead of CommitFailedException.

Generated-by: Claude Code (Claude Opus 5.5)

[korsakowii]

Thanks for the fix. This review is limited to the lost-response detection and manifest-cleanup paths at 8153d7878bb5659f87a065d6889100ef92672f0b; it is not a full review of the commit/concurrency implementation.

Two additional regression cases are suggested inline: timeout exhaustion with retry-count budget remaining, and a no-op delete on a nonempty table followed by an append.

Local validation using the existing memory/SQL/SQL-without-rowcount fixtures:

• Both new cases fail on main eb30d8ae6ac4054f751b91cb04d4ae5026354cc2 and pass on this PR (6 parametrized cases).
• The commit-retry, validation, and snapshot test modules pass: 210 tests.
• Ruff check/format and file-scoped mypy pass.

Environment: Python 3.11.9, PyArrow 26.0.0, pure Python Avro decoder, with only its fallback warning ignored. Full-suite, Cython, and remote catalog integration checks were not run.

No additional actionable correctness issue was identified within this scope. These are non-blocking test suggestions; please use them if helpful.

[korsakowii]

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:

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}]