Add concurrency safety validation checks

pyiceberg/table/delete_file_index.py

@@ -54,6 +54,10 @@ def filter_by_seq(self, seq: int) -> list[DataFile]:
  start_idx = bisect_left(self._seqs, seq)
  return [delete_file for delete_file, _ in self._files[start_idx:]]
 
+ def referenced_delete_files(self) -> list[DataFile]:
+ self._ensure_indexed()
+ return [data_file for data_file, _ in self._files]
+
 
 def _has_path_bounds(delete_file: DataFile) -> bool:
  lower = delete_file.lower_bounds
@@ -140,3 +144,14 @@ def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record
  deletes.update(path_deletes.filter_by_seq(seq_num))
 
  return deletes
+
+ def referenced_delete_files(self) -> list[DataFile]:
+ data_files: list[DataFile] = []
+
+ for deletes in self._by_partition.values():
+ data_files.extend(deletes.referenced_delete_files())
+
+ for deletes in self._by_path.values():
+ data_files.extend(deletes.referenced_delete_files())
+
+ return data_files

pyiceberg/table/update/validate.py

@@ -19,14 +19,23 @@
 from pyiceberg.exceptions import ValidationException
 from pyiceberg.expressions import BooleanExpression
 from pyiceberg.expressions.visitors import ROWS_CANNOT_MATCH, _InclusiveMetricsEvaluator
-from pyiceberg.manifest import ManifestContent, ManifestEntry, ManifestEntryStatus, ManifestFile
+from pyiceberg.manifest import (
+ INITIAL_SEQUENCE_NUMBER,
+ DataFile,
+ ManifestContent,
+ ManifestEntry,
+ ManifestEntryStatus,
+ ManifestFile,
+)
 from pyiceberg.schema import Schema
 from pyiceberg.table import Table
+from pyiceberg.table.delete_file_index import DeleteFileIndex
 from pyiceberg.table.snapshots import Operation, Snapshot, ancestors_between
 from pyiceberg.typedef import Record
 
 VALIDATE_DATA_FILES_EXIST_OPERATIONS: set[Operation] = {Operation.OVERWRITE, Operation.REPLACE, Operation.DELETE}
 VALIDATE_ADDED_DATA_FILES_OPERATIONS: set[Operation] = {Operation.APPEND, Operation.OVERWRITE}
+VALIDATE_ADDED_DELETE_FILES_OPERATIONS: set[Operation] = {Operation.DELETE, Operation.OVERWRITE}
 
 
 def _validation_history(
@@ -216,6 +225,60 @@ def _added_data_files(
  yield entry
 
 
+def _added_delete_files(
+ table: Table,
+ starting_snapshot: Snapshot,
+ data_filter: BooleanExpression | None,
+ partition_set: dict[int, set[Record]] | None,
+ parent_snapshot: Snapshot | None,
+) -> DeleteFileIndex:
+ """Return matching delete files that have been added to the table since a starting snapshot.
+
+ Args:
+ table: Table to get the history from
+ starting_snapshot: Starting snapshot to get the history from
+ data_filter: Optional filter to match data files
+ partition_set: Optional set of partitions to match data files
+ parent_snapshot: Parent snapshot to get the history from
+
+ Returns:
+ DeleteFileIndex
+ """
+ if parent_snapshot is None or table.format_version < 2:
+ return DeleteFileIndex()
+
+ manifests, snapshot_ids = _validation_history(
+ table, parent_snapshot, starting_snapshot, VALIDATE_ADDED_DELETE_FILES_OPERATIONS, ManifestContent.DELETES
+ )
+
+ dfi = DeleteFileIndex()
+
+ for manifest in manifests:
+ for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=True):
+ if _filter_manifest_entries(
+ entry, snapshot_ids, data_filter, partition_set, ManifestEntryStatus.ADDED, table.schema()
+ ):
+ dfi.add_delete_file(entry, entry.data_file.partition)
+
+ return dfi
+
+
+def _starting_sequence_number(table: Table, starting_snapshot: Snapshot) -> int:
+ """Find the starting sequence number from a snapshot.
+
+ Args:
+ table: Table to find snapshot from
+ starting_snapshot: Snapshot from where to start looking
+
+ Returns
+ Sequence number as int
+ """
+ if starting_snapshot is not None:
+ if seq := starting_snapshot.sequence_number:
+ return seq
+ return INITIAL_SEQUENCE_NUMBER
+
+
 def _validate_added_data_files(
  table: Table,
  starting_snapshot: Snapshot,
@@ -235,3 +298,60 @@ def _validate_added_data_files(
  if any(conflicting_entries):
  conflicting_snapshots = {entry.snapshot_id for entry in conflicting_entries if entry.snapshot_id is not None}
  raise ValidationException(f"Added data files were found matching the filter for snapshots {conflicting_snapshots}!")
+
+
+def _validate_no_new_delete_files(
+ table: Table,
+ starting_snapshot: Snapshot,
+ data_filter: BooleanExpression | None,
+ partition_set: dict[int, set[Record]] | None,
+ parent_snapshot: Snapshot | None,
+) -> None:
+ """Validate no new delete files matching a filter have been added to the table since starting a snapshot.
+
+ Args:
+ table: Table to validate
+ starting_snapshot: Snapshot current at the start of the operation
+ data_filter: Expression used to find added data files
+ partition_set: Dictionary of partition spec to set of partition records
+ parent_snapshot: Ending snapshot on the branch being validated
+ """
+ deletes = _added_delete_files(table, starting_snapshot, data_filter, partition_set, parent_snapshot)
+
+ if deletes.is_empty():
+ return
+
+ conflicting_delete_paths = [file.file_path for file in deletes.referenced_delete_files()]
+ raise ValidationException(
+ f"Found new conflicting delete files that can apply to records matching {data_filter}: {conflicting_delete_paths}"
+ )
+
+
+def _validate_no_new_deletes_for_data_files(
+ table: Table,
+ starting_snapshot: Snapshot,
+ data_filter: BooleanExpression | None,
+ data_files: set[DataFile],
+ parent_snapshot: Snapshot | None,
+) -> None:
+ """Validate no new delete files must be applied for data files that have been added to the table since a starting snapshot.
+
+ Args:
+ table: Table to validate
+ starting_snapshot: Snapshot current at the start of the operation
+ data_filter: Expression used to find added data files
+ data_files: data files to validate have no new deletes
+ 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:
+ return
+
+ deletes = _added_delete_files(table, starting_snapshot, data_filter, None, parent_snapshot)
+ seq_num = _starting_sequence_number(table, starting_snapshot)
+
+ # Fail to any delete file found that applies to files written in or before the starting snapshot
+ for data_file in data_files:
+ delete_files = deletes.for_data_file(seq_num, data_file, data_file.partition)
+ if len(delete_files) > 0:
+ raise ValidationException(f"Cannot commit, found new delete for replace data file {data_file}")

tests/table/test_validate.py

@@ -22,14 +22,17 @@
 
 from pyiceberg.exceptions import ValidationException
 from pyiceberg.io import FileIO
-from pyiceberg.manifest import ManifestContent, ManifestEntry, ManifestEntryStatus, ManifestFile
+from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestEntry, ManifestEntryStatus, ManifestFile
 from pyiceberg.table import Table
 from pyiceberg.table.snapshots import Operation, Snapshot, Summary
 from pyiceberg.table.update.validate import (
  _added_data_files,
+ _added_delete_files,
  _deleted_data_files,
  _validate_added_data_files,
  _validate_deleted_data_files,
+ _validate_no_new_delete_files,
+ _validate_no_new_deletes_for_data_files,
  _validation_history,
 )
 
@@ -350,3 +353,153 @@ class DummyEntry:
  data_filter=None,
  parent_snapshot=oldest_snapshot,
  )
+
+
+@pytest.mark.parametrize("operation", [Operation.APPEND, Operation.REPLACE])
+def test_added_delete_files_non_conflicting_count(
+ table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]],
+ operation: Operation,
+) -> None:
+ table, mock_manifests = table_v2_with_extensive_snapshots_and_manifests
+
+ snapshot_history = 100
+ snapshots = table.snapshots()
+ for i in range(1, snapshot_history + 1):
+ altered_snapshot = snapshots[-i]
+ altered_snapshot = altered_snapshot.model_copy(update={"summary": Summary(operation=operation)})
+ snapshots[-i] = altered_snapshot
+
+ table.metadata = table.metadata.model_copy(
+ update={"snapshots": snapshots},
+ )
+
+ oldest_snapshot = table.snapshots()[-snapshot_history]
+ newest_snapshot = cast(Snapshot, table.current_snapshot())
+
+ 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
+ if snapshot_id in mock_manifests:
+ return mock_manifests[snapshot_id]
+ return []
+
+ def mock_fetch_manifest_entry(self: ManifestFile, io: FileIO, discard_deleted: bool = True) -> list[ManifestEntry]:
+ return [
+ ManifestEntry.from_args(
+ status=ManifestEntryStatus.ADDED, snapshot_id=self.added_snapshot_id, sequence_number=self.sequence_number
+ )
+ ]
+
+ with (
+ patch("pyiceberg.table.snapshots.Snapshot.manifests", new=mock_read_manifest_side_effect),
+ patch("pyiceberg.manifest.ManifestFile.fetch_manifest_entry", new=mock_fetch_manifest_entry),
+ ):
+ dfi = _added_delete_files(
+ table=table,
+ starting_snapshot=newest_snapshot,
+ data_filter=None,
+ parent_snapshot=oldest_snapshot,
+ partition_set=None,
+ )
+
+ assert dfi.is_empty()
+ assert len(dfi.referenced_delete_files()) == 0
+
+
+@pytest.mark.parametrize("operation", [Operation.DELETE, Operation.OVERWRITE])
+def test_added_delete_files_conflicting_count(
+ table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]],
+ operation: Operation,
+) -> None:
+ table, mock_manifests = table_v2_with_extensive_snapshots_and_manifests
+
+ snapshot_history = 100
+ snapshots = table.snapshots()
+ for i in range(1, snapshot_history + 1):
+ altered_snapshot = snapshots[-i]
+ altered_snapshot = altered_snapshot.model_copy(update={"summary": Summary(operation=operation)})
+ snapshots[-i] = altered_snapshot
+
+ table.metadata = table.metadata.model_copy(
+ update={"snapshots": snapshots},
+ )
+
+ oldest_snapshot = table.snapshots()[-snapshot_history]
+ newest_snapshot = cast(Snapshot, table.current_snapshot())
+
+ mock_delete_file = DataFile.from_args(
+ content=DataFileContent.POSITION_DELETES,
+ file_path="s3://dummy/path",
+ )
+
+ mock_delete_file.spec_id = 0
+
+ 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
+ if snapshot_id in mock_manifests:
+ return mock_manifests[snapshot_id]
+ return []
+
+ def mock_fetch_manifest_entry(self: ManifestFile, io: FileIO, discard_deleted: bool = True) -> list[ManifestEntry]:
+ return [
+ ManifestEntry.from_args(
+ status=ManifestEntryStatus.ADDED,
+ snapshot_id=self.added_snapshot_id,
+ sequence_number=self.min_sequence_number,
+ data_file=mock_delete_file,
+ )
+ ]
+
+ with (
+ patch("pyiceberg.table.snapshots.Snapshot.manifests", new=mock_read_manifest_side_effect),
+ patch("pyiceberg.manifest.ManifestFile.fetch_manifest_entry", new=mock_fetch_manifest_entry),
+ ):
+ dfi = _added_delete_files(
+ table=table,
+ starting_snapshot=newest_snapshot,
+ data_filter=None,
+ parent_snapshot=oldest_snapshot,
+ partition_set=None,
+ )
+
+ assert not dfi.is_empty()
+ assert dfi.referenced_delete_files()[0] == mock_delete_file
+
+
+def test_validate_no_new_delete_files_raises_on_conflict(
+ table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]],
+) -> None:
+ table, _ = table_v2_with_extensive_snapshots_and_manifests
+ oldest_snapshot = table.snapshots()[0]
+ newest_snapshot = cast(Snapshot, table.current_snapshot())
+
+ with patch("pyiceberg.table.update.validate.DeleteFileIndex.is_empty", return_value=False):
+ with pytest.raises(ValidationException):
+ _validate_no_new_delete_files(
+ table=table,
+ starting_snapshot=newest_snapshot,
+ data_filter=None,
+ partition_set=None,
+ parent_snapshot=oldest_snapshot,
+ )
+
+
+def test_validate_no_new_delete_files_for_data_files_raises_on_conflict(
+ table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]],
+) -> None:
+ table, _ = table_v2_with_extensive_snapshots_and_manifests
+ oldest_snapshot = table.snapshots()[0]
+ newest_snapshot = cast(Snapshot, table.current_snapshot())
+
+ mocked_data_file = DataFile.from_args()
+
+ with patch("pyiceberg.table.update.validate.DeleteFileIndex.for_data_file", return_value=[mocked_data_file]):
+ with pytest.raises(ValidationException):
+ _validate_no_new_deletes_for_data_files(
+ table=table,
+ starting_snapshot=newest_snapshot,
+ data_filter=None,
+ data_files={mocked_data_file},
+ parent_snapshot=oldest_snapshot,
+ )