feat(catalog): support write.metadata.delete-after-commit.enabled

crates/catalog/glue/src/catalog.rs

@@ -34,7 +34,7 @@ use iceberg::spec::{TableMetadata, TableMetadataBuilder};
 use iceberg::table::Table;
 use iceberg::{
     Catalog, CatalogBuilder, Error, ErrorKind, MetadataLocation, Namespace, NamespaceIdent, Result,
-    Runtime, TableCommit, TableCreation, TableIdent,
+    Runtime, TableCommit, TableCreation, TableIdent, delete_removed_metadata_files,
 };
 use iceberg_property_macro::Properties;
 use iceberg_storage_opendal::OpenDalStorageFactory;
@@ -871,6 +871,7 @@ impl Catalog for GlueCatalog {
         let (current_table, current_version_id) =
             self.load_table_with_version_id(&table_ident).await?;
         let current_metadata_location = current_table.metadata_location_result()?.to_string();
+        let base_metadata = current_table.metadata_ref();
 
         let staged_table = commit.apply(current_table)?;
         let staged_metadata_location_str = staged_table.metadata_location_result()?;
@@ -923,6 +924,13 @@ impl Catalog for GlueCatalog {
             .with_source(anyhow!("aws sdk error: {error:?}"))
         })?;
 
+        delete_removed_metadata_files(
+            staged_table.file_io(),
+            &base_metadata,
+            staged_table.metadata(),
+        )
+        .await;
+
         Ok(staged_table)
     }
 }

crates/catalog/loader/tests/table_suite.rs

@@ -24,7 +24,7 @@ mod common;
 use std::collections::HashMap;
 
 use common::{CatalogKind, TableMode, cleanup_namespace_dyn, load_catalog, table_creation};
-use iceberg::spec::{DataContentType, DataFileBuilder, DataFileFormat, Struct};
+use iceberg::spec::{DataContentType, DataFileBuilder, DataFileFormat, Struct, TableProperties};
 use iceberg::transaction::{ApplyTransactionAction, Transaction};
 use iceberg::{ErrorKind, NamespaceIdent, Result, TableIdent};
 use iceberg_test_utils::normalize_test_name_with_parts;
@@ -211,6 +211,69 @@ async fn test_catalog_update_table_properties(#[case] kind: CatalogKind) -> Resu
     Ok(())
 }
 
+// Common behavior: catalogs that write metadata files delete the ones that fall out of the
+// metadata log when `write.metadata.delete-after-commit.enabled` is set.
+#[rstest]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_update_table_deletes_metadata_files_removed_from_log(
+    #[case] kind: CatalogKind,
+) -> Result<()> {
+    let Some(harness) = load_catalog(kind).await else {
+        return Ok(());
+    };
+    let catalog = harness.catalog;
+    let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+        "catalog_update_table_deletes_metadata_files",
+        harness.label
+    ));
+
+    cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+    catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+    let table_name = normalize_test_name_with_parts!(
+        "catalog_update_table_deletes_metadata_files",
+        harness.label,
+        "table"
+    );
+    let mut creation = table_creation(table_name);
+    creation.properties = HashMap::from([
+        (
+            TableProperties::PROPERTY_METADATA_DELETE_AFTER_COMMIT_ENABLED.to_string(),
+            "true".to_string(),
+        ),
+        (
+            TableProperties::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX.to_string(),
+            "1".to_string(),
+        ),
+    ]);
+    let mut table = catalog.create_table(&namespace, creation).await?;
+
+    let mut locations = vec![table.metadata_location_result()?.to_string()];
+    for i in 0..3 {
+        let tx = Transaction::new(&table);
+        let tx = tx
+            .update_table_properties()
+            .set("test_property".to_string(), i.to_string())
+            .apply(tx)?;
+        table = tx.commit(catalog.as_ref()).await?;
+        locations.push(table.metadata_location_result()?.to_string());
+    }
+
+    let mut existing = Vec::new();
+    for location in &locations {
+        if table.file_io().exists(location).await? {
+            existing.push(location.clone());
+        }
+    }
+    // The current metadata file and the one entry left in the metadata log remain.
+    assert_eq!(existing, locations[2..].to_vec());
+
+    Ok(())
+}
+
 // Common behavior: update_table_properties is rejected when unsupported.
 #[rstest]
 #[case::hms_catalog(CatalogKind::Hms)]

crates/catalog/sql/src/catalog.rs

@@ -27,7 +27,7 @@ use iceberg::spec::{TableMetadata, TableMetadataBuilder};
 use iceberg::table::Table;
 use iceberg::{
     Catalog, CatalogBuilder, Error, ErrorKind, MetadataLocation, Namespace, NamespaceIdent, Result,
-    Runtime, TableCommit, TableCreation, TableIdent,
+    Runtime, TableCommit, TableCreation, TableIdent, delete_removed_metadata_files,
 };
 use iceberg_property_macro::Properties;
 use sqlx::any::{AnyPoolOptions, AnyQueryResult, AnyRow, install_default_drivers};
@@ -1173,6 +1173,7 @@ impl Catalog for SqlCatalog {
         let table_ident = commit.identifier().clone();
         let current_table = self.load_table(&table_ident).await?;
         let current_metadata_location = current_table.metadata_location_result()?.to_string();
+        let base_metadata = current_table.metadata_ref();
 
         let staged_table = commit.apply(current_table)?;
         let staged_metadata_location_str = staged_table.metadata_location_result()?;
@@ -1216,6 +1217,13 @@ impl Catalog for SqlCatalog {
             .with_retryable(true));
         }
 
+        delete_removed_metadata_files(
+            staged_table.file_io(),
+            &base_metadata,
+            staged_table.metadata(),
+        )
+        .await;
+
         Ok(staged_table)
     }
 }

crates/iceberg/public-api.txt

@@ -3035,6 +3035,8 @@ pub const iceberg::spec::TableProperties<'_>::PROPERTY_MAX_SNAPSHOT_AGE_MS: &'st
 pub const iceberg::spec::TableProperties<'_>::PROPERTY_MAX_SNAPSHOT_AGE_MS_DEFAULT: i64
 pub const iceberg::spec::TableProperties<'_>::PROPERTY_METADATA_COMPRESSION_CODEC: &'static str
 pub const iceberg::spec::TableProperties<'_>::PROPERTY_METADATA_COMPRESSION_CODEC_DEFAULT: &'static str
+pub const iceberg::spec::TableProperties<'_>::PROPERTY_METADATA_DELETE_AFTER_COMMIT_ENABLED: &'static str
+pub const iceberg::spec::TableProperties<'_>::PROPERTY_METADATA_DELETE_AFTER_COMMIT_ENABLED_DEFAULT: bool
 pub const iceberg::spec::TableProperties<'_>::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX: &'static str
 pub const iceberg::spec::TableProperties<'_>::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX_DEFAULT: usize
 pub const iceberg::spec::TableProperties<'_>::PROPERTY_MIN_SNAPSHOTS_TO_KEEP: &'static str
@@ -3090,6 +3092,7 @@ pub fn iceberg::spec::TableProperties<'properties>::gc_enabled(&self) -> iceberg
 pub fn iceberg::spec::TableProperties<'properties>::max_ref_age_ms(&self) -> iceberg::Result<i64>
 pub fn iceberg::spec::TableProperties<'properties>::max_snapshot_age_ms(&self) -> iceberg::Result<i64>
 pub fn iceberg::spec::TableProperties<'properties>::metadata_compression_codec(&self) -> iceberg::Result<iceberg::compression::CompressionCodec>
+pub fn iceberg::spec::TableProperties<'properties>::metadata_delete_after_commit_enabled(&self) -> iceberg::Result<bool>
 pub fn iceberg::spec::TableProperties<'properties>::min_snapshots_to_keep(&self) -> iceberg::Result<usize>
 pub fn iceberg::spec::TableProperties<'properties>::new(properties: &'properties std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> Self
 pub fn iceberg::spec::TableProperties<'properties>::parquet_compression_codec(&self) -> iceberg::Result<iceberg::compression::CompressionCodec>
@@ -4179,5 +4182,6 @@ pub fn iceberg::SessionCatalog::rename_table<'life0, 'life1, 'life2, 'life3, 'as
 pub fn iceberg::SessionCatalog::table_exists<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, table: &'life2 iceberg::TableIdent) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<bool>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg::SessionCatalog::update_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, namespace: &'life2 iceberg::NamespaceIdent, properties: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg::SessionCatalog::update_table<'life0, 'life1, 'async_trait>(&'life0 self, context: &'life1 iceberg::SessionContext, commit: iceberg::TableCommit) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::table::Table>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait
+pub async fn iceberg::delete_removed_metadata_files(io: &iceberg::io::FileIO, base: &iceberg::spec::TableMetadata, new: &iceberg::spec::TableMetadata)
 pub async fn iceberg::drop_table_data(table_info: &iceberg::table::Table) -> iceberg::Result<()>
 pub type iceberg::Result<T> = core::result::Result<T, iceberg::Error>