Core: Preserve explicit delete_data_file() in _DeleteFiles._compute_deletes

Fixes #3857

_compute_deletes resets self._deleted_data_files before scanning manifests by predicate. Files added via the inherited delete_data_file() method were silently dropped. Preserve them and include them in the should_delete check alongside predicate-matched files.

Author: qzyu999

pyiceberg/table/update/snapshot.py

@@ -594,6 +594,8 @@ def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) ->
         existing_manifests = []
         total_deleted_entries = []
         partial_rewrites_needed = False
+        # Preserve files explicitly requested via delete_data_file() before resetting.
+        explicit_deletes = set(self._deleted_data_files)
         self._deleted_data_files = set()
 
         # Determine the snapshot to read manifests from for deletion
@@ -620,7 +622,10 @@ def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) ->
                                 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
+                                    should_delete = (
+                                        entry.data_file in explicit_deletes
+                                        or strict_metrics_evaluator(entry.data_file) == ROWS_MUST_MATCH
+                                    )
 
                                 if should_delete:
                                     # Based on the metadata, it can be dropped right away

tests/table/test_delete_data_file_on_delete_files.py

@@ -0,0 +1,67 @@
+# 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.
+"""Regression tests for delete_data_file() on _DeleteFiles instances."""
+
+import pyarrow as pa
+
+from pyiceberg.catalog import Catalog
+from pyiceberg.schema import Schema
+from pyiceberg.types import LongType, NestedField
+
+
+def test_delete_files_with_explicit_delete_data_file(catalog: Catalog) -> None:
+    """Calling delete_data_file() on a _DeleteFiles instance should delete the file."""
+    catalog.create_namespace("default")
+    table = catalog.create_table(
+        "default.delete_explicit",
+        Schema(NestedField(1, "x", LongType(), required=False)),
+    )
+    table.append(pa.table({"x": [1, 2, 3]}))
+
+    data_file = next(iter(table.scan().plan_files())).file
+
+    with table.transaction() as tx:
+        with tx.update_snapshot().delete() as delete_snapshot:
+            delete_snapshot.delete_data_file(data_file)
+
+    reloaded = catalog.load_table("default.delete_explicit")
+    assert reloaded.scan().to_arrow()["x"].to_pylist() == []
+
+
+def test_delete_files_explicit_file_and_predicate_combined(catalog: Catalog) -> None:
+    """delete_data_file() and delete_by_predicate() should both take effect."""
+    catalog.create_namespace("default")
+    table = catalog.create_table(
+        "default.delete_combined",
+        Schema(NestedField(1, "x", LongType(), required=False)),
+    )
+    # Write two separate files
+    table.append(pa.table({"x": [1]}))
+    table.append(pa.table({"x": [2]}))
+
+    files = [task.file for task in table.scan().plan_files()]
+    assert len(files) == 2
+
+    # Delete one file explicitly, leave the other
+    with table.transaction() as tx:
+        with tx.update_snapshot().delete() as delete_snapshot:
+            delete_snapshot.delete_data_file(files[0])
+
+    reloaded = catalog.load_table("default.delete_combined")
+    remaining = reloaded.scan().to_arrow()["x"].to_pylist()
+    # Only the second file's data should remain
+    assert len(remaining) == 1