Add an annotation framework for typed table properties

Cargo.toml

@@ -21,6 +21,7 @@ members = [
   "crates/catalog/*",
   "crates/examples",
   "crates/iceberg",
+  "crates/property-macro",
   "crates/integration_tests",
   "crates/integrations/*",
   "crates/sqllogictest",

crates/iceberg/Cargo.toml

@@ -59,6 +59,7 @@ flate2 = { workspace = true }
 fnv = { workspace = true }
 form_urlencoded = { workspace = true }
 futures = { workspace = true }
+iceberg-property-macro = { version = "0.10.0", path = "../property-macro" }
 itertools = { workspace = true }
 moka = { version = "0.12.10", features = ["future"] }
 murmur3 = { workspace = true }

crates/iceberg/src/catalog/utils.rs

@@ -61,7 +61,7 @@ pub async fn drop_table_data(table_info: &Table) -> Result<()> {
     }
 
     // Delete data files only if gc.enabled is true, to avoid corrupting shared tables
-    if metadata.table_properties()?.gc_enabled {
+    if *metadata.table_properties()?.gc_enabled() {
         delete_data_files(io, &manifests_to_delete).await?;
     }
 

crates/iceberg/src/compression.rs

@@ -17,6 +17,7 @@
 
 //! Compression codec support for data compression and decompression.
 
+use std::collections::HashMap;
 use std::fmt;
 use std::io::{Read, Write};
 
@@ -40,8 +41,12 @@ pub enum CompressionCodec {
     #[default]
     /// No compression
     None,
+    /// Brotli compression
+    Brotli,
     /// LZ4 single compression frame with content size present
     Lz4,
+    /// LZO compression
+    Lzo,
     /// Zstandard single compression frame with content size present.
     /// Level range is 0–22, where 0 means default compression level (not no compression).
     /// Use [`CompressionCodec::zstd_default`] to construct with the default level.
@@ -51,6 +56,8 @@ pub enum CompressionCodec {
     Gzip(u8),
     /// Snappy compression
     Snappy,
+    /// Zlib compression
+    Zlib,
 }
 
 impl CompressionCodec {
@@ -68,10 +75,80 @@ impl CompressionCodec {
     pub fn name(&self) -> &'static str {
         match self {
             CompressionCodec::None => "none",
+            CompressionCodec::Brotli => "brotli",
             CompressionCodec::Lz4 => "lz4",
+            CompressionCodec::Lzo => "lzo",
             CompressionCodec::Zstd(_) => "zstd",
             CompressionCodec::Gzip(_) => "gzip",
             CompressionCodec::Snappy => "snappy",
+            CompressionCodec::Zlib => "zlib",
+        }
+    }
+
+    /// Parses a compression codec name used by an Iceberg table property.
+    pub(crate) fn parse_property(value: &str) -> Result<Self> {
+        serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| {
+            Error::new(
+                ErrorKind::DataInvalid,
+                format!("Invalid compression codec: {value}"),
+            )
+        })
+    }
+
+    /// Parses the metadata-file compression codec table property.
+    pub(crate) fn parse_metadata_property(value: &str) -> Result<Self> {
+        match value.to_ascii_lowercase().as_str() {
+            "" | "none" => Ok(Self::None),
+            "gzip" => Ok(Self::gzip_default()),
+            _ => Err(Error::new(
+                ErrorKind::DataInvalid,
+                format!(
+                    "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.",
+                    Self::None.name(),
+                    Self::gzip_default().name()
+                ),
+            )),
+        }
+    }
+
+    /// Returns the codec name used by an Iceberg table property.
+    pub(crate) fn property_value(&self) -> String {
+        self.name().to_string()
+    }
+
+    /// Parses a codec and its optional companion compression-level property.
+    pub(crate) fn parse_properties(
+        properties: &HashMap<String, String>,
+        codec_key: &str,
+        level_key: &str,
+        default: Self,
+    ) -> Result<Self> {
+        let codec = properties
+            .get(codec_key)
+            .map(|value| Self::parse_property(value))
+            .transpose()?
+            .unwrap_or(default);
+        let Some(level) = properties.get(level_key) else {
+            return Ok(codec);
+        };
+        let level = level.parse::<u8>().map_err(|error| {
+            Error::new(
+                ErrorKind::DataInvalid,
+                format!("Invalid compression level for {level_key}: {level}"),
+            )
+            .with_source(error)
+        })?;
+
+        match codec {
+            Self::Gzip(_) => Ok(Self::Gzip(level)),
+            Self::Zstd(_) => Ok(Self::Zstd(level)),
+            _ => Err(Error::new(
+                ErrorKind::DataInvalid,
+                format!(
+                    "Compression level {level_key} is not supported for codec '{}'",
+                    codec.name()
+                ),
+            )),
         }
     }
 }
@@ -90,13 +167,24 @@ impl<'de> Deserialize<'de> for CompressionCodec {
     fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
         let s = String::deserialize(deserializer)?;
         match s.to_lowercase().as_str() {
-            "none" => Ok(CompressionCodec::None),
+            "none" | "uncompressed" => Ok(CompressionCodec::None),
+            "brotli" => Ok(CompressionCodec::Brotli),
             "lz4" => Ok(CompressionCodec::Lz4),
+            "lzo" => Ok(CompressionCodec::Lzo),
             "zstd" => Ok(CompressionCodec::zstd_default()),
             "gzip" => Ok(CompressionCodec::gzip_default()),
             "snappy" => Ok(CompressionCodec::Snappy),
+            "zlib" => Ok(CompressionCodec::Zlib),
             other => Err(serde::de::Error::unknown_variant(other, &[
-                "none", "lz4", "zstd", "gzip", "snappy",
+                "none",
+                "uncompressed",
+                "brotli",
+                "lz4",
+                "lzo",
+                "zstd",
+                "gzip",
+                "snappy",
+                "zlib",
             ])),
         }
     }
@@ -106,10 +194,13 @@ impl fmt::Display for CompressionCodec {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             CompressionCodec::None => write!(f, "None"),
+            CompressionCodec::Brotli => write!(f, "Brotli"),
             CompressionCodec::Lz4 => write!(f, "Lz4"),
+            CompressionCodec::Lzo => write!(f, "Lzo"),
             CompressionCodec::Zstd(level) => write!(f, "Zstd(level={level})"),
             CompressionCodec::Gzip(level) => write!(f, "Gzip(level={level})"),
             CompressionCodec::Snappy => write!(f, "Snappy"),
+            CompressionCodec::Zlib => write!(f, "Zlib"),
         }
     }
 }
@@ -122,17 +213,23 @@ impl CompressionCodec {
                 ErrorKind::FeatureUnsupported,
                 "LZ4 decompression is not supported currently",
             )),
+            CompressionCodec::Snappy => Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                "Snappy decompression is not supported currently",
+            )),
+            codec @ (CompressionCodec::Brotli | CompressionCodec::Lzo | CompressionCodec::Zlib) => {
+                Err(Error::new(
+                    ErrorKind::FeatureUnsupported,
+                    format!("{codec} decompression is not supported currently"),
+                ))
+            }
             CompressionCodec::Zstd(_) => Ok(zstd::stream::decode_all(&bytes[..])?),
             CompressionCodec::Gzip(_) => {
                 let mut decoder = GzDecoder::new(&bytes[..]);
                 let mut decompressed = Vec::new();
                 decoder.read_to_end(&mut decompressed)?;
                 Ok(decompressed)
             }
-            CompressionCodec::Snappy => Err(Error::new(
-                ErrorKind::FeatureUnsupported,
-                "Snappy decompression is not supported currently",
-            )),
         }
     }
 
@@ -143,6 +240,16 @@ impl CompressionCodec {
                 ErrorKind::FeatureUnsupported,
                 "LZ4 compression is not supported currently",
             )),
+            CompressionCodec::Snappy => Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                "Snappy compression is not supported currently",
+            )),
+            codec @ (CompressionCodec::Brotli | CompressionCodec::Lzo | CompressionCodec::Zlib) => {
+                Err(Error::new(
+                    ErrorKind::FeatureUnsupported,
+                    format!("{codec} compression is not supported currently"),
+                ))
+            }
             CompressionCodec::Zstd(level) => {
                 let writer = Vec::<u8>::new();
                 let mut encoder = zstd::stream::Encoder::new(writer, *level as i32)?;
@@ -157,10 +264,6 @@ impl CompressionCodec {
                 encoder.write_all(&bytes)?;
                 Ok(encoder.finish()?)
             }
-            CompressionCodec::Snappy => Err(Error::new(
-                ErrorKind::FeatureUnsupported,
-                "Snappy compression is not supported currently",
-            )),
         }
     }
 
@@ -173,14 +276,17 @@ impl CompressionCodec {
     ///
     /// # Errors
     ///
-    /// Returns an error for Lz4 and Zstd as they are not fully supported.
+    /// Returns an error when the codec does not have a defined suffix.
     pub fn suffix(&self) -> Result<&'static str> {
         match self {
             CompressionCodec::None => Ok(""),
             CompressionCodec::Gzip(_) => Ok(".gz"),
-            codec @ (CompressionCodec::Lz4
+            codec @ (CompressionCodec::Brotli
+            | CompressionCodec::Lz4
+            | CompressionCodec::Lzo
             | CompressionCodec::Zstd(_)
-            | CompressionCodec::Snappy) => Err(Error::new(
+            | CompressionCodec::Snappy
+            | CompressionCodec::Zlib) => Err(Error::new(
                 ErrorKind::FeatureUnsupported,
                 format!("suffix not defined for {codec:?}"),
             )),
@@ -196,10 +302,9 @@ mod tests {
     async fn test_compression_codec_none() {
         let bytes_vec = [0_u8; 100].to_vec();
 
-        let codec = CompressionCodec::None;
-        let compressed = codec.compress(bytes_vec.clone()).unwrap();
+        let compressed = CompressionCodec::None.compress(bytes_vec.clone()).unwrap();
         assert_eq!(bytes_vec, compressed);
-        let decompressed = codec.decompress(compressed).unwrap();
+        let decompressed = CompressionCodec::None.decompress(compressed).unwrap();
         assert_eq!(bytes_vec, decompressed);
     }
 
@@ -225,6 +330,9 @@ mod tests {
         let unsupported_codecs = [
             (CompressionCodec::Lz4, "LZ4"),
             (CompressionCodec::Snappy, "Snappy"),
+            (CompressionCodec::Brotli, "Brotli"),
+            (CompressionCodec::Lzo, "Lzo"),
+            (CompressionCodec::Zlib, "Zlib"),
         ];
         let bytes_vec = [0_u8; 100].to_vec();
 
@@ -273,4 +381,31 @@ mod tests {
         assert_eq!(CompressionCodec::Gzip(9).to_string(), "Gzip(level=9)");
         assert_eq!(CompressionCodec::Snappy.to_string(), "Snappy");
     }
+
+    #[test]
+    fn test_serde_names() {
+        let codecs = [
+            ("none", CompressionCodec::None),
+            ("brotli", CompressionCodec::Brotli),
+            ("lz4", CompressionCodec::Lz4),
+            ("lzo", CompressionCodec::Lzo),
+            ("zstd", CompressionCodec::zstd_default()),
+            ("gzip", CompressionCodec::gzip_default()),
+            ("snappy", CompressionCodec::Snappy),
+            ("zlib", CompressionCodec::Zlib),
+        ];
+
+        for (name, codec) in codecs {
+            assert_eq!(serde_json::to_value(codec).unwrap(), name);
+            assert_eq!(
+                serde_json::from_value::<CompressionCodec>(serde_json::json!(name)).unwrap(),
+                codec
+            );
+        }
+
+        assert_eq!(
+            serde_json::from_value::<CompressionCodec>(serde_json::json!("uncompressed")).unwrap(),
+            CompressionCodec::None
+        );
+    }
 }

crates/iceberg/src/encryption/manager.rs

@@ -119,7 +119,7 @@ impl EncryptionManager {
         }
 
         let table_properties = metadata.table_properties()?;
-        let Some(table_key_id) = table_properties.encryption_key_id else {
+        let Some(table_key_id) = table_properties.encryption_key_id() else {
             if kms_client.is_some() {
                 tracing::warn!(
                     "KeyManagementClient provided but table does not have encryption.key-id set"
@@ -140,7 +140,7 @@ impl EncryptionManager {
             .table_key_id(table_key_id)
             .encryption_keys(metadata.encryption_keys.clone())
             .key_size(AesKeySize::from_key_length(
-                table_properties.encryption_data_key_length,
+                *table_properties.encryption_data_key_length(),
             )?)
             .build();
         Ok(Some(Arc::new(em)))

crates/iceberg/src/spec/mod.rs

@@ -30,7 +30,7 @@ mod sort;
 mod statistic_file;
 mod table_metadata;
 mod table_metadata_builder;
-mod table_properties;
+mod table_props;
 mod transform;
 mod values;
 mod view_metadata;
@@ -50,8 +50,7 @@ pub use sort::*;
 pub use statistic_file::*;
 pub use table_metadata::*;
 pub(crate) use table_metadata_builder::FIRST_FIELD_ID;
-pub(crate) use table_properties::parse_metadata_file_compression;
-pub use table_properties::*;
+pub use table_props::*;
 pub use transform::*;
 pub(crate) use values::decimal_utils;
 pub use values::*;