Add commit retry and concurrency validation for writes

[lawofcycles]

Closes #3319
Closes #819
Closes #269

Rationale for this change

PyIceberg currently fails immediately with CommitFailedException when a concurrent transaction commits first, regardless of whether the writes actually conflict. Java Iceberg handles this transparently through its retry loop in SnapshotProducer.commit().

This PR adds automatic commit retry with exponential backoff and data conflict validation to PyIceberg, matching Java Iceberg's behavior. On CommitFailedException, the retry loop refreshes table metadata, re-runs validation, and regenerates manifests. If validation detects a real data conflict, the operation aborts with ValidationException instead of retrying.

The retry loop is placed in Transaction.commit_transaction() rather than in individual snapshot producers. This is necessary because Transaction.delete() uses two producers (_DeleteFiles + _OverwriteFiles) that must be committed atomically. Retrying at the producer level would break this atomicity.

Validation behavior follows Java's BaseOverwriteFiles.validate(), using the existing validation functions from validate.py that were contributed through #1935, #1938, #2050, and #3049.

Are these changes tested?

Yes. Unit tests and integration tests covering retry success, ValidationException abort, retry exhaustion, isolation levels, partition-level conflict detection, manifest cleanup, and producer state reset.

Are there any user-facing changes?

Yes. Previously, all concurrent write conflicts resulted in CommitFailedException.

Now:

• Compatible concurrent writes (e.g. concurrent appends) are retried automatically and succeed transparently
• Incompatible concurrent writes (e.g. concurrent deletes on the same data) raise ValidationException instead of CommitFailedException

The following new table properties are supported.

• commit.retry.num-retries (default: 4)
• commit.retry.min-wait-ms (default: 100)
• commit.retry.max-wait-ms (default: 60000)
• write.delete.isolation-level (default: serializable)
• write.update.isolation-level (default: serializable)

[lawofcycles]

Benchmark results

This PR brings three capabilities to PyIceberg's write path.

1. Transparent retry for concurrent writes. Users no longer need to implement retry logic around table.append() or table.delete().
2. Data conflict validation. Incompatible concurrent modifications (e.g. concurrent deletes on the same data) are detected and rejected with ValidationException, preventing silent data corruption.
3. Efficient retry via data file reuse. On retry, only manifests are regenerated. Data files already written to S3 are reused, avoiding redundant Parquet writes.

To validate (3), I benchmarked concurrent appends using the NYC Yellow Taxi dataset (2024-01, 2.9M rows, 19 columns) with Glue Data Catalog + S3.

Before vs After

Without this PR, concurrent appends fail immediately with CommitFailedException. Only one writer succeeds per batch, regardless of parallelism.

Workers	Before (no retry)	After (this PR)
2	50.0%	100.0%
4	25.0%	100.0%
8	12.5%	100.0%

(N workers x 10 batches x 1K rows, commit.retry.num-retries=10, commit.retry.min-wait-ms=500)

Internal retry vs user-side retry

Compared the internal retry (this PR) against a user-side retry that catches CommitFailedException and re-does load_table + append from scratch. Both use the same backoff parameters (retries=15, min-wait=500ms).

Workers	Internal retry	User-side retry	Speedup
2	33s	46s	1.4x
4	68s	87s	1.3x
8	167s	299s	1.8x
16	399s	588s	1.5x

(3 batches per worker, ~370K-1.5M rows per batch depending on worker count)

Internal retry is faster because it reuses data files already written to S3 and only regenerates manifests on retry. User-side retry rewrites Parquet files on every attempt.

Interestingly, internal retry actually performs more retries than user-side retry (88 vs 50 total retries at 8 workers), because the shorter retry window increases commit attempt density. Despite more retries, the total time is lower because each retry is much cheaper.

Tuning commit.retry.min-wait-ms

Tested different min-wait-ms values with 8 workers to find the optimal backoff for Glue.

min-wait-ms	Total time	Total retries
100	158s	78
500	126s	115
1000	238s	67
2000	235s	63
3000	206s	41

The default (100ms, matching Java Iceberg) works reasonably well, but 500ms is optimal for Glue. Too short causes contention storms, too long wastes time waiting. The optimal value depends on the catalog's commit latency.

[qzyu999]

Hi @lawofcycles, thanks so much for this amazing PR. I took a look and saw two spaces so far where there are some minor gaps that can be easily patched.

The first is regarding AssertTableUUID, where I notice a pattern of repetitively adding/removing it inside the retry loop for commit_transaction(). I believe this can be resolved simply by moving the addition part outside the for-loop.

The second is also regarding commit_transaction(), where in the case of an abort (e.g., ValidationException), there will be some orphaned manifest files. This can be easily fixed by adding a try/except around the for-loop itself, making sure upon failure that both _uncommitted_manifests and _written_manifests are cleared.

Thanks again for the great work, I look forward to #3320 merging so that I may integrate the changes into #3131, PTAL!

[qzyu999]

suggestion: Here AssertTableUUID is appended to self._requirements within each retry loop, but below in _rebuild_snapshot_updates it's removed again with:

self._requirements = tuple(r for r in self._requirements if not isinstance(r, (AssertRefSnapshotId, AssertTableUUID)))

This can be simplified by moving self._requirements += (AssertTableUUID(uuid=self.table_metadata.table_uuid),) outside the for-loop and updating the line in _rebuild_snapshot_updates to simply:

self._requirements = tuple(r for r in self._requirements if not isinstance(r, AssertRefSnapshotId))

The reason being is that AssertTableUUID would remain constant the whole time, so we're simply adding and removing it within each retry.

[lawofcycles]

Thanks for the suggestion. I moved it outside the loop and removed the AssertTableUUID filter from _rebuild_snapshot_updates.

[qzyu999]

suggestion: We could also add a second similar function as follows:

    def _clean_all_uncommitted(self) -> None:
        """Clean up all manifests written during this producer's lifecycle on abort."""
        for path in itertools.chain(self._uncommitted_manifests, self._written_manifests):
            try:
                self._io.delete(path)
            except Exception:
                logger.warning("Failed to delete uncommitted manifest: %s", path, exc_info=True)
        self._uncommitted_manifests.clear()
        self._written_manifests.clear()

then in Transaciton.commit_transaction(), we can add a try/except to the for-loop as follows:

        try:
            for attempt in range(num_retries + 1):
                try:
                    self._table._do_commit(...)
                    self._cleanup_uncommitted_manifests()
                    break
                except CommitFailedException:
                    ... # retry logic
        except Exception:
            # Catch ValidationException or retry exhaustion
            for producer in self._snapshot_producers:
                producer._clean_all_uncommitted()
            raise

this would then allow the PyIceberg implementation to mirror the cleanAll() method in Java. In the current implementation, the for-loop for retrying will only clear out the _uncommitted_manifests from the previous failed retries, but we can extend this with _clean_all_uncommitted which will clear out that and _written_manifests from the current attempt in the case of a permanent abort. This would fix the gap for orphaned manifests from ValidationException (or other permanent failures) that are not cleaned up. I also think it's worth mentioning that this fix could be cleanly added to this PR without waiting for a full Delete orphaned files implementation in PyIceberg. WDYT about adding this into the current PR?

[lawofcycles]

Good catch. I added _clean_all_uncommitted() that cleans up both _uncommitted_manifests and _written_manifests, and wrapped the retry loop with try/except so it gets called on any permanent failure (ValidationException, retry exhaustion, etc.).

[antoniogplobato]

Any update on this? We are currently facing a production issue that this PR would solve.

[lawofcycles]

@qzyu999 Thanks for the review and sorry for the late response. Both addressed in the latest commit. Looking forward to seeing this integrated with #3131!

[qzyu999]

Hi @lawofcycles, thank you so much for accepting the suggested changes. I reviewed those and they all look correct.

[rambleraptor]

Thanks for writing this! This is such a useful feature and there's a ton of nuance.

Here's a short Gist explaining this issue that I'm seeing. Code is often easier to parse than writing.

We try to commit and there's a conflict. Now, while we're waiting to retry, a second conflicting commit comes in. We're now two commits behind. We have to make sure that there's no issues against both of these commits.

We should have this example as a test. As it stands, we're ignoring one of the commits.

[rambleraptor]

Why are both of these parent_snapshot?

[lawofcycles]

It shouldn't be. This was a bug. _parent_snapshot_id gets updated on each retry, so passing it for both collapsed the validation window to zero. Fixed by introducing _starting_snapshot_id that stays fixed across retries.

[lawofcycles]

@rambleraptor Thanks for the repro, this made the issue very clear. Confirmed and fixed. The validation now pins the original base snapshot at init time so the window covers all concurrent commits, not just the latest head.

[abnobdoss]

Could we make these short-circuits a bit more targeted?

self._parent_snapshot_id is None seems like a valid empty-table/new-branch case, but if a non-null parent or starting snapshot id cannot be resolved, I’m not sure we should silently skip validation. Would it make sense to raise in those cases, or otherwise distinguish expected no-snapshot cases from unexpected missing snapshots?

[lawofcycles]

Good catch. Now raises ValidationException when the ID is non-null but unresolvable. The _parent_snapshot_id is None early return stays for the empty-table case.

[abnobdoss]

Could we add coverage for operations that stage more than one snapshot? For example, Transaction.delete() can create both _DeleteFiles and _OverwriteFiles when one file is fully deleted and another is partially rewritten.

For example:

    def test_mixed_delete_overwrite_starts_from_catalog_snapshot(catalog: Catalog) -> None:
        """Mixed full-file and partial deletes should validate from the original table snapshot."""
        catalog.create_namespace("default")
        schema = _test_schema()
        table = catalog.create_table("default.mixed_delete_start_snapshot", schema=schema)
    
        import pyarrow as pa
        from pyiceberg.table.update.snapshot import _DeleteFiles, _OverwriteFiles
    
        table.append(pa.table({"x": [1, 2]}))
        table.append(pa.table({"x": [2, 3]}))
    
        base_snapshot_id = table.metadata.current_snapshot_id
    
        tx = Transaction(table, autocommit=False)
        tx.delete("x <= 2")
    
        assert len(tx._snapshot_producers) == 2
    
        delete_producer, overwrite_producer = tx._snapshot_producers
        assert isinstance(delete_producer, _DeleteFiles)
        assert isinstance(overwrite_producer, _OverwriteFiles)
    
        assert delete_producer._starting_snapshot_id == base_snapshot_id
        assert overwrite_producer._starting_snapshot_id == base_snapshot_id

[lawofcycles]

Added. Writing this test exposed a bug: _OverwriteFiles was picking up the post-_DeleteFiles snapshot as its starting point. Fixed by propagating _starting_snapshot_id from the delete producer.

[lawofcycles]

On _SnapshotProducer._validate_concurrency, this line:

conflict_detection_filter = self._predicate if self._predicate != AlwaysFalse() else None

When a producer has no predicate (_predicate == AlwaysFalse()), conflict_detection_filter becomes None, and under the default serializable isolation we then call:

_validate_added_data_files(table, catalog_head, None, starting_snapshot)

Since _filter_manifest_entries treats data_filter=None + partition_set=None as "match everything," this validates against every data file added anywhere in the table during the commit window. So a predicate-less overwrite gets aborted by any concurrent append — even one to a completely disjoint partition.

The high-level paths are unaffected — delete()/overwrite(overwrite_filter)/upsert() all populate _predicate, so they scope correctly. The gap is the low-level update_snapshot().overwrite() + explicit delete_data_file() path (file/row-level overwrites that don't set a predicate), and presumably the _RewriteFiles integration discussed above — i.e. compaction/rewrite-style operations that touch a known, bounded set of files.

For those, Java doesn't fall back to whole-table: RewriteFiles/OverwriteFiles derive the conflict scope from the operation's own added/deleted files (a partition set), so a file-level rewrite isn't invalidated by an unrelated concurrent append. With the current fallback, these operations become effectively un-retryable under any concurrent write on the table, which works against the PR's goal of making non-conflicting commits seamless.

Two questions:

1. Is the whole-table fallback for predicate-less producers intentional for now (e.g. deferring scoped detection to the _RewriteFiles work), or worth addressing here?
2. Would deriving a partition_set from the producer's _added_data_files/_deleted_data_files when _predicate is absent be a reasonable way to scope it, matching the Java behavior?

Good analysis. You're right that predicate-less producers currently fall back to whole-table validation under serializable isolation.

This doesn't affect the paths exposed by this PR since delete(), overwrite(), and append() all set _predicate before reaching _validate_concurrency. The gap is in the low-level update_snapshot().overwrite() + explicit file deletion path, and the future _RewriteFiles integration for compaction.

Deriving a partition set from _added_data_files/_deleted_data_files when _predicate is absent (matching Java's approach) makes sense as the fix. This will become relevant when _RewriteFiles lands and the path becomes user-reachable.

[qzyu999]

Hi @lawofcycles, verified the manifest list tracking/cleanup changes I requested — LGTM on those. The CommitWindow refactoring and branch fixes look reasonable from a quick read, but I haven't reviewed those in the same depth as of yet.

[qzyu999]

Hi @lawofcycles, a few minor nits inline (all non-blocking):

1. Class-level annotation: _isolation_level_property is the only instance attribute on _SnapshotProducer not declared at the class body level. One-liner change for consistency.

2. super().commit(): The override duplicates the base class logic instead of delegating. Safer to wrap super() so future changes to UpdateTableMetadata.commit() propagate automatically.

3. TYPE_CHECKING import for _SnapshotProducer (3 locations): _snapshot_producers and _register_snapshot_producer use Any where they could be properly typed via a TYPE_CHECKING guard.

[qzyu999]

Nit: _isolation_level_property should be declared as a class-level type annotation alongside the other instance variable declarations. The rest of _SnapshotProducer's fields follow this pattern of declaring the attribute's type at the class body level for discoverability and static analysis, then assigning in __init__. Missing it here breaks the convention and makes the field invisible when scanning the class interface.

Currently, it's only set in the __init__ as follows:

self._isolation_level_property: str = TableProperties.WRITE_DELETE_ISOLATION_LEVEL

we should also add it above in the class-level annotations block:

_isolation_level_property: str

[lawofcycles]

Done.

[qzyu999]

Nit: commit() duplicates the base class implementation rather than delegating via super().

Consider:

def commit(self) -> None:
    self._transaction._register_snapshot_producer(self)
    super().commit()

This preserves the override's additional behavior (producer registration) while keeping the base class as the single source of truth for the commit mechanics. If UpdateTableMetadata.commit() ever gains additional logic (logging, hooks, etc.), this override will pick it up automatically rather than silently diverging.

[lawofcycles]

Done.

[qzyu999]

Nit (1/3) - Proper typing for _snapshot_producers

_snapshot_producers and _register_snapshot_producer (introduced in this PR) currently use Any as their type. This means mypy and IDEs treat producer objects as completely untyped, no error if you call a nonexistent method, no autocompletion, no verification that the right object type is being stored.

We can't do a normal from pyiceberg.table.update.snapshot import _SnapshotProducer at the top of the file because that module already imports from pyiceberg.table, and it would create a circular import at runtime. However, there's already a TYPE_CHECKING block here for exactly this situation. Imports inside this block are only seen by type checkers (mypy, Pyright, IDEs) so they're completely skipped at runtime, therefore no circular import occurs.

Add at the end of this block:

    from pyiceberg.table.update.snapshot import _SnapshotProducer

This makes _SnapshotProducer available as a type annotation (for 2/3 and 3/3 below) without affecting runtime behavior.

[lawofcycles]

Done.

[qzyu999]

Nit (2/3): With the TYPE_CHECKING import from (1/3), this can be properly annotated:

self._snapshot_producers: list[_SnapshotProducer[Any]] = []

This matters since the retry loop in _rebuild_snapshot_updates() calls producer._refresh_for_retry(), producer._validate_concurrency(), producer._clean_all_uncommitted() etc. on items from this list. With list[Any], mypy treats those calls as untyped, and it won't catch a misspelled method name, a wrong argument, or a missing attribute. With list[_SnapshotProducer[Any]], all those calls are statically verified against the actual class definition.

[lawofcycles]

Done.

[qzyu999]

Nit (3/3): Same reasoning, type the parameter so callers passing the wrong type are caught:

def _register_snapshot_producer(self, producer: _SnapshotProducer[Any]) -> None:

This completes the type chain: _SnapshotProducer.commit() calls self._transaction._register_snapshot_producer(self), with this annotation, mypy confirms that self (a _SnapshotProducer) satisfies the parameter type. If someone accidentally tried to register a non-producer object, it would be flagged immediately.

[lawofcycles]

Done.

[qzyu999]

Hi @lawofcycles, LGTM, thanks for addressing the feedback!

[wallacms]

Is this ready to merge? I'm waiting on this change to clean up some of my pipelines

[lawofcycles]

I'm waiting for reviews @Fokko and @rambleraptor. I would appreciate it if you could take a look.

[rambleraptor]

Alright, I think I'm good on this. Thank you so much for all the hard work on this PR!

[rambleraptor]

you can just use `num_retries = property_as_int(properties, TableProperties.COMMIT_NUM_RETRIES, TableProperties.COMMIT_NUM_RETRIES_DEFAULT) to make this easier to understand. The if/else makes this a bit confusing.

[lawofcycles]

addressed.

[rambleraptor]

Same for all of these.

[lawofcycles]

addressed.

[rambleraptor]

nit: It looks like Java uses 0.1 instead of 0.25.

[lawofcycles]

addressed.

[lawofcycles]

@Fokko This PR has been open for about two months and has gone through multiple rounds of review with all requested changes addressed. Would you be able to take a look?

[Fokko]

Thanks for working on this @lawofcycles This looks pretty solid by piecing everything together. I've left some comments. Sorry for the nit picking sometime, but the SnapshotProducer is known for its complexity, so it is important that everything goes in is properly reviewed to keep it managable.

[Fokko]

This is dead code because snapshot_by_id will throw a StopIteration before it gets here.

iceberg-python/pyiceberg/table/metadata.py

Lines 239 to 241 in 48e710d

	def snapshot_by_id(self, snapshot_id: int) -> Snapshot | None:
	"""Get the snapshot by snapshot_id."""
	return next((snapshot for snapshot in self.snapshots if snapshot.snapshot_id == snapshot_id), None)

[lawofcycles]

I think this branch is actually reachable.

snapshot_by_id returns None rather than raising here, since it is implemented as next((...), None), so execution does reach the guard.

base_id is the branch head snapshot id captured when the operation starts (_starting_snapshot_id), fixed across retries. On retry we look it up in the refreshed metadata. It resolves to None when the starting snapshot has been expired or removed between the start of the operation and the retry. In that case we can no longer trace the validation ancestry from it, so raising ValidationException is the safe behavior.

[Fokko]

Ah, you're right! Sorry for the confusion. Thanks for clarifying this, I was suprised that this wasn't being caught by the linter, but I missed the , None part.

[Fokko]

I really dislike that we have to pass this in and around everywhere, it really clutters up the code. Earlier I suggested to adding metadata.isolation_level(Operation), I think this would solve this and we can just resolve it from the table.

Also, this will also take the latest properties into account. So, if in the conflict, the isolation level has been updated, this will be taken into account since the underlying properties will be refreshed 😄

[lawofcycles]

Added TableMetadata.isolation_level(operation) and dropped the _isolation_level_property field. The producer just carries the operation now, and _validate_concurrency resolves the level from the metadata, so the lookup lives in one place and still picks up refreshed properties on retry.

I based it on the logical operation (delete vs update) rather than the snapshot operation, to match Java. SparkRowLevelOperationBuilder.isolationLevel(properties, command) picks the property from the SQL command and applies it to the whole operation, including copy on write, so a copy on write delete stays on write.delete.isolation-level. Going by the snapshot operation would push the rewrite part of a delete onto write.update, which diverges.

The tradeoff is that a small operation marker still travels with the producer.

[lawofcycles]

Thanks for the thorough review @Fokko. I have addressed the comments and left replies inline. On a couple of them I made a design choice that could reasonably go either way, so let me know your thoughts if you would prefer a different direction.

[abnobdoss]

This is looking great; I'm excited for this to get added! I just left a few comments on some edge cases I dug into, mostly around what happens after a commit fails or its outcome is unknown.

[abnobdoss]

I'm a bit wary of the bare except Exception here, so I traced what can actually reach it. I count three groups with very different meanings:

1. The commit definitely did not happen: ValidationException from the conflict checks during rebuild, IO errors while writing the next attempt's manifests.
2. The outcome is unknown: CommitStateUnknownException (REST raises it for 500/502/504, which is what a gateway timeout looks like when the backend finished the commit), raw requests connection errors and timeouts (nothing in the client catches these), and SQLAlchemy OperationalError, which covers both "database locked" and "connection dropped after COMMIT".
3. The commit definitely happened: a pydantic ValidationError while parsing the 200 commit response, and the ValueError from _check_uuid, both raised after commit_table already succeeded.

The handler treats all three the same way and deletes every written manifest and manifest list. For groups 2 and 3 those files can already be referenced by the catalog's current snapshot, and deleting them makes the table unreadable for every reader, not just this writer.

import pyarrow as pa
import pytest
from unittest.mock import patch

from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.exceptions import CommitStateUnknownException
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField


def test_unknown_commit_outcome_keeps_the_committed_files(tmp_path):
    catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
    catalog.create_namespace("default")
    table = catalog.create_table(
        "default.t", schema=Schema(NestedField(1, "x", LongType(), required=False))
    )
    real_commit = catalog.commit_table

    def commit_then_lose_response(*args, **kwargs):
        real_commit(*args, **kwargs)
        raise CommitStateUnknownException("response lost after the commit landed")

    with patch.object(catalog, "commit_table", side_effect=commit_then_lose_response):
        with pytest.raises(CommitStateUnknownException):
            table.append(pa.table({"x": [1]}))

    # The commit landed, so the table must still be readable.
    assert catalog.load_table("default.t").scan().to_arrow().to_pylist() == [{"x": 1}]

Fails with FileNotFoundError on the committed snapshot's manifest list.

I think the safe rule is to invert the default: only delete files for exceptions that guarantee the commit did not happen (CommitFailedException, ValidationException) and keep them for everything else. Orphan files are recoverable, a deleted manifest list behind a live snapshot is not. That matches Java, which rethrows CommitStateUnknownException before any cleanup runs (SnapshotProducer.java).

Longer term, group 2 can only be classified truthfully by asking the catalog whether the commit landed, the way Java's checkCommitStatus polls for its own metadata location. That needs a stable snapshot id across attempts, which I have raised separately.

[abnobdoss]

On a similar note to my comment on the bare except Exception in Transaction.commit_transaction:

A commit call over a network has three possible outcomes: it applied and we got the response, it did not apply and we got an error, or it applied and we still got an error because the response was lost. The retry loop only models the first two. In the third case it refreshes and commits the same data again, like retrying a payment without an idempotency key.

The snapshot id could be that idempotency key, but this line mints a new one on every attempt, so the client has no way to recognize its own commit in the refreshed metadata. The third case is not hypothetical: Glue's boto client silently resends UpdateTable on connection errors and the resend's version conflict is reported as CommitFailedException, and a retrying proxy in front of a REST catalog does the same via a 409.

import pyarrow as pa
from unittest.mock import patch

from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.exceptions import CommitFailedException
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField


def test_commit_that_landed_but_was_reported_failed_is_not_committed_twice(tmp_path):
    catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
    catalog.create_namespace("default")
    table = catalog.create_table(
        "default.t",
        schema=Schema(NestedField(1, "x", LongType(), required=False)),
        properties={"commit.retry.min-wait-ms": "1", "commit.retry.max-wait-ms": "2"},
    )
    real_commit = catalog.commit_table
    calls = []

    def commit_then_report_conflict(*args, **kwargs):
        result = real_commit(*args, **kwargs)
        calls.append(1)
        if len(calls) == 1:
            raise CommitFailedException("transport layer retried; first response was lost")
        return result

    with patch.object(catalog, "commit_table", side_effect=commit_then_report_conflict):
        table.append(pa.table({"x": [1]}))

    assert catalog.load_table("default.t").scan().to_arrow().to_pylist() == [{"x": 1}]

Fails with FileNotFoundError: the data is committed twice, and the post-success cleanup then deletes manifests the first committed snapshot references.

Java generates the snapshot id once, reuses it for every attempt, and checks the refreshed metadata for it before re-committing (SnapshotProducer.java). Doing the same here would close this for every catalog at once.

[abnobdoss]

Is it safe to clear the computed deletes on retry? Here's a scenario: a writer plans delete("x == 1"), and before it commits, someone else appends a file whose rows all match that predicate. The commit conflicts, the retry replans the delete against the refreshed head, and the concurrent file is now a delete target, so it gets dropped whole. The writer deletes rows it never saw, under the isolation setting whose promise is that concurrent appends are preserved:

import pyarrow as pa

from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField


def test_snapshot_isolation_delete_does_not_remove_rows_it_never_saw(tmp_path):
    catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
    catalog.create_namespace("default")
    table = catalog.create_table(
        "default.t",
        schema=Schema(NestedField(1, "x", LongType(), required=False)),
        properties={
            "write.delete.isolation-level": "snapshot",
            "commit.retry.min-wait-ms": "1",
            "commit.retry.max-wait-ms": "2",
        },
    )
    table.append(pa.table({"x": [0, 1]}))

    stale = catalog.load_table("default.t")
    tx = stale.transaction()
    tx.delete("x == 1")

    # Lands after the delete was planned. The stale writer never saw this row.
    catalog.load_table("default.t").append(pa.table({"x": [1]}))

    tx.commit_transaction()

    rows = sorted(catalog.load_table("default.t").scan().to_arrow()["x"].to_pylist())
    assert rows == [0, 1]  # fails: the concurrent row is gone

Maybe that is the intent, and a retried delete is supposed to re-execute against the new table state and remove the concurrent row too. That reading seems defensible as well, but I do not think the current behavior lands on it either: only whole-file drops are replanned, while partial-file rewrites are planned once in Transaction.delete and not recomputed. So a single concurrent commit can end up half incorporated:

import uuid

from pyiceberg.io.pyarrow import _dataframe_to_data_files


def test_retried_delete_treats_a_concurrent_commit_atomically(tmp_path):
    catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
    catalog.create_namespace("default")
    table = catalog.create_table(
        "default.t",
        schema=Schema(NestedField(1, "x", LongType(), required=False)),
        properties={
            "write.delete.isolation-level": "snapshot",
            "commit.retry.min-wait-ms": "1",
            "commit.retry.max-wait-ms": "2",
        },
    )
    table.append(pa.table({"x": [0, 1]}))

    stale = catalog.load_table("default.t")
    tx = stale.transaction()
    tx.delete("x == 1")

    # One concurrent commit carrying two files: [1] wholly matches, [1, 5] partially.
    b = catalog.load_table("default.t")
    btx = b.transaction()
    with btx.update_snapshot().fast_append() as append:
        for df in (pa.table({"x": [1]}), pa.table({"x": [1, 5]})):
            for f in _dataframe_to_data_files(
                table_metadata=btx.table_metadata, io=b.io, write_uuid=uuid.uuid4(), df=df
            ):
                append.append_data_file(f)
    btx.commit_transaction()

    tx.commit_transaction()

    final = sorted(catalog.load_table("default.t").scan().to_arrow()["x"].to_pylist())
    # The concurrent commit is atomic, so the delete may apply to all of it or none of it.
    assert final in ([0, 1, 1, 5], [0, 5]), f"half of the concurrent commit was deleted: {final}"

The second test fails with [0, 1, 5]: one of the commit's matching rows deleted, the other preserved, which I do not think either reading intends. Since users cannot see or control which file a row lands in, the outcome of the race is effectively random. (test_snapshot_isolation_allows_concurrent_append_delete in this PR is the partial-match instance of the same race and expects the row to survive.)

For reference, Java fixes the set of files a delete operates on at planning time; retries revalidate but never replan, so concurrent commits are consistently preserved under snapshot isolation. Would freezing the planned file set here be an option? As far as I can tell it would make all of these tests pass, including the existing one.

[abnobdoss]

Is a Transaction object meant to be safe to use after commit_transaction? If it is then I think clearing _snapshot_producers on success and resetting or invalidating the state on failure is missing.

Here are some test scenarios:

Scenario 1: a commit fails once for a transient reason (the kinds traced in my except Exception comment), and the caller retries the same transaction. The staged AddSnapshotUpdate survived the failure, its manifest list did not, and the catalog accepts it:

import pyarrow as pa
import pytest
from unittest.mock import patch

from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.exceptions import CommitFailedException
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField


def test_reusing_a_failed_transaction_cannot_publish_deleted_files(tmp_path):
    catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
    catalog.create_namespace("default")
    table = catalog.create_table(
        "default.t",
        schema=Schema(NestedField(1, "x", LongType(), required=False)),
        properties={"commit.retry.num-retries": "0"},
    )
    table.append(pa.table({"x": [1]}))

    tx = table.transaction()
    tx.append(pa.table({"x": [2]}))

    # The commit fails once for a transient reason.
    with patch.object(catalog, "commit_table", side_effect=CommitFailedException("transient")):
        with pytest.raises(CommitFailedException):
            tx.commit_transaction()

    # The caller retries the same transaction, which was safe before this PR.
    try:
        tx.commit_transaction()
    except Exception:
        # Refusing reuse would be fine, as long as the table is untouched.
        assert catalog.load_table("default.t").scan().to_arrow().to_pylist() == [{"x": 1}]
        return

    # If the re-commit is accepted, the table must still be readable.
    rows = sorted(catalog.load_table("default.t").scan().to_arrow()["x"].to_pylist())
    assert rows == [1, 2]

Fails with FileNotFoundError scanning the table: the catalog head points at a manifest list the cleanup deleted.

Scenario 2: no failure, but reuse after a successful commit plus one concurrent writer:

def test_reusing_a_committed_transaction_does_not_damage_the_first_commit(tmp_path):
    catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
    catalog.create_namespace("default")
    table = catalog.create_table(
        "default.t",
        schema=Schema(NestedField(1, "x", LongType(), required=False)),
        properties={"commit.retry.min-wait-ms": "1", "commit.retry.max-wait-ms": "2"},
    )

    tx = table.transaction()
    tx.append(pa.table({"x": [1, 2, 3]}))
    tx.commit_transaction()

    tx.append(pa.table({"x": [10, 20]}))
    catalog.load_table("default.t").append(pa.table({"x": [100]}))  # forces a retry

    tx.commit_transaction()

    rows = sorted(catalog.load_table("default.t").scan().to_arrow()["x"].to_pylist())
    assert rows == [1, 2, 3, 10, 20, 100]

Also fails with FileNotFoundError: the retry replays the first producer, which is still registered, so its batch is committed a second time and its already-committed manifests get moved into the uncommitted list and deleted by the post-success cleanup.

[abnobdoss]

Is base is None safe to treat as an empty window? It also means the writer started on a table with no snapshots, and if another writer lands the first snapshot in the meantime, the retry skips validation entirely. Two jobs racing to initialize the same table is a common bootstrap pattern:

import pyarrow as pa
import pytest

from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.exceptions import ValidationException
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField


def test_writer_that_started_on_an_empty_table_still_validates(tmp_path):
    catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
    catalog.create_namespace("default")
    table = catalog.create_table(
        "default.t",
        schema=Schema(NestedField(1, "x", LongType(), required=False)),
        properties={"commit.retry.min-wait-ms": "1", "commit.retry.max-wait-ms": "2"},
    )

    stale = catalog.load_table("default.t")
    tx = stale.transaction()
    with pytest.warns(UserWarning):  # the delete matches nothing on an empty table
        tx.overwrite(pa.table({"x": [2]}), overwrite_filter="x == 1")

    # The first ever snapshot lands concurrently, with a row matching the filter.
    catalog.load_table("default.t").append(pa.table({"x": [1]}))

    with pytest.raises(ValidationException):
        tx.commit_transaction()

Fails because no exception is raised: the commit goes through unvalidated and the table ends up as just [2]. The concurrent writer's row matched the filter and was deleted, with no error on either side. Java treats a null starting snapshot as validate against the entire history (MergingSnapshotProducer walks every ancestor of the head), and treating a None base the same way here would close this.

[lawofcycles]

Good catch. A None base now triggers full-history validation instead of skipping, matching Java's treatment of a null starting snapshot. _validation_history and the added-file lookups now accept a None lower bound and walk every ancestor of the head. Added your test as a regression.

[abnobdoss]

With commit.retry.num-retries=-1 this is range(0): no commit is attempted, no error is raised, the staged updates are cleared after the loop, and the call returns as success. Since this is a table property, one writer setting it turns every writer's commits into silent no-ops. Some tools use -1 to mean retry forever, so it is a plausible value.

import pyarrow as pa

from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField


def test_invalid_retry_count_does_not_silently_skip_the_commit(tmp_path):
    catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
    catalog.create_namespace("default")
    table = catalog.create_table(
        "default.t",
        schema=Schema(NestedField(1, "x", LongType(), required=False)),
        properties={"commit.retry.num-retries": "-1"},
    )
    table.append(pa.table({"x": [1]}))
    assert catalog.load_table("default.t").scan().to_arrow().to_pylist() == [{"x": 1}]

Fails with an empty table. Validating the property when read (raise on negatives, or clamp to zero) makes this a loud config error instead of silent data loss. Java runs one attempt even for a negative value.

[lawofcycles]

Good catch. Clamped num-retries to a minimum of 0, so a negative value now runs one attempt instead of silently skipping the commit, which matches Java. Added a regression test.

[Fokko]

This PR looks good to me now. I'll wait until the comments by @abnobdoss are resolved before merging