This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new ffdf043e1 feat(catalog): support
write.metadata.delete-after-commit.enabled (#3582)
ffdf043e1 is described below
commit ffdf043e182e0bb492b9b6f026d49b00991e48ba
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Mon Nov 9 19:41:44 2026 +0000
feat(catalog): support write.metadata.delete-after-commit.enabled (#3582)
* 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 <[email protected]>
* 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 <[email protected]>
---------
Co-authored-by: Isaac <[email protected]>
---
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 | 209 ++++++++++++++++++++++++++-
crates/iceberg/src/catalog/utils.rs | 43 +++++-
crates/iceberg/src/lib.rs | 2 +-
crates/iceberg/src/spec/table_properties.rs | 14 ++
8 files changed, 349 insertions(+), 8 deletions(-)
diff --git a/crates/catalog/glue/src/catalog.rs
b/crates/catalog/glue/src/catalog.rs
index 99fb2a51e..d6a05000f 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 7bef71a2b..fd836d4b2 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 e3104dff7..7aeba57c8 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 1c5af7cae..649356eec 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -3038,6 +3038,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
@@ -3093,6 +3095,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>
@@ -4182,5 +4185,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, [...]
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>
diff --git a/crates/iceberg/src/catalog/memory/catalog.rs
b/crates/iceberg/src/catalog/memory/catalog.rs
index dc5079a1a..b73a1d468 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<C: Catalog>(
+ catalog: &C,
+ properties: HashMap<String, String>,
+ ) -> 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<C: Catalog>(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<C: Catalog>(
+ catalog: &C,
+ mut table: Table,
+ num_commits: usize,
+ ) -> Vec<String> {
+ 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<String> {
+ 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 6e6997b21..3fb9b007d 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 72c2acd3e..4bc9e8895 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 f418d3a1d..392c5cf0a 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";
