feat: Honor compression settings for metadata.json on write

crates/catalog/glue/src/catalog.rs

@@ -17,6 +17,7 @@
 
 use std::collections::HashMap;
 use std::fmt::Debug;
+use std::str::FromStr;
 
 use anyhow::anyhow;
 use async_trait::async_trait;
@@ -526,14 +527,14 @@ impl Catalog for GlueCatalog {
         let metadata = TableMetadataBuilder::from_table_creation(creation)?
             .build()?
             .metadata;
-        let metadata_location =
-            MetadataLocation::new_with_table_location(location.clone()).to_string();
+        let metadata_location = MetadataLocation::new_with_metadata(location.clone(), &metadata);
 
         metadata.write_to(&self.file_io, &metadata_location).await?;
 
+        let metadata_location_str = metadata_location.to_string();
         let glue_table = convert_to_glue_table(
             &table_name,
-            metadata_location.clone(),
+            metadata_location_str.clone(),
             &metadata,
             metadata.properties(),
             None,
@@ -551,7 +552,7 @@ impl Catalog for GlueCatalog {
 
         Table::builder()
             .file_io(self.file_io())
-            .metadata_location(metadata_location)
+            .metadata_location(metadata_location_str)
             .metadata(metadata)
             .identifier(TableIdent::new(NamespaceIdent::new(db_name), table_name))
             .build()
@@ -789,12 +790,13 @@ impl Catalog for GlueCatalog {
         let current_metadata_location = current_table.metadata_location_result()?.to_string();
 
         let staged_table = commit.apply(current_table)?;
-        let staged_metadata_location = staged_table.metadata_location_result()?;
+        let staged_metadata_location_str = staged_table.metadata_location_result()?;
+        let staged_metadata_location = MetadataLocation::from_str(staged_metadata_location_str)?;
 
         // Write new metadata
         staged_table
             .metadata()
-            .write_to(staged_table.file_io(), staged_metadata_location)
+            .write_to(staged_table.file_io(), &staged_metadata_location)
             .await?;
 
         // Persist staged table to Glue with optimistic locking

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

@@ -442,17 +442,17 @@ impl Catalog for HmsCatalog {
             .build()?
             .metadata;
 
-        let metadata_location =
-            MetadataLocation::new_with_table_location(location.clone()).to_string();
+        let metadata_location = MetadataLocation::new_with_metadata(location.clone(), &metadata);
 
         metadata.write_to(&self.file_io, &metadata_location).await?;
 
+        let metadata_location_str = metadata_location.to_string();
         let hive_table = convert_to_hive_table(
             db_name.clone(),
             metadata.current_schema(),
             table_name.clone(),
             location,
-            metadata_location.clone(),
+            metadata_location_str.clone(),
             metadata.properties(),
         )?;
 
@@ -464,7 +464,7 @@ impl Catalog for HmsCatalog {
 
         Table::builder()
             .file_io(self.file_io())
-            .metadata_location(metadata_location)
+            .metadata_location(metadata_location_str)
             .metadata(metadata)
             .identifier(TableIdent::new(NamespaceIdent::new(db_name), table_name))
             .build()

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

@@ -17,6 +17,7 @@
 
 use std::collections::HashMap;
 use std::future::Future;
+use std::str::FromStr;
 
 use async_trait::async_trait;
 use aws_sdk_s3tables::operation::create_table::CreateTableOutput;
@@ -476,25 +477,26 @@ impl Catalog for S3TablesCatalog {
         let metadata = TableMetadataBuilder::from_table_creation(creation)?
             .build()?
             .metadata;
-        let metadata_location =
-            MetadataLocation::new_with_table_location(table_location).to_string();
+
+        let metadata_location = MetadataLocation::new_with_metadata(table_location, &metadata);
         metadata.write_to(&self.file_io, &metadata_location).await?;
 
         // update metadata location
+        let metadata_location_str = metadata_location.to_string();
         self.s3tables_client
             .update_table_metadata_location()
             .table_bucket_arn(self.config.table_bucket_arn.clone())
             .namespace(namespace.to_url_string())
             .name(table_ident.name())
-            .metadata_location(metadata_location.clone())
+            .metadata_location(metadata_location_str.clone())
             .version_token(create_resp.version_token())
             .send()
             .await
             .map_err(from_aws_sdk_error)?;
 
         let table = Table::builder()
             .identifier(table_ident)
-            .metadata_location(metadata_location)
+            .metadata_location(metadata_location_str)
             .metadata(metadata)
             .file_io(self.file_io.clone())
             .build()?;
@@ -605,11 +607,12 @@ impl Catalog for S3TablesCatalog {
             self.load_table_with_version_token(&table_ident).await?;
 
         let staged_table = commit.apply(current_table)?;
-        let staged_metadata_location = staged_table.metadata_location_result()?;
+        let staged_metadata_location_str = staged_table.metadata_location_result()?;
+        let staged_metadata_location = MetadataLocation::from_str(staged_metadata_location_str)?;
 
         staged_table
             .metadata()
-            .write_to(staged_table.file_io(), staged_metadata_location)
+            .write_to(staged_table.file_io(), &staged_metadata_location)
             .await?;
 
         let builder = self
@@ -619,7 +622,7 @@ impl Catalog for S3TablesCatalog {
             .namespace(table_namespace.to_url_string())
             .name(table_ident.name())
             .version_token(version_token)
-            .metadata_location(staged_metadata_location);
+            .metadata_location(staged_metadata_location_str);
 
         let _ = builder.send().await.map_err(|e| {
             let error = e.into_service_error();

crates/catalog/sql/src/catalog.rs

@@ -829,21 +829,22 @@ 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);
 
         tbl_metadata
             .write_to(&self.fileio, &tbl_metadata_location)
             .await?;
 
+        let tbl_metadata_location_str = tbl_metadata_location.to_string();
         self.execute(&format!(
             "INSERT INTO {CATALOG_TABLE_NAME}
              ({CATALOG_FIELD_CATALOG_NAME}, {CATALOG_FIELD_TABLE_NAMESPACE}, {CATALOG_FIELD_TABLE_NAME}, {CATALOG_FIELD_METADATA_LOCATION_PROP}, {CATALOG_FIELD_RECORD_TYPE})
              VALUES (?, ?, ?, ?, ?)
-            "), vec![Some(&self.name), Some(&namespace.join(".")), Some(&tbl_name.clone()), Some(&tbl_metadata_location), Some(CATALOG_FIELD_TABLE_RECORD_TYPE)], None).await?;
+            "), vec![Some(&self.name), Some(&namespace.join(".")), Some(&tbl_name.clone()), Some(&tbl_metadata_location_str), Some(CATALOG_FIELD_TABLE_RECORD_TYPE)], None).await?;
 
         Ok(Table::builder()
             .file_io(self.fileio.clone())
-            .metadata_location(tbl_metadata_location)
+            .metadata_location(tbl_metadata_location_str)
             .identifier(tbl_ident)
             .metadata(tbl_metadata)
             .build()?)
@@ -927,13 +928,15 @@ impl Catalog for SqlCatalog {
         let current_metadata_location = current_table.metadata_location_result()?.to_string();
 
         let staged_table = commit.apply(current_table)?;
-        let staged_metadata_location = staged_table.metadata_location_result()?;
+        let staged_metadata_location_str = staged_table.metadata_location_result()?;
+        let staged_metadata_location = MetadataLocation::from_str(staged_metadata_location_str)?;
 
         staged_table
             .metadata()
             .write_to(staged_table.file_io(), &staged_metadata_location)
             .await?;
 
+        let staged_metadata_location_str = staged_metadata_location.to_string();
         let update_result = self
             .execute(
                 &format!(
@@ -949,7 +952,7 @@ impl Catalog for SqlCatalog {
                       AND {CATALOG_FIELD_METADATA_LOCATION_PROP} = ?"
                 ),
                 vec![
-                    Some(staged_metadata_location),
+                    Some(&staged_metadata_location_str),
                     Some(current_metadata_location.as_str()),
                     Some(&self.name),
                     Some(table_ident.name()),

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.