Fix summary generation

Author: Fokko

pyiceberg/table/__init__.py

@@ -2888,6 +2888,15 @@ def _summary(self, snapshot_properties: Dict[str, str] = EMPTY_DICT) -> Summary:
                 schema=self._transaction.table_metadata.schema(),
             )
 
+        if len(self._deleted_data_files) > 0:
+            specs = self._transaction.table_metadata.specs()
+            for data_file in self._deleted_data_files:
+                ssc.remove_file(
+                    data_file=data_file,
+                    partition_spec=specs.get(data_file.spec_id),
+                    schema=self._transaction.table_metadata.schema(),
+                )
+
         previous_snapshot = (
             self._transaction.table_metadata.snapshot_by_id(self._parent_snapshot_id)
             if self._parent_snapshot_id is not None
@@ -3028,6 +3037,7 @@ def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) ->
         existing_manifests = []
         total_deleted_entries = []
         partial_rewrites_needed = False
+        self._deleted_data_files = set()
         if snapshot := self._transaction.table_metadata.current_snapshot():
             for manifest_file in snapshot.manifests(io=self._io):
                 if manifest_file.content == ManifestContent.DATA:
@@ -3041,6 +3051,7 @@ def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) ->
                         for entry in manifest_file.fetch_manifest_entry(io=self._io, discard_deleted=True):
                             if strict_metrics_evaluator(entry.data_file) == ROWS_MUST_MATCH:
                                 deleted_entries.append(_copy_with_new_status(entry, ManifestEntryStatus.DELETED))
+                                self._deleted_data_files.add(entry.data_file)
                             elif inclusive_metrics_evaluator(entry.data_file) == ROWS_CANNOT_MATCH:
                                 existing_entries.append(_copy_with_new_status(entry, ManifestEntryStatus.EXISTING))
                             else:

tests/integration/test_inspect_table.py

@@ -103,13 +103,14 @@ def test_inspect_snapshots(
         assert isinstance(snapshot_id.as_py(), int)
 
     assert df['parent_id'][0].as_py() is None
-    assert df['parent_id'][1:] == df['snapshot_id'][:2]
+    assert df['parent_id'][1:].to_pylist() == df['snapshot_id'][:-1].to_pylist()
 
-    assert [operation.as_py() for operation in df['operation']] == ['append', 'overwrite', 'append']
+    assert [operation.as_py() for operation in df['operation']] == ['append', 'delete', 'overwrite', 'append']
 
     for manifest_list in df['manifest_list']:
         assert manifest_list.as_py().startswith("s3://")
 
+    # Append
     assert df['summary'][0].as_py() == [
         ('added-files-size', '5459'),
         ('added-data-files', '1'),
@@ -122,6 +123,19 @@ def test_inspect_snapshots(
         ('total-equality-deletes', '0'),
     ]
 
+    # Delete
+    assert df['summary'][1].as_py() == [
+        ('removed-files-size', '5459'),
+        ('deleted-data-files', '1'),
+        ('deleted-records', '3'),
+        ('total-data-files', '0'),
+        ('total-delete-files', '0'),
+        ('total-records', '0'),
+        ('total-files-size', '0'),
+        ('total-position-deletes', '0'),
+        ('total-equality-deletes', '0'),
+    ]
+
     lhs = spark.table(f"{identifier}.snapshots").toPandas()
     rhs = df.to_pandas()
     for column in df.column_names: