feat: Honor write.metadata.path for metadata file locations

crates/catalog/glue/src/catalog.rs

@@ -615,7 +615,8 @@ impl Catalog for GlueCatalog {
         let metadata = TableMetadataBuilder::from_table_creation(creation)?
             .build()?
             .metadata;
-        let metadata_location = MetadataLocation::new_with_metadata(location.clone(), &metadata);
+        let metadata_location =
+            MetadataLocation::try_new_with_metadata(location.clone(), &metadata)?;
 
         metadata.write_to(&self.file_io, &metadata_location).await?;
 

crates/catalog/glue/src/utils.rs

@@ -356,7 +356,7 @@ mod tests {
 
         let metadata = create_metadata(schema)?;
         let metadata_location =
-            MetadataLocation::new_with_metadata(location, &metadata).to_string();
+            MetadataLocation::try_new_with_metadata(location, &metadata)?.to_string();
 
         let parameters = HashMap::from([
             (ICEBERG_FIELD_ID.to_string(), "1".to_string()),

crates/catalog/hms/src/catalog.rs

@@ -519,7 +519,8 @@ impl Catalog for HmsCatalog {
             .build()?
             .metadata;
 
-        let metadata_location = MetadataLocation::new_with_metadata(location.clone(), &metadata);
+        let metadata_location =
+            MetadataLocation::try_new_with_metadata(location.clone(), &metadata)?;
 
         metadata.write_to(&self.file_io, &metadata_location).await?;
 

crates/catalog/hms/src/utils.rs

@@ -362,7 +362,7 @@ mod tests {
             .build()?
             .metadata;
         let metadata_location =
-            MetadataLocation::new_with_metadata(location.clone(), &metadata).to_string();
+            MetadataLocation::try_new_with_metadata(location.clone(), &metadata)?.to_string();
 
         let result = convert_to_hive_table(
             db_name.clone(),

crates/catalog/s3tables/src/catalog.rs

@@ -553,7 +553,7 @@ impl Catalog for S3TablesCatalog {
         let metadata = TableMetadataBuilder::from_table_creation(creation)?
             .build()?
             .metadata;
-        let metadata_location = MetadataLocation::new_with_metadata(table_location, &metadata);
+        let metadata_location = MetadataLocation::try_new_with_metadata(table_location, &metadata)?;
         metadata.write_to(&self.file_io, &metadata_location).await?;
 
         // update metadata location

crates/catalog/sql/src/catalog.rs

@@ -876,7 +876,7 @@ impl Catalog for SqlCatalog {
             .build()?
             .metadata;
         let tbl_metadata_location =
-            MetadataLocation::new_with_metadata(location.clone(), &tbl_metadata);
+            MetadataLocation::try_new_with_metadata(location.clone(), &tbl_metadata)?;
 
         tbl_metadata
             .write_to(&self.fileio, &tbl_metadata_location)

crates/iceberg/public-api.txt

@@ -2608,6 +2608,7 @@ pub fn iceberg::spec::TableMetadata::last_updated_ms(&self) -> i64
 pub fn iceberg::spec::TableMetadata::last_updated_timestamp(&self) -> iceberg::Result<chrono::datetime::DateTime<chrono::offset::utc::Utc>>
 pub fn iceberg::spec::TableMetadata::location(&self) -> &str
 pub fn iceberg::spec::TableMetadata::metadata_compression_codec(&self) -> iceberg::Result<iceberg::compression::CompressionCodec>
+pub fn iceberg::spec::TableMetadata::metadata_location(&self) -> iceberg::Result<alloc::string::String>
 pub fn iceberg::spec::TableMetadata::metadata_log(&self) -> &[iceberg::spec::MetadataLog]
 pub fn iceberg::spec::TableMetadata::next_row_id(&self) -> u64
 pub fn iceberg::spec::TableMetadata::next_sequence_number(&self) -> i64
@@ -2711,6 +2712,7 @@ pub iceberg::spec::TableProperties::metadata_compression_codec: iceberg::compres
 pub iceberg::spec::TableProperties::min_snapshots_to_keep: usize
 pub iceberg::spec::TableProperties::write_datafusion_fanout_enabled: bool
 pub iceberg::spec::TableProperties::write_format_default: alloc::string::String
+pub iceberg::spec::TableProperties::write_metadata_path: core::option::Option<alloc::string::String>
 pub iceberg::spec::TableProperties::write_target_file_size_bytes: usize
 impl iceberg::spec::TableProperties
 pub const iceberg::spec::TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS: &str
@@ -2758,6 +2760,7 @@ pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL: &str
 pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT: i32
 pub const iceberg::spec::TableProperties::PROPERTY_SNAPSHOT_COUNT: &str
 pub const iceberg::spec::TableProperties::PROPERTY_UUID: &str
+pub const iceberg::spec::TableProperties::PROPERTY_WRITE_METADATA_PATH: &str
 pub const iceberg::spec::TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT: &str
 pub const iceberg::spec::TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT: u64
 pub const iceberg::spec::TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES: &str
@@ -3504,9 +3507,8 @@ pub fn iceberg::memory::MemoryCatalog::update_table<'life0, 'async_trait>(&'life
 pub struct iceberg::MetadataLocation
 impl iceberg::MetadataLocation
 pub fn iceberg::MetadataLocation::compression_codec(&self) -> iceberg::compression::CompressionCodec
-pub fn iceberg::MetadataLocation::new_with_metadata(table_location: impl alloc::string::ToString, metadata: &iceberg::spec::TableMetadata) -> Self
-pub fn iceberg::MetadataLocation::new_with_table_location(table_location: impl alloc::string::ToString) -> Self
-pub fn iceberg::MetadataLocation::with_new_metadata(&self, new_metadata: &iceberg::spec::TableMetadata) -> Self
+pub fn iceberg::MetadataLocation::try_new_with_metadata(table_location: impl alloc::string::ToString, metadata: &iceberg::spec::TableMetadata) -> iceberg::Result<Self>
+pub fn iceberg::MetadataLocation::try_with_new_metadata(&self, new_metadata: &iceberg::spec::TableMetadata) -> iceberg::Result<Self>
 pub fn iceberg::MetadataLocation::with_next_version(&self) -> Self
 impl core::clone::Clone for iceberg::MetadataLocation
 pub fn iceberg::MetadataLocation::clone(&self) -> iceberg::MetadataLocation

crates/iceberg/src/catalog/memory/catalog.rs

@@ -309,7 +309,7 @@ impl Catalog for MemoryCatalog {
         let metadata = TableMetadataBuilder::from_table_creation(table_creation)?
             .build()?
             .metadata;
-        let metadata_location = MetadataLocation::new_with_metadata(location, &metadata);
+        let metadata_location = MetadataLocation::try_new_with_metadata(location, &metadata)?;
 
         metadata.write_to(&self.file_io, &metadata_location).await?;