first attempt at init_manifests

pyiceberg/manifest.py

@@ -612,6 +612,24 @@ def has_added_files(self) -> bool:
     def has_existing_files(self) -> bool:
         return self.existing_files_count is None or self.existing_files_count > 0
 
+    def copy_with_snapshot_id(self, snapshot_id: int) -> ManifestFile:
+        return ManifestFile(
+            manifest_path=self.manifest_path,
+            manifest_length=self.manifest_length,
+            partition_spec_id=self.partition_spec_id,
+            content=self.content,
+            sequence_number=self.sequence_number,
+            min_sequence_number=self.min_sequence_number,
+            added_snapshot_id=snapshot_id,
+            added_files_count=self.added_files_count,
+            existing_files_count=self.existing_files_count,
+            deleted_files_count=self.deleted_files_count,
+            added_rows_count=self.added_rows_count,
+            existing_rows_count=self.existing_rows_count,
+            deleted_rows_count=self.deleted_rows_count,
+            partitions=self.partitions,
+            key_metadata=self.key_metadata,
+        )
     def fetch_manifest_entry(self, io: FileIO, discard_deleted: bool = True) -> List[ManifestEntry]:
         """
         Read the manifest entries from the manifest file.
@@ -635,7 +653,9 @@ def fetch_manifest_entry(self, io: FileIO, discard_deleted: bool = True) -> List
                 for entry in reader
                 if not discard_deleted or entry.status != ManifestEntryStatus.DELETED
             ]
-
+    def __hash__(self) -> int:
+        """Return the hash of the file path."""
+        return hash(self.manifest_path)
 
 @cached(cache=LRUCache(maxsize=128), key=lambda io, manifest_list: hashkey(manifest_list))
 def _manifests(io: FileIO, manifest_list: str) -> Tuple[ManifestFile, ...]:

pyiceberg/table/__init__.py

@@ -116,6 +116,7 @@
     ManageSnapshots,
     UpdateSnapshot,
     _FastAppendFiles,
+    _ManifestMergeManager,
 )
 from pyiceberg.table.update.spec import UpdateSpec
 from pyiceberg.table.update.statistics import UpdateStatistics
@@ -135,7 +136,7 @@
 )
 from pyiceberg.utils.concurrent import ExecutorFactory
 from pyiceberg.utils.config import Config
-from pyiceberg.utils.properties import property_as_bool
+from pyiceberg.utils.properties import property_as_bool, property_as_int
 
 if TYPE_CHECKING:
     import daft
@@ -420,7 +421,29 @@ def update_snapshot(self, snapshot_properties: Dict[str, str] = EMPTY_DICT) -> U
             A new UpdateSnapshot
         """
         return UpdateSnapshot(self, io=self._table.io, snapshot_properties=snapshot_properties)
-
+    def rewrite_manifests(self) -> None:
+
+        # snapshot = self._table.current_snapshot()
+        # manifests = []
+        # for manifest in snapshot.manifests(self.io):
+        #     if manifest.content == ManifestContent.DATA:
+        #         manifests.append(manifest)
+
+        with self.update_snapshot().rewrite() as rewrite:
+            rewrite.commit()
+
+        # data_manifest_merge_manager = _ManifestMergeManager(
+        #     target_size_bytes=property_as_int(
+        #         self.properties,
+        #         TableProperties.MANIFEST_TARGET_SIZE_BYTES,
+        #         TableProperties.MANIFEST_TARGET_SIZE_BYTES_DEFAULT,
+        #     ),
+        #     min_count_to_merge=2,
+        #     merge_enabled=True,
+        #     snapshot_producer=self,
+        # )
+
+        # data_manifest_merge_manager.merge_manifests(manifests)
     def append(self, df: pa.Table, snapshot_properties: Dict[str, str] = EMPTY_DICT) -> None:
         """
         Shorthand API for appending a PyArrow table to a table transaction.
@@ -1084,6 +1107,9 @@ def name_mapping(self) -> Optional[NameMapping]:
         """Return the table's field-id NameMapping."""
         return self.metadata.name_mapping()
 
+    def rewrite_manifests(self) -> None:
+        with self.transaction() as tx:
+            tx.rewrite_manifests()
     def append(self, df: pa.Table, snapshot_properties: Dict[str, str] = EMPTY_DICT) -> None:
         """
         Shorthand API for appending a PyArrow table to the table.
@@ -1168,6 +1194,53 @@ def add_files(
                 file_paths=file_paths, snapshot_properties=snapshot_properties, check_duplicate_files=check_duplicate_files
             )
 
+    def rewrite_manifests(
+        self,
+        spec_id: Optional[int] = None,
+        rewrite_all: bool = False,
+        max_manifest_size: Optional[int] = None,
+    ) -> "Table":
+
+        with self.transaction() as tx:
+            tx.rewrite_manifests()
+        ...
+        """Rewrite manifests in the table.
+
+        Args:
+            spec_id: Spec ID to be used for the rewritten manifests
+            rewrite_all: If True, rewrite all manifests. If False, only rewrite small manifests
+            max_manifest_size: Target size for manifests in bytes
+
+        Returns:
+            An updated version of the table with rewritten manifests
+        #"""
+        # return RewriteManifests(
+        #     self,
+        #     spec_id=spec_id,
+        #     rewrite_all=rewrite_all,
+        #     max_manifest_size=max_manifest_size,
+        # ).commit()
+
+        # snapshot = self.current_snapshot()
+        # manifests = []
+        # for manifest in snapshot.manifests(self.io):
+        #     if manifest.content == ManifestContent.DATA:
+        #         manifests.append(manifest)
+        #
+        # data_manifest_merge_manager = _ManifestMergeManager(
+        #     target_size_bytes=property_as_int(
+        #         self.properties,
+        #         TableProperties.MANIFEST_TARGET_SIZE_BYTES,
+        #         TableProperties.MANIFEST_TARGET_SIZE_BYTES_DEFAULT,
+        #     ),
+        #     min_count_to_merge=2,
+        #     merge_enabled=True,
+        #     snapshot_producer=self,
+        # )
+        #
+        # data_manifest_merge_manager.merge_manifests(manifests)
+        # entries = self.inspect.entries().filter("status < 2").selectExpr("input_file_name() as manifest")
+
     def update_spec(self, case_sensitive: bool = True) -> UpdateSpec:
         return UpdateSpec(Transaction(self, autocommit=True), case_sensitive=case_sensitive)