From 4ad0c29a9b1647a05472ce97a334c6a5cc2d5a70 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Thu, 29 Oct 2026 16:53:46 -0700 Subject: [PATCH 1/2] feat(catalog): support write.metadata.delete-after-commit.enabled Add the `write.metadata.delete-after-commit.enabled` table property (default false). When it is set, catalogs that write metadata files themselves delete the previous metadata files that fall out of the metadata log (bounded by `write.metadata.previous-versions-max`) after a successful commit, matching Java's `CatalogUtil.deleteRemovedMetadataFiles`. The shared `delete_removed_metadata_files` helper diffs the base and committed metadata logs and deletes best-effort, logging failures without failing the commit. It is wired into the memory, SQL and Glue catalogs. REST and S3 Tables are left out because the server or service owns metadata file cleanup, and HMS does not support commits yet. Co-authored-by: Isaac --- crates/catalog/glue/src/catalog.rs | 10 +- crates/catalog/loader/tests/table_suite.rs | 65 ++++++- crates/catalog/sql/src/catalog.rs | 10 +- crates/iceberg/public-api.txt | 4 + crates/iceberg/src/catalog/memory/catalog.rs | 170 ++++++++++++++++++- crates/iceberg/src/catalog/utils.rs | 43 ++++- crates/iceberg/src/lib.rs | 2 +- crates/iceberg/src/spec/table_properties.rs | 14 ++ 8 files changed, 310 insertions(+), 8 deletions(-) diff --git a/crates/catalog/glue/src/catalog.rs b/crates/catalog/glue/src/catalog.rs index 99fb2a51e9..d6a05000f9 100644 --- a/crates/catalog/glue/src/catalog.rs +++ b/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) } } diff --git a/crates/catalog/loader/tests/table_suite.rs b/crates/catalog/loader/tests/table_suite.rs index 7bef71a2b0..fd836d4b25 100644 --- a/crates/catalog/loader/tests/table_suite.rs +++ b/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)] diff --git a/crates/catalog/sql/src/catalog.rs b/crates/catalog/sql/src/catalog.rs index e3104dff76..7aeba57c89 100644 --- a/crates/catalog/sql/src/catalog.rs +++ b/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) } } diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 92267f0a99..6217aaf2b5 100644 --- a/crates/iceberg/public-api.txt +++ b/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 pub fn iceberg::spec::TableProperties<'properties>::max_snapshot_age_ms(&self) -> iceberg::Result pub fn iceberg::spec::TableProperties<'properties>::metadata_compression_codec(&self) -> iceberg::Result +pub fn iceberg::spec::TableProperties<'properties>::metadata_delete_after_commit_enabled(&self) -> iceberg::Result pub fn iceberg::spec::TableProperties<'properties>::min_snapshots_to_keep(&self) -> iceberg::Result pub fn iceberg::spec::TableProperties<'properties>::new(properties: &'properties std::collections::hash::map::HashMap) -> Self pub fn iceberg::spec::TableProperties<'properties>::parquet_compression_codec(&self) -> iceberg::Result @@ -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> + 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) -> core::pin::Pin> + 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> + 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 = core::result::Result diff --git a/crates/iceberg/src/catalog/memory/catalog.rs b/crates/iceberg/src/catalog/memory/catalog.rs index dc5079a1ae..0963d90baf 100644 --- a/crates/iceberg/src/catalog/memory/catalog.rs +++ b/crates/iceberg/src/catalog/memory/catalog.rs @@ -35,7 +35,7 @@ use crate::spec::{TableMetadata, TableMetadataBuilder}; use crate::table::Table; use crate::{ Catalog, CatalogBuilder, MetadataLocation, Namespace, NamespaceIdent, Result, TableCommit, - TableCreation, TableIdent, + TableCreation, TableIdent, delete_removed_metadata_files, }; /// Memory catalog warehouse location @@ -406,6 +406,7 @@ impl Catalog for MemoryCatalog { let current_table = self .load_table_from_locked_state(commit.identifier(), &root_namespace_state) .await?; + let base_metadata = current_table.metadata_ref(); // Apply TableCommit to get staged table let staged_table = commit.apply(current_table)?; @@ -420,6 +421,14 @@ impl Catalog for MemoryCatalog { // Flip the pointer to reference the new metadata file. let updated_table = root_namespace_state.commit_table_update(staged_table)?; + drop(root_namespace_state); + + delete_removed_metadata_files( + updated_table.file_io(), + &base_metadata, + updated_table.metadata(), + ) + .await; Ok(updated_table) } @@ -439,7 +448,9 @@ pub(crate) mod tests { use crate::ErrorKind; use crate::encryption::kms::MemoryKmsClientFactory; use crate::io::{FileIO, LocalFsStorageFactory}; - use crate::spec::{NestedField, PartitionSpec, PrimitiveType, Schema, SortOrder, Type}; + use crate::spec::{ + NestedField, PartitionSpec, PrimitiveType, Schema, SortOrder, TableProperties, Type, + }; use crate::test_utils::test_runtime; use crate::transaction::{ApplyTransactionAction, Transaction}; @@ -1983,6 +1994,161 @@ pub(crate) mod tests { ); } + async fn create_table_with_properties( + catalog: &C, + properties: HashMap, + ) -> Table { + let namespace_ident = NamespaceIdent::new("abc".into()); + create_namespace(catalog, &namespace_ident).await; + catalog + .create_table( + &namespace_ident, + TableCreation::builder() + .name("test".into()) + .schema(simple_table_schema()) + .properties(properties) + .build(), + ) + .await + .unwrap() + } + + async fn commit_property_update(catalog: &C, table: &Table, value: &str) -> Table { + let tx = Transaction::new(table); + tx.update_table_properties() + .set("key".to_string(), value.to_string()) + .apply(tx) + .unwrap() + .commit(catalog) + .await + .unwrap() + } + + /// Commits `num_commits` property updates and returns the metadata location of every + /// table version, oldest first. + async fn commit_metadata_versions( + catalog: &C, + mut table: Table, + num_commits: usize, + ) -> Vec { + let mut locations = vec![table.metadata_location().unwrap().to_string()]; + for i in 0..num_commits { + table = commit_property_update(catalog, &table, &i.to_string()).await; + locations.push(table.metadata_location().unwrap().to_string()); + } + locations + } + + async fn existing_files(file_io: &FileIO, locations: &[String]) -> Vec { + let mut existing = Vec::new(); + for location in locations { + if file_io.exists(location).await.unwrap() { + existing.push(location.clone()); + } + } + existing + } + + #[tokio::test] + async fn test_update_table_deletes_metadata_files_removed_from_log_when_enabled() { + let catalog = new_memory_catalog().await; + let table = create_table_with_properties( + &catalog, + 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(), + ), + ]), + ) + .await; + let file_io = table.file_io().clone(); + + let locations = commit_metadata_versions(&catalog, table, 3).await; + + // The current metadata file and the one entry left in the metadata log remain. + assert_eq!( + existing_files(&file_io, &locations).await, + locations[2..].to_vec() + ); + } + + #[tokio::test] + async fn test_update_table_keeps_metadata_files_removed_from_log_when_not_enabled() { + for delete_after_commit in [None, Some("false")] { + let catalog = new_memory_catalog().await; + let mut properties = HashMap::from([( + TableProperties::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX.to_string(), + "1".to_string(), + )]); + if let Some(value) = delete_after_commit { + properties.insert( + TableProperties::PROPERTY_METADATA_DELETE_AFTER_COMMIT_ENABLED.to_string(), + value.to_string(), + ); + } + let table = create_table_with_properties(&catalog, properties).await; + let file_io = table.file_io().clone(); + + let locations = commit_metadata_versions(&catalog, table, 3).await; + + assert_eq!(existing_files(&file_io, &locations).await, locations); + } + } + + #[tokio::test] + async fn test_update_table_succeeds_when_deleting_removed_metadata_file_fails() { + let warehouse = TempDir::new().unwrap(); + let catalog = MemoryCatalogBuilder::default() + .with_storage_factory(Arc::new(LocalFsStorageFactory)) + .load( + "memory", + HashMap::from([( + MEMORY_CATALOG_WAREHOUSE.to_string(), + warehouse.path().to_str().unwrap().to_string(), + )]), + ) + .await + .unwrap(); + let table = create_table_with_properties( + &catalog, + 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(), + ), + ]), + ) + .await; + let file_io = table.file_io().clone(); + let v0 = table.metadata_location().unwrap().to_string(); + let v1_table = commit_property_update(&catalog, &table, "1").await; + let v1 = v1_table.metadata_location().unwrap().to_string(); + + // A non-empty directory in place of v0 makes its deletion fail. + std::fs::remove_file(&v0).unwrap(); + std::fs::create_dir(&v0).unwrap(); + std::fs::write(format!("{v0}/blocker"), b"").unwrap(); + + let v2_table = commit_property_update(&catalog, &v1_table, "2").await; + let v2 = v2_table.metadata_location().unwrap().to_string(); + let v3_table = commit_property_update(&catalog, &v2_table, "3").await; + let v3 = v3_table.metadata_location().unwrap().to_string(); + + assert_eq!( + existing_files(&file_io, &[v0.clone(), v1, v2.clone(), v3.clone()]).await, + vec![v0, v2, v3] + ); + } + #[tokio::test] async fn test_update_table_fails_if_table_doesnt_exist() { let catalog = new_memory_catalog().await; diff --git a/crates/iceberg/src/catalog/utils.rs b/crates/iceberg/src/catalog/utils.rs index 6e6997b214..5b3df6cea6 100644 --- a/crates/iceberg/src/catalog/utils.rs +++ b/crates/iceberg/src/catalog/utils.rs @@ -19,11 +19,11 @@ use std::collections::{HashMap, HashSet}; -use futures::{TryStreamExt, stream}; +use futures::{StreamExt, TryStreamExt, stream}; use crate::Result; use crate::io::FileIO; -use crate::spec::{ManifestFile, ManifestReader}; +use crate::spec::{ManifestFile, ManifestReader, TableMetadata}; use crate::table::Table; const DELETE_CONCURRENCY: usize = 10; @@ -104,6 +104,45 @@ pub async fn drop_table_data(table_info: &Table) -> Result<()> { Ok(()) } +/// Deletes the previous metadata files tracked by `base` that `new` no longer tracks, if +/// `write.metadata.delete-after-commit.enabled` is set on `new`. +/// +/// This mirrors the Java implementation's `CatalogUtil.deleteRemovedMetadataFiles`. Call it +/// only after `new` has been committed over `base`. Deletion is best-effort: failures are +/// logged rather than returned, since the commit has already succeeded. +pub async fn delete_removed_metadata_files(io: &FileIO, base: &TableMetadata, new: &TableMetadata) { + match new + .table_properties() + .metadata_delete_after_commit_enabled() + { + Ok(true) => {} + Ok(false) => return, + Err(e) => { + tracing::warn!("Skipping deletion of removed metadata files: {e}"); + return; + } + } + + let retained: HashSet<&str> = new + .metadata_log() + .iter() + .map(|log| log.metadata_file.as_str()) + .collect(); + let removed = base + .metadata_log() + .iter() + .map(|log| log.metadata_file.as_str()) + .filter(|path| !retained.contains(path)); + + stream::iter(removed) + .for_each_concurrent(DELETE_CONCURRENCY, |path| async move { + if let Err(e) = io.delete(path).await { + tracing::warn!("Failed to delete removed metadata file {path}: {e}"); + } + }) + .await; +} + /// Reads manifests concurrently and deletes the data files referenced within. async fn delete_data_files( io: &FileIO, diff --git a/crates/iceberg/src/lib.rs b/crates/iceberg/src/lib.rs index 72c2acd3e6..4bc9e8895b 100644 --- a/crates/iceberg/src/lib.rs +++ b/crates/iceberg/src/lib.rs @@ -75,7 +75,7 @@ pub mod sensitive; mod catalog; -pub use catalog::utils::drop_table_data; +pub use catalog::utils::{delete_removed_metadata_files, drop_table_data}; pub use catalog::*; pub mod table; diff --git a/crates/iceberg/src/spec/table_properties.rs b/crates/iceberg/src/spec/table_properties.rs index f418d3a1de..392c5cf0af 100644 --- a/crates/iceberg/src/spec/table_properties.rs +++ b/crates/iceberg/src/spec/table_properties.rs @@ -331,6 +331,13 @@ pub struct TableProperties { getter )] write_object_storage_partitioned_paths: bool, + /// Whether to delete the oldest tracked metadata files after each commit. + #[property( + key = Self::PROPERTY_METADATA_DELETE_AFTER_COMMIT_ENABLED, + default = Self::PROPERTY_METADATA_DELETE_AFTER_COMMIT_ENABLED_DEFAULT, + getter + )] + metadata_delete_after_commit_enabled: bool, /// The table's default name mapping, used to assign field ids when reading data files /// that carry no field id metadata. `None` if `schema.name-mapping.default` is not set. #[property( @@ -376,6 +383,13 @@ impl TableProperties<'_> { /// Default value for max number of previous versions to keep. pub const PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX_DEFAULT: usize = 100; + /// Property key for deleting the previous metadata files that fall out of the metadata log + /// (bounded by [`TableProperties::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX`]) after a commit. + pub const PROPERTY_METADATA_DELETE_AFTER_COMMIT_ENABLED: &'static str = + "write.metadata.delete-after-commit.enabled"; + /// Default value for [`TableProperties::PROPERTY_METADATA_DELETE_AFTER_COMMIT_ENABLED`]. + pub const PROPERTY_METADATA_DELETE_AFTER_COMMIT_ENABLED_DEFAULT: bool = false; + /// Property key for max number of partitions to keep summary stats for. pub const PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT: &'static str = "write.summary.partition-limit"; From 7169ba7688fa31bda0bc1d38f0bac9c18afe12c7 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Fri, 6 Nov 2026 22:19:40 -0800 Subject: [PATCH 2/2] test(catalog): cover enabling delete-after-commit in the same commit Address review: test that cleanup reads the property from the new metadata, and say the property must be true in the helper's doc. Co-authored-by: Isaac --- crates/iceberg/src/catalog/memory/catalog.rs | 39 ++++++++++++++++++++ crates/iceberg/src/catalog/utils.rs | 2 +- 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/crates/iceberg/src/catalog/memory/catalog.rs b/crates/iceberg/src/catalog/memory/catalog.rs index 0963d90baf..b73a1d468e 100644 --- a/crates/iceberg/src/catalog/memory/catalog.rs +++ b/crates/iceberg/src/catalog/memory/catalog.rs @@ -2077,6 +2077,45 @@ pub(crate) mod tests { ); } + #[tokio::test] + async fn test_update_table_deletes_removed_metadata_files_on_commit_that_enables_cleanup() { + let catalog = new_memory_catalog().await; + let table = create_table_with_properties( + &catalog, + HashMap::from([( + TableProperties::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX.to_string(), + "1".to_string(), + )]), + ) + .await; + let file_io = table.file_io().clone(); + let v0 = table.metadata_location().unwrap().to_string(); + let v1_table = commit_property_update(&catalog, &table, "1").await; + let v1 = v1_table.metadata_location().unwrap().to_string(); + let v2_table = commit_property_update(&catalog, &v1_table, "2").await; + let v2 = v2_table.metadata_location().unwrap().to_string(); + + let tx = Transaction::new(&v2_table); + let v3_table = tx + .update_table_properties() + .set( + TableProperties::PROPERTY_METADATA_DELETE_AFTER_COMMIT_ENABLED.to_string(), + "true".to_string(), + ) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + let v3 = v3_table.metadata_location().unwrap().to_string(); + + // v0 fell out of the log while cleanup was disabled, so only v1 is deleted. + assert_eq!( + existing_files(&file_io, &[v0.clone(), v1, v2.clone(), v3.clone()]).await, + vec![v0, v2, v3] + ); + } + #[tokio::test] async fn test_update_table_keeps_metadata_files_removed_from_log_when_not_enabled() { for delete_after_commit in [None, Some("false")] { diff --git a/crates/iceberg/src/catalog/utils.rs b/crates/iceberg/src/catalog/utils.rs index 5b3df6cea6..3fb9b007d6 100644 --- a/crates/iceberg/src/catalog/utils.rs +++ b/crates/iceberg/src/catalog/utils.rs @@ -105,7 +105,7 @@ pub async fn drop_table_data(table_info: &Table) -> Result<()> { } /// Deletes the previous metadata files tracked by `base` that `new` no longer tracks, if -/// `write.metadata.delete-after-commit.enabled` is set on `new`. +/// `write.metadata.delete-after-commit.enabled` is `true` on `new`. /// /// This mirrors the Java implementation's `CatalogUtil.deleteRemovedMetadataFiles`. Call it /// only after `new` has been committed over `base`. Deletion is best-effort: failures are