diff --git a/mkdocs/docs/configuration.md b/mkdocs/docs/configuration.md index 44b5e395a9..d42309a3b7 100644 --- a/mkdocs/docs/configuration.md +++ b/mkdocs/docs/configuration.md @@ -108,6 +108,26 @@ Iceberg tables support table properties to configure table behavior. +### Commit retry options + +When a concurrent commit is detected, PyIceberg automatically retries the operation with exponential backoff. If the retry detects a real data conflict (e.g. concurrent deletes on the same partition), it raises `ValidationException` instead of retrying. + +| Key | Options | Default | Description | +| -------------------------------- | ---------------- | --------- | ------------------------------------------------------------------ | +| `commit.retry.num-retries` | Integer | 4 | Maximum number of retry attempts after a commit conflict | +| `commit.retry.min-wait-ms` | Integer (ms) | 100 | Minimum wait time before the first retry | +| `commit.retry.max-wait-ms` | Integer (ms) | 60000 | Maximum wait time between retries (caps exponential backoff) | +| `commit.retry.total-timeout-ms` | Integer (ms) | 1800000 | Total time allowed for all retry attempts before giving up | + +### Isolation level options + +These properties control conflict detection behavior during concurrent writes. + +| Key | Options | Default | Description | +| -------------------------------- | -------------------------------- | ------------ | ----------------------------------------------------------------------------------------------- | +| `write.delete.isolation-level` | `{serializable,snapshot}` | serializable | Isolation level for delete operations. Under `serializable`, concurrent appends to affected partitions cause `ValidationException`. Under `snapshot`, only conflicting deletes are rejected. | +| `write.update.isolation-level` | `{serializable,snapshot}` | serializable | Isolation level for overwrite operations. Same semantics as `write.delete.isolation-level`. | + ## FileIO Iceberg works with the concept of a FileIO which is a pluggable module for reading, writing, and deleting files. By default, PyIceberg will try to initialize the FileIO that's suitable for the scheme (`s3://`, `gs://`, etc.) and will use the first one that's installed. diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 63b87d290e..df8b4a9971 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -17,7 +17,10 @@ from __future__ import annotations import itertools +import logging import os +import random +import time import uuid import warnings from abc import ABC, abstractmethod @@ -31,6 +34,7 @@ from pydantic import Field import pyiceberg.expressions.parser as parser +from pyiceberg.exceptions import CommitFailedException, ValidationException from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, And, BooleanExpression, EqualTo, IsNull, Or, Reference from pyiceberg.expressions.visitors import ( ResidualEvaluator, @@ -100,7 +104,7 @@ from pyiceberg.types import strtobool from pyiceberg.utils.concurrent import ExecutorFactory from pyiceberg.utils.config import Config -from pyiceberg.utils.properties import property_as_bool +from pyiceberg.utils.properties import property_as_bool, property_as_int if TYPE_CHECKING: import bodo.pandas as bd @@ -114,6 +118,9 @@ from pyiceberg.catalog import Catalog from pyiceberg.catalog.rest.scan_planning import RESTContentFile, RESTDeleteFile, RESTFileScanTask + from pyiceberg.table.update.snapshot import _SnapshotProducer + +logger = logging.getLogger(__name__) ALWAYS_TRUE = AlwaysTrue() DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE = "downcast-ns-timestamp-to-us-on-write" @@ -212,6 +219,22 @@ class TableProperties: MIN_SNAPSHOTS_TO_KEEP = "history.expire.min-snapshots-to-keep" MIN_SNAPSHOTS_TO_KEEP_DEFAULT = 1 + COMMIT_NUM_RETRIES = "commit.retry.num-retries" + COMMIT_NUM_RETRIES_DEFAULT = 4 + + COMMIT_MIN_RETRY_WAIT_MS = "commit.retry.min-wait-ms" + COMMIT_MIN_RETRY_WAIT_MS_DEFAULT = 100 + + COMMIT_MAX_RETRY_WAIT_MS = "commit.retry.max-wait-ms" + COMMIT_MAX_RETRY_WAIT_MS_DEFAULT = 60000 + + COMMIT_TOTAL_RETRY_TIME_MS = "commit.retry.total-timeout-ms" + COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT = 1800000 # 30 minutes + + WRITE_DELETE_ISOLATION_LEVEL = "write.delete.isolation-level" + WRITE_UPDATE_ISOLATION_LEVEL = "write.update.isolation-level" + WRITE_ISOLATION_LEVEL_DEFAULT = "serializable" + class Transaction: _table: Table @@ -230,6 +253,8 @@ def __init__(self, table: Table, autocommit: bool = False): self._autocommit = autocommit self._updates = () self._requirements = () + self._snapshot_producers: list[_SnapshotProducer[Any]] = [] + self._failed = False @property def table_metadata(self) -> TableMetadata: @@ -272,6 +297,10 @@ def _stage( return self + def _register_snapshot_producer(self, producer: _SnapshotProducer[Any]) -> None: + """Register a snapshot producer for retry support.""" + self._snapshot_producers.append(producer) + def _apply( self, updates: tuple[TableUpdate, ...], @@ -596,7 +625,12 @@ def dynamic_partition_overwrite( delete_filter = self._build_partition_predicate( partition_records=partitions_to_overwrite, spec=self.table_metadata.spec(), schema=self.table_metadata.schema() ) - self.delete(delete_filter=delete_filter, snapshot_properties=snapshot_properties, branch=branch) + self.delete( + delete_filter=delete_filter, + snapshot_properties=snapshot_properties, + branch=branch, + _isolation_operation=Operation.OVERWRITE, + ) with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files: append_files.commit_uuid = append_snapshot_commit_uuid @@ -689,6 +723,7 @@ def overwrite( case_sensitive=case_sensitive, snapshot_properties=snapshot_properties, branch=branch, + _isolation_operation=Operation.OVERWRITE, ) with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files: @@ -706,6 +741,7 @@ def delete( snapshot_properties: dict[str, str] = EMPTY_DICT, case_sensitive: bool = True, branch: str | None = MAIN_BRANCH, + _isolation_operation: Operation | None = None, ) -> None: """ Shorthand for deleting record from a table. @@ -733,6 +769,8 @@ def delete( delete_filter = _parse_row_filter(delete_filter) with self.update_snapshot(snapshot_properties=snapshot_properties, branch=branch).delete() as delete_snapshot: + if _isolation_operation is not None: + delete_snapshot._isolation_operation = _isolation_operation delete_snapshot.delete_by_predicate(delete_filter, case_sensitive) # Check if there are any files that require an actual rewrite of a data file @@ -788,7 +826,11 @@ def delete( with self.update_snapshot( snapshot_properties=snapshot_properties, branch=branch ).overwrite() as overwrite_snapshot: + if _isolation_operation is not None: + overwrite_snapshot._isolation_operation = _isolation_operation + overwrite_snapshot._starting_snapshot_id = delete_snapshot._starting_snapshot_id overwrite_snapshot.commit_uuid = commit_uuid + overwrite_snapshot.delete_by_predicate(delete_filter, case_sensitive) for original_data_file, replaced_data_files in replaced_files: overwrite_snapshot.delete_data_file(original_data_file) for replaced_data_file in replaced_data_files: @@ -1042,18 +1084,146 @@ def commit_transaction(self) -> Table: Returns: The table with the updates applied. """ + if self._failed: + raise RuntimeError("This transaction failed to commit and cannot be reused; create a new transaction.") if len(self._updates) > 0: - self._requirements += (AssertTableUUID(uuid=self.table_metadata.table_uuid),) - self._table._do_commit( # pylint: disable=W0212 - updates=self._updates, - requirements=self._requirements, + properties = self._table.metadata.properties + num_retries: int = max( + 0, + property_as_int( # type: ignore # The default is set with non-None value. + properties, TableProperties.COMMIT_NUM_RETRIES, TableProperties.COMMIT_NUM_RETRIES_DEFAULT + ), ) + # All retry properties are clamped to non-negative values: a negative wait + # would raise ValueError from time.sleep mid-retry and mask the original + # CommitFailedException. + min_wait_ms: int = max( + 0, + property_as_int( # type: ignore # The default is set with non-None value. + properties, TableProperties.COMMIT_MIN_RETRY_WAIT_MS, TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT + ), + ) + max_wait_ms: int = max( + 0, + property_as_int( # type: ignore # The default is set with non-None value. + properties, TableProperties.COMMIT_MAX_RETRY_WAIT_MS, TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT + ), + ) + total_timeout_ms: int = max( + 0, + property_as_int( # type: ignore # The default is set with non-None value. + properties, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT + ), + ) + start_time = time.monotonic() + self._requirements += (AssertTableUUID(uuid=self.table_metadata.table_uuid),) + + try: + try: + for attempt in range(num_retries + 1): + try: + self._table._do_commit( # pylint: disable=W0212 + updates=self._updates, + requirements=self._requirements, + ) + self._cleanup_uncommitted_manifests() + break + except CommitFailedException: + elapsed_ms = (time.monotonic() - start_time) * 1000 + if attempt == num_retries or not self._snapshot_producers or elapsed_ms >= total_timeout_ms: + raise + + wait = min(min_wait_ms * (2**attempt), max_wait_ms) + jitter = random.uniform(0, 0.1 * wait) + logger.warning( + "Commit failed due to a concurrent update, retrying (%s/%s) in %s ms", + attempt + 1, + num_retries, + round(wait + jitter), + ) + time.sleep((wait + jitter) / 1000.0) + + self._table.refresh() + if all( + self._table.metadata.snapshot_by_id(producer._snapshot_id) is not None + for producer in self._snapshot_producers + ): + # A previous attempt actually landed even though it was reported as + # failed (for example a lost response that the transport layer retried). + # The snapshot id is stable across attempts, so finding it in the + # refreshed metadata means the commit is already applied. Stop here + # instead of committing the same data again. + self._cleanup_uncommitted_manifests() + break + self._rebuild_snapshot_updates() + except (CommitFailedException, ValidationException): + # These exceptions guarantee the commit did not land, so it is safe to delete the + # files written for it. Any other exception (unknown outcome, or a commit that already + # succeeded) is re-raised without deleting, since those files may be referenced by the + # catalog's current snapshot and deleting them would corrupt the table for all readers. + for producer in self._snapshot_producers: + producer._clean_all_uncommitted() + raise + except Exception: + # Any failure leaves the transaction in an indeterminate state (files deleted, or the + # commit outcome unknown), so mark it as failed to refuse reuse. + self._failed = True + raise + + self._snapshot_producers = [] + + elif self._snapshot_producers: + # An empty staged output (e.g. a delete whose plan matched nothing) skips the + # commit loop above, so run concurrency validation explicitly before reporting success. + from pyiceberg.table.update.snapshot import CommitWindow + + try: + self._table.refresh() + commit_window = CommitWindow.resolve( + self._table.metadata, + self._snapshot_producers[0]._starting_snapshot_id, + self._snapshot_producers[0]._target_branch, + ) + for producer in self._snapshot_producers: + producer._commit_window = commit_window + producer._validate_concurrency() + except Exception: + for producer in self._snapshot_producers: + producer._clean_all_uncommitted() + self._failed = True + raise + + self._snapshot_producers = [] self._updates = () self._requirements = () return self._table + def _cleanup_uncommitted_manifests(self) -> None: + """Clean up manifests from failed retry attempts after a successful commit.""" + for producer in self._snapshot_producers: + producer._cleanup_uncommitted() + + def _rebuild_snapshot_updates(self) -> None: + """Rebuild snapshot updates for retry by re-executing registered producers.""" + from pyiceberg.table.update import AddSnapshotUpdate, AssertRefSnapshotId, SetSnapshotRefUpdate + from pyiceberg.table.update.snapshot import CommitWindow + + self._updates = tuple(u for u in self._updates if not isinstance(u, (AddSnapshotUpdate, SetSnapshotRefUpdate))) + self._requirements = tuple(r for r in self._requirements if not isinstance(r, AssertRefSnapshotId)) + + starting_id = self._snapshot_producers[0]._starting_snapshot_id if self._snapshot_producers else None + target_branch = self._snapshot_producers[0]._target_branch if self._snapshot_producers else None + commit_window = CommitWindow.resolve(self._table.metadata, starting_id, target_branch) + + for producer in self._snapshot_producers: + producer._commit_window = commit_window + producer._refresh_for_retry() + producer._validate_concurrency() + updates, requirements = producer._commit() + self._stage(updates, requirements) + class CreateTableTransaction(Transaction): """A transaction that involves the creation of a new table.""" diff --git a/pyiceberg/table/metadata.py b/pyiceberg/table/metadata.py index 26b6e3d3ad..84be7b07bf 100644 --- a/pyiceberg/table/metadata.py +++ b/pyiceberg/table/metadata.py @@ -30,7 +30,7 @@ from pyiceberg.schema import Schema, assign_fresh_schema_ids from pyiceberg.table.name_mapping import NameMapping, parse_mapping_from_json from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRef, SnapshotRefType -from pyiceberg.table.snapshots import MetadataLogEntry, Snapshot, SnapshotLogEntry +from pyiceberg.table.snapshots import IsolationLevel, MetadataLogEntry, Operation, Snapshot, SnapshotLogEntry from pyiceberg.table.sorting import ( UNSORTED_SORT_ORDER, UNSORTED_SORT_ORDER_ID, @@ -309,6 +309,16 @@ def snapshot_by_name(self, name: str | None) -> Snapshot | None: return self.snapshot_by_id(ref.snapshot_id) return None + def isolation_level(self, operation: Operation) -> IsolationLevel: + """Resolve the isolation level for the given operation from the table properties.""" + from pyiceberg.table import TableProperties + + if operation == Operation.OVERWRITE: + property_name = TableProperties.WRITE_UPDATE_ISOLATION_LEVEL + else: + property_name = TableProperties.WRITE_DELETE_ISOLATION_LEVEL + return IsolationLevel(self.properties.get(property_name, TableProperties.WRITE_ISOLATION_LEVEL_DEFAULT)) + def current_snapshot(self) -> Snapshot | None: """Get the current snapshot for this table, or None if there is no current snapshot.""" if self.current_snapshot_id is not None: diff --git a/pyiceberg/table/snapshots.py b/pyiceberg/table/snapshots.py index 68dfcc868d..5e9e519a01 100644 --- a/pyiceberg/table/snapshots.py +++ b/pyiceberg/table/snapshots.py @@ -86,6 +86,13 @@ def __repr__(self) -> str: return f"Operation.{self.name}" +class IsolationLevel(str, Enum): + """Transaction isolation level for concurrent write validation.""" + + SERIALIZABLE = "serializable" + SNAPSHOT = "snapshot" + + class UpdateMetrics: added_file_size: int removed_file_size: int diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 7931edacdd..3c58f8ff44 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -17,15 +17,18 @@ from __future__ import annotations import itertools +import logging import uuid from abc import abstractmethod from collections import defaultdict from collections.abc import Callable +from dataclasses import dataclass from datetime import datetime from functools import cached_property from typing import TYPE_CHECKING, Generic from pyiceberg.avro.codecs import AvroCompressionCodec +from pyiceberg.exceptions import ValidationException from pyiceberg.expressions import AlwaysFalse, BooleanExpression, Or from pyiceberg.expressions.visitors import ( ROWS_MIGHT_NOT_MATCH, @@ -79,6 +82,9 @@ if TYPE_CHECKING: from pyiceberg.table import Transaction + from pyiceberg.table.metadata import TableMetadata + +logger = logging.getLogger(__name__) def _new_manifest_file_name(num: int, commit_uuid: uuid.UUID) -> str: @@ -91,12 +97,43 @@ def _new_manifest_list_file_name(snapshot_id: int, attempt: int, commit_uuid: uu return f"snap-{snapshot_id}-{attempt}-{commit_uuid}.avro" +@dataclass(frozen=True) +class CommitWindow: + """Tracks the commit range to validate against during retry. + + base: The snapshot when the operation began (exclusive lower bound, fixed across retries). + head: The branch HEAD snapshot after metadata refresh (inclusive upper bound). + """ + + base: Snapshot | None + head: Snapshot | None + + @classmethod + def resolve(cls, metadata: TableMetadata, base_id: int | None, branch: str | None) -> CommitWindow: + """Resolve a CommitWindow from metadata, starting snapshot ID, and target branch.""" + head = metadata.snapshot_by_name(branch) + base = metadata.snapshot_by_id(base_id) if base_id is not None else None + if base_id is not None and base is None: + raise ValidationException(f"Cannot find starting snapshot {base_id}") + return cls(base=base, head=head) + + def is_empty(self) -> bool: + """Return True if no concurrent commits occurred (validation can be skipped). + + A None base means the operation started on a table with no snapshots. If a head + exists, another writer landed the first snapshot concurrently, so the window is not + empty and validation must run against the whole history. + """ + return self.head is None or (self.base is not None and self.base.snapshot_id == self.head.snapshot_id) + + class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]): commit_uuid: uuid.UUID _io: FileIO _operation: Operation _snapshot_id: int _parent_snapshot_id: int | None + _starting_snapshot_id: int | None _added_data_files: list[DataFile] _manifest_num_counter: itertools.count[int] _deleted_data_files: set[DataFile] @@ -104,6 +141,11 @@ class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]): _target_branch: str | None _predicate: BooleanExpression _case_sensitive: bool + _commit_window: CommitWindow | None + _written_manifests: list[str] + _uncommitted_manifests: list[str] + _written_manifest_lists: list[str] + _isolation_operation: Operation def __init__( self, @@ -123,17 +165,21 @@ def __init__( self._deleted_data_files = set() self.snapshot_properties = snapshot_properties self._manifest_num_counter = itertools.count(0) + self._written_manifests = [] + self._uncommitted_manifests = [] + self._written_manifest_lists = [] from pyiceberg.table import TableProperties self._compression = self._transaction.table_metadata.properties.get( # type: ignore TableProperties.WRITE_AVRO_COMPRESSION, TableProperties.WRITE_AVRO_COMPRESSION_DEFAULT ) self._target_branch = self._validate_target_branch(branch=branch) - self._parent_snapshot_id = ( - snapshot.snapshot_id if (snapshot := self._transaction.table_metadata.snapshot_by_name(self._target_branch)) else None - ) + self._parent_snapshot_id = self._current_branch_head_id() + self._starting_snapshot_id = self._parent_snapshot_id self._predicate = AlwaysFalse() self._case_sensitive = True + self._commit_window = None + self._isolation_operation: Operation = Operation.DELETE def _validate_target_branch(self, branch: str | None) -> str | None: # if branch is none, write will be written into a staging snapshot @@ -144,6 +190,11 @@ def _validate_target_branch(self, branch: str | None) -> str | None: raise ValueError(f"{branch} is a tag, not a branch. Tags cannot be targets for producing snapshots") return branch + def _current_branch_head_id(self) -> int | None: + """Return the snapshot id at the head of the target branch, or None if the branch has no snapshot.""" + snapshot = self._transaction.table_metadata.snapshot_by_name(self._target_branch) + return snapshot.snapshot_id if snapshot else None + def append_data_file(self, data_file: DataFile) -> _SnapshotProducer[U]: self._added_data_files.append(data_file) return self @@ -275,6 +326,7 @@ def _commit(self) -> UpdatesAndRequirements: ) location_provider = self._transaction._table.location_provider() manifest_list_file_path = location_provider.new_metadata_location(file_name) + self._written_manifest_lists.append(manifest_list_file_path) with write_manifest_list( format_version=self._transaction.table_metadata.format_version, @@ -353,11 +405,106 @@ def new_manifest_output(self) -> OutputFile: location_provider = self._transaction._table.location_provider() file_name = _new_manifest_file_name(num=next(self._manifest_num_counter), commit_uuid=self.commit_uuid) file_path = location_provider.new_metadata_location(file_name) + self._written_manifests.append(file_path) return self._io.new_output(file_path) def fetch_manifest_entry(self, manifest: ManifestFile, discard_deleted: bool = True) -> list[ManifestEntry]: return manifest.fetch_manifest_entry(io=self._io, discard_deleted=discard_deleted) + def commit(self) -> None: + self._transaction._register_snapshot_producer(self) + super().commit() + + def _cleanup_uncommitted(self) -> None: + """Delete manifest files and manifest lists from failed retry attempts.""" + for path in self._uncommitted_manifests: + try: + self._io.delete(path) + except Exception: + logger.warning("Failed to delete uncommitted manifest: %s", path, exc_info=True) + self._uncommitted_manifests.clear() + # Delete all manifest lists except the last one (which the committed snapshot references) + if len(self._written_manifest_lists) > 1: + for path in self._written_manifest_lists[:-1]: + try: + self._io.delete(path) + except Exception: + logger.warning("Failed to delete uncommitted manifest list: %s", path, exc_info=True) + self._written_manifest_lists = self._written_manifest_lists[-1:] + + def _clean_all_uncommitted(self) -> None: + """Clean up all manifests and manifest lists 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) + for path in self._written_manifest_lists: + try: + self._io.delete(path) + except Exception: + logger.warning("Failed to delete uncommitted manifest list: %s", path, exc_info=True) + self._uncommitted_manifests.clear() + self._written_manifests.clear() + self._written_manifest_lists.clear() + + def _refresh_for_retry(self) -> None: + """Reset state for a retry attempt with refreshed metadata.""" + self._uncommitted_manifests.extend(self._written_manifests) + self._written_manifests.clear() + self._parent_snapshot_id = self._current_branch_head_id() + # The snapshot id is kept stable across attempts so it acts as an idempotency key: if a + # previous attempt actually landed but its response was lost, the id can be found in the + # refreshed metadata and the commit is not repeated. + self._manifest_num_counter = itertools.count(0) + self.commit_uuid = uuid.uuid4() + + def _validate_concurrency(self) -> None: + """Validate that concurrent changes do not conflict with this operation. + + Uses the CommitWindow to determine which catalog commits to validate against. + The window spans from base (when the operation began) to head (latest branch + HEAD), covering all external concurrent commits. + + Subclasses that do not require validation (e.g. fast append) should override + with a no-op. + """ + from pyiceberg.table.snapshots import IsolationLevel + from pyiceberg.table.update.validate import ( + _validate_added_data_files, + _validate_deleted_data_files, + _validate_no_new_delete_files, + _validate_no_new_deletes_for_data_files, + ) + + if self._commit_window is None or self._commit_window.is_empty(): + return + + catalog_head = self._commit_window.head + starting_snapshot = self._commit_window.base + + if catalog_head is None: + return + + table = self._transaction._table + isolation_level = table.metadata.isolation_level(self._isolation_operation) + conflict_detection_filter = self._predicate if self._predicate != AlwaysFalse() else None + + if isolation_level == IsolationLevel.SERIALIZABLE: + _validate_added_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot) + + if self._predicate != AlwaysFalse(): + # partition_set is None, so any concurrently committed delete file counts as + # a conflict even in unrelated partitions. Scope this via partition projection + # (as Java does) when delete-file write support lands. + _validate_no_new_delete_files(table, catalog_head, conflict_detection_filter, None, starting_snapshot) + _validate_deleted_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot) + + if self._deleted_data_files: + _validate_no_new_deletes_for_data_files( + table, catalog_head, conflict_detection_filter, self._deleted_data_files, starting_snapshot + ) + def _build_partition_projection(self, spec_id: int) -> BooleanExpression: project = inclusive_projection(self.schema(), self.spec(spec_id), self._case_sensitive) return project(self._predicate) @@ -384,7 +531,8 @@ def _build_delete_files_partition_predicate(self) -> None: self.delete_by_predicate( self._transaction._build_partition_predicate( partition_records=partition_records, schema=self.schema(), spec=self.spec(spec_id) - ) + ), + self._case_sensitive, ) @@ -398,6 +546,10 @@ class _DeleteFiles(_SnapshotProducer["_DeleteFiles"]): From the specification """ + # The set of data files this delete was planned to drop, frozen at first plan. + # Reused across retries so the delete is not replanned against a newer head. + _planned_delete_files: set[DataFile] | None = None + def _commit(self) -> UpdatesAndRequirements: # Only produce a commit when there is something to delete if self.files_affected: @@ -457,14 +609,27 @@ def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) -> deleted_entries = [] existing_entries = [] for entry in manifest_file.fetch_manifest_entry(io=self._io, discard_deleted=True): - if strict_metrics_evaluator(entry.data_file) == ROWS_MUST_MATCH: + # On the first plan the predicate decides which files are dropped. On a + # retry the planned set is frozen, so a file is dropped only if it was + # planned for deletion originally. This keeps concurrently added files + # (which the writer never saw) from being dropped when the delete is + # replanned against a newer head under snapshot isolation. + if self._planned_delete_files is not None: + should_delete = entry.data_file in self._planned_delete_files + else: + should_delete = strict_metrics_evaluator(entry.data_file) == ROWS_MUST_MATCH + + if should_delete: # Based on the metadata, it can be dropped right away deleted_entries.append(_copy_with_new_status(entry, ManifestEntryStatus.DELETED)) self._deleted_data_files.add(entry.data_file) else: # Based on the metadata, we cannot determine if it can be deleted existing_entries.append(_copy_with_new_status(entry, ManifestEntryStatus.EXISTING)) - if inclusive_metrics_evaluator(entry.data_file) != ROWS_MIGHT_NOT_MATCH: + if ( + self._planned_delete_files is None + and inclusive_metrics_evaluator(entry.data_file) != ROWS_MIGHT_NOT_MATCH + ): partial_rewrites_needed = True if len(deleted_entries) > 0: @@ -481,6 +646,10 @@ def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) -> else: existing_manifests.append(manifest_file) + if self._planned_delete_files is None: + # Freeze the set of files this delete operates on so retries reuse it instead of replanning. + self._planned_delete_files = set(self._deleted_data_files) + return existing_manifests, total_deleted_entries, partial_rewrites_needed def _existing_manifests(self) -> list[ManifestFile]: @@ -499,8 +668,19 @@ def files_affected(self) -> bool: """Indicate if any manifest-entries can be dropped.""" return len(self._deleted_entries()) > 0 + def _refresh_for_retry(self) -> None: + """Reset state for a retry attempt, clearing the cached delete computation.""" + super()._refresh_for_retry() + # Clear @cached_property by removing it from the instance __dict__. + # _compute_deletes depends on _parent_snapshot_id which changes on retry. + if "_compute_deletes" in self.__dict__: + del self.__dict__["_compute_deletes"] + class _FastAppendFiles(_SnapshotProducer["_FastAppendFiles"]): + def _validate_concurrency(self) -> None: + """Skip validation; appends do not conflict with other operations.""" + def _existing_manifests(self) -> list[ManifestFile]: """To determine if there are any existing manifest files. diff --git a/pyiceberg/table/update/validate.py b/pyiceberg/table/update/validate.py index 8178ed6ee0..df8506aab4 100644 --- a/pyiceberg/table/update/validate.py +++ b/pyiceberg/table/update/validate.py @@ -40,17 +40,21 @@ def _validation_history( table: Table, - from_snapshot: Snapshot, + from_snapshot: Snapshot | None, to_snapshot: Snapshot, matching_operations: set[Operation], manifest_content_filter: ManifestContent, ) -> tuple[list[ManifestFile], set[int]]: """Return newly added manifests and snapshot IDs between the starting snapshot and parent snapshot. + Walks from to_snapshot backwards towards from_snapshot, collecting manifests from + snapshots whose operations match. from_snapshot is excluded from results. A None + from_snapshot walks the entire history of to_snapshot down to the root. + Args: table: Table to get the history from - from_snapshot: Parent snapshot to get the history from - to_snapshot: Starting snapshot + from_snapshot: Snapshot where the walk stops (exclusive), or None to walk the whole history + to_snapshot: Snapshot where the walk starts matching_operations: Operations to match on manifest_content_filter: Manifest content type to filter @@ -60,11 +64,17 @@ def _validation_history( Returns: List of manifest files and set of snapshots ID's matching conditions """ + if from_snapshot is not None and from_snapshot.snapshot_id == to_snapshot.snapshot_id: + return [], set() + manifests_files: list[ManifestFile] = [] snapshots: set[int] = set() last_snapshot = None for snapshot in ancestors_between(from_snapshot, to_snapshot, table.metadata): + if from_snapshot is not None and snapshot.snapshot_id == from_snapshot.snapshot_id: + last_snapshot = snapshot + break last_snapshot = snapshot summary = snapshot.summary if summary is None: @@ -73,7 +83,6 @@ def _validation_history( continue snapshots.add(snapshot.snapshot_id) - # TODO: Maybe do the IO in a separate thread at some point, and collect at the bottom (we can easily merge the sets manifests_files.extend( [ manifest @@ -82,7 +91,7 @@ def _validation_history( ] ) - if last_snapshot is not None and last_snapshot.snapshot_id != from_snapshot.snapshot_id: + if from_snapshot is not None and (last_snapshot is None or last_snapshot.snapshot_id != from_snapshot.snapshot_id): raise ValidationException("No matching snapshot found.") return manifests_files, snapshots @@ -149,9 +158,6 @@ def _deleted_data_files( List of conflicting manifest-entries """ # if there is no current table state, no files have been deleted - if parent_snapshot is None: - return - manifests, snapshot_ids = _validation_history( table, parent_snapshot, @@ -172,7 +178,7 @@ def _validate_deleted_data_files( table: Table, starting_snapshot: Snapshot, data_filter: BooleanExpression | None, - parent_snapshot: Snapshot, + parent_snapshot: Snapshot | None, ) -> None: """Validate that no files matching a filter have been deleted from the table since a starting snapshot. @@ -208,9 +214,6 @@ def _added_data_files( Returns: Iterator of manifest entries for added data files matching the conditions """ - if parent_snapshot is None: - return - manifests, snapshot_ids = _validation_history( table, parent_snapshot, @@ -244,7 +247,7 @@ def _added_delete_files( Returns: DeleteFileIndex """ - if parent_snapshot is None or table.format_version < 2: + if table.format_version < 2: return DeleteFileIndex() manifests, snapshot_ids = _validation_history( @@ -344,7 +347,7 @@ def _validate_no_new_deletes_for_data_files( parent_snapshot: Ending snapshot on the branch being validated """ # If there is no current state, or no files has been added - if parent_snapshot is None or table.format_version < 2: + if table.format_version < 2: return deletes = _added_delete_files(table, starting_snapshot, data_filter, None, parent_snapshot) diff --git a/tests/integration/test_writes/test_optimistic_concurrency.py b/tests/integration/test_writes/test_optimistic_concurrency.py index 6ddf4c11d5..299c58c833 100644 --- a/tests/integration/test_writes/test_optimistic_concurrency.py +++ b/tests/integration/test_writes/test_optimistic_concurrency.py @@ -20,7 +20,7 @@ from pyspark.sql import SparkSession from pyiceberg.catalog import Catalog -from pyiceberg.exceptions import CommitFailedException +from pyiceberg.exceptions import ValidationException from utils import _create_table @@ -29,15 +29,14 @@ def test_conflict_delete_delete( spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int ) -> None: - """This test should start passing once optimistic concurrency control has been implemented.""" + """Concurrent deletes on the same data should fail with ValidationException.""" identifier = "default.test_conflict" tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version}, [arrow_table_with_null]) tbl2 = session_catalog.load_table(identifier) tbl1.delete("string == 'z'") - with pytest.raises(CommitFailedException, match="(branch main has changed: expected id ).*"): - # tbl2 isn't aware of the commit by tbl1 + with pytest.raises(ValidationException): tbl2.delete("string == 'z'") @@ -46,17 +45,13 @@ def test_conflict_delete_delete( def test_conflict_delete_append( spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int ) -> None: - """This test should start passing once optimistic concurrency control has been implemented.""" + """Append after a concurrent delete should succeed via retry.""" identifier = "default.test_conflict" tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version}, [arrow_table_with_null]) tbl2 = session_catalog.load_table(identifier) - # This is allowed tbl1.delete("string == 'z'") - - with pytest.raises(CommitFailedException, match="(branch main has changed: expected id ).*"): - # tbl2 isn't aware of the commit by tbl1 - tbl2.append(arrow_table_with_null) + tbl2.append(arrow_table_with_null) @pytest.mark.integration @@ -64,15 +59,14 @@ def test_conflict_delete_append( def test_conflict_append_delete( spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int ) -> None: - """This test should start passing once optimistic concurrency control has been implemented.""" + """Delete after a concurrent append fails with ValidationException under serializable isolation.""" identifier = "default.test_conflict" tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version}, [arrow_table_with_null]) tbl2 = session_catalog.load_table(identifier) tbl1.append(arrow_table_with_null) - with pytest.raises(CommitFailedException, match="(branch main has changed: expected id ).*"): - # tbl2 isn't aware of the commit by tbl1 + with pytest.raises(ValidationException): tbl2.delete("string == 'z'") @@ -81,13 +75,10 @@ def test_conflict_append_delete( def test_conflict_append_append( spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int ) -> None: - """This test should start passing once optimistic concurrency control has been implemented.""" + """Concurrent appends should both succeed via retry.""" identifier = "default.test_conflict" tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version}, [arrow_table_with_null]) tbl2 = session_catalog.load_table(identifier) tbl1.append(arrow_table_with_null) - - with pytest.raises(CommitFailedException, match="(branch main has changed: expected id ).*"): - # tbl2 isn't aware of the commit by tbl1 - tbl2.append(arrow_table_with_null) + tbl2.append(arrow_table_with_null) diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py new file mode 100644 index 0000000000..ce5dca96aa --- /dev/null +++ b/tests/table/test_commit_retry.py @@ -0,0 +1,1220 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from typing import Any +from unittest.mock import patch + +import pytest + +from pyiceberg.catalog import Catalog +from pyiceberg.exceptions import CommitFailedException, CommitStateUnknownException, ValidationException +from pyiceberg.schema import Schema +from pyiceberg.table import TableProperties, Transaction +from pyiceberg.table.snapshots import IsolationLevel, Operation +from pyiceberg.types import LongType, NestedField, StringType + + +def test_isolation_level_enum() -> None: + assert IsolationLevel.SERIALIZABLE.value == "serializable" + assert IsolationLevel.SNAPSHOT.value == "snapshot" + assert IsolationLevel("serializable") is IsolationLevel.SERIALIZABLE + assert IsolationLevel("snapshot") is IsolationLevel.SNAPSHOT + + +def test_commit_retry_table_properties() -> None: + assert TableProperties.COMMIT_NUM_RETRIES == "commit.retry.num-retries" + assert TableProperties.COMMIT_NUM_RETRIES_DEFAULT == 4 + assert TableProperties.COMMIT_MIN_RETRY_WAIT_MS == "commit.retry.min-wait-ms" + assert TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT == 100 + assert TableProperties.COMMIT_MAX_RETRY_WAIT_MS == "commit.retry.max-wait-ms" + assert TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT == 60000 + assert TableProperties.COMMIT_TOTAL_RETRY_TIME_MS == "commit.retry.total-timeout-ms" + assert TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT == 1800000 + + +def test_isolation_level_table_properties() -> None: + assert TableProperties.WRITE_DELETE_ISOLATION_LEVEL == "write.delete.isolation-level" + assert TableProperties.WRITE_UPDATE_ISOLATION_LEVEL == "write.update.isolation-level" + assert TableProperties.WRITE_ISOLATION_LEVEL_DEFAULT == "serializable" + + +def _test_schema() -> Schema: + return Schema(NestedField(1, "x", LongType(), required=False)) + + +def test_commit_retry_on_commit_failed(catalog: Catalog) -> None: + """Verify that CommitFailedException triggers retry for append operations.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.retry_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + # Load two references to the same table to simulate concurrent access + tbl1 = catalog.load_table("default.retry_test") + tbl2 = catalog.load_table("default.retry_test") + + # First append succeeds + tbl1.append(df) + + # Second append should succeed via retry (append vs append never conflicts) + import pyiceberg.table as _table_module + + RuntimeTransaction = _table_module.Transaction + original_rebuild = RuntimeTransaction._rebuild_snapshot_updates + rebuild_count = 0 + + def counting_rebuild(self_tx: Any) -> None: + nonlocal rebuild_count + rebuild_count += 1 + original_rebuild(self_tx) + + with patch.object(RuntimeTransaction, "_rebuild_snapshot_updates", counting_rebuild): + tbl2.append(df) + + assert rebuild_count == 1, "Expected exactly one retry via _rebuild_snapshot_updates" + + # Both appends should be visible + refreshed = catalog.load_table("default.retry_test") + result = refreshed.scan().to_arrow() + assert len(result) == 6 + + +def test_no_retry_without_snapshot_producers(catalog: Catalog) -> None: + """Verify that a transaction with no snapshot producers has an empty producer list.""" + catalog.create_namespace("default") + schema = _test_schema() + table = catalog.create_table("default.no_retry_test", schema=schema) + + tx = Transaction(table, autocommit=False) + tx.set_properties({"key": "value"}) + + # No snapshot producers registered + assert len(tx._snapshot_producers) == 0 + + +def test_rebuild_snapshot_updates_preserves_non_snapshot_updates(catalog: Catalog) -> None: + """Verify that non-snapshot updates survive retry.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.rebuild_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1]}) + + tbl1 = catalog.load_table("default.rebuild_test") + tbl2 = catalog.load_table("default.rebuild_test") + + # tbl1 commits first + tbl1.append(df) + + # tbl2 does both property change and append in one transaction + with tbl2.transaction() as tx: + tx.set_properties({"test_key": "test_value"}) + tx.append(df) + + # Both the property and the data should be committed + refreshed = catalog.load_table("default.rebuild_test") + assert refreshed.metadata.properties.get("test_key") == "test_value" + assert len(refreshed.scan().to_arrow()) == 2 + + +def test_refresh_for_retry_resets_producer_state(catalog: Catalog) -> None: + """Verify that _refresh_for_retry resets the necessary fields.""" + catalog.create_namespace("default") + schema = _test_schema() + table = catalog.create_table("default.refresh_test", schema=schema) + + from pyiceberg.table.update.snapshot import _FastAppendFiles + + tx = Transaction(table, autocommit=False) + producer = _FastAppendFiles( + operation=Operation.APPEND, + transaction=tx, + io=table.io, + ) + + original_snapshot_id = producer._snapshot_id + original_uuid = producer.commit_uuid + + producer._refresh_for_retry() + + # The snapshot id is kept stable across retries so it acts as an idempotency key. + assert producer._snapshot_id == original_snapshot_id + assert producer.commit_uuid != original_uuid + # parent stays None for empty table + assert producer._parent_snapshot_id is None + + +def test_concurrent_delete_delete_raises_validation_exception(catalog: Catalog) -> None: + """Concurrent deletes on the same data should fail with ValidationException.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.del_del_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl = catalog.load_table("default.del_del_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.del_del_test") + tbl2 = catalog.load_table("default.del_del_test") + + tbl1.delete("x == 1") + + with pytest.raises(ValidationException): + tbl2.delete("x == 1") + + +def test_concurrent_append_delete_raises_validation_exception(catalog: Catalog) -> None: + """Delete after a concurrent append fails with ValidationException under serializable isolation.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.app_del_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl = catalog.load_table("default.app_del_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.app_del_test") + tbl2 = catalog.load_table("default.app_del_test") + + tbl1.append(df) + + with pytest.raises(ValidationException): + tbl2.delete("x == 1") + + +def test_concurrent_delete_append_retries_successfully(catalog: Catalog) -> None: + """Append after a concurrent delete should succeed via retry.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.del_app_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl = catalog.load_table("default.del_app_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.del_app_test") + tbl2 = catalog.load_table("default.del_app_test") + + tbl1.delete("x == 1") + + import pyiceberg.table as _table_module + + RuntimeTransaction = _table_module.Transaction + original_rebuild = RuntimeTransaction._rebuild_snapshot_updates + rebuild_count = 0 + + def counting_rebuild(self_tx: Any) -> None: + nonlocal rebuild_count + rebuild_count += 1 + original_rebuild(self_tx) + + with patch.object(RuntimeTransaction, "_rebuild_snapshot_updates", counting_rebuild): + tbl2.append(df) + + assert rebuild_count == 1 + + refreshed = catalog.load_table("default.del_app_test") + result = refreshed.scan().to_arrow() + # Original 3 rows, minus 1 deleted, plus 3 appended = 5 + assert len(result) == 5 + + +def test_retry_exhaustion_raises_commit_failed(catalog: Catalog) -> None: + """When retries are exhausted, CommitFailedException should be raised.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table( + "default.exhaust_test", + schema=schema, + properties={"commit.retry.num-retries": "0"}, + ) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl1 = catalog.load_table("default.exhaust_test") + tbl2 = catalog.load_table("default.exhaust_test") + + tbl1.append(df) + + with pytest.raises(CommitFailedException): + tbl2.append(df) + + +def test_delete_files_refresh_clears_compute_deletes_cache(catalog: Catalog) -> None: + """Verify that _refresh_for_retry clears the _compute_deletes cached property.""" + catalog.create_namespace("default") + schema = _test_schema() + table = catalog.create_table("default.cache_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + table.append(df) + table = catalog.load_table("default.cache_test") + + from pyiceberg.expressions import EqualTo + from pyiceberg.table.update.snapshot import _DeleteFiles + + tx = Transaction(table, autocommit=False) + producer = _DeleteFiles( + operation=Operation.DELETE, + transaction=tx, + io=table.io, + ) + producer.delete_by_predicate(EqualTo("x", 1)) + + # Access _compute_deletes to populate the cache + _ = producer._compute_deletes + + assert "_compute_deletes" in producer.__dict__ + + producer._refresh_for_retry() + + assert "_compute_deletes" not in producer.__dict__ + + +def test_concurrent_overwrite_overwrite_raises_validation_exception(catalog: Catalog) -> None: + """Concurrent overwrites on the same data should fail with ValidationException.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.ow_ow_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl = catalog.load_table("default.ow_ow_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.ow_ow_test") + tbl2 = catalog.load_table("default.ow_ow_test") + + tbl1.overwrite(pa.table({"x": [10, 20, 30]}), overwrite_filter="x > 0") + with pytest.raises(ValidationException): + tbl2.overwrite(pa.table({"x": [40, 50, 60]}), overwrite_filter="x > 0") + + +def test_concurrent_overwrite_append_retries_successfully(catalog: Catalog) -> None: + """Append after a concurrent overwrite should succeed via retry.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.ow_app_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl = catalog.load_table("default.ow_app_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.ow_app_test") + tbl2 = catalog.load_table("default.ow_app_test") + + tbl1.overwrite(pa.table({"x": [10, 20, 30]}), overwrite_filter="x > 0") + tbl2.append(pa.table({"x": [4, 5, 6]})) + + refreshed = catalog.load_table("default.ow_app_test") + result = refreshed.scan().to_arrow() + # overwrite replaced 3 rows with 3 new rows, then append added 3 more = 6 + assert len(result) == 6 + + +def test_snapshot_isolation_allows_concurrent_append_delete(catalog: Catalog) -> None: + """Under snapshot isolation, delete after a concurrent append should succeed via retry.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table( + "default.snapshot_iso_test", + schema=schema, + properties={"write.delete.isolation-level": "snapshot"}, + ) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl = catalog.load_table("default.snapshot_iso_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.snapshot_iso_test") + tbl2 = catalog.load_table("default.snapshot_iso_test") + + tbl1.append(df) + + # Under serializable this would raise ValidationException, + # but under snapshot isolation _validate_added_data_files is skipped + tbl2.delete("x == 1") + + refreshed = catalog.load_table("default.snapshot_iso_test") + result = refreshed.scan().to_arrow() + # Original 3, delete removes x==1 from original (1 row), append adds 3 = 5 + assert len(result) == 5 + + +def test_uncommitted_manifests_tracked_correctly(catalog: Catalog) -> None: + """Verify that uncommitted manifests are moved to _uncommitted_manifests on retry.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.manifest_track_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl = catalog.load_table("default.manifest_track_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.manifest_track_test") + tbl2 = catalog.load_table("default.manifest_track_test") + + tbl1.append(df) + + import pyiceberg.table as _table_module2 + + RuntimeTransaction2 = _table_module2.Transaction + original_rebuild = RuntimeTransaction2._rebuild_snapshot_updates + uncommitted_count_during_rebuild = 0 + + def checking_rebuild(self_tx: Any) -> None: + nonlocal uncommitted_count_during_rebuild + original_rebuild(self_tx) + for producer in self_tx._snapshot_producers: + uncommitted_count_during_rebuild += len(producer._uncommitted_manifests) + + with patch.object(RuntimeTransaction2, "_rebuild_snapshot_updates", checking_rebuild): + tbl2.append(df) + + # After rebuild, the first attempt's manifests should be in _uncommitted_manifests + assert uncommitted_count_during_rebuild > 0 + + +def test_concurrent_deletes_on_different_partitions_succeed(catalog: Catalog) -> None: + """Concurrent deletes on different partitions should succeed via retry thanks to conflict detection filter.""" + 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")) + catalog.create_table("default.part_del_test", schema=schema, partition_spec=spec) + + import pyarrow as pa + + df = pa.table( + { + "category": ["a", "a", "b", "b"], + "value": [1, 2, 3, 4], + } + ) + + tbl = catalog.load_table("default.part_del_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.part_del_test") + tbl2 = catalog.load_table("default.part_del_test") + + # Delete from different partitions should not conflict + tbl1.delete("category == 'a'") + tbl2.delete("category == 'b'") + + refreshed = catalog.load_table("default.part_del_test") + result = refreshed.scan().to_arrow() + assert len(result) == 0 + + +def test_concurrent_partial_deletes_on_different_partitions_succeed(catalog: Catalog) -> None: + """Concurrent partial deletes (CoW rewrite) on different partitions should succeed. + + This tests the auto-computed partition predicate from _build_delete_files_partition_predicate. + """ + 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")) + catalog.create_table("default.part_partial_del_test", schema=schema, partition_spec=spec) + + import pyarrow as pa + + df = pa.table( + { + "category": ["a", "a", "b", "b"], + "value": [1, 2, 3, 4], + } + ) + + tbl = catalog.load_table("default.part_partial_del_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.part_partial_del_test") + tbl2 = catalog.load_table("default.part_partial_del_test") + + # Partial delete: only value==1 in partition a, triggers CoW rewrite + tbl1.delete("value == 1") + # Partial delete: only value==3 in partition b, triggers CoW rewrite + tbl2.delete("value == 3") + + refreshed = catalog.load_table("default.part_partial_del_test") + result = refreshed.scan().to_arrow() + # Original 4 rows, minus value==1 and value==3 = 2 rows remaining + assert len(result) == 2 + + +def test_overwrite_uses_update_isolation_level(catalog: Catalog) -> None: + """Verify that overwrite() reads write.update.isolation-level, not write.delete.isolation-level.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table( + "default.update_iso_test", + schema=schema, + properties={ + "write.delete.isolation-level": "serializable", + "write.update.isolation-level": "snapshot", + }, + ) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl = catalog.load_table("default.update_iso_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.update_iso_test") + tbl2 = catalog.load_table("default.update_iso_test") + + tbl1.append(df) + + # Under write.delete.isolation-level=serializable this would raise ValidationException. + # But overwrite() uses write.update.isolation-level=snapshot, so it succeeds. + tbl2.overwrite(pa.table({"x": [10, 20, 30]}), overwrite_filter="x > 0") + + refreshed = catalog.load_table("default.update_iso_test") + result = refreshed.scan().to_arrow() + # overwrite() uses write.update.isolation-level=snapshot, so it does not raise on the concurrent + # append. Under snapshot isolation that concurrent append (tbl1's rows) is preserved: the overwrite + # only replaces the rows it saw and then adds 3 new rows, leaving 6 in total. + assert len(result) == 6 + + +def test_overwrite_with_serializable_update_isolation_raises(catalog: Catalog) -> None: + """Verify that overwrite() raises ValidationException when write.update.isolation-level=serializable.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table( + "default.update_serial_test", + schema=schema, + properties={ + "write.update.isolation-level": "serializable", + }, + ) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl = catalog.load_table("default.update_serial_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.update_serial_test") + tbl2 = catalog.load_table("default.update_serial_test") + + tbl1.append(df) + + with pytest.raises(ValidationException): + tbl2.overwrite(pa.table({"x": [10, 20, 30]}), overwrite_filter="x > 0") + + +def test_clean_all_uncommitted_on_validation_exception(catalog: Catalog) -> None: + """Verify that all manifests are cleaned up when commit aborts with ValidationException.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.clean_abort_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + tbl = catalog.load_table("default.clean_abort_test") + tbl.append(df) + + tbl1 = catalog.load_table("default.clean_abort_test") + tbl2 = catalog.load_table("default.clean_abort_test") + + tbl1.delete("x == 1") + + from pyiceberg.table.update.snapshot import _SnapshotProducer + + captured_producers: list[Any] = [] + original_clean_all = _SnapshotProducer._clean_all_uncommitted + + def capturing_clean_all(self_producer: Any) -> None: + captured_producers.append(self_producer) + original_clean_all(self_producer) + + with patch.object(_SnapshotProducer, "_clean_all_uncommitted", capturing_clean_all): + with pytest.raises(ValidationException): + tbl2.delete("x == 1") + + # _clean_all_uncommitted was called on abort + assert len(captured_producers) > 0 + # All manifest lists should be cleared + for producer in captured_producers: + assert producer._written_manifests == [] + assert producer._uncommitted_manifests == [] + + +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.""" + from pyiceberg.partitioning import PartitionField, PartitionSpec + from pyiceberg.table.update.snapshot import _DeleteFiles, _OverwriteFiles + 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("default.mixed_delete_start_snapshot", schema=schema, partition_spec=spec) + + import pyarrow as pa + + # Partition "a" has one row (will be fully deleted) and partition "b" has two rows (partial delete) + table.append(pa.table({"category": ["a", "b", "b"], "value": [1, 2, 3]})) + + base_snapshot_id = table.metadata.current_snapshot_id + + tx = Transaction(table, autocommit=False) + # "value == 1" deletes the entire file in partition "a" (full-file delete) + # "value == 2" partially deletes from partition "b" (CoW rewrite) + tx.delete("value <= 2") + + assert len(tx._snapshot_producers) == 2 + + delete_producer = next(p for p in tx._snapshot_producers if isinstance(p, _DeleteFiles)) + overwrite_producer = next(p for p in tx._snapshot_producers if isinstance(p, _OverwriteFiles)) + + assert delete_producer._starting_snapshot_id == base_snapshot_id + assert overwrite_producer._starting_snapshot_id == base_snapshot_id + + +def test_validate_concurrency_skips_when_commit_window_is_empty(catalog: Catalog) -> None: + """Validation should be skipped when CommitWindow.is_empty() is True (no concurrent commits).""" + catalog.create_namespace("default") + schema = _test_schema() + table = catalog.create_table("default.missing_parent_test", schema=schema) + + import pyarrow as pa + + table.append(pa.table({"x": [1, 2, 3]})) + + from pyiceberg.table.update.snapshot import CommitWindow, _DeleteFiles + + tx = Transaction(table, autocommit=False) + producer = _DeleteFiles( + operation=Operation.DELETE, + transaction=tx, + io=table.io, + ) + + # CommitWindow where base == head means no concurrent commits occurred + assert table.metadata.current_snapshot_id is not None + current = table.metadata.snapshot_by_id(table.metadata.current_snapshot_id) + producer._commit_window = CommitWindow(base=current, head=current) + + # Should not raise (validation is skipped) + producer._validate_concurrency() + + +def test_validate_concurrency_raises_on_missing_starting_snapshot(catalog: Catalog) -> None: + """CommitWindow.resolve should raise when starting_snapshot_id cannot be resolved.""" + catalog.create_namespace("default") + schema = _test_schema() + table = catalog.create_table("default.missing_starting_test", schema=schema) + + import pyarrow as pa + + table.append(pa.table({"x": [1, 2, 3]})) + + from pyiceberg.table.update.snapshot import CommitWindow + + with pytest.raises(ValidationException, match="Cannot find starting snapshot"): + CommitWindow.resolve(table.metadata, base_id=99999999, branch="main") + + +def test_mixed_delete_overwrite_retries_successfully(catalog: Catalog) -> None: + """A mixed full-file + partial delete should succeed via retry, not raise ValidationException.""" + 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")) + catalog.create_table("default.mixed_retry_test", schema=schema, partition_spec=spec) + + import pyarrow as pa + + tbl = catalog.load_table("default.mixed_retry_test") + + # 3 partitions, one data file each: a->[1,2], b->[3,4], c->[5,6] + tbl.append(pa.table({"category": ["a", "a", "b", "b", "c", "c"], "value": [1, 2, 3, 4, 5, 6]})) + + tbl1 = catalog.load_table("default.mixed_retry_test") + tbl2 = catalog.load_table("default.mixed_retry_test") + + # Concurrent append to partition 'c' (commits first, advances the HEAD) + tbl1.append(pa.table({"category": ["c"], "value": [7]})) + + # Mixed delete on tbl2 (stale snapshot): + # partition 'a' is a partial rewrite (value==1 deleted, value==2 remains) -> _OverwriteFiles + # partition 'b' is a full delete (category == 'b') -> _DeleteFiles + # This should NOT conflict with the append to 'c', so retry should succeed. + tbl2.delete("value == 1 or category == 'b'") + + result = catalog.load_table("default.mixed_retry_test").scan().to_arrow() + assert sorted(result.column("value").to_pylist()) == [2, 5, 6, 7] + + +def test_manifest_list_cleanup_on_retry(catalog: Catalog) -> None: + """Verify that manifest list files from failed retry attempts are cleaned up.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.manifest_list_cleanup_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + # Seed the table so there's a branch HEAD to conflict with + tbl = catalog.load_table("default.manifest_list_cleanup_test") + tbl.append(df) + + # Two writers see the same snapshot + tbl1 = catalog.load_table("default.manifest_list_cleanup_test") + tbl2 = catalog.load_table("default.manifest_list_cleanup_test") + + # tbl1 commits first, advancing catalog HEAD + tbl1.append(df) + + # Track deletes on tbl2 + deleted_paths: list[str] = [] + original_delete = tbl2.io.delete + + def tracking_delete(path: str) -> None: + deleted_paths.append(path) + original_delete(path) + + with patch.object(tbl2.io, "delete", side_effect=tracking_delete): + tbl2.append(df) + + # At least one manifest list (snap-*.avro) from the failed attempt should be deleted + manifest_list_deletes = [p for p in deleted_paths if "snap-" in p and p.endswith(".avro")] + assert len(manifest_list_deletes) >= 1, ( + f"Expected at least one orphaned manifest list to be cleaned up. Deleted paths: {deleted_paths}" + ) + + # Sanity check: all data committed + refreshed = catalog.load_table("default.manifest_list_cleanup_test") + assert len(refreshed.scan().to_arrow()) == 9 + + +def test_manifest_list_cleanup_on_abort(catalog: Catalog) -> None: + """Verify that ALL manifest lists are cleaned up when a commit permanently fails.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.manifest_list_abort_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + # Seed the table + tbl = catalog.load_table("default.manifest_list_abort_test") + tbl.append(df) + + # Two writers see the same snapshot + tbl1 = catalog.load_table("default.manifest_list_abort_test") + tbl2 = catalog.load_table("default.manifest_list_abort_test") + + # tbl1 deletes x==1, so tbl2's same delete will conflict + tbl1.delete("x == 1") + + # Track deletes on tbl2 + deleted_paths: list[str] = [] + original_delete = tbl2.io.delete + + def tracking_delete(path: str) -> None: + deleted_paths.append(path) + original_delete(path) + + with patch.object(tbl2.io, "delete", side_effect=tracking_delete): + with pytest.raises(ValidationException): + tbl2.delete("x == 1") + + # Manifest list files should be cleaned up on abort + manifest_list_deletes = [p for p in deleted_paths if "snap-" in p and p.endswith(".avro")] + assert len(manifest_list_deletes) >= 1, f"Expected manifest list cleanup on abort. Deleted paths: {deleted_paths}" + + +def test_commit_retry_on_non_main_branch(catalog: Catalog) -> None: + """Verify that commit retry works correctly on a non-main branch.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.branch_retry_test", schema=schema) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + # Seed the table and create a branch + tbl = catalog.load_table("default.branch_retry_test") + tbl.append(df) + assert tbl.metadata.current_snapshot_id is not None + tbl.manage_snapshots().create_branch(snapshot_id=tbl.metadata.current_snapshot_id, branch_name="test-branch").commit() + + # Two writers targeting the same branch + tbl1 = catalog.load_table("default.branch_retry_test") + tbl2 = catalog.load_table("default.branch_retry_test") + + # tbl1 appends to the branch first + tbl1.append(df, branch="test-branch") + + # tbl2 appends to the same branch (stale ref), should retry and succeed + import pyiceberg.table as _table_module + + RuntimeTransaction = _table_module.Transaction + original_rebuild = RuntimeTransaction._rebuild_snapshot_updates + rebuild_count = 0 + + def counting_rebuild(self_tx: Any) -> None: + nonlocal rebuild_count + rebuild_count += 1 + original_rebuild(self_tx) + + with patch.object(RuntimeTransaction, "_rebuild_snapshot_updates", counting_rebuild): + tbl2.append(df, branch="test-branch") + + assert rebuild_count == 1, "Expected exactly one retry on non-main branch" + + # Both branch appends should be visible when scanning the branch + refreshed = catalog.load_table("default.branch_retry_test") + branch_snapshot_id = refreshed.metadata.refs["test-branch"].snapshot_id + result = refreshed.scan(snapshot_id=branch_snapshot_id).to_arrow() + assert len(result) == 9 + + +def test_commit_retry_delete_on_non_main_branch(catalog: Catalog) -> None: + """Verify that delete with retry works correctly on a non-main branch.""" + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table( + "default.branch_delete_retry_test", + schema=schema, + properties={"write.delete.isolation-level": "snapshot"}, + ) + + import pyarrow as pa + + df = pa.table({"x": [1, 2, 3]}) + + # Seed and create branch + tbl = catalog.load_table("default.branch_delete_retry_test") + tbl.append(df) + assert tbl.metadata.current_snapshot_id is not None + tbl.manage_snapshots().create_branch(snapshot_id=tbl.metadata.current_snapshot_id, branch_name="test-branch").commit() + + # Append more data to the branch + tbl = catalog.load_table("default.branch_delete_retry_test") + tbl.append(pa.table({"x": [4, 5, 6]}), branch="test-branch") + + # Two writers: one appends, one deletes on the same branch + tbl1 = catalog.load_table("default.branch_delete_retry_test") + tbl2 = catalog.load_table("default.branch_delete_retry_test") + + # tbl1 appends to branch (non-conflicting with delete on different data) + tbl1.append(pa.table({"x": [7, 8, 9]}), branch="test-branch") + + # tbl2 deletes x==1 on branch. Should retry (stale ref) and succeed + # because the concurrent append (x=7,8,9) does not conflict under snapshot isolation. + tbl2.delete("x == 1", branch="test-branch") + + refreshed = catalog.load_table("default.branch_delete_retry_test") + branch_snapshot_id = refreshed.metadata.refs["test-branch"].snapshot_id + result = refreshed.scan(snapshot_id=branch_snapshot_id).to_arrow() + # Original: 1,2,3,4,5,6 + append 7,8,9 - delete x==1 = 2,3,4,5,6,7,8,9 + assert sorted(result.column("x").to_pylist()) == [2, 3, 4, 5, 6, 7, 8, 9] + + +def test_negative_num_retries_still_commits(catalog: Catalog) -> None: + """A negative commit.retry.num-retries is clamped to 0, so one attempt still runs (matches Java). + + Without the clamp, range(num_retries + 1) becomes range(0), the commit is never attempted, + the staged updates are cleared, and the call returns as a silent no-op. + """ + import pyarrow as pa + + catalog.create_namespace("default") + schema = Schema(NestedField(1, "x", LongType(), required=False)) + catalog.create_table( + "default.negative_retries_test", + schema=schema, + properties={TableProperties.COMMIT_NUM_RETRIES: "-1"}, + ) + + table = catalog.load_table("default.negative_retries_test") + table.append(pa.table({"x": [1]})) + + assert catalog.load_table("default.negative_retries_test").scan().to_arrow().to_pylist() == [{"x": 1}] + + +def test_writer_that_started_on_an_empty_table_still_validates(catalog: Catalog) -> None: + """A writer that started on an empty table must still validate against a concurrently-landed first snapshot. + + When the starting snapshot is None (empty table), the commit window must not be treated as empty + if a head exists. Otherwise a concurrent first snapshot bypasses validation and its rows can be + deleted with no error. + """ + import pyarrow as pa + + catalog.create_namespace("default") + schema = Schema(NestedField(1, "x", LongType(), required=False)) + catalog.create_table( + "default.empty_start_validate", + schema=schema, + properties={ + TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1", + TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2", + }, + ) + + stale = catalog.load_table("default.empty_start_validate") + 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.empty_start_validate").append(pa.table({"x": [1]})) + + with pytest.raises(ValidationException): + tx.commit_transaction() + + +def test_unknown_commit_outcome_keeps_the_committed_files(catalog: Catalog) -> None: + """An exception with an unknown outcome must not delete files the committed snapshot may reference. + + The cleanup on the retry loop should only run for exceptions that guarantee the commit did not + happen. A CommitStateUnknownException (e.g. a lost response after the catalog already committed) + is re-raised without deleting, so the live snapshot's manifests stay intact. + """ + import pyarrow as pa + + catalog.create_namespace("default") + schema = Schema(NestedField(1, "x", LongType(), required=False)) + catalog.create_table("default.unknown_outcome", schema=schema) + + table = catalog.load_table("default.unknown_outcome") + real_commit = catalog.commit_table + + def commit_then_lose_response(*args: Any, **kwargs: Any) -> Any: + 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.unknown_outcome").scan().to_arrow().to_pylist() == [{"x": 1}] + + +def test_reusing_a_failed_transaction_cannot_publish_deleted_files(catalog: Catalog) -> None: + """A transaction whose commit failed is invalidated, so reusing it cannot publish deleted files. + + On failure the retry loop deletes the written manifests. Reusing the transaction would otherwise + re-stage updates that point at those deleted files, so the transaction refuses reuse instead. + """ + import pyarrow as pa + + catalog.create_namespace("default") + schema = Schema(NestedField(1, "x", LongType(), required=False)) + table = catalog.create_table("default.reuse_failed", schema=schema, properties={TableProperties.COMMIT_NUM_RETRIES: "0"}) + table.append(pa.table({"x": [1]})) + + tx = table.transaction() + tx.append(pa.table({"x": [2]})) + + with patch.object(catalog, "commit_table", side_effect=CommitFailedException("transient")): + with pytest.raises(CommitFailedException): + tx.commit_transaction() + + # Reusing the failed transaction must not corrupt the table. + with pytest.raises(RuntimeError): + tx.commit_transaction() + assert catalog.load_table("default.reuse_failed").scan().to_arrow().to_pylist() == [{"x": 1}] + + +def test_reusing_a_committed_transaction_does_not_damage_the_first_commit(catalog: Catalog) -> None: + """Reusing a transaction after a successful commit must not replay the first commit's producers.""" + import pyarrow as pa + + catalog.create_namespace("default") + schema = Schema(NestedField(1, "x", LongType(), required=False)) + table = catalog.create_table( + "default.reuse_committed", + schema=schema, + properties={ + TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1", + TableProperties.COMMIT_MAX_RETRY_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.reuse_committed").append(pa.table({"x": [100]})) # forces a retry + + tx.commit_transaction() + + rows = sorted(catalog.load_table("default.reuse_committed").scan().to_arrow()["x"].to_pylist()) + assert rows == [1, 2, 3, 10, 20, 100] + + +def test_snapshot_isolation_delete_does_not_remove_rows_it_never_saw(catalog: Catalog) -> None: + """Under snapshot isolation, a retried delete must not drop concurrently added rows it never saw. + + The delete's planned file set is frozen at plan time, so replanning against a newer head on retry + does not turn a concurrently added, matching file into a delete target. + """ + import pyarrow as pa + + catalog.create_namespace("default") + schema = Schema(NestedField(1, "x", LongType(), required=False)) + table = catalog.create_table( + "default.snap_iso_delete", + schema=schema, + properties={ + TableProperties.WRITE_DELETE_ISOLATION_LEVEL: "snapshot", + TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1", + TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2", + }, + ) + table.append(pa.table({"x": [0, 1]})) + + stale = catalog.load_table("default.snap_iso_delete") + tx = stale.transaction() + tx.delete("x == 1") + + # Lands after the delete was planned. The stale writer never saw this row. + catalog.load_table("default.snap_iso_delete").append(pa.table({"x": [1]})) + + tx.commit_transaction() + + rows = sorted(catalog.load_table("default.snap_iso_delete").scan().to_arrow()["x"].to_pylist()) + assert rows == [0, 1] + + +def test_retried_delete_treats_a_concurrent_commit_atomically(catalog: Catalog) -> None: + """A retried delete under snapshot isolation must treat a concurrent commit atomically.""" + import uuid + + import pyarrow as pa + + from pyiceberg.io.pyarrow import _dataframe_to_data_files + + catalog.create_namespace("default") + schema = Schema(NestedField(1, "x", LongType(), required=False)) + table = catalog.create_table( + "default.retried_delete_atomic", + schema=schema, + properties={ + TableProperties.WRITE_DELETE_ISOLATION_LEVEL: "snapshot", + TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1", + TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2", + }, + ) + table.append(pa.table({"x": [0, 1]})) + + stale = catalog.load_table("default.retried_delete_atomic") + tx = stale.transaction() + tx.delete("x == 1") + + # One concurrent commit carrying two files: [1] wholly matches, [1, 5] partially. + b = catalog.load_table("default.retried_delete_atomic") + 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.retried_delete_atomic").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}" + + +def test_commit_that_landed_but_was_reported_failed_is_not_committed_twice(catalog: Catalog) -> None: + """A commit that landed but was reported as failed must not be committed a second time. + + The snapshot id is stable across attempts, so a landed commit can be found in the refreshed + metadata on retry and is not repeated. Without this, the data would be committed twice and the + post-success cleanup would delete manifests the first committed snapshot references. + """ + import pyarrow as pa + + catalog.create_namespace("default") + schema = Schema(NestedField(1, "x", LongType(), required=False)) + catalog.create_table( + "default.landed_but_reported_failed", + schema=schema, + properties={ + TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1", + TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2", + }, + ) + + table = catalog.load_table("default.landed_but_reported_failed") + real_commit = catalog.commit_table + calls: list[int] = [] + + def commit_then_report_conflict(*args: Any, **kwargs: Any) -> Any: + 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.landed_but_reported_failed").scan().to_arrow().to_pylist() == [{"x": 1}] + + +def test_delete_that_matched_nothing_at_staging_still_validates(catalog: Catalog) -> None: + """A delete whose plan was empty at staging must still validate at commit time. + + An empty-plan delete stages no updates, so commit_transaction would return before the + retry loop and validation. If a row matching the writer's own predicate lands between + staging and commit, serializable isolation requires a ValidationException rather than + a silent success that leaves the matching row in place. + """ + import pyarrow as pa + + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table( + "default.empty_plan_delete_test", + schema=schema, + properties={ + TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1", + TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2", + }, + ) + + tx = catalog.load_table("default.empty_plan_delete_test").transaction() + with pytest.warns(UserWarning): # the delete matches nothing at staging time + tx.delete("x > 45") + + # A row matching the writer's own predicate lands concurrently. + catalog.load_table("default.empty_plan_delete_test").append(pa.table({"x": [50]})) + + with pytest.raises(ValidationException): + tx.commit_transaction() + + # The concurrent row is intact. + result = catalog.load_table("default.empty_plan_delete_test").scan().to_arrow() + assert result["x"].to_pylist() == [50] + + +def test_delete_that_matched_nothing_without_concurrent_commit_succeeds(catalog: Catalog) -> None: + """An empty-plan delete with no concurrent activity commits cleanly as a no-op.""" + import pyarrow as pa + + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table("default.empty_plan_noop_test", schema=schema) + + table = catalog.load_table("default.empty_plan_noop_test") + table.append(pa.table({"x": [1]})) + + tx = catalog.load_table("default.empty_plan_noop_test").transaction() + with pytest.warns(UserWarning): # the delete matches nothing + tx.delete("x > 45") + tx.commit_transaction() + + result = catalog.load_table("default.empty_plan_noop_test").scan().to_arrow() + assert result["x"].to_pylist() == [1] + + +def test_negative_wait_properties_do_not_mask_commit_failure(catalog: Catalog) -> None: + """Negative wait properties are clamped so a retry does not die in time.sleep. + + Without the clamp, a negative min-wait raises ValueError from time.sleep mid-retry, + masking the original CommitFailedException. + """ + import pyarrow as pa + + catalog.create_namespace("default") + schema = _test_schema() + catalog.create_table( + "default.negative_wait_test", + schema=schema, + properties={ + TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "-100", + TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "-1", + }, + ) + + df = pa.table({"x": [1, 2, 3]}) + tbl1 = catalog.load_table("default.negative_wait_test") + tbl2 = catalog.load_table("default.negative_wait_test") + + tbl1.append(df) + # Succeeds via retry; a negative wait would raise ValueError from time.sleep instead. + tbl2.append(df) + + result = catalog.load_table("default.negative_wait_test").scan().to_arrow() + assert len(result) == 6 diff --git a/tests/table/test_validate.py b/tests/table/test_validate.py index cb9d80e196..a19983fd66 100644 --- a/tests/table/test_validate.py +++ b/tests/table/test_validate.py @@ -65,11 +65,18 @@ def test_validation_history(table_v2_with_extensive_snapshots_and_manifests: tup """Test the validation history function.""" table, mock_manifests = table_v2_with_extensive_snapshots_and_manifests - expected_manifest_data_counts = len([m for m in mock_manifests.values() if m[0].content == ManifestContent.DATA]) - oldest_snapshot = table.snapshots()[0] newest_snapshot = cast(Snapshot, table.current_snapshot()) + # from_snapshot (oldest) is excluded from the results since it's the base + expected_manifest_data_counts = len( + [ + m + for snap_id, m in mock_manifests.items() + if m[0].content == ManifestContent.DATA and snap_id != oldest_snapshot.snapshot_id + ] + ) + def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]: """Mock the manifests method to use the snapshot_id for lookup.""" snapshot_id = self.snapshot_id @@ -246,7 +253,8 @@ def test_validate_added_data_files_conflicting_count( update={"snapshots": snapshots}, ) - oldest_snapshot = table.snapshots()[-snapshot_history] + # Use one snapshot before the altered range as the boundary (exclusive stop point) + boundary_snapshot = table.snapshots()[-(snapshot_history + 1)] newest_snapshot = cast(Snapshot, table.current_snapshot()) def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]: @@ -273,7 +281,7 @@ def mock_fetch_manifest_entry(self: ManifestFile, io: FileIO, discard_deleted: b table=table, starting_snapshot=newest_snapshot, data_filter=None, - parent_snapshot=oldest_snapshot, + parent_snapshot=boundary_snapshot, partition_set=None, ) )