From b6f88b4a5271d980393d746a732bff8356623bc3 Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Tue, 18 Aug 2026 22:35:42 -0700 Subject: [PATCH 1/7] Validate replaced files during commit retries --- pyiceberg/table/update/snapshot.py | 2 + pyiceberg/table/update/validate.py | 27 +++++++ tests/table/test_commit_retry.py | 111 ++++++++++++++++++++++++++++- 3 files changed, 139 insertions(+), 1 deletion(-) diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 57215dca04..e500e09fac 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -472,6 +472,7 @@ def _validate_concurrency(self) -> None: from pyiceberg.table.snapshots import IsolationLevel from pyiceberg.table.update.validate import ( _validate_added_data_files, + _validate_data_files_exist, _validate_deleted_data_files, _validate_no_new_delete_files, _validate_no_new_deletes_for_data_files, @@ -501,6 +502,7 @@ def _validate_concurrency(self) -> None: _validate_deleted_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot) if self._deleted_data_files: + _validate_data_files_exist(table, catalog_head, self._deleted_data_files, starting_snapshot) _validate_no_new_deletes_for_data_files( table, catalog_head, conflict_detection_filter, self._deleted_data_files, starting_snapshot ) diff --git a/pyiceberg/table/update/validate.py b/pyiceberg/table/update/validate.py index df8506aab4..79d17cef9e 100644 --- a/pyiceberg/table/update/validate.py +++ b/pyiceberg/table/update/validate.py @@ -195,6 +195,33 @@ def _validate_deleted_data_files( raise ValidationException(f"Deleted data files were found matching the filter for snapshots {conflicting_snapshots}!") +def _validate_data_files_exist( + table: Table, + starting_snapshot: Snapshot, + data_files: set[DataFile], + parent_snapshot: Snapshot | None, +) -> None: + """Validate that explicitly replaced data files have not been concurrently deleted. + + Args: + table: Table to validate + starting_snapshot: Snapshot current at the start of the operation + data_files: Data files that must still exist + parent_snapshot: Ending snapshot on the branch being validated + """ + partition_set: dict[int, set[Record]] = {} + for data_file in data_files: + partition_set.setdefault(data_file.spec_id, set()).add(data_file.partition) + + conflicting_paths = { + entry.data_file.file_path + for entry in _deleted_data_files(table, starting_snapshot, None, partition_set, parent_snapshot) + if entry.data_file in data_files + } + if conflicting_paths: + raise ValidationException(f"Data files were concurrently deleted: {sorted(conflicting_paths)}") + + def _added_data_files( table: Table, starting_snapshot: Snapshot, diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index ce5dca96aa..b91eb23bd6 100644 --- a/tests/table/test_commit_retry.py +++ b/tests/table/test_commit_retry.py @@ -14,6 +14,7 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +from dataclasses import dataclass from typing import Any from unittest.mock import patch @@ -21,8 +22,9 @@ from pyiceberg.catalog import Catalog from pyiceberg.exceptions import CommitFailedException, CommitStateUnknownException, ValidationException +from pyiceberg.manifest import DataFile from pyiceberg.schema import Schema -from pyiceberg.table import TableProperties, Transaction +from pyiceberg.table import Table, TableProperties, Transaction from pyiceberg.table.snapshots import IsolationLevel, Operation from pyiceberg.types import LongType, NestedField, StringType @@ -323,6 +325,113 @@ def test_concurrent_overwrite_overwrite_raises_validation_exception(catalog: Cat tbl2.overwrite(pa.table({"x": [40, 50, 60]}), overwrite_filter="x > 0") +@dataclass(frozen=True) +class _FileOverwriteScenario: + identifier: str + deleting_table: Table + replacing_transaction: Transaction + target_file: DataFile + same_partition_file: DataFile + different_partition_file: DataFile + + +def _prepare_file_overwrite_scenario(catalog: Catalog) -> _FileOverwriteScenario: + """Prepare a file replacement and candidates for concurrent deletion.""" + import uuid + + import pyarrow as pa + + from pyiceberg.io.pyarrow import _dataframe_to_data_files + from pyiceberg.partitioning import PartitionField, PartitionSpec + from pyiceberg.transforms import IdentityTransform + + catalog.create_namespace("default") + schema = Schema( + NestedField(1, "category", StringType(), required=False), + NestedField(2, "value", LongType(), required=False), + ) + spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category")) + identifier = "default.concurrent_file_delete" + table = catalog.create_table(identifier, schema=schema, partition_spec=spec) + table.append(pa.table({"category": ["a", "b"], "value": [0, 1]})) + original_file = next(task.file for task in table.scan().plan_files() if task.file.partition[0] == "a") + table.append(pa.table({"category": ["a"], "value": [3]})) + + deleting_table = catalog.load_table(identifier) + replacing_table = catalog.load_table(identifier) + active_files = [task.file for task in replacing_table.scan().plan_files()] + same_partition_file = next( + data_file for data_file in active_files if data_file.partition[0] == "a" and data_file != original_file + ) + different_partition_file = next(data_file for data_file in active_files if data_file.partition[0] == "b") + replacement_files = list( + _dataframe_to_data_files( + table_metadata=replacing_table.metadata, + df=pa.table({"category": ["a"], "value": [2]}), + io=replacing_table.io, + write_uuid=uuid.uuid4(), + ) + ) + + replacing_transaction = replacing_table.transaction() + with replacing_transaction.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(original_file) + for replacement_file in replacement_files: + overwrite.append_data_file(replacement_file) + + return _FileOverwriteScenario( + identifier=identifier, + deleting_table=deleting_table, + replacing_transaction=replacing_transaction, + target_file=original_file, + same_partition_file=same_partition_file, + different_partition_file=different_partition_file, + ) + + +def test_file_overwrite_fails_when_target_file_is_concurrently_deleted(catalog: Catalog) -> None: + """A file replacement must fail if the original file was concurrently deleted.""" + scenario = _prepare_file_overwrite_scenario(catalog) + + with scenario.deleting_table.transaction() as deleting_transaction: + with deleting_transaction.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(scenario.target_file) + + with pytest.raises(ValidationException, match="Data files were concurrently deleted"): + scenario.replacing_transaction.commit_transaction() + + result = catalog.load_table(scenario.identifier).scan().to_arrow() + assert sorted(result["value"].to_pylist()) == [1, 3] + + +def test_file_overwrite_allows_concurrent_delete_in_same_partition(catalog: Catalog) -> None: + """A file replacement must allow another file in its partition to be concurrently deleted.""" + scenario = _prepare_file_overwrite_scenario(catalog) + + with scenario.deleting_table.transaction() as deleting_transaction: + with deleting_transaction.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(scenario.same_partition_file) + + scenario.replacing_transaction.commit_transaction() + + result = catalog.load_table(scenario.identifier).scan().to_arrow() + assert sorted(result["value"].to_pylist()) == [1, 2] + + +def test_file_overwrite_allows_concurrent_delete_in_different_partition(catalog: Catalog) -> None: + """A file replacement must allow a file in another partition to be concurrently deleted.""" + scenario = _prepare_file_overwrite_scenario(catalog) + + with scenario.deleting_table.transaction() as deleting_transaction: + with deleting_transaction.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(scenario.different_partition_file) + + scenario.replacing_transaction.commit_transaction() + + result = catalog.load_table(scenario.identifier).scan().to_arrow() + assert sorted(result["value"].to_pylist()) == [2, 3] + + def test_concurrent_overwrite_append_retries_successfully(catalog: Catalog) -> None: """Append after a concurrent overwrite should succeed via retry.""" catalog.create_namespace("default") From 144c5bd5bdee4a66b2d413dd26db6326c28575f6 Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Tue, 18 Aug 2026 22:42:31 -0700 Subject: [PATCH 2/7] Simplify concurrent file overwrite tests --- tests/table/test_commit_retry.py | 124 ++++++++++++++++--------------- 1 file changed, 65 insertions(+), 59 deletions(-) diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index b91eb23bd6..3489a46ef0 100644 --- a/tests/table/test_commit_retry.py +++ b/tests/table/test_commit_retry.py @@ -14,7 +14,6 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -from dataclasses import dataclass from typing import Any from unittest.mock import patch @@ -325,23 +324,13 @@ def test_concurrent_overwrite_overwrite_raises_validation_exception(catalog: Cat tbl2.overwrite(pa.table({"x": [40, 50, 60]}), overwrite_filter="x > 0") -@dataclass(frozen=True) -class _FileOverwriteScenario: - identifier: str - deleting_table: Table - replacing_transaction: Transaction - target_file: DataFile - same_partition_file: DataFile - different_partition_file: DataFile +_FILE_OVERWRITE_TABLE = "default.concurrent_file_delete" -def _prepare_file_overwrite_scenario(catalog: Catalog) -> _FileOverwriteScenario: - """Prepare a file replacement and candidates for concurrent deletion.""" - import uuid - +def _create_file_overwrite_table(catalog: Catalog) -> DataFile: + """Create the file-overwrite test table and return the file to replace.""" import pyarrow as pa - from pyiceberg.io.pyarrow import _dataframe_to_data_files from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.transforms import IdentityTransform @@ -351,85 +340,102 @@ def _prepare_file_overwrite_scenario(catalog: Catalog) -> _FileOverwriteScenario NestedField(2, "value", LongType(), required=False), ) spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category")) - identifier = "default.concurrent_file_delete" - table = catalog.create_table(identifier, schema=schema, partition_spec=spec) + table = catalog.create_table(_FILE_OVERWRITE_TABLE, schema=schema, partition_spec=spec) + table.append(pa.table({"category": ["a", "b"], "value": [0, 1]})) - original_file = next(task.file for task in table.scan().plan_files() if task.file.partition[0] == "a") + file_to_replace = next(task.file for task in table.scan().plan_files() if task.file.partition[0] == "a") + table.append(pa.table({"category": ["a"], "value": [3]})) + return file_to_replace + - deleting_table = catalog.load_table(identifier) - replacing_table = catalog.load_table(identifier) - active_files = [task.file for task in replacing_table.scan().plan_files()] - same_partition_file = next( - data_file for data_file in active_files if data_file.partition[0] == "a" and data_file != original_file +def _data_file_in_partition(table: Table, partition: str, excluded_file: DataFile | None = None) -> DataFile: + """Return a data file in a partition, optionally excluding one file.""" + return next( + task.file for task in table.scan().plan_files() if task.file.partition[0] == partition and task.file != excluded_file ) - different_partition_file = next(data_file for data_file in active_files if data_file.partition[0] == "b") + + +def _stage_file_replacement(table: Table, file_to_replace: DataFile) -> Transaction: + """Stage replacing one data file without committing the transaction.""" + import uuid + + import pyarrow as pa + + from pyiceberg.io.pyarrow import _dataframe_to_data_files + replacement_files = list( _dataframe_to_data_files( - table_metadata=replacing_table.metadata, + table_metadata=table.metadata, df=pa.table({"category": ["a"], "value": [2]}), - io=replacing_table.io, + io=table.io, write_uuid=uuid.uuid4(), ) ) - replacing_transaction = replacing_table.transaction() - with replacing_transaction.update_snapshot().overwrite() as overwrite: - overwrite.delete_data_file(original_file) + transaction = table.transaction() + with transaction.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(file_to_replace) for replacement_file in replacement_files: overwrite.append_data_file(replacement_file) - return _FileOverwriteScenario( - identifier=identifier, - deleting_table=deleting_table, - replacing_transaction=replacing_transaction, - target_file=original_file, - same_partition_file=same_partition_file, - different_partition_file=different_partition_file, - ) + return transaction + + +def _delete_data_file(table: Table, data_file: DataFile) -> None: + """Commit the deletion of one data file.""" + with table.transaction() as transaction: + with transaction.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(data_file) + + +def _file_overwrite_values(catalog: Catalog) -> list[int]: + """Return the sorted values in the file-overwrite test table.""" + result = catalog.load_table(_FILE_OVERWRITE_TABLE).scan().to_arrow() + return sorted(result["value"].to_pylist()) def test_file_overwrite_fails_when_target_file_is_concurrently_deleted(catalog: Catalog) -> None: """A file replacement must fail if the original file was concurrently deleted.""" - scenario = _prepare_file_overwrite_scenario(catalog) + file_to_replace = _create_file_overwrite_table(catalog) + replacing_table = catalog.load_table(_FILE_OVERWRITE_TABLE) + deleting_table = catalog.load_table(_FILE_OVERWRITE_TABLE) - with scenario.deleting_table.transaction() as deleting_transaction: - with deleting_transaction.update_snapshot().overwrite() as overwrite: - overwrite.delete_data_file(scenario.target_file) + replacing_transaction = _stage_file_replacement(replacing_table, file_to_replace) + _delete_data_file(deleting_table, file_to_replace) with pytest.raises(ValidationException, match="Data files were concurrently deleted"): - scenario.replacing_transaction.commit_transaction() + replacing_transaction.commit_transaction() - result = catalog.load_table(scenario.identifier).scan().to_arrow() - assert sorted(result["value"].to_pylist()) == [1, 3] + assert _file_overwrite_values(catalog) == [1, 3] def test_file_overwrite_allows_concurrent_delete_in_same_partition(catalog: Catalog) -> None: """A file replacement must allow another file in its partition to be concurrently deleted.""" - scenario = _prepare_file_overwrite_scenario(catalog) + file_to_replace = _create_file_overwrite_table(catalog) + replacing_table = catalog.load_table(_FILE_OVERWRITE_TABLE) + deleting_table = catalog.load_table(_FILE_OVERWRITE_TABLE) + file_to_delete = _data_file_in_partition(deleting_table, "a", excluded_file=file_to_replace) - with scenario.deleting_table.transaction() as deleting_transaction: - with deleting_transaction.update_snapshot().overwrite() as overwrite: - overwrite.delete_data_file(scenario.same_partition_file) + replacing_transaction = _stage_file_replacement(replacing_table, file_to_replace) + _delete_data_file(deleting_table, file_to_delete) + replacing_transaction.commit_transaction() - scenario.replacing_transaction.commit_transaction() - - result = catalog.load_table(scenario.identifier).scan().to_arrow() - assert sorted(result["value"].to_pylist()) == [1, 2] + assert _file_overwrite_values(catalog) == [1, 2] def test_file_overwrite_allows_concurrent_delete_in_different_partition(catalog: Catalog) -> None: """A file replacement must allow a file in another partition to be concurrently deleted.""" - scenario = _prepare_file_overwrite_scenario(catalog) - - with scenario.deleting_table.transaction() as deleting_transaction: - with deleting_transaction.update_snapshot().overwrite() as overwrite: - overwrite.delete_data_file(scenario.different_partition_file) + file_to_replace = _create_file_overwrite_table(catalog) + replacing_table = catalog.load_table(_FILE_OVERWRITE_TABLE) + deleting_table = catalog.load_table(_FILE_OVERWRITE_TABLE) + file_to_delete = _data_file_in_partition(deleting_table, "b") - scenario.replacing_transaction.commit_transaction() + replacing_transaction = _stage_file_replacement(replacing_table, file_to_replace) + _delete_data_file(deleting_table, file_to_delete) + replacing_transaction.commit_transaction() - result = catalog.load_table(scenario.identifier).scan().to_arrow() - assert sorted(result["value"].to_pylist()) == [2, 3] + assert _file_overwrite_values(catalog) == [2, 3] def test_concurrent_overwrite_append_retries_successfully(catalog: Catalog) -> None: From 3fb9cfeb53d81e4038c5847d89bb81183d084e4c Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Tue, 18 Aug 2026 22:48:12 -0700 Subject: [PATCH 3/7] Align concurrency tests with existing style --- tests/table/test_commit_retry.py | 141 ++++++++++++++++++------------- 1 file changed, 80 insertions(+), 61 deletions(-) diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index 3489a46ef0..9cde49702c 100644 --- a/tests/table/test_commit_retry.py +++ b/tests/table/test_commit_retry.py @@ -324,38 +324,6 @@ def test_concurrent_overwrite_overwrite_raises_validation_exception(catalog: Cat tbl2.overwrite(pa.table({"x": [40, 50, 60]}), overwrite_filter="x > 0") -_FILE_OVERWRITE_TABLE = "default.concurrent_file_delete" - - -def _create_file_overwrite_table(catalog: Catalog) -> DataFile: - """Create the file-overwrite test table and return the file to replace.""" - import pyarrow as pa - - from pyiceberg.partitioning import PartitionField, PartitionSpec - from pyiceberg.transforms import IdentityTransform - - catalog.create_namespace("default") - schema = Schema( - NestedField(1, "category", StringType(), required=False), - NestedField(2, "value", LongType(), required=False), - ) - spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category")) - table = catalog.create_table(_FILE_OVERWRITE_TABLE, schema=schema, partition_spec=spec) - - table.append(pa.table({"category": ["a", "b"], "value": [0, 1]})) - file_to_replace = next(task.file for task in table.scan().plan_files() if task.file.partition[0] == "a") - - table.append(pa.table({"category": ["a"], "value": [3]})) - return file_to_replace - - -def _data_file_in_partition(table: Table, partition: str, excluded_file: DataFile | None = None) -> DataFile: - """Return a data file in a partition, optionally excluding one file.""" - return next( - task.file for task in table.scan().plan_files() if task.file.partition[0] == partition and task.file != excluded_file - ) - - def _stage_file_replacement(table: Table, file_to_replace: DataFile) -> Transaction: """Stage replacing one data file without committing the transaction.""" import uuid @@ -382,60 +350,111 @@ def _stage_file_replacement(table: Table, file_to_replace: DataFile) -> Transact return transaction -def _delete_data_file(table: Table, data_file: DataFile) -> None: - """Commit the deletion of one data file.""" - with table.transaction() as transaction: - with transaction.update_snapshot().overwrite() as overwrite: - overwrite.delete_data_file(data_file) - +def test_file_overwrite_fails_when_target_file_is_concurrently_deleted(catalog: Catalog) -> None: + """A file replacement must fail if the original file was concurrently deleted.""" + import pyarrow as pa -def _file_overwrite_values(catalog: Catalog) -> list[int]: - """Return the sorted values in the file-overwrite test table.""" - result = catalog.load_table(_FILE_OVERWRITE_TABLE).scan().to_arrow() - return sorted(result["value"].to_pylist()) + from pyiceberg.partitioning import PartitionField, PartitionSpec + from pyiceberg.transforms import IdentityTransform + catalog.create_namespace("default") + schema = Schema( + NestedField(1, "category", StringType(), required=False), + NestedField(2, "value", LongType(), required=False), + ) + spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category")) + identifier = "default.concurrent_target_file_delete" + table = catalog.create_table(identifier, schema=schema, partition_spec=spec) + table.append(pa.table({"category": ["a", "b"], "value": [0, 1]})) + file_to_replace = next(task.file for task in table.scan().plan_files() if task.file.partition[0] == "a") + table.append(pa.table({"category": ["a"], "value": [3]})) -def test_file_overwrite_fails_when_target_file_is_concurrently_deleted(catalog: Catalog) -> None: - """A file replacement must fail if the original file was concurrently deleted.""" - file_to_replace = _create_file_overwrite_table(catalog) - replacing_table = catalog.load_table(_FILE_OVERWRITE_TABLE) - deleting_table = catalog.load_table(_FILE_OVERWRITE_TABLE) + replacing_table = catalog.load_table(identifier) + deleting_table = catalog.load_table(identifier) replacing_transaction = _stage_file_replacement(replacing_table, file_to_replace) - _delete_data_file(deleting_table, file_to_replace) + + with deleting_table.transaction() as deleting_transaction: + with deleting_transaction.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(file_to_replace) with pytest.raises(ValidationException, match="Data files were concurrently deleted"): replacing_transaction.commit_transaction() - assert _file_overwrite_values(catalog) == [1, 3] + result = catalog.load_table(identifier).scan().to_arrow() + assert sorted(result["value"].to_pylist()) == [1, 3] def test_file_overwrite_allows_concurrent_delete_in_same_partition(catalog: Catalog) -> None: """A file replacement must allow another file in its partition to be concurrently deleted.""" - file_to_replace = _create_file_overwrite_table(catalog) - replacing_table = catalog.load_table(_FILE_OVERWRITE_TABLE) - deleting_table = catalog.load_table(_FILE_OVERWRITE_TABLE) - file_to_delete = _data_file_in_partition(deleting_table, "a", excluded_file=file_to_replace) + import pyarrow as pa + + from pyiceberg.partitioning import PartitionField, PartitionSpec + from pyiceberg.transforms import IdentityTransform + + catalog.create_namespace("default") + schema = Schema( + NestedField(1, "category", StringType(), required=False), + NestedField(2, "value", LongType(), required=False), + ) + spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category")) + identifier = "default.concurrent_same_partition_file_delete" + table = catalog.create_table(identifier, schema=schema, partition_spec=spec) + table.append(pa.table({"category": ["a", "b"], "value": [0, 1]})) + file_to_replace = next(task.file for task in table.scan().plan_files() if task.file.partition[0] == "a") + table.append(pa.table({"category": ["a"], "value": [3]})) + + replacing_table = catalog.load_table(identifier) + deleting_table = catalog.load_table(identifier) + file_to_delete = next( + task.file for task in deleting_table.scan().plan_files() if task.file.partition[0] == "a" and task.file != file_to_replace + ) replacing_transaction = _stage_file_replacement(replacing_table, file_to_replace) - _delete_data_file(deleting_table, file_to_delete) + + with deleting_table.transaction() as deleting_transaction: + with deleting_transaction.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(file_to_delete) + replacing_transaction.commit_transaction() - assert _file_overwrite_values(catalog) == [1, 2] + result = catalog.load_table(identifier).scan().to_arrow() + assert sorted(result["value"].to_pylist()) == [1, 2] def test_file_overwrite_allows_concurrent_delete_in_different_partition(catalog: Catalog) -> None: """A file replacement must allow a file in another partition to be concurrently deleted.""" - file_to_replace = _create_file_overwrite_table(catalog) - replacing_table = catalog.load_table(_FILE_OVERWRITE_TABLE) - deleting_table = catalog.load_table(_FILE_OVERWRITE_TABLE) - file_to_delete = _data_file_in_partition(deleting_table, "b") + import pyarrow as pa + + from pyiceberg.partitioning import PartitionField, PartitionSpec + from pyiceberg.transforms import IdentityTransform + + catalog.create_namespace("default") + schema = Schema( + NestedField(1, "category", StringType(), required=False), + NestedField(2, "value", LongType(), required=False), + ) + spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category")) + identifier = "default.concurrent_different_partition_file_delete" + table = catalog.create_table(identifier, schema=schema, partition_spec=spec) + table.append(pa.table({"category": ["a", "b"], "value": [0, 1]})) + file_to_replace = next(task.file for task in table.scan().plan_files() if task.file.partition[0] == "a") + table.append(pa.table({"category": ["a"], "value": [3]})) + + replacing_table = catalog.load_table(identifier) + deleting_table = catalog.load_table(identifier) + file_to_delete = next(task.file for task in deleting_table.scan().plan_files() if task.file.partition[0] == "b") replacing_transaction = _stage_file_replacement(replacing_table, file_to_replace) - _delete_data_file(deleting_table, file_to_delete) + + with deleting_table.transaction() as deleting_transaction: + with deleting_transaction.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(file_to_delete) + replacing_transaction.commit_transaction() - assert _file_overwrite_values(catalog) == [2, 3] + result = catalog.load_table(identifier).scan().to_arrow() + assert sorted(result["value"].to_pylist()) == [2, 3] def test_concurrent_overwrite_append_retries_successfully(catalog: Catalog) -> None: From 5052b88acd155da3eb65e02cc1e6addd45c38b0a Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Tue, 18 Aug 2026 22:54:01 -0700 Subject: [PATCH 4/7] Parameterize concurrent file delete scenarios --- tests/table/test_commit_retry.py | 149 +++++++++---------------------- 1 file changed, 44 insertions(+), 105 deletions(-) diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index 9cde49702c..a2f01f6496 100644 --- a/tests/table/test_commit_retry.py +++ b/tests/table/test_commit_retry.py @@ -21,9 +21,8 @@ from pyiceberg.catalog import Catalog from pyiceberg.exceptions import CommitFailedException, CommitStateUnknownException, ValidationException -from pyiceberg.manifest import DataFile from pyiceberg.schema import Schema -from pyiceberg.table import Table, TableProperties, Transaction +from pyiceberg.table import TableProperties, Transaction from pyiceberg.table.snapshots import IsolationLevel, Operation from pyiceberg.types import LongType, NestedField, StringType @@ -324,36 +323,26 @@ def test_concurrent_overwrite_overwrite_raises_validation_exception(catalog: Cat tbl2.overwrite(pa.table({"x": [40, 50, 60]}), overwrite_filter="x > 0") -def _stage_file_replacement(table: Table, file_to_replace: DataFile) -> Transaction: - """Stage replacing one data file without committing the transaction.""" +@pytest.mark.parametrize( + ("concurrently_deleted_file", "expect_conflict", "expected_values"), + [ + pytest.param("target", True, [1, 3], id="target-file"), + pytest.param("same-partition", False, [1, 2], id="same-partition-file"), + pytest.param("different-partition", False, [2, 3], id="different-partition-file"), + ], +) +def test_file_overwrite_validates_concurrent_file_delete( + catalog: Catalog, + concurrently_deleted_file: str, + expect_conflict: bool, + expected_values: list[int], +) -> None: + """A file replacement must fail only when its target file was concurrently deleted.""" import uuid import pyarrow as pa from pyiceberg.io.pyarrow import _dataframe_to_data_files - - replacement_files = list( - _dataframe_to_data_files( - table_metadata=table.metadata, - df=pa.table({"category": ["a"], "value": [2]}), - io=table.io, - write_uuid=uuid.uuid4(), - ) - ) - - transaction = table.transaction() - with transaction.update_snapshot().overwrite() as overwrite: - overwrite.delete_data_file(file_to_replace) - for replacement_file in replacement_files: - overwrite.append_data_file(replacement_file) - - return transaction - - -def test_file_overwrite_fails_when_target_file_is_concurrently_deleted(catalog: Catalog) -> None: - """A file replacement must fail if the original file was concurrently deleted.""" - import pyarrow as pa - from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.transforms import IdentityTransform @@ -363,7 +352,7 @@ def test_file_overwrite_fails_when_target_file_is_concurrently_deleted(catalog: NestedField(2, "value", LongType(), required=False), ) spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category")) - identifier = "default.concurrent_target_file_delete" + identifier = "default.concurrent_file_delete" table = catalog.create_table(identifier, schema=schema, partition_spec=spec) table.append(pa.table({"category": ["a", "b"], "value": [0, 1]})) file_to_replace = next(task.file for task in table.scan().plan_files() if task.file.partition[0] == "a") @@ -371,90 +360,40 @@ def test_file_overwrite_fails_when_target_file_is_concurrently_deleted(catalog: replacing_table = catalog.load_table(identifier) deleting_table = catalog.load_table(identifier) - - replacing_transaction = _stage_file_replacement(replacing_table, file_to_replace) - - with deleting_table.transaction() as deleting_transaction: - with deleting_transaction.update_snapshot().overwrite() as overwrite: - overwrite.delete_data_file(file_to_replace) - - with pytest.raises(ValidationException, match="Data files were concurrently deleted"): - replacing_transaction.commit_transaction() - - result = catalog.load_table(identifier).scan().to_arrow() - assert sorted(result["value"].to_pylist()) == [1, 3] - - -def test_file_overwrite_allows_concurrent_delete_in_same_partition(catalog: Catalog) -> None: - """A file replacement must allow another file in its partition to be concurrently deleted.""" - import pyarrow as pa - - from pyiceberg.partitioning import PartitionField, PartitionSpec - from pyiceberg.transforms import IdentityTransform - - catalog.create_namespace("default") - schema = Schema( - NestedField(1, "category", StringType(), required=False), - NestedField(2, "value", LongType(), required=False), - ) - spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category")) - identifier = "default.concurrent_same_partition_file_delete" - table = catalog.create_table(identifier, schema=schema, partition_spec=spec) - table.append(pa.table({"category": ["a", "b"], "value": [0, 1]})) - file_to_replace = next(task.file for task in table.scan().plan_files() if task.file.partition[0] == "a") - table.append(pa.table({"category": ["a"], "value": [3]})) - - replacing_table = catalog.load_table(identifier) - deleting_table = catalog.load_table(identifier) - file_to_delete = next( - task.file for task in deleting_table.scan().plan_files() if task.file.partition[0] == "a" and task.file != file_to_replace - ) - - replacing_transaction = _stage_file_replacement(replacing_table, file_to_replace) - - with deleting_table.transaction() as deleting_transaction: - with deleting_transaction.update_snapshot().overwrite() as overwrite: - overwrite.delete_data_file(file_to_delete) - - replacing_transaction.commit_transaction() - - result = catalog.load_table(identifier).scan().to_arrow() - assert sorted(result["value"].to_pylist()) == [1, 2] - - -def test_file_overwrite_allows_concurrent_delete_in_different_partition(catalog: Catalog) -> None: - """A file replacement must allow a file in another partition to be concurrently deleted.""" - import pyarrow as pa - - from pyiceberg.partitioning import PartitionField, PartitionSpec - from pyiceberg.transforms import IdentityTransform - - catalog.create_namespace("default") - schema = Schema( - NestedField(1, "category", StringType(), required=False), - NestedField(2, "value", LongType(), required=False), - ) - spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category")) - identifier = "default.concurrent_different_partition_file_delete" - table = catalog.create_table(identifier, schema=schema, partition_spec=spec) - table.append(pa.table({"category": ["a", "b"], "value": [0, 1]})) - file_to_replace = next(task.file for task in table.scan().plan_files() if task.file.partition[0] == "a") - table.append(pa.table({"category": ["a"], "value": [3]})) - - replacing_table = catalog.load_table(identifier) - deleting_table = catalog.load_table(identifier) - file_to_delete = next(task.file for task in deleting_table.scan().plan_files() if task.file.partition[0] == "b") - - replacing_transaction = _stage_file_replacement(replacing_table, file_to_replace) + data_files = [task.file for task in deleting_table.scan().plan_files()] + file_to_delete = { + "target": file_to_replace, + "same-partition": next( + data_file for data_file in data_files if data_file.partition[0] == "a" and data_file != file_to_replace + ), + "different-partition": next(data_file for data_file in data_files if data_file.partition[0] == "b"), + }[concurrently_deleted_file] + + replacement_file = list( + _dataframe_to_data_files( + table_metadata=replacing_table.metadata, + df=pa.table({"category": ["a"], "value": [2]}), + io=replacing_table.io, + write_uuid=uuid.uuid4(), + ) + )[0] + replacing_transaction = replacing_table.transaction() + with replacing_transaction.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(file_to_replace) + overwrite.append_data_file(replacement_file) with deleting_table.transaction() as deleting_transaction: with deleting_transaction.update_snapshot().overwrite() as overwrite: overwrite.delete_data_file(file_to_delete) - replacing_transaction.commit_transaction() + if expect_conflict: + with pytest.raises(ValidationException, match="Data files were concurrently deleted"): + replacing_transaction.commit_transaction() + else: + replacing_transaction.commit_transaction() result = catalog.load_table(identifier).scan().to_arrow() - assert sorted(result["value"].to_pylist()) == [2, 3] + assert sorted(result["value"].to_pylist()) == expected_values def test_concurrent_overwrite_append_retries_successfully(catalog: Catalog) -> None: From 68ba3ac216b433d87568255ee03f9bab2a1215da Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Tue, 18 Aug 2026 23:16:22 -0700 Subject: [PATCH 5/7] Clarify validation snapshot window --- pyiceberg/table/update/snapshot.py | 7 ++++++- pyiceberg/table/update/validate.py | 10 +++++----- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index e500e09fac..b9d30081ea 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -502,7 +502,12 @@ def _validate_concurrency(self) -> None: _validate_deleted_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot) if self._deleted_data_files: - _validate_data_files_exist(table, catalog_head, self._deleted_data_files, starting_snapshot) + _validate_data_files_exist( + table=table, + to_snapshot=catalog_head, + data_files=self._deleted_data_files, + from_snapshot=starting_snapshot, + ) _validate_no_new_deletes_for_data_files( table, catalog_head, conflict_detection_filter, self._deleted_data_files, starting_snapshot ) diff --git a/pyiceberg/table/update/validate.py b/pyiceberg/table/update/validate.py index 79d17cef9e..aca965ce65 100644 --- a/pyiceberg/table/update/validate.py +++ b/pyiceberg/table/update/validate.py @@ -197,17 +197,17 @@ def _validate_deleted_data_files( def _validate_data_files_exist( table: Table, - starting_snapshot: Snapshot, + to_snapshot: Snapshot, data_files: set[DataFile], - parent_snapshot: Snapshot | None, + from_snapshot: Snapshot | None, ) -> None: """Validate that explicitly replaced data files have not been concurrently deleted. Args: table: Table to validate - starting_snapshot: Snapshot current at the start of the operation + to_snapshot: Current branch head to scan from data_files: Data files that must still exist - parent_snapshot: Ending snapshot on the branch being validated + from_snapshot: Snapshot where the scan stops, excluding this snapshot """ partition_set: dict[int, set[Record]] = {} for data_file in data_files: @@ -215,7 +215,7 @@ def _validate_data_files_exist( conflicting_paths = { entry.data_file.file_path - for entry in _deleted_data_files(table, starting_snapshot, None, partition_set, parent_snapshot) + for entry in _deleted_data_files(table, to_snapshot, None, partition_set, from_snapshot) if entry.data_file in data_files } if conflicting_paths: From 74e74f50e08705ec6f86660708c6011c47e39c81 Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Tue, 18 Aug 2026 23:17:33 -0700 Subject: [PATCH 6/7] Align validation helper signature --- pyiceberg/table/update/snapshot.py | 7 +------ pyiceberg/table/update/validate.py | 10 +++++----- 2 files changed, 6 insertions(+), 11 deletions(-) diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index b9d30081ea..e500e09fac 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -502,12 +502,7 @@ def _validate_concurrency(self) -> None: _validate_deleted_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot) if self._deleted_data_files: - _validate_data_files_exist( - table=table, - to_snapshot=catalog_head, - data_files=self._deleted_data_files, - from_snapshot=starting_snapshot, - ) + _validate_data_files_exist(table, catalog_head, self._deleted_data_files, starting_snapshot) _validate_no_new_deletes_for_data_files( table, catalog_head, conflict_detection_filter, self._deleted_data_files, starting_snapshot ) diff --git a/pyiceberg/table/update/validate.py b/pyiceberg/table/update/validate.py index aca965ce65..cf957f101c 100644 --- a/pyiceberg/table/update/validate.py +++ b/pyiceberg/table/update/validate.py @@ -197,17 +197,17 @@ def _validate_deleted_data_files( def _validate_data_files_exist( table: Table, - to_snapshot: Snapshot, + starting_snapshot: Snapshot, data_files: set[DataFile], - from_snapshot: Snapshot | None, + parent_snapshot: Snapshot | None, ) -> None: """Validate that explicitly replaced data files have not been concurrently deleted. Args: table: Table to validate - to_snapshot: Current branch head to scan from + starting_snapshot: Snapshot at the end of the validation window data_files: Data files that must still exist - from_snapshot: Snapshot where the scan stops, excluding this snapshot + parent_snapshot: Snapshot at the start of the validation window, excluded from the scan """ partition_set: dict[int, set[Record]] = {} for data_file in data_files: @@ -215,7 +215,7 @@ def _validate_data_files_exist( conflicting_paths = { entry.data_file.file_path - for entry in _deleted_data_files(table, to_snapshot, None, partition_set, from_snapshot) + for entry in _deleted_data_files(table, starting_snapshot, None, partition_set, parent_snapshot) if entry.data_file in data_files } if conflicting_paths: From de9514812894f74186583eae6b8c3adb4777dfcf Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Wed, 19 Aug 2026 18:56:06 -0700 Subject: [PATCH 7/7] Address review nits --- pyiceberg/table/update/validate.py | 5 +++-- tests/table/test_commit_retry.py | 3 +-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/pyiceberg/table/update/validate.py b/pyiceberg/table/update/validate.py index cf957f101c..0545182bf0 100644 --- a/pyiceberg/table/update/validate.py +++ b/pyiceberg/table/update/validate.py @@ -14,6 +14,7 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +from collections import defaultdict from collections.abc import Iterator from pyiceberg.exceptions import ValidationException @@ -209,9 +210,9 @@ def _validate_data_files_exist( data_files: Data files that must still exist parent_snapshot: Snapshot at the start of the validation window, excluded from the scan """ - partition_set: dict[int, set[Record]] = {} + partition_set: dict[int, set[Record]] = defaultdict(set) for data_file in data_files: - partition_set.setdefault(data_file.spec_id, set()).add(data_file.partition) + partition_set[data_file.spec_id].add(data_file.partition) conflicting_paths = { entry.data_file.file_path diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index a2f01f6496..ab95457e23 100644 --- a/tests/table/test_commit_retry.py +++ b/tests/table/test_commit_retry.py @@ -14,6 +14,7 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +import uuid from typing import Any from unittest.mock import patch @@ -338,8 +339,6 @@ def test_file_overwrite_validates_concurrent_file_delete( expected_values: list[int], ) -> None: """A file replacement must fail only when its target file was concurrently deleted.""" - import uuid - import pyarrow as pa from pyiceberg.io.pyarrow import _dataframe_to_data_files