Implement write.metadata.delete-after-commit.enabled to clean up old metadata files

pyiceberg/catalog/__init__.py

@@ -55,6 +55,7 @@
     CreateTableTransaction,
     StagedTable,
     Table,
+    TableProperties,
 )
 from pyiceberg.table.metadata import TableMetadata, TableMetadataV1, new_table_metadata
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
@@ -72,6 +73,7 @@
 from pyiceberg.utils.config import Config, merge_config
 from pyiceberg.utils.deprecated import deprecated as deprecated
 from pyiceberg.utils.deprecated import deprecation_message
+from pyiceberg.utils.properties import property_as_bool
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -859,14 +861,20 @@ def _update_and_stage_table(
             metadata_location=current_table.metadata_location if current_table else None,
         )
 
+        # https://github.com/apache/iceberg/blob/f6faa58/core/src/main/java/org/apache/iceberg/CatalogUtil.java#L527
+        # delete old metadata if METADATA_DELETE_AFTER_COMMIT_ENABLED is set to true
+        io = self._load_file_io(properties=updated_metadata.properties, location=updated_metadata.location)
+        if current_table is not None:
+            self._delete_old_metadata(io, current_table.metadata, updated_metadata)
+
         new_metadata_version = self._parse_metadata_version(current_table.metadata_location) + 1 if current_table else 0
         new_metadata_location = self._get_metadata_location(updated_metadata.location, new_metadata_version)
 
         return StagedTable(
             identifier=table_identifier,
             metadata=updated_metadata,
             metadata_location=new_metadata_location,
-            io=self._load_file_io(properties=updated_metadata.properties, location=new_metadata_location),
+            io=io,
             catalog=self,
         )
 
@@ -913,6 +921,20 @@ def _get_default_warehouse_location(self, database_name: str, table_name: str) -
 
         raise ValueError("No default path is set, please specify a location when creating a table")
 
+    def _delete_old_metadata(self, io: FileIO, base: TableMetadata, metadata: TableMetadata) -> None:
+        """Delete oldest metadata if config is set to true."""
+        delete_after_commit: bool = property_as_bool(
+            metadata.properties,
+            TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED,
+            TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED_DEFAULT,
+        )
+
+        if delete_after_commit:
+            removed_previous_metadata_files: set[str] = {log.metadata_file for log in base.metadata_log}
+            current_metadata_files: set[str] = {log.metadata_file for log in metadata.metadata_log}
+            removed_previous_metadata_files.difference_update(current_metadata_files)
+            delete_files(io, removed_previous_metadata_files, METADATA)
+
     @staticmethod
     def _write_metadata(metadata: TableMetadata, io: FileIO, metadata_path: str) -> None:
         ToOutputFile.table_metadata(metadata, io.new_output(metadata_path))

pyiceberg/table/__init__.py

@@ -217,6 +217,9 @@ class TableProperties:
     METADATA_PREVIOUS_VERSIONS_MAX = "write.metadata.previous-versions-max"
     METADATA_PREVIOUS_VERSIONS_MAX_DEFAULT = 100
 
+    METADATA_DELETE_AFTER_COMMIT_ENABLED = "write.metadata.delete-after-commit.enabled"
+    METADATA_DELETE_AFTER_COMMIT_ENABLED_DEFAULT = False
+
     MAX_SNAPSHOT_AGE_MS = "history.expire.max-snapshot-age-ms"
     MAX_SNAPSHOT_AGE_MS_DEFAULT = 5 * 24 * 60 * 60 * 1000  # 5 days
 

tests/catalog/test_base.py

@@ -141,6 +141,10 @@ def commit_table(
             requirement.validate(base_metadata)
 
         updated_metadata = update_table_metadata(base_metadata, updates)
+
+        # delete old metadata
+        self._delete_old_metadata(current_table.io, base_metadata, updated_metadata)
+
         if updated_metadata == base_metadata:
             # no changes, do nothing
             return CommitTableResponse(metadata=base_metadata, metadata_location=current_table.metadata_location)

tests/catalog/test_sql.py

@@ -50,6 +50,7 @@
 from pyiceberg.io.pyarrow import _dataframe_to_data_files, schema_to_pyarrow
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC
 from pyiceberg.schema import Schema
+from pyiceberg.table import TableProperties
 from pyiceberg.table.snapshots import Operation
 from pyiceberg.table.sorting import (
     NullOrder,
@@ -1613,3 +1614,60 @@ def test_merge_manifests_local_file_system(catalog: SqlCatalog, arrow_table_with
         tbl.append(arrow_table_with_null)
 
     assert len(tbl.scan().to_arrow()) == 5 * len(arrow_table_with_null)
+
+
+@pytest.mark.parametrize(
+    "catalog",
+    [
+        lazy_fixture("catalog_memory"),
+        lazy_fixture("catalog_sqlite"),
+        lazy_fixture("catalog_sqlite_without_rowcount"),
+    ],
+)
+@pytest.mark.parametrize(
+    "table_identifier",
+    [
+        lazy_fixture("random_table_identifier"),
+        lazy_fixture("random_hierarchical_identifier"),
+    ],
+)
+def test_delete_metadata_multiple(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
+    namespace = Catalog.namespace_from(table_identifier)
+    catalog.create_namespace(namespace)
+    table = catalog.create_table(table_identifier, table_schema_nested)
+
+    original_metadata_location = table.metadata_location
+
+    for i in range(5):
+        transaction = table.transaction()
+        update = transaction.update_schema()
+        update.add_column(path=f"new_column_{i}", field_type=IntegerType())
+        update.commit()
+        transaction.commit_transaction()
+
+    assert len(table.metadata.metadata_log) == 5
+    assert os.path.exists(original_metadata_location[len("file://") :])
+
+    # Set the max versions property to 2, and delete after commit
+    new_property = {
+        TableProperties.METADATA_PREVIOUS_VERSIONS_MAX: "2",
+        TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED: "true",
+    }
+
+    transaction = table.transaction()
+    transaction.set_properties(properties=new_property).commit_transaction()
+
+    # Verify that only the most recent metadata files are kept
+    assert len(table.metadata.metadata_log) == 2
+    updated_metadata_1, updated_metadata_2 = table.metadata.metadata_log
+
+    transaction = table.transaction()
+    update = transaction.update_schema()
+    update.add_column(path="new_column_x", field_type=IntegerType())
+    update.commit()
+    transaction.commit_transaction()
+
+    assert len(table.metadata.metadata_log) == 2
+    assert not os.path.exists(original_metadata_location[len("file://") :])
+    assert not os.path.exists(updated_metadata_1.metadata_file[len("file://") :])
+    assert os.path.exists(updated_metadata_2.metadata_file[len("file://") :])