feat: write gzipped metadata when the table asks for it

catalogs/iceberg-file-catalog/src/lib.rs

@@ -396,8 +396,11 @@ impl Catalog for FileCatalog {
             .put_metadata(&temp_metadata_location, metadata.as_ref())
             .await?;
 
-        let metadata_location =
-            new_filesystem_metadata_location(&metadata.location, &previous_metadata_location)?;
+        let metadata_location = new_filesystem_metadata_location(
+            &metadata.location,
+            &previous_metadata_location,
+            temp_metadata_location.ends_with(".gz.metadata.json"),
+        )?;
 
         object_store
             .copy_if_not_exists(
@@ -453,6 +456,7 @@ impl Catalog for FileCatalog {
                 let metadata_location = new_filesystem_metadata_location(
                     &metadata.location,
                     &previous_metadata_location,
+                    temp_metadata_location.ends_with(".gz.metadata.json"),
                 )?;
 
                 object_store
@@ -514,6 +518,7 @@ impl Catalog for FileCatalog {
                 let metadata_location = new_filesystem_metadata_location(
                     &metadata.location,
                     &previous_metadata_location,
+                    temp_metadata_location.ends_with(".gz.metadata.json"),
                 )?;
 
                 object_store
@@ -612,12 +617,14 @@ impl FileCatalog {
                 .trim_start_matches((strip_prefix(&path) + "/v").trim_start_matches("/"))
                 .trim_end_matches("/")
                 .trim_end_matches(".metadata.json")
+                .trim_end_matches(".gz")
                 .parse::<usize>()
                 .unwrap();
             let y = y
                 .trim_start_matches((strip_prefix(&path) + "/v").trim_start_matches("/"))
                 .trim_end_matches("/")
                 .trim_end_matches(".metadata.json")
+                .trim_end_matches(".gz")
                 .parse::<usize>()
                 .unwrap();
             x.cmp(&y)
@@ -668,19 +675,100 @@ fn parse_version(path: &str) -> Result<u64, IcebergError> {
         .ok_or(IcebergError::InvalidFormat("Metadata location".to_owned()))?
         .trim_start_matches('v')
         .trim_end_matches(".metadata.json")
+        .trim_end_matches(".gz")
         .parse()
         .map_err(IcebergError::from)
 }
 
 fn new_filesystem_metadata_location(
     metadata_location: &str,
     previous_metadata_location: &str,
+    gzipped: bool,
 ) -> Result<String, IcebergError> {
     let current_version = parse_version(previous_metadata_location)? + 1;
-    Ok(metadata_location.to_string()
-        + "/metadata/v"
-        + &current_version.to_string()
-        + ".metadata.json")
+    // The name carries the encoding: the reader decides whether to decompress
+    // from the suffix alone, so a gzipped file must be named `.gz.metadata.json`
+    // even though this catalog numbers versions rather than using the metastore
+    // name. Kept in sync with the temp file actually written, not re-derived
+    // from properties, so the two can never disagree.
+    let suffix = if gzipped {
+        "gz.metadata.json"
+    } else {
+        "metadata.json"
+    };
+    Ok(format!(
+        "{}/metadata/v{}.{}",
+        metadata_location, current_version, suffix
+    ))
+}
+
+#[cfg(test)]
+mod metadata_naming_tests {
+    use super::*;
+
+    /// The version number must be recovered from both the plain and the
+    /// gzipped name, since a gzipped table's previous location carries the
+    /// `.gz.metadata.json` suffix and drives the next version number.
+    #[test]
+    fn parse_version_reads_plain_and_gzipped_names() {
+        assert_eq!(
+            parse_version("/wh/ns/t/metadata/v7.metadata.json").unwrap(),
+            7
+        );
+        assert_eq!(
+            parse_version("/wh/ns/t/metadata/v7.gz.metadata.json").unwrap(),
+            7
+        );
+    }
+
+    /// The final versioned name must carry the same encoding as the file that
+    /// was actually written; otherwise the reader, which decides purely from
+    /// the suffix, would read gzip bytes as plain JSON.
+    #[test]
+    fn the_versioned_name_carries_the_gzip_suffix() {
+        let plain = new_filesystem_metadata_location(
+            "/wh/ns/t",
+            "/wh/ns/t/metadata/v3.metadata.json",
+            false,
+        )
+        .unwrap();
+        assert_eq!(plain, "/wh/ns/t/metadata/v4.metadata.json");
+
+        let gzipped = new_filesystem_metadata_location(
+            "/wh/ns/t",
+            "/wh/ns/t/metadata/v3.gz.metadata.json",
+            true,
+        )
+        .unwrap();
+        assert_eq!(gzipped, "/wh/ns/t/metadata/v4.gz.metadata.json");
+    }
+
+    /// The encoding may be toggled on an existing table, so the previous name's
+    /// suffix and the requested encoding are independent: whichever the caller
+    /// asks for wins, and the version still advances.
+    #[test]
+    fn the_encoding_can_change_between_commits() {
+        // Was plain, now gzipped.
+        assert_eq!(
+            new_filesystem_metadata_location(
+                "/wh/ns/t",
+                "/wh/ns/t/metadata/v1.metadata.json",
+                true
+            )
+            .unwrap(),
+            "/wh/ns/t/metadata/v2.gz.metadata.json"
+        );
+        // Was gzipped, now plain.
+        assert_eq!(
+            new_filesystem_metadata_location(
+                "/wh/ns/t",
+                "/wh/ns/t/metadata/v1.gz.metadata.json",
+                false
+            )
+            .unwrap(),
+            "/wh/ns/t/metadata/v2.metadata.json"
+        );
+    }
 }
 
 #[derive(Debug)]

iceberg-rust-spec/src/spec/table_metadata.rs

@@ -63,6 +63,9 @@ pub const WRITE_METADATA_METRICS_COLUMN_PREFIX: &str = "write.metadata.metrics.c
 /// it are only measured when named explicitly. Defaults to 100.
 pub const WRITE_METADATA_METRICS_MAX_INFERRED_COLUMN_DEFAULTS: &str =
     "write.metadata.metrics.max-inferred-column-defaults";
+/// Codec for `metadata.json`: `none` (default) or `gzip`. A gzipped metadata
+/// file is named `*.gz.metadata.json`, which is how readers detect it.
+pub const WRITE_METADATA_COMPRESSION_CODEC: &str = "write.metadata.compression-codec";
 
 pub use _serde::{TableMetadataV1, TableMetadataV2, TableMetadataV3};
 
@@ -371,6 +374,11 @@ pub fn partition_fields<'a>(
 
 /// Creates a new metadata file location for a table
 ///
+/// The name carries the encoding: a table whose
+/// `write.metadata.compression-codec` is `gzip` gets a `.gz.metadata.json`
+/// suffix, which is how readers already decide whether to decompress. Any
+/// other value, including the default `none`, produces a plain name.
+///
 /// # Arguments
 /// * `metadata` - The table metadata to create a location for
 ///
@@ -380,15 +388,28 @@ pub fn new_metadata_location<'a, T: Into<TabularMetadataRef<'a>>>(metadata: T) -
     let metadata: TabularMetadataRef = metadata.into();
     let transaction_uuid = Uuid::new_v4();
     let version = metadata.sequence_number();
+    let suffix = if metadata_is_gzipped(metadata.properties()) {
+        "gz.metadata.json"
+    } else {
+        "metadata.json"
+    };
 
     format!(
-        "{}/metadata/{:05}-{}.metadata.json",
+        "{}/metadata/{:05}-{}.{}",
         metadata.location(),
         version,
-        transaction_uuid
+        transaction_uuid,
+        suffix
     )
 }
 
+/// Whether a table's properties ask for gzipped metadata.
+pub fn metadata_is_gzipped(properties: &HashMap<String, String>) -> bool {
+    properties
+        .get(WRITE_METADATA_COMPRESSION_CODEC)
+        .is_some_and(|codec| codec.trim().eq_ignore_ascii_case("gzip"))
+}
+
 impl fmt::Display for TableMetadata {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
@@ -1095,6 +1116,43 @@ impl From<FormatVersion> for i32 {
     }
 }
 
+#[cfg(test)]
+mod metadata_compression_tests {
+    use super::*;
+
+    #[test]
+    fn gzip_is_recognized_whatever_the_casing() {
+        for value in ["gzip", "GZIP", " Gzip "] {
+            let properties = HashMap::from([(
+                WRITE_METADATA_COMPRESSION_CODEC.to_string(),
+                value.to_string(),
+            )]);
+            assert!(
+                metadata_is_gzipped(&properties),
+                "{value:?} should mean gzip"
+            );
+        }
+    }
+
+    /// The spec's default is `none`, and an unrecognized codec must not be
+    /// guessed at -- a name claiming gzip that is not gzip is unreadable.
+    #[test]
+    fn anything_else_means_uncompressed() {
+        assert!(!metadata_is_gzipped(&HashMap::new()));
+
+        for value in ["none", "", "zstd", "gzip2"] {
+            let properties = HashMap::from([(
+                WRITE_METADATA_COMPRESSION_CODEC.to_string(),
+                value.to_string(),
+            )]);
+            assert!(
+                !metadata_is_gzipped(&properties),
+                "{value:?} should not mean gzip"
+            );
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
 

iceberg-rust-spec/src/spec/tabular.rs

@@ -131,6 +131,19 @@ impl TabularMetadataRef<'_> {
         }
     }
 
+    /// Returns the properties of the tabular object
+    ///
+    /// # Returns
+    /// * The `write.*` and other properties that configure how this table,
+    ///   view, or materialized view is written
+    pub fn properties(&self) -> &std::collections::HashMap<String, String> {
+        match self {
+            TabularMetadataRef::Table(table) => &table.properties,
+            TabularMetadataRef::View(view) => &view.properties,
+            TabularMetadataRef::MaterializedView(matview) => &matview.properties,
+        }
+    }
+
     /// Returns the current sequence number or version ID of the tabular object
     ///
     /// # Returns