Commits & concurrency

4 h
requires LKH-01

From your world. Overlapping dbt runs writing the same table

Resolve a write conflict with refresh-and-reapply

Two overlapping writers each read the same account balance and write a new one. If you do this naively, one update is silently lost. You implement the refresh-and-reapply loop. The loop makes concurrent read-modify-write commits behave as if they ran serially. Every Iceberg writer must preserve this invariant.

The account table has one row (id=1, bal=<int>). Implement this in solution.py:

`commit_with_retry(catalog, identifier, mutate, max_retries=5, before_commit=None) -> int`

Apply a read-modify-write to account id=1:

Loop up to max_retries times. Each attempt does steps 2 to 7.
Load the table fresh from the catalog inside the loop (catalog.load_table(identifier)). Read the current balance for id=1 from that fresh table.
Compute new = mutate(current).
If before_commit is not None, call before_commit(attempt). Call it after you read and before you write. The tests use it to inject a competing writer. This writer commits between your read and your write. attempt is the 0-based attempt number.
Overwrite the row for id=1 with the new balance (table.overwrite(arrow_table_with_one_row, overwrite_filter=EqualTo('id', 1))).
If the overwrite raises pyiceberg.exceptions.CommitFailedException, the table moved under you. Refresh and reapply: loop again. The next loop reloads the table and re-reads the updated balance. Do not reuse the stale balance. Do not swallow the exception.
On success, return the committed balance. If every attempt conflicts, let the final CommitFailedException propagate.

`read_balance(table) -> int`

Return the current balance for id=1 from the given table. The table is already loaded.

Rules and traps:

The fresh load must be inside the loop. The reload lets attempt 2 see the value of the winner. If you load once before the loop, you get the "no refresh" bug.
If you overwrite with a value that came from a stale read, you get a lost update.
If you catch CommitFailedException and return anyway, you get a silent lost write.

The test runs a serial oracle. The oracle applies the same deltas one after another. The test asserts that your concurrent-with-retry result equals the oracle result. This means no lost update and no double-apply.

Starter code

solution.py

"""LKH-02 / 01-retry-commit — starter. See README.md."""


def read_balance(table):
    # TODO: return the current balance for id=1 from the given table.
    raise NotImplementedError


def commit_with_retry(catalog, identifier, mutate, max_retries=5, before_commit=None):
    # TODO: refresh-and-reapply loop. Load the table fresh inside the loop, read the
    # current balance, compute new = mutate(current), call before_commit(attempt) if
    # given (after reading, before writing), overwrite id=1, and retry on
    # CommitFailedException.
    raise NotImplementedError

Hints

Hint 1

Each attempt must start from a FRESH load: call catalog.load_table(identifier) inside the loop, not once before it. Read the current balance from that fresh table.

Hint 2

Compute new = mutate(current), then call before_commit(attempt) if given, then write with table.overwrite(arrow_row, overwrite_filter=EqualTo('id', 1)). Wrap the overwrite in try/except pyiceberg.exceptions.CommitFailedException.

Hint 3

On CommitFailedException, do NOT reuse the old value and do NOT swallow the error: loop again (which reloads the table and re-reads the now-updated balance). Re-raise only after max_retries attempts.

Run it locally

uv run de start LKH-02/01-retry-commit
# edit workspace/LKH-02/01-retry-commit/
uv run de test