fixes according to pr

Author: amitgilad3

pyiceberg/manifest.py

@@ -865,7 +865,6 @@ def existing(self, entry: ManifestEntry) -> ManifestWriter:
                 entry.snapshot_id, entry.sequence_number, entry.file_sequence_number, entry.data_file
             )
         )
-        self._existing_files += 1
         return self
 
 

pyiceberg/table/__init__.py

@@ -219,9 +219,6 @@ class TableProperties:
     MIN_SNAPSHOTS_TO_KEEP = "history.expire.min-snapshots-to-keep"
     MIN_SNAPSHOTS_TO_KEEP_DEFAULT = 1
 
-    SNAPSHOT_ID_INHERITANCE_ENABLED = "compatibility.snapshot-id-inheritance.enabled"
-    SNAPSHOT_ID_INHERITANCE_ENABLED_DEFAULT = False
-
 
 class Transaction:
     _table: Table

pyiceberg/table/update/snapshot.py

@@ -22,7 +22,7 @@
 from abc import abstractmethod
 from collections import defaultdict
 from concurrent.futures import Future
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from functools import cached_property
 from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, List, Optional, Set, Tuple
 
@@ -480,8 +480,8 @@ def _deleted_entries(self) -> List[ManifestEntry]:
 
 @dataclass(init=False)
 class RewriteManifestsResult:
-    rewritten_manifests: List[ManifestFile]
-    added_manifests: List[ManifestFile]
+    rewritten_manifests: List[ManifestFile] = field(default_factory=list)
+    added_manifests: List[ManifestFile] = field(default_factory=list)
 
     def __init__(
         self,
@@ -544,7 +544,11 @@ def _process_manifests(self, manifests: List[ManifestFile]) -> List[ManifestFile
 
 
 class _RewriteManifests(_SnapshotProducer["_RewriteManifests"]):
+    _table: Table
+    _spec_id: int
     _target_size_bytes: int
+    _min_count_to_merge: int
+    _merge_enabled: bool
     rewritten_manifests: List[ManifestFile] = []
     added_manifests: List[ManifestFile] = []
     kept_manifests: List[ManifestFile] = []
@@ -559,10 +563,15 @@ def __init__(
     ):
         from pyiceberg.table import TableProperties
 
-        _table: Table
-        _spec: PartitionSpec
-
         super().__init__(Operation.REPLACE, transaction, io, snapshot_properties=snapshot_properties)
+
+        snapshot = self._table.current_snapshot()
+        if self._spec_id and self._spec_id not in self._table.specs():
+            raise ValueError(f"Cannot find spec with id: {self._spec_id}")
+
+        if not snapshot:
+            raise ValueError("Cannot rewrite manifests without a current snapshot")
+
         self._target_size_bytes = property_as_int(
             self._transaction.table_metadata.properties,
             TableProperties.MANIFEST_TARGET_SIZE_BYTES,
@@ -571,6 +580,17 @@ def __init__(
         self._table = table
         self._spec_id = spec_id or table.spec().spec_id
 
+        self._min_count_to_merge = property_as_int(
+            self._transaction.table_metadata.properties,
+            TableProperties.MANIFEST_MIN_MERGE_COUNT,
+            TableProperties.MANIFEST_MIN_MERGE_COUNT_DEFAULT,
+        )  # type: ignore
+        self._merge_enabled = property_as_bool(
+            self._transaction.table_metadata.properties,
+            TableProperties.MANIFEST_MERGE_ENABLED,
+            TableProperties.MANIFEST_MERGE_ENABLED_DEFAULT,
+        )
+
     def _summary(self, snapshot_properties: Dict[str, str] = EMPTY_DICT) -> Summary:
         from pyiceberg.table import TableProperties
 
@@ -583,10 +603,10 @@ def _summary(self, snapshot_properties: Dict[str, str] = EMPTY_DICT) -> Summary:
         ssc.set_partition_summary_limit(partition_summary_limit)
 
         props = {
-            "manifests-kept": str(len([])),
+            "manifests-kept": "0",
             "manifests-created": str(len(self.added_manifests)),
             "manifests-replaced": str(len(self.rewritten_manifests)),
-            "entries-processed": str(len([])),
+            "entries-processed": "0",
         }
         previous_snapshot = (
             self._transaction.table_metadata.snapshot_by_id(self._parent_snapshot_id)
@@ -601,28 +621,25 @@ def _summary(self, snapshot_properties: Dict[str, str] = EMPTY_DICT) -> Summary:
         )
 
     def rewrite_manifests(self) -> RewriteManifestsResult:
-        data_result = self._find_matching_manifests(ManifestContent.DATA)
+        snapshot = self._table.current_snapshot()
+        if not snapshot:
+            raise ValueError("Cannot rewrite manifests without a current snapshot")
+
+        data_result = self._find_matching_manifests(snapshot, ManifestContent.DATA)
 
         self.rewritten_manifests.extend(data_result.rewritten_manifests)
         self.added_manifests.extend(data_result.added_manifests)
 
-        deletes_result = self._find_matching_manifests(ManifestContent.DELETES)
+        deletes_result = self._find_matching_manifests(snapshot, ManifestContent.DELETES)
         self.rewritten_manifests.extend(deletes_result.rewritten_manifests)
         self.added_manifests.extend(deletes_result.added_manifests)
 
-        if not self.rewritten_manifests:
+        if len(self.rewritten_manifests) == 0:
             return RewriteManifestsResult(rewritten_manifests=[], added_manifests=[])
 
         return RewriteManifestsResult(rewritten_manifests=self.rewritten_manifests, added_manifests=self.added_manifests)
 
-    def _find_matching_manifests(self, content: ManifestContent) -> RewriteManifestsResult:
-        snapshot = self._table.current_snapshot()
-        if self._spec_id and self._spec_id not in self._table.specs():
-            raise ValueError(f"Cannot find spec with id: {self._spec_id}")
-
-        if not snapshot:
-            raise ValueError("Cannot rewrite manifests without a current snapshot")
-
+    def _find_matching_manifests(self, snapshot: Snapshot, content: ManifestContent) -> RewriteManifestsResult:
         manifests = [
             manifest
             for manifest in snapshot.manifests(io=self._io)
@@ -631,8 +648,8 @@ def _find_matching_manifests(self, content: ManifestContent) -> RewriteManifests
 
         data_manifest_merge_manager = _ManifestMergeManager(
             target_size_bytes=self._target_size_bytes,
-            min_count_to_merge=2,
-            merge_enabled=True,
+            min_count_to_merge=self._min_count_to_merge,
+            merge_enabled=self._merge_enabled,
             snapshot_producer=self,
         )
         new_manifests = data_manifest_merge_manager.merge_manifests(manifests=manifests)
@@ -668,10 +685,7 @@ def _existing_manifests(self) -> List[ManifestFile]:
         return [self._copy_manifest_file(manifest, self.snapshot_id) for manifest in self.added_manifests]
 
     def _deleted_entries(self) -> List[ManifestEntry]:
-        """To determine if we need to record any deleted manifest entries.
-
-        In case of an append, nothing is deleted.
-        """
+        """To determine if we need to record any deleted manifest entries."""
         return []
 
 

tests/integration/test_writes/test_rewrite_manifests.py

@@ -180,8 +180,9 @@ def compute_manifest_entry_size_bytes(manifests: List[ManifestFile]) -> float:
 
     for manifest in manifests:
         total_size += manifest.manifest_length
-        num_entries += manifest.added_files_count + manifest.existing_files_count + manifest.deleted_files_count
-
+        num_entries += (
+            (manifest.added_files_count or 0) + (manifest.existing_files_count or 0) + (manifest.deleted_files_count or 0)
+        )
     return total_size / num_entries if num_entries > 0 else 0
 
 
@@ -212,7 +213,7 @@ def test_rewrite_small_manifests_partitioned_table(session_catalog: Catalog) ->
     tbl.refresh()
 
     tbl.refresh()
-    manifests = tbl.current_snapshot().manifests(tbl.io)
+    manifests = tbl.current_snapshot().manifests(tbl.io)  # type: ignore
     assert len(manifests) == 4, "Should have 4 manifests before rewrite"
 
     # manifest_entry_size_bytes = compute_manifest_entry_size_bytes(manifests)
@@ -229,7 +230,7 @@ def test_rewrite_small_manifests_partitioned_table(session_catalog: Catalog) ->
     assert len(result.rewritten_manifests) == 4, "Action should rewrite 4 manifests"
     assert len(result.added_manifests) == 2, "Action should add 2 manifests"
 
-    new_manifests = tbl.current_snapshot().manifests(tbl.io)
+    new_manifests = tbl.current_snapshot().manifests(tbl.io)  # type: ignore
     assert len(new_manifests) == 2, "Should have 2 manifests after rewrite"
 
     assert new_manifests[0].existing_files_count == 4