feat: Honor compression settings for metadata.json on write

crates/catalog/glue/src/catalog.rs

@@ -527,7 +527,7 @@ impl Catalog for GlueCatalog {
             .build()?
             .metadata;
         let metadata_location =
-            MetadataLocation::new_with_table_location(location.clone()).to_string();
+            MetadataLocation::new_with_metadata(location.clone(), &metadata).to_string();
 
         metadata.write_to(&self.file_io, &metadata_location).await?;
 

crates/catalog/glue/src/utils.rs

@@ -306,8 +306,6 @@ mod tests {
     fn test_convert_to_glue_table() -> Result<()> {
         let table_name = "my_table".to_string();
         let location = "s3a://warehouse/hive".to_string();
-        let metadata_location = MetadataLocation::new_with_table_location(location).to_string();
-        let properties = HashMap::new();
         let schema = Schema::builder()
             .with_schema_id(1)
             .with_fields(vec![
@@ -316,6 +314,8 @@ mod tests {
             .build()?;
 
         let metadata = create_metadata(schema)?;
+        let metadata_location =
+            MetadataLocation::new_with_metadata(location, &metadata).to_string();
 
         let parameters = HashMap::from([
             (ICEBERG_FIELD_ID.to_string(), "1".to_string()),
@@ -336,8 +336,13 @@ mod tests {
             .location(metadata.location())
             .build();
 
-        let result =
-            convert_to_glue_table(&table_name, metadata_location, &metadata, &properties, None)?;
+        let result = convert_to_glue_table(
+            &table_name,
+            metadata_location,
+            &metadata,
+            metadata.properties(),
+            None,
+        )?;
 
         assert_eq!(result.name(), &table_name);
         assert_eq!(result.description(), None);

crates/catalog/hms/src/catalog.rs

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

crates/catalog/hms/src/utils.rs

@@ -311,8 +311,8 @@ fn get_current_time() -> Result<i32> {
 
 #[cfg(test)]
 mod tests {
-    use iceberg::spec::{NestedField, PrimitiveType, Type};
-    use iceberg::{MetadataLocation, Namespace, NamespaceIdent};
+    use iceberg::spec::{NestedField, PrimitiveType, TableMetadataBuilder, Type};
+    use iceberg::{MetadataLocation, Namespace, NamespaceIdent, TableCreation};
 
     use super::*;
 
@@ -343,8 +343,6 @@ mod tests {
         let db_name = "my_db".to_string();
         let table_name = "my_table".to_string();
         let location = "s3a://warehouse/hms".to_string();
-        let metadata_location =
-            MetadataLocation::new_with_table_location(location.clone()).to_string();
         let properties = HashMap::new();
         let schema = Schema::builder()
             .with_schema_id(1)
@@ -354,6 +352,18 @@ mod tests {
             ])
             .build()?;
 
+        let table_creation = TableCreation::builder()
+            .name(table_name.clone())
+            .location(location.clone())
+            .schema(schema.clone())
+            .properties(properties.clone())
+            .build();
+        let metadata = TableMetadataBuilder::from_table_creation(table_creation)?
+            .build()?
+            .metadata;
+        let metadata_location =
+            MetadataLocation::new_with_metadata(location.clone(), &metadata).to_string();
+
         let result = convert_to_hive_table(
             db_name.clone(),
             &schema,

crates/catalog/s3tables/src/catalog.rs

@@ -450,7 +450,7 @@ impl Catalog for S3TablesCatalog {
 
         // prepare metadata location. the warehouse location is generated by s3tables catalog,
         // which looks like: s3://e6c9bf20-991a-46fb-kni5xs1q2yxi3xxdyxzjzigdeop1quse2b--table-s3
-        let metadata_location = match &creation.location {
+        let table_location = match &creation.location {
             Some(_) => {
                 return Err(Error::new(
                     ErrorKind::DataInvalid,
@@ -467,16 +467,17 @@ impl Catalog for S3TablesCatalog {
                     .send()
                     .await
                     .map_err(from_aws_sdk_error)?;
-                let warehouse_location = get_resp.warehouse_location().to_string();
-                MetadataLocation::new_with_table_location(warehouse_location).to_string()
+                get_resp.warehouse_location().to_string()
             }
         };
 
         // write metadata to file
-        creation.location = Some(metadata_location.clone());
+        creation.location = Some(table_location.clone());
         let metadata = TableMetadataBuilder::from_table_creation(creation)?
             .build()?
             .metadata;
+        let metadata_location =
+            MetadataLocation::new_with_metadata(table_location, &metadata).to_string();
         metadata.write_to(&self.file_io, &metadata_location).await?;
 
         // update metadata location

crates/catalog/sql/src/catalog.rs

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

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

@@ -18,6 +18,7 @@
 //! This module contains memory catalog implementation.
 
 use std::collections::HashMap;
+use std::str::FromStr;
 
 use async_trait::async_trait;
 use futures::lock::{Mutex, MutexGuard};
@@ -279,15 +280,15 @@ impl Catalog for MemoryCatalog {
         let metadata = TableMetadataBuilder::from_table_creation(table_creation)?
             .build()?
             .metadata;
-        let metadata_location = MetadataLocation::new_with_table_location(location).to_string();
+        let metadata_location = MetadataLocation::new_with_metadata(location, &metadata);
 
         metadata.write_to(&self.file_io, &metadata_location).await?;
 
-        root_namespace_state.insert_new_table(&table_ident, metadata_location.clone())?;
+        root_namespace_state.insert_new_table(&table_ident, metadata_location.to_string())?;
 
         Table::builder()
             .file_io(self.file_io.clone())
-            .metadata_location(metadata_location)
+            .metadata_location(metadata_location.to_string())
             .metadata(metadata)
             .identifier(table_ident)
             .build()
@@ -365,12 +366,11 @@ impl Catalog for MemoryCatalog {
         let staged_table = commit.apply(current_table)?;
 
         // Write table metadata to the new location
+        let metadata_location =
+            MetadataLocation::from_str(staged_table.metadata_location_result()?)?;
         staged_table
             .metadata()
-            .write_to(
-                staged_table.file_io(),
-                staged_table.metadata_location_result()?,
-            )
+            .write_to(staged_table.file_io(), &metadata_location)
             .await?;
 
         // Flip the pointer to reference the new metadata file.