diff --git a/crates/iceberg/src/catalog/metadata_location.rs b/crates/iceberg/src/catalog/metadata_location.rs index 8240a326f3..d4776d9653 100644 --- a/crates/iceberg/src/catalog/metadata_location.rs +++ b/crates/iceberg/src/catalog/metadata_location.rs @@ -27,9 +27,21 @@ use crate::{Error, ErrorKind, Result}; /// Default folder name for metadata files under the table location, used when the /// `write.metadata.path` table property is not set. pub(crate) const METADATA_FOLDER_NAME: &str = "metadata"; +const METADATA_SUFFIX: &str = ".metadata.json"; + +fn strip_metadata_suffix<'a>(file_name: &'a str, codec_suffix: &str) -> Option<&'a str> { + file_name + .strip_suffix(METADATA_SUFFIX) + .and_then(|stripped| stripped.strip_suffix(codec_suffix)) + .or_else(|| { + file_name + .strip_suffix(codec_suffix) + .and_then(|stripped| stripped.strip_suffix(METADATA_SUFFIX)) + }) +} /// Helper for parsing a location of the format: `/-.metadata.json` -/// or with compression: `/-.gz.metadata.json` +/// or with compression: `/-..metadata.json` /// /// `` is set to the `write.metadata.path` table property and /// it defaults to the `/metadata` when the property is not set. @@ -83,21 +95,26 @@ impl MetadataLocation { } /// Parses a file name of the format `-.metadata.json` - /// or with compression: `-.gz.metadata.json`. - /// Parse errors for compression codec result in CompressionCodec::None. + /// or with compression before or after `.metadata.json`. fn parse_file_name(file_name: &str) -> Result<(i32, Uuid, CompressionCodec)> { - let stripped = file_name.strip_suffix(".metadata.json").ok_or(Error::new( - ErrorKind::Unexpected, - format!("Invalid metadata file ending: {file_name}"), - ))?; - - // Check for compression suffix (e.g., .gz) - let gzip_suffix = CompressionCodec::gzip_default().suffix()?; - let (stripped, compression_codec) = if let Some(s) = stripped.strip_suffix(gzip_suffix) { - (s, CompressionCodec::gzip_default()) - } else { - (stripped, CompressionCodec::None) - }; + let (stripped, compression_codec) = CompressionCodec::table_metadata_codecs() + .into_iter() + .filter(|codec| !codec.is_none()) + .find_map(|codec| { + strip_metadata_suffix(file_name, codec.table_metadata_suffix()?) + .map(|stripped| (stripped, codec)) + }) + .or_else(|| { + file_name + .strip_suffix(METADATA_SUFFIX) + .map(|stripped| (stripped, CompressionCodec::None)) + }) + .ok_or_else(|| { + Error::new( + ErrorKind::Unexpected, + format!("Invalid metadata file ending: {file_name}"), + ) + })?; let (version, id) = stripped.split_once('-').ok_or(Error::new( ErrorKind::Unexpected, @@ -241,6 +258,36 @@ mod test { compression_codec: CompressionCodec::gzip_default(), }), ), + // With trailing gzip compression suffix + ( + "/abc/metadata/1234567-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json.gz", + Ok(MetadataLocation { + location: "/abc/metadata".to_string(), + version: 1234567, + id: Uuid::from_str("2cd22b57-5127-4198-92ba-e4e67c79821b").unwrap(), + compression_codec: CompressionCodec::gzip_default(), + }), + ), + // With zstd compression + ( + "/abc/metadata/1234567-2cd22b57-5127-4198-92ba-e4e67c79821b.zstd.metadata.json", + Ok(MetadataLocation { + location: "/abc/metadata".to_string(), + version: 1234567, + id: Uuid::from_str("2cd22b57-5127-4198-92ba-e4e67c79821b").unwrap(), + compression_codec: CompressionCodec::zstd_default(), + }), + ), + // With trailing zstd compression suffix + ( + "/abc/metadata/1234567-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json.zstd", + Ok(MetadataLocation { + location: "/abc/metadata".to_string(), + version: 1234567, + id: Uuid::from_str("2cd22b57-5127-4198-92ba-e4e67c79821b").unwrap(), + compression_codec: CompressionCodec::zstd_default(), + }), + ), // Negative version ( "/metadata/-123-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json", @@ -288,6 +335,25 @@ mod test { } } + #[test] + fn test_metadata_location_canonicalizes_compression_suffixes() { + for (input, expected) in [ + ( + "/abc/metadata/00001-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json.gz", + "/abc/metadata/00001-2cd22b57-5127-4198-92ba-e4e67c79821b.gz.metadata.json", + ), + ( + "/abc/metadata/00001-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json.zstd", + "/abc/metadata/00001-2cd22b57-5127-4198-92ba-e4e67c79821b.zstd.metadata.json", + ), + ] { + assert_eq!( + MetadataLocation::from_str(input).unwrap().to_string(), + expected + ); + } + } + #[test] fn test_metadata_location_with_next_version() { let metadata = create_test_metadata(HashMap::new()); @@ -311,33 +377,34 @@ mod test { #[test] fn test_with_next_version_preserves_compression() { - // Start from a parsed location with no compression - let location_none = MetadataLocation::from_str( - "/test/table/metadata/00000-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json", - ) - .unwrap(); - assert_eq!(location_none.compression_codec, CompressionCodec::None); - - let next_none = location_none.with_next_version(); - assert_eq!(next_none.compression_codec, CompressionCodec::None); - assert_eq!(next_none.version, 1); - - // Start from a parsed location with gzip compression - let location_gzip = MetadataLocation::from_str( - "/test/table/metadata/00005-81056704-ce5b-41c4-bb83-eb6408081af6.gz.metadata.json", - ) - .unwrap(); - assert_eq!( - location_gzip.compression_codec, - CompressionCodec::gzip_default() - ); - - let next_gzip = location_gzip.with_next_version(); - assert_eq!( - next_gzip.compression_codec, - CompressionCodec::gzip_default() - ); - assert_eq!(next_gzip.version, 6); + for (input, expected_codec, expected_version, expected_suffix) in [ + ( + "/test/table/metadata/00000-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json", + CompressionCodec::None, + 1, + ".metadata.json", + ), + ( + "/test/table/metadata/00005-81056704-ce5b-41c4-bb83-eb6408081af6.metadata.json.gz", + CompressionCodec::gzip_default(), + 6, + ".gz.metadata.json", + ), + ( + "/test/table/metadata/00009-81056704-ce5b-41c4-bb83-eb6408081af6.metadata.json.zstd", + CompressionCodec::zstd_default(), + 10, + ".zstd.metadata.json", + ), + ] { + let location = MetadataLocation::from_str(input).unwrap(); + assert_eq!(location.compression_codec, expected_codec); + + let next = location.with_next_version(); + assert_eq!(next.compression_codec, expected_codec); + assert_eq!(next.version, expected_version); + assert!(next.to_string().ends_with(expected_suffix)); + } } #[test] @@ -367,10 +434,26 @@ mod test { "/test/table/metadata/00000-2cd22b57-5127-4198-92ba-e4e67c79821b.gz.metadata.json" ); + // Transition from gzip to zstd compression + let props_zstd = HashMap::from([( + TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(), + "zstd".to_string(), + )]); + let metadata_zstd = create_test_metadata(props_zstd); + let updated_zstd = updated_gzip.try_with_new_metadata(&metadata_zstd).unwrap(); + assert_eq!( + updated_zstd.compression_codec, + CompressionCodec::zstd_default() + ); + assert_eq!( + updated_zstd.to_string(), + "/test/table/metadata/00000-2cd22b57-5127-4198-92ba-e4e67c79821b.zstd.metadata.json" + ); + // Update back to no compression let props_none = HashMap::new(); let metadata_none = create_test_metadata(props_none); - let updated_none = updated_gzip.try_with_new_metadata(&metadata_none).unwrap(); + let updated_none = updated_zstd.try_with_new_metadata(&metadata_none).unwrap(); assert_eq!(updated_none.compression_codec, CompressionCodec::None); assert_eq!(updated_none.version, 0); assert_eq!( @@ -429,15 +512,30 @@ mod test { #[test] fn test_new_with_metadata_honors_write_metadata_path() { - // Test metadata lives under `/metadata` by default - let default_meta = create_test_metadata(HashMap::new()); - let default_loc = MetadataLocation::try_new_with_metadata(&default_meta).unwrap(); - assert!( - default_loc - .to_string() - .starts_with("/test/table/metadata/00000-"), - "unexpected location: {default_loc}" - ); + // Test metadata lives under `/metadata` with canonical compression suffixes. + for (compression, expected_suffix) in [ + (None, ".metadata.json"), + (Some("gzip"), ".gz.metadata.json"), + (Some("zstd"), ".zstd.metadata.json"), + ] { + let properties = compression + .map(|compression| { + HashMap::from([( + TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(), + compression.to_string(), + )]) + }) + .unwrap_or_default(); + let metadata = create_test_metadata(properties); + let location = MetadataLocation::try_new_with_metadata(&metadata).unwrap(); + assert!( + location + .to_string() + .starts_with("/test/table/metadata/00000-"), + "unexpected location: {location}" + ); + assert!(location.to_string().ends_with(expected_suffix)); + } // Test a configured `write.metadata.path` is honored let props = HashMap::from([( diff --git a/crates/iceberg/src/compression.rs b/crates/iceberg/src/compression.rs index dbd2c97881..e8657fd615 100644 --- a/crates/iceberg/src/compression.rs +++ b/crates/iceberg/src/compression.rs @@ -35,6 +35,10 @@ const GZIP_DEFAULT_LEVEL: u8 = 6; const GZIP_MAX_LEVEL: u8 = 9; /// Default compression level for Brotli. const BROTLI_DEFAULT_LEVEL: u8 = 1; +const GZIP_SUFFIX: &str = ".gz"; +const ZSTD_SUFFIX: &str = ".zstd"; +const GZIP_MAGIC: &[u8] = &[0x1F, 0x8B]; +const ZSTD_MAGIC: &[u8] = &[0x28, 0xB5, 0x2F, 0xFD]; /// Data compression formats #[derive(Debug, PartialEq, Eq, Clone, Copy, Default)] @@ -78,6 +82,38 @@ impl CompressionCodec { CompressionCodec::Brotli(BROTLI_DEFAULT_LEVEL) } + pub(crate) const fn table_metadata_codecs() -> [Self; 3] { + [ + CompressionCodec::None, + CompressionCodec::Gzip(GZIP_DEFAULT_LEVEL), + CompressionCodec::Zstd(ZSTD_DEFAULT_LEVEL), + ] + } + + pub(crate) const fn is_supported_for_table_metadata(&self) -> bool { + matches!( + self, + CompressionCodec::None | CompressionCodec::Gzip(_) | CompressionCodec::Zstd(_) + ) + } + + pub(crate) const fn table_metadata_suffix(&self) -> Option<&'static str> { + match self { + CompressionCodec::None => Some(""), + CompressionCodec::Zstd(_) => Some(ZSTD_SUFFIX), + CompressionCodec::Gzip(_) => Some(GZIP_SUFFIX), + _ => None, + } + } + + pub(crate) const fn table_metadata_magic(&self) -> Option<&'static [u8]> { + match self { + CompressionCodec::Zstd(_) => Some(ZSTD_MAGIC), + CompressionCodec::Gzip(_) => Some(GZIP_MAGIC), + _ => None, + } + } + /// Returns the codec name as used in serialization and error messages. pub fn name(&self) -> &'static str { match self { @@ -224,25 +260,18 @@ impl CompressionCodec { } /// Returns the file extension suffix for this compression codec. - /// Returns empty string for None, ".gz" for Gzip. + /// Returns empty string for None, ".zstd" for Zstd, and ".gz" for Gzip. /// /// # Errors /// - /// Returns an error for Lz4 and Zstd as they are not fully supported. + /// Returns an error for codecs without a metadata file suffix. pub fn suffix(&self) -> Result<&'static str> { - match self { - CompressionCodec::None => Ok(""), - CompressionCodec::Gzip(_) => Ok(".gz"), - codec @ (CompressionCodec::Lz4 - | CompressionCodec::Lz4Raw - | CompressionCodec::Zstd(_) - | CompressionCodec::Brotli(_) - | CompressionCodec::Lzo - | CompressionCodec::Snappy) => Err(Error::new( + self.table_metadata_suffix().ok_or_else(|| { + Error::new( ErrorKind::FeatureUnsupported, - format!("suffix not defined for {codec:?}"), - )), - } + format!("suffix not defined for {self:?}"), + ) + }) } } @@ -302,17 +331,24 @@ mod tests { #[test] fn test_suffix() { assert_eq!(CompressionCodec::None.suffix().unwrap(), ""); + assert_eq!(CompressionCodec::zstd_default().suffix().unwrap(), ".zstd"); + assert_eq!(CompressionCodec::Zstd(5).suffix().unwrap(), ".zstd"); assert_eq!(CompressionCodec::gzip_default().suffix().unwrap(), ".gz"); + assert_eq!(CompressionCodec::Gzip(9).suffix().unwrap(), ".gz"); assert!(CompressionCodec::Lz4.suffix().is_err()); - assert!(CompressionCodec::zstd_default().suffix().is_err()); assert!(CompressionCodec::Snappy.suffix().is_err()); let lz4_err = CompressionCodec::Lz4.suffix().unwrap_err(); assert!(lz4_err.to_string().contains("suffix not defined for Lz4")); + } - let zstd_err = CompressionCodec::zstd_default().suffix().unwrap_err(); - assert!(zstd_err.to_string().contains("suffix not defined for Zstd")); + #[test] + fn test_table_metadata_compression_support_ignores_level() { + assert!(CompressionCodec::None.is_supported_for_table_metadata()); + assert!(CompressionCodec::Gzip(9).is_supported_for_table_metadata()); + assert!(CompressionCodec::Zstd(5).is_supported_for_table_metadata()); + assert!(!CompressionCodec::Lz4.is_supported_for_table_metadata()); } #[test] diff --git a/crates/iceberg/src/spec/table_metadata.rs b/crates/iceberg/src/spec/table_metadata.rs index 2b9d1d140e..e7f6830a56 100644 --- a/crates/iceberg/src/spec/table_metadata.rs +++ b/crates/iceberg/src/spec/table_metadata.rs @@ -392,7 +392,7 @@ impl TableMetadata { /// Returns the metadata compression codec from table properties. /// /// Returns `CompressionCodec::None` if compression is disabled or not configured. - /// Returns `CompressionCodec::Gzip` if gzip compression is enabled. + /// Returns the configured gzip or zstd codec when compression is enabled. /// /// # Errors /// @@ -479,21 +479,24 @@ impl TableMetadata { let input_file = file_io.new_input(metadata_location)?; let metadata_content = input_file.read().await?; - // 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 decompressed_data = CompressionCodec::gzip_default() - .decompress(metadata_content.to_vec()) - .map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - "Trying to read compressed metadata file", - ) - .with_context("file_path", metadata_location) - .with_source(e) - })?; + let compression_codec = + CompressionCodec::table_metadata_codecs() + .into_iter() + .find(|codec| { + codec + .table_metadata_magic() + .is_some_and(|magic| metadata_content.starts_with(magic)) + }); + + let metadata = if let Some(codec) = compression_codec { + let decompressed_data = codec.decompress(metadata_content.to_vec()).map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + "Trying to read compressed metadata file", + ) + .with_context("file_path", metadata_location) + .with_source(e) + })?; serde_json::from_slice(&decompressed_data)? } else { serde_json::from_slice(&metadata_content)? @@ -524,17 +527,7 @@ impl TableMetadata { )); } - // Apply compression based on codec - let data_to_write = match codec { - CompressionCodec::Gzip(_) => codec.compress(json_data)?, - CompressionCodec::None => json_data, - _ => { - return Err(Error::new( - ErrorKind::DataInvalid, - format!("Unsupported metadata compression codec: {codec:?}"), - )); - } - }; + let data_to_write = codec.compress(json_data)?; file_io .new_output(metadata_location.to_string())? @@ -1672,6 +1665,10 @@ mod tests { serde_json::from_str(&metadata).unwrap() } + fn table_metadata_magic(codec: CompressionCodec) -> &'static [u8] { + codec.table_metadata_magic().unwrap() + } + /// Loads a test table metadata and relocates it to `location`, so that derived /// metadata paths point at a writable (e.g. temp) directory. fn get_test_table_metadata_at(file_name: &str, location: &str) -> TableMetadata { @@ -3693,6 +3690,80 @@ mod tests { assert_eq!(read_metadata, original_metadata); } + #[tokio::test] + async fn test_table_metadata_read_iceberg_go_zstd_fixture() { + let fixture = + fs::read("testdata/table_metadata/TableMetadataV2Valid.zstd.metadata.json").unwrap(); + assert!(fixture.starts_with(table_metadata_magic(CompressionCodec::zstd_default()))); + + let temp_dir = TempDir::new().unwrap(); + let file_io = FileIO::new_with_fs(); + let mut expected_metadata = None; + + for file_name in [ + "v1.zstd.metadata.json", + "v1.metadata.json.zstd", + "v1.metadata.json", + ] { + let metadata_location = temp_dir.path().join(file_name); + fs::write(&metadata_location, &fixture).unwrap(); + + let read_metadata = + TableMetadata::read_from(&file_io, metadata_location.to_str().unwrap()) + .await + .unwrap(); + if let Some(expected_metadata) = &expected_metadata { + assert_eq!(&read_metadata, expected_metadata); + } else { + assert_eq!(read_metadata.format_version(), FormatVersion::V2); + assert_eq!( + read_metadata.uuid(), + Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap() + ); + assert_eq!(read_metadata.location(), "s3://bucket/test/location"); + expected_metadata = Some(read_metadata); + } + } + } + + #[tokio::test] + async fn test_table_metadata_read_plain_json_with_zstd_file_name() { + let temp_dir = TempDir::new().unwrap(); + let metadata_location = temp_dir.path().join("v1.zstd.metadata.json"); + let expected_metadata: TableMetadata = get_test_table_metadata("TableMetadataV2Valid.json"); + fs::write( + &metadata_location, + serde_json::to_vec(&expected_metadata).unwrap(), + ) + .unwrap(); + + let file_io = FileIO::new_with_fs(); + let read_metadata = TableMetadata::read_from(&file_io, metadata_location.to_str().unwrap()) + .await + .unwrap(); + assert_eq!(read_metadata, expected_metadata); + } + + #[tokio::test] + async fn test_table_metadata_read_truncated_zstd() { + let mut fixture = + fs::read("testdata/table_metadata/TableMetadataV2Valid.zstd.metadata.json").unwrap(); + fixture.truncate(fixture.len() / 2); + assert!(fixture.starts_with(table_metadata_magic(CompressionCodec::zstd_default()))); + + let temp_dir = TempDir::new().unwrap(); + let metadata_location = temp_dir.path().join("truncated.metadata.json"); + fs::write(&metadata_location, fixture).unwrap(); + + let file_io = FileIO::new_with_fs(); + let metadata_location = metadata_location.to_str().unwrap(); + let err = TableMetadata::read_from(&file_io, metadata_location) + .await + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.to_string().contains(metadata_location)); + } + #[tokio::test] async fn test_table_metadata_read_nonexistent_file() { // Create a FileIO instance @@ -3750,9 +3821,7 @@ mod tests { // Read the raw file and check it's gzip compressed let raw_content = fs::read(&metadata_location_str).unwrap(); - assert!(raw_content.len() > 2); - assert_eq!(raw_content[0], 0x1F); // gzip magic number - assert_eq!(raw_content[1], 0x8B); // gzip magic number + assert!(raw_content.starts_with(table_metadata_magic(CompressionCodec::gzip_default()))); // Read the metadata back using the compressed location let read_metadata = TableMetadata::read_from(&file_io, &metadata_location_str) @@ -3763,6 +3832,47 @@ mod tests { assert_eq!(read_metadata, compressed_metadata); } + #[tokio::test] + async fn test_table_metadata_write_with_zstd_compression() { + let temp_dir = TempDir::new().unwrap(); + let temp_path = temp_dir.path().to_str().unwrap(); + let file_io = FileIO::new_with_fs(); + let original_metadata: TableMetadata = + get_test_table_metadata_at("TableMetadataV2Valid.json", temp_path); + + let mut props = original_metadata.properties.clone(); + props.insert( + TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(), + "ZsTd".to_string(), + ); + let compressed_metadata = + TableMetadataBuilder::new_from_metadata(original_metadata.clone(), None) + .assign_uuid(original_metadata.table_uuid) + .set_properties(props) + .unwrap() + .build() + .unwrap() + .metadata; + + let expected_location = + format!("{temp_path}/00000-2cd22b57-5127-4198-92ba-e4e67c79821b.zstd.metadata.json"); + let metadata_location = expected_location.parse::().unwrap(); + + compressed_metadata + .write_to(&file_io, &metadata_location) + .await + .unwrap(); + + assert_eq!(metadata_location.to_string(), expected_location); + let raw_content = fs::read(&expected_location).unwrap(); + assert!(raw_content.starts_with(table_metadata_magic(CompressionCodec::zstd_default()))); + + let read_metadata = TableMetadata::read_from(&file_io, &expected_location) + .await + .unwrap(); + assert_eq!(read_metadata, compressed_metadata); + } + #[test] fn test_partition_name_exists() { let schema = Schema::builder() diff --git a/crates/iceberg/src/spec/table_properties.rs b/crates/iceberg/src/spec/table_properties.rs index 01a330e28d..8144b75eb9 100644 --- a/crates/iceberg/src/spec/table_properties.rs +++ b/crates/iceberg/src/spec/table_properties.rs @@ -25,6 +25,32 @@ use crate::error::{Error, ErrorKind, Result}; use crate::spec::NameMapping; use crate::util::location::strip_trailing_slash; +fn supported_metadata_compression_names() -> String { + let names = CompressionCodec::table_metadata_codecs() + .into_iter() + .map(|codec| format!("'{}'", codec.name())) + .collect::>(); + let (last, rest) = names + .split_last() + .expect("metadata compression codec list must not be empty"); + + if rest.is_empty() { + last.clone() + } else { + format!("{}, and {last}", rest.join(", ")) + } +} + +fn invalid_metadata_compression_codec(value: &str) -> Error { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid metadata compression codec: {value}. Only {} are supported for metadata files.", + supported_metadata_compression_names() + ), + ) +} + fn parse_location_property(path: &str) -> Result { if path.is_empty() { return Err(Error::new(ErrorKind::DataInvalid, "path must not be empty")); @@ -43,31 +69,14 @@ fn parse_metadata_compression(value: &str) -> Result { let lowercase_value = value.to_lowercase(); // Use serde to parse the codec (which has rename_all = "lowercase") - let codec: CompressionCodec = serde_json::from_value(serde_json::Value::String( - lowercase_value, - )) - .map_err(|_| { - Error::new( - ErrorKind::DataInvalid, - format!( - "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported.", - CompressionCodec::None.name(), - CompressionCodec::gzip_default().name() - ), - ) - })?; - - // Validate that only None and Gzip are used for metadata - match codec { - CompressionCodec::None | CompressionCodec::Gzip(_) => Ok(codec), - _ => Err(Error::new( - ErrorKind::DataInvalid, - format!( - "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.", - CompressionCodec::None.name(), - CompressionCodec::gzip_default().name() - ), - )), + let codec: CompressionCodec = + serde_json::from_value(serde_json::Value::String(lowercase_value)) + .map_err(|_| invalid_metadata_compression_codec(value))?; + + if codec.is_supported_for_table_metadata() { + Ok(codec) + } else { + Err(invalid_metadata_compression_codec(value)) } } @@ -701,15 +710,20 @@ mod tests { #[test] fn test_table_properties_compression() { - let props = HashMap::from([( - TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(), - "gzip".to_string(), - )]); - let table_properties = TableProperties::new(&props); - assert_eq!( - table_properties.metadata_compression_codec().unwrap(), - CompressionCodec::gzip_default() - ); + for (value, expected) in [ + ("gzip", CompressionCodec::gzip_default()), + ("zstd", CompressionCodec::zstd_default()), + ] { + let props = HashMap::from([( + TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(), + value.to_string(), + )]); + let table_properties = TableProperties::new(&props); + assert_eq!( + table_properties.metadata_compression_codec().unwrap(), + expected + ); + } } #[test] @@ -727,38 +741,23 @@ mod tests { #[test] fn test_table_properties_compression_case_insensitive() { - // Test uppercase - let props_upper = HashMap::from([( - TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(), - "GZIP".to_string(), - )]); - let table_properties = TableProperties::new(&props_upper); - assert_eq!( - table_properties.metadata_compression_codec().unwrap(), - CompressionCodec::gzip_default() - ); - - // Test mixed case - let props_mixed = HashMap::from([( - TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(), - "GzIp".to_string(), - )]); - let table_properties = TableProperties::new(&props_mixed); - assert_eq!( - table_properties.metadata_compression_codec().unwrap(), - CompressionCodec::gzip_default() - ); - - // Test "NONE" should also be case-insensitive - let props_none_upper = HashMap::from([( - TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(), - "NONE".to_string(), - )]); - let table_properties = TableProperties::new(&props_none_upper); - assert_eq!( - table_properties.metadata_compression_codec().unwrap(), - CompressionCodec::None - ); + for (value, expected) in [ + ("GZIP", CompressionCodec::gzip_default()), + ("GzIp", CompressionCodec::gzip_default()), + ("ZSTD", CompressionCodec::zstd_default()), + ("ZsTd", CompressionCodec::zstd_default()), + ("NONE", CompressionCodec::None), + ] { + let props = HashMap::from([( + TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(), + value.to_string(), + )]); + let table_properties = TableProperties::new(&props); + assert_eq!( + table_properties.metadata_compression_codec().unwrap(), + expected + ); + } } #[test] @@ -859,7 +858,7 @@ mod tests { #[test] fn test_table_properties_compression_invalid_rejected() { - let invalid_codecs = ["lz4", "zstd", "snappy"]; + let invalid_codecs = ["lz4", "snappy"]; for codec in invalid_codecs { let props = HashMap::from([( @@ -875,7 +874,7 @@ mod tests { "Expected error message to contain codec '{codec}', got: {err_msg}" ); assert!( - err_msg.contains("Only 'none' and 'gzip' are supported"), + err_msg.contains("Only 'none', 'gzip', and 'zstd' are supported"), "Expected error message to contain supported codecs, got: {err_msg}" ); } diff --git a/crates/iceberg/testdata/table_metadata/TableMetadataV2Valid.zstd.metadata.json b/crates/iceberg/testdata/table_metadata/TableMetadataV2Valid.zstd.metadata.json new file mode 100644 index 0000000000..4709076c12 Binary files /dev/null and b/crates/iceberg/testdata/table_metadata/TableMetadataV2Valid.zstd.metadata.json differ