feat: Add metadata-only REPLACE API via _RewriteFiles snapshot producer

Adds a _RewriteFiles snapshot producer that implements Operation.REPLACE
for data file compaction (bin-packing, sort, format migration). This
mirrors Java Iceberg's BaseRewriteFiles and is accessed via:

    with table.transaction() as tx:
        with tx.update_snapshot().replace() as rewrite:
            rewrite.delete_data_file(old_file)
            rewrite.append_data_file(new_file)

Key behaviors:
- Validates all files-to-delete exist in the table
- Enforces added_records <= deleted_records invariant
- No-op on empty input (no snapshot produced)
- Reuses unaffected manifests, rewrites only dirty ones
- Passes through delete-file manifests unchanged
- Scoped _validate_concurrency: only checks no new delete files
  conflict with replaced data (concurrent appends are NOT conflicts)
- Retry-safe via _refresh_for_retry clearing @cached_property

Scoped to data-file rewriting only. Delete-file rewriting,
dataSequenceNumber override, and validateFromSnapshot are documented
as future additive extensions.

Fixes #3130

Author: qzyu999

pyiceberg/table/snapshots.py

@@ -351,7 +351,7 @@ def _partition_summary(self, update_metrics: UpdateMetrics) -> str:
 
 
 def update_snapshot_summaries(summary: Summary, previous_summary: Mapping[str, str] | None = None) -> Summary:
-    if summary.operation not in {Operation.APPEND, Operation.OVERWRITE, Operation.DELETE}:
+    if summary.operation not in {Operation.APPEND, Operation.OVERWRITE, Operation.DELETE, Operation.REPLACE}:
         raise ValueError(f"Operation not implemented: {summary.operation}")
 
     if not previous_summary:

pyiceberg/table/update/snapshot.py

@@ -216,8 +216,75 @@ def _calculate_added_rows(self, manifests: list[ManifestFile]) -> int:
                 added_rows += manifest.added_rows_count
         return added_rows
 
-    @abstractmethod
-    def _deleted_entries(self) -> list[ManifestEntry]: ...
+    def _get_existing_manifests(self, should_use_manifest_pruning: bool) -> list[ManifestFile]:
+        """Filter existing manifests and rewrite those containing deleted data files."""
+        existing_files: list[ManifestFile] = []
+        manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)
+
+        if snapshot := self._transaction.table_metadata.snapshot_by_name(name=self._target_branch):
+            for manifest_file in snapshot.manifests(io=self._io):
+                if should_use_manifest_pruning and not manifest_evaluators[manifest_file.partition_spec_id](manifest_file):
+                    existing_files.append(manifest_file)
+                    continue
+
+                entries_to_write: list[ManifestEntry] = []
+                found_deleted_entries = False
+
+                for entry in manifest_file.fetch_manifest_entry(io=self._io, discard_deleted=True):
+                    if entry.data_file in self._deleted_data_files:
+                        found_deleted_entries = True
+                    else:
+                        entries_to_write.append(entry)
+
+                if not found_deleted_entries:
+                    existing_files.append(manifest_file)
+                    continue
+
+                if len(entries_to_write) > 0:
+                    with self.new_manifest_writer(self.spec(manifest_file.partition_spec_id)) as writer:
+                        for entry in entries_to_write:
+                            writer.add_entry(
+                                ManifestEntry.from_args(
+                                    status=ManifestEntryStatus.EXISTING,
+                                    snapshot_id=entry.snapshot_id,
+                                    sequence_number=entry.sequence_number,
+                                    file_sequence_number=entry.file_sequence_number,
+                                    data_file=entry.data_file,
+                                )
+                            )
+                    existing_files.append(writer.to_manifest_file())
+
+        return existing_files
+
+    def _get_deleted_manifest_entries(self, manifest: ManifestFile) -> list[ManifestEntry]:
+        """Return the entries from the given manifest that should be marked as DELETED.
+
+        Subclasses override this to control which entries are selected for deletion
+        and whether partition-level pruning is applied. The default returns no entries.
+        """
+        return []
+
+    @cached_property
+    def _cached_deleted_entries(self) -> list[ManifestEntry]:
+        """Scan the parent snapshot's manifests and collect entries to delete."""
+        if self._parent_snapshot_id is not None:
+            previous_snapshot = self._transaction.table_metadata.snapshot_by_id(self._parent_snapshot_id)
+            if previous_snapshot is None:
+                raise ValueError(f"Could not find the previous snapshot: {self._parent_snapshot_id}")
+
+            executor = ExecutorFactory.get_or_create()
+            list_of_entries = executor.map(self._get_deleted_manifest_entries, previous_snapshot.manifests(self._io))
+            return list(itertools.chain(*list_of_entries))
+        else:
+            return []
+
+    def _deleted_entries(self) -> list[ManifestEntry]:
+        return self._cached_deleted_entries
+
+    def _refresh_deleted_entries_cache(self) -> None:
+        """Clear the cached deleted entries so they are recomputed on next access."""
+        if "_cached_deleted_entries" in self.__dict__:
+            del self.__dict__["_cached_deleted_entries"]
 
     @abstractmethod
     def _existing_manifests(self) -> list[ManifestFile]: ...
@@ -773,89 +840,28 @@ class _OverwriteFiles(_SnapshotProducer["_OverwriteFiles"]):
 
     def _existing_manifests(self) -> list[ManifestFile]:
         """Determine if there are any existing manifest files."""
-        existing_files = []
+        return self._get_existing_manifests(should_use_manifest_pruning=True)
 
+    def _get_deleted_manifest_entries(self, manifest: ManifestFile) -> list[ManifestEntry]:
         manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)
-        if snapshot := self._transaction.table_metadata.snapshot_by_name(name=self._target_branch):
-            for manifest_file in snapshot.manifests(io=self._io):
-                # Manifest does not contain rows that match the files to delete partitions
-                if not manifest_evaluators[manifest_file.partition_spec_id](manifest_file):
-                    existing_files.append(manifest_file)
-                    continue
-
-                entries_to_write: set[ManifestEntry] = set()
-                found_deleted_entries: set[ManifestEntry] = set()
+        if not manifest_evaluators[manifest.partition_spec_id](manifest):
+            return []
 
-                for entry in manifest_file.fetch_manifest_entry(io=self._io, discard_deleted=True):
-                    if entry.data_file in self._deleted_data_files:
-                        found_deleted_entries.add(entry)
-                    else:
-                        entries_to_write.add(entry)
-
-                # Is the intercept the empty set?
-                if len(found_deleted_entries) == 0:
-                    existing_files.append(manifest_file)
-                    continue
-
-                # Delete all files from manifest
-                if len(entries_to_write) == 0:
-                    continue
-
-                # We have to rewrite the manifest file without the deleted data files
-                with self.new_manifest_writer(self.spec(manifest_file.partition_spec_id)) as writer:
-                    for entry in entries_to_write:
-                        writer.add_entry(
-                            ManifestEntry.from_args(
-                                status=ManifestEntryStatus.EXISTING,
-                                snapshot_id=entry.snapshot_id,
-                                sequence_number=entry.sequence_number,
-                                file_sequence_number=entry.file_sequence_number,
-                                data_file=entry.data_file,
-                            )
-                        )
-                existing_files.append(writer.to_manifest_file())
-
-        return existing_files
+        return [
+            ManifestEntry.from_args(
+                status=ManifestEntryStatus.DELETED,
+                snapshot_id=self._snapshot_id,
+                sequence_number=entry.sequence_number,
+                file_sequence_number=entry.file_sequence_number,
+                data_file=entry.data_file,
+            )
+            for entry in manifest.fetch_manifest_entry(self._io, discard_deleted=True)
+            if entry.data_file.content == DataFileContent.DATA and entry.data_file in self._deleted_data_files
+        ]
 
     def _deleted_entries(self) -> list[ManifestEntry]:
-        """To determine if we need to record any deleted entries.
-
-        With a full overwrite all the entries are considered deleted.
-        With partial overwrites we have to use the predicate to evaluate
-        which entries are affected.
-        """
-        if self._parent_snapshot_id is not None:
-            previous_snapshot = self._transaction.table_metadata.snapshot_by_id(self._parent_snapshot_id)
-            if previous_snapshot is None:
-                # This should never happen since you cannot overwrite an empty table
-                raise ValueError(f"Could not find the previous snapshot: {self._parent_snapshot_id}")
-
-            executor = ExecutorFactory.get_or_create()
-            manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)
-
-            def _get_entries(manifest: ManifestFile) -> list[ManifestEntry]:
-                if not manifest_evaluators[manifest.partition_spec_id](manifest):
-                    return []
-
-                return [
-                    ManifestEntry.from_args(
-                        status=ManifestEntryStatus.DELETED,
-                        snapshot_id=self._snapshot_id,
-                        sequence_number=entry.sequence_number,
-                        file_sequence_number=entry.file_sequence_number,
-                        data_file=entry.data_file,
-                    )
-                    for entry in manifest.fetch_manifest_entry(self._io, discard_deleted=True)
-                    if entry.data_file.content == DataFileContent.DATA and entry.data_file in self._deleted_data_files
-                ]
-
-            list_of_entries = executor.map(_get_entries, previous_snapshot.manifests(self._io))
-            deleted_entries = list(itertools.chain(*list_of_entries))
-        else:
-            deleted_entries = []
-
+        deleted_entries = self._cached_deleted_entries
         self._validate_required_deletes(deleted_entries)
-
         return deleted_entries
 
     def _validate_required_deletes(self, deleted_entries: list[ManifestEntry]) -> None:
@@ -876,6 +882,97 @@ def _validate_required_deletes(self, deleted_entries: list[ManifestEntry]) -> No
             raise ValidationException(f"Missing required files to delete: {', '.join(sorted(missing))}")
 
 
+class _RewriteFiles(_SnapshotProducer["_RewriteFiles"]):
+    """A snapshot producer that rewrites data files.
+
+    Produces a REPLACE snapshot that swaps existing data files for new ones without
+    changing the logical contents of the table. This is the metadata-only operation
+    used by compaction (bin-packing, sort, format migration).
+
+    Current scope:
+        - Data file rewriting only (delete + add DataFiles)
+        - Validates: files-to-delete exist, added_records <= deleted_records,
+          no new delete files conflict with replaced data files
+
+    Future work (additive — no structural changes needed):
+        - Delete-file rewriting (add _deleted_delete_files set + separate manifest handling)
+        - dataSequenceNumber override (pin new files' seq to match replaced, for eq-delete safety)
+        - validateFromSnapshot (expose _starting_snapshot_id setter for long-running planners)
+        - ignoreEqualityDeletes in validation (coupled with dataSequenceNumber)
+    """
+
+    def _commit(self) -> UpdatesAndRequirements:
+        if not self._deleted_data_files and not self._added_data_files:
+            return (), ()
+
+        deleted_entries = self._deleted_entries()
+        found_deleted_files = {entry.data_file for entry in deleted_entries}
+
+        if len(found_deleted_files) != len(self._deleted_data_files):
+            raise ValidationException("Cannot commit, missing data files to be rewritten that are not in the table")
+
+        added_records = sum(f.record_count for f in self._added_data_files)
+        deleted_records = sum(entry.data_file.record_count for entry in deleted_entries)
+
+        if added_records > deleted_records:
+            raise ValidationException(
+                f"Invalid replace: records added ({added_records}) exceeds records removed ({deleted_records})"
+            )
+
+        return super()._commit()
+
+    def _get_deleted_manifest_entries(self, manifest: ManifestFile) -> list[ManifestEntry]:
+        return [
+            ManifestEntry.from_args(
+                status=ManifestEntryStatus.DELETED,
+                snapshot_id=self.snapshot_id,
+                sequence_number=entry.sequence_number,
+                file_sequence_number=entry.file_sequence_number,
+                data_file=entry.data_file,
+            )
+            for entry in manifest.fetch_manifest_entry(self._io, discard_deleted=True)
+            if entry.data_file.content == DataFileContent.DATA and entry.data_file in self._deleted_data_files
+        ]
+
+    def _existing_manifests(self) -> list[ManifestFile]:
+        return self._get_existing_manifests(should_use_manifest_pruning=False)
+
+    def _validate_concurrency(self) -> None:
+        """Validate that concurrent changes do not conflict with this replace.
+
+        Unlike overwrite/delete, a replace operation only needs to validate that no new
+        delete files have been added that would apply to the data files being replaced.
+        Concurrent data file additions (appends) do NOT conflict with a replace because
+        the replace only touches files it explicitly planned to rewrite.
+
+        This matches Java's BaseRewriteFiles.validate() which only calls
+        validateNoNewDeletesForDataFiles, not validateAddedDataFiles or
+        validateDeletedDataFiles.
+        """
+        from pyiceberg.table.update.validate import _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
+
+        if self._deleted_data_files:
+            table = self._transaction._table
+            conflict_detection_filter = self._predicate if self._predicate != AlwaysFalse() else None
+            _validate_no_new_deletes_for_data_files(
+                table, catalog_head, conflict_detection_filter, self._deleted_data_files, starting_snapshot
+            )
+
+    def _refresh_for_retry(self) -> None:
+        """Reset state for a retry attempt, clearing the cached deleted entries."""
+        super()._refresh_for_retry()
+        self._refresh_deleted_entries_cache()
+
+
 class UpdateSnapshot:
     _transaction: Transaction
     _io: FileIO
@@ -933,6 +1030,15 @@ def delete(self) -> _DeleteFiles:
             snapshot_properties=self._snapshot_properties,
         )
 
+    def replace(self) -> _RewriteFiles:
+        return _RewriteFiles(
+            operation=Operation.REPLACE,
+            transaction=self._transaction,
+            io=self._io,
+            branch=self._branch,
+            snapshot_properties=self._snapshot_properties,
+        )
+
 
 class _ManifestMergeManager(Generic[U]):
     _target_size_bytes: int