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..b73a1d468e 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,200 @@ 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_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")] { + 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..3fb9b007d6 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 `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 +/// 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";