Table: Implement replace() API for data compaction

Replaces the use of .overwrite() in MaintenanceTable.compact() with a new .replace() API backed by a _RewriteFiles producer. This ensures compaction now generates an Operation.REPLACE snapshot instead of Operation.OVERWRITE, preserving logical table state for downstream consumers.

Fixes #1092

Author: qzyu999

pyiceberg/table/__init__.py

@@ -614,6 +614,54 @@ def overwrite(
                 for data_file in data_files:
                     append_files.append_data_file(data_file)
 
+    def replace(
+        self,
+        df: pa.Table,
+        files_to_delete: Iterable[DataFile],
+        snapshot_properties: dict[str, str] = EMPTY_DICT,
+        branch: str | None = MAIN_BRANCH,
+    ) -> None:
+        """
+        Shorthand for replacing existing files.
+
+        A replace will produce a REPLACE snapshot that will ignore existing
+        files and replace them with the new files.
+
+        Args:
+            df: The Arrow dataframe that will be used to generate the new data files
+            files_to_delete: The files to delete
+            snapshot_properties: Custom properties to be added to the snapshot summary
+            branch: Branch Reference to run the replace operation
+        """
+        try:
+            import pyarrow as pa
+        except ModuleNotFoundError as e:
+            raise ModuleNotFoundError("For writes PyArrow needs to be installed") from e
+
+        from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, _dataframe_to_data_files
+
+        if not isinstance(df, pa.Table):
+            raise ValueError(f"Expected PyArrow table, got: {df}")
+
+        downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
+        _check_pyarrow_schema_compatible(
+            self.table_metadata.schema(),
+            provided_schema=df.schema,
+            downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us,
+            format_version=self.table_metadata.format_version,
+        )
+
+        with self.update_snapshot(snapshot_properties=snapshot_properties, branch=branch).replace() as replace_snapshot:
+            for file_to_delete in files_to_delete:
+                replace_snapshot.delete_data_file(file_to_delete)
+
+            if df.shape[0] > 0:
+                data_files = _dataframe_to_data_files(
+                    table_metadata=self.table_metadata, write_uuid=replace_snapshot.commit_uuid, df=df, io=self._table.io
+                )
+                for data_file in data_files:
+                    replace_snapshot.append_data_file(data_file)
+
     def delete(
         self,
         delete_filter: str | BooleanExpression,
@@ -1432,6 +1480,33 @@ def overwrite(
                 branch=branch,
             )
 
+    def replace(
+        self,
+        df: pa.Table,
+        files_to_delete: Iterable[DataFile],
+        snapshot_properties: dict[str, str] = EMPTY_DICT,
+        branch: str | None = MAIN_BRANCH,
+    ) -> None:
+        """
+        Shorthand for replacing existing files.
+
+        A replace will produce a REPLACE snapshot that will ignore existing
+        files and replace them with the new files.
+
+        Args:
+            df: The Arrow dataframe that will be used to generate the new data files
+            files_to_delete: The files to delete
+            snapshot_properties: Custom properties to be added to the snapshot summary
+            branch: Branch Reference to run the replace operation
+        """
+        with self.transaction() as tx:
+            tx.replace(
+                df=df,
+                files_to_delete=files_to_delete,
+                snapshot_properties=snapshot_properties,
+                branch=branch,
+            )
+
     def delete(
         self,
         delete_filter: BooleanExpression | str = ALWAYS_TRUE,

pyiceberg/table/maintenance.py

@@ -63,6 +63,11 @@ def compact(self) -> None:
             logger.info("Table contains no rows, skipping compaction.")
             return
 
-        # Overwrite the table atomically (REPLACE operation)
+        # Replace existing files with new compacted files
         with self.tbl.transaction() as txn:
-            txn.overwrite(arrow_table, snapshot_properties={"snapshot-type": "replace", "replace-operation": "compaction"})
+            files_to_delete = [task.file for task in self.tbl.scan().plan_files()]
+            txn.replace(
+                df=arrow_table,
+                files_to_delete=files_to_delete,
+                snapshot_properties={"snapshot-type": "replace", "replace-operation": "compaction"},
+            )

pyiceberg/table/snapshots.py

@@ -344,7 +344,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

@@ -667,6 +667,91 @@ def _get_entries(manifest: ManifestFile) -> list[ManifestEntry]:
             return []
 
 
+class _RewriteFiles(_SnapshotProducer["_RewriteFiles"]):
+    """Rewrites data in the table. This will produce a REPLACE snapshot.
+
+    Data files were logically rearranged, but no new logical records were
+    added or removed (e.g. compaction).
+    """
+
+    def _existing_manifests(self) -> list[ManifestFile]:
+        """Determine if there are any existing manifest files."""
+        existing_files = []
+
+        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()
+
+                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
+
+    def _deleted_entries(self) -> list[ManifestEntry]:
+        """To determine if we need to record any deleted entries."""
+        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()
+            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=entry.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))
+            return list(itertools.chain(*list_of_entries))
+        else:
+            return []
+
+
 class UpdateSnapshot:
     _transaction: Transaction
     _io: FileIO
@@ -715,6 +800,16 @@ def overwrite(self, commit_uuid: uuid.UUID | None = None) -> _OverwriteFiles:
             snapshot_properties=self._snapshot_properties,
         )
 
+    def replace(self, commit_uuid: uuid.UUID | None = None) -> _RewriteFiles:
+        return _RewriteFiles(
+            commit_uuid=commit_uuid,
+            operation=Operation.REPLACE,
+            transaction=self._transaction,
+            io=self._io,
+            branch=self._branch,
+            snapshot_properties=self._snapshot_properties,
+        )
+
     def delete(self) -> _DeleteFiles:
         return _DeleteFiles(
             operation=Operation.DELETE,

tests/table/test_snapshots.py

@@ -398,8 +398,8 @@ def test_merge_snapshot_summaries_overwrite_summary() -> None:
 
 def test_invalid_operation() -> None:
     with pytest.raises(ValueError) as e:
-        update_snapshot_summaries(summary=Summary(Operation.REPLACE))
-    assert "Operation not implemented: Operation.REPLACE" in str(e.value)
+        update_snapshot_summaries(summary=Summary(Operation("invalid")))
+    assert "'invalid' is not a valid Operation" in str(e.value)
 
 
 def test_invalid_type() -> None: