Apply every delete file when rewriting files for a partial delete

pyiceberg/table/__init__.py

@@ -339,12 +339,6 @@ def _apply(
 
         return self
 
-    def _scan(self, row_filter: str | BooleanExpression = ALWAYS_TRUE, case_sensitive: bool = True) -> DataScan:
-        """Minimal data scan of the table with the current state of the transaction."""
-        return DataScan(
-            table_metadata=self.table_metadata, io=self._table.io, row_filter=row_filter, case_sensitive=case_sensitive
-        )
-
     def upgrade_table_version(self, format_version: TableVersion) -> Transaction:
         """Set the table to a certain version.
 
@@ -807,10 +801,17 @@ def delete(
             bound_delete_filter = bind(self.table_metadata.schema(), delete_filter, case_sensitive)
             preserve_row_filter = _expression_to_complementary_pyarrow(bound_delete_filter, self.table_metadata.schema())
 
-            file_scan = self._scan(row_filter=delete_filter, case_sensitive=case_sensitive)
-            if branch is not None:
-                file_scan = file_scan.use_ref(branch)
-            files = file_scan.plan_files()
+            # Each file is rewritten in full, so it needs every delete file that applies to it,
+            # not only those whose bounds can match the delete filter.
+            planner = ManifestGroupPlanner(
+                table_metadata=self.table_metadata,
+                io=self._table.io,
+                row_filter=delete_filter,
+                case_sensitive=case_sensitive,
+                _ignore_residuals=True,
+            )
+            snapshot = self.table_metadata.snapshot_by_name(branch)
+            files = planner.plan_files(snapshot.manifests(self._table.io)) if snapshot is not None else []
 
             commit_uuid = uuid.uuid4()
             counter = itertools.count(0)
@@ -2863,12 +2864,16 @@ def __init__(
         row_filter: str | BooleanExpression = ALWAYS_TRUE,
         case_sensitive: bool = True,
         options: Properties = EMPTY_DICT,
+        _ignore_residuals: bool = False,
     ):
         self.table_metadata = table_metadata
         self.io = io
         self.row_filter = _parse_row_filter(row_filter)
         self.case_sensitive = case_sensitive
         self.options = options
+        # Set when the planned data files are read in full, as in a rewrite. Residuals are then always true, and delete
+        # files are kept even if their bounds cannot match the row filter, as no residual drops the rows they delete.
+        self._ignore_residuals = _ignore_residuals
 
     @cached_property
     def partition_filters(self) -> KeyDefaultDict[int, BooleanExpression]:
@@ -2906,7 +2911,9 @@ def plan_manifest_entries(self, manifests: Iterable[ManifestFile]) -> Iterator[l
                     self.io,
                     manifest,
                     partition_evaluators[manifest.partition_spec_id],
-                    metrics_evaluator,
+                    metrics_evaluator
+                    if manifest.content == ManifestContent.DATA or not self._ignore_residuals
+                    else lambda _: True,
                 )
                 for manifest in manifests
                 if self._check_sequence_number(min_sequence_number, manifest)
@@ -2985,7 +2992,7 @@ def residual_for(data_file: DataFile) -> BooleanExpression:
                     data_entry.data_file,
                     partition_key=data_entry.data_file.partition,
                 ),
-                residual=residual_for(data_entry.data_file),
+                residual=ALWAYS_TRUE if self._ignore_residuals else residual_for(data_entry.data_file),
             )
             for data_entry in data_entries
         ]

tests/table/test_commit_retry.py

@@ -453,11 +453,17 @@ def test_file_overwrite_validates_concurrent_file_delete(
     assert sorted(result["value"].to_pylist()) == expected_values
 
 
-def _commit_partition_position_delete(table: Table, partition: Record, file_path: str, pos: int) -> None:
-    """PyIceberg cannot write delete files, so commit one without path bounds, as other engines write them."""
+def commit_partition_position_delete(
+    table: Table, partition: Record, file_path: str, pos: int, row: dict[str, Any] | None = None
+) -> None:
+    """PyIceberg cannot write delete files, so commit one without path bounds, as other engines write them.
+
+    If ``row`` is given, it is stored as the deleted row, which gives the delete file bounds on the data columns.
+    """
     import pyarrow as pa
     import pyarrow.parquet as pq
 
+    from pyiceberg.conversions import to_bytes
     from pyiceberg.manifest import DataFile, DataFileContent, FileFormat, ManifestContent, ManifestWriter, ManifestWriterV2
     from pyiceberg.partitioning import PartitionSpec
     from pyiceberg.table.update.snapshot import _OverwriteFiles
@@ -475,16 +481,24 @@ def new_delete_manifest_writer(producer: _OverwriteFiles, spec: PartitionSpec) -
             spec, producer.schema(), producer.new_manifest_output(), producer._snapshot_id, producer._compression
         )
 
+    columns: dict[str, list[Any]] = {"file_path": [file_path], "pos": [pos]}
+    bounds = None
+    if row is not None:
+        columns["row"] = [row]
+        bounds = {field.field_id: to_bytes(field.field_type, row[field.name]) for field in table.schema().fields}
+
     delete_path = table.location_provider().new_data_location(f"pos-delete-{uuid.uuid4()}.parquet")
     with table.io.new_output(delete_path).create() as output:
-        pq.write_table(pa.table({"file_path": [file_path], "pos": [pos]}), output)
+        pq.write_table(pa.table(columns), output)
     delete_file = DataFile.from_args(
         content=DataFileContent.POSITION_DELETES,
         file_path=delete_path,
         file_format=FileFormat.PARQUET,
         partition=partition,
         record_count=1,
         file_size_in_bytes=len(table.io.new_input(delete_path)),
+        lower_bounds=bounds,
+        upper_bounds=bounds,
         spec_id=table.spec().spec_id,
     )
 
@@ -524,7 +538,7 @@ def test_file_overwrite_validates_concurrent_position_delete_in_nan_partition(ca
         overwrite.delete_data_file(file_to_replace)
         overwrite.append_data_file(replacement_file)
 
-    _commit_partition_position_delete(deleting_table, Record(float("nan")), file_to_replace.file_path, pos=0)
+    commit_partition_position_delete(deleting_table, Record(float("nan")), file_to_replace.file_path, pos=0)
 
     with pytest.raises(ValidationException, match="found new delete for replace data file"):
         replacing_transaction.commit_transaction()

tests/table/test_init.py

@@ -124,6 +124,7 @@
     TimeType,
     UUIDType,
 )
+from tests.table.test_commit_retry import commit_partition_position_delete
 
 
 def test_schema(table_v2: Table) -> None:
@@ -2390,6 +2391,25 @@ def test_overwrite_delete_data_file_on_bucket_partition(catalog: Catalog) -> Non
     assert table.scan().to_arrow().num_rows == 0
 
 
+def test_partial_delete_keeps_earlier_deletes_outside_the_filter(catalog: Catalog) -> None:
+    """Rewriting a file applies all of its delete files, not only those whose bounds can match the delete filter."""
+    import pyarrow as pa
+
+    catalog.create_namespace("default")
+    schema = Schema(NestedField(1, "k", LongType(), required=False), NestedField(2, "v", StringType(), required=False))
+    table = catalog.create_table("default.partial_delete", schema=schema, properties={"format-version": "2"})
+    table.append(pa.table({"k": [1, 2, 3], "v": ["a", "b", "c"]}))
+    data_file = next(iter(table.scan().plan_files())).file
+
+    # The delete file stores the deleted row (2, 'b'), so a scan for k = 1 skips it.
+    commit_partition_position_delete(table, Record(), data_file.file_path, pos=1, row={"k": 2, "v": "b"})
+    assert [task.delete_files for task in table.scan(row_filter="k = 1").plan_files()] == [set()]
+
+    table.delete("k = 1")
+
+    assert table.scan().to_arrow()["k"].to_pylist() == [3]
+
+
 def test_static_table_forwards_location_to_table_file_io(metadata_location: str, monkeypatch: pytest.MonkeyPatch) -> None:
     seen_locations: list[str | None] = []
     real_load_file_io = load_file_io