This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3582-8e8f14459da3d1697fb37bf9514440c1554f5370 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
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";
