fix: support reading compressed metadata

The spec mentions that metadata files "may be compressed with GZIP",
here:

    https://iceberg.apache.org/spec/#table-metadata-and-snapshots

Author: colinmarc

Cargo.lock

@@ -71,6 +71,7 @@ enum-ordinalize = "4.3.0"
 env_logger = "0.11.8"
 expect-test = "1"
 faststr = "0.2.31"
+flate2 = "1.1.5"
 fnv = "1.0.7"
 fs-err = "3.1.0"
 futures = "0.3"

Cargo.toml

@@ -63,6 +63,7 @@ bytes = { workspace = true }
 chrono = { workspace = true }
 derive_builder = { workspace = true }
 expect-test = { workspace = true }
+flate2 = { workspace = true }
 fnv = { workspace = true }
 futures = { workspace = true }
 itertools = { workspace = true }

crates/iceberg/Cargo.toml

@@ -22,10 +22,12 @@ use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::fmt::{Display, Formatter};
 use std::hash::Hash;
+use std::io::Read as _;
 use std::sync::Arc;
 
 use _serde::TableMetadataEnum;
 use chrono::{DateTime, Utc};
+use flate2::read::GzDecoder;
 use serde::{Deserialize, Serialize};
 use serde_repr::{Deserialize_repr, Serialize_repr};
 use uuid::Uuid;
@@ -413,9 +415,24 @@ impl TableMetadata {
         file_io: &FileIO,
         metadata_location: impl AsRef<str>,
     ) -> Result<TableMetadata> {
-        let input_file = file_io.new_input(metadata_location)?;
+        let input_file = file_io.new_input(metadata_location.as_ref())?;
         let metadata_content = input_file.read().await?;
-        let metadata = serde_json::from_slice::<TableMetadata>(&metadata_content)?;
+
+        // Check if the file is compressed by looking for the gzip "magic number".
+        let metadata = if metadata_content.len() > 2
+            && metadata_content[0] == 0x1F
+            && metadata_content[1] == 0x8B
+        {
+            let mut decoder = GzDecoder::new(metadata_content.as_ref());
+            let mut decompressed_data = Vec::new();
+            decoder
+                .read_to_end(&mut decompressed_data)
+                .map_err(|e| Error::new(ErrorKind::DataInvalid, e.to_string()))?;
+            serde_json::from_slice(&decompressed_data)?
+        } else {
+            serde_json::from_slice(&metadata_content)?
+        };
+
         Ok(metadata)
     }
 
@@ -1318,6 +1335,7 @@ impl SnapshotLog {
 mod tests {
     use std::collections::HashMap;
     use std::fs;
+    use std::io::Write as _;
     use std::sync::Arc;
 
     use anyhow::Result;
@@ -3053,6 +3071,30 @@ mod tests {
         assert_eq!(read_metadata, original_metadata);
     }
 
+    #[tokio::test]
+    async fn test_table_metadata_read_compressed() {
+        let temp_dir = TempDir::new().unwrap();
+        let metadata_location = temp_dir.path().join("v1.gz.metadata.json");
+
+        let original_metadata: TableMetadata = get_test_table_metadata("TableMetadataV2Valid.json");
+        let json = serde_json::to_string(&original_metadata).unwrap();
+
+        let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
+        encoder.write_all(json.as_bytes()).unwrap();
+        std::fs::write(&metadata_location, encoder.finish().unwrap())
+            .expect("failed to write metadata");
+
+        // Read the metadata back
+        let file_io = FileIOBuilder::new_fs_io().build().unwrap();
+        let metadata_location = metadata_location.to_str().unwrap();
+        let read_metadata = TableMetadata::read_from(&file_io, metadata_location)
+            .await
+            .unwrap();
+
+        // Verify the metadata matches
+        assert_eq!(read_metadata, original_metadata);
+    }
+
     #[tokio::test]
     async fn test_table_metadata_read_nonexistent_file() {
         // Create a FileIO instance