feat!: Support compression codecs for Avro Files (including manifest and manifest lists)

Cargo.toml

@@ -43,7 +43,7 @@ rust-version = "1.92"
 aes = { version = "0.8", features = ["zeroize"] }
 aes-gcm = "0.10"
 anyhow = "1.0.72"
-apache-avro = { version = "0.21", features = ["zstandard"] }
+apache-avro = { version = "0.21", features = ["zstandard", "snappy"] }
 array-init = "2"
 arrow-arith = "58"
 arrow-array = "58"
@@ -98,6 +98,7 @@ log = "0.4.28"
 metainfo = "0.7.14"
 mimalloc = "0.1.46"
 minijinja = "2.12.0"
+miniz_oxide = "0.8"
 mockall = "0.13.1"
 mockito = "1"
 motore-macros = "0.4.3"

crates/iceberg/Cargo.toml

@@ -58,6 +58,8 @@ flate2 = { workspace = true }
 fnv = { workspace = true }
 futures = { workspace = true }
 itertools = { workspace = true }
+log = { workspace = true }
+miniz_oxide = { workspace = true }
 moka = { version = "0.12.10", features = ["future"] }
 murmur3 = { workspace = true }
 once_cell = { workspace = true }
@@ -89,6 +91,7 @@ mockall = { workspace = true }
 pretty_assertions = { workspace = true }
 rand = { workspace = true }
 regex = { workspace = true }
+rstest = { workspace = true }
 tempfile = { workspace = true }
 minijinja = { workspace = true }
 

crates/iceberg/src/compression.rs

@@ -173,7 +173,7 @@ impl CompressionCodec {
     ///
     /// # Errors
     ///
-    /// Returns an error for Lz4 and Zstd as they are not fully supported.
+    /// Returns an error for Lz4, Zstd, and Snappy as they are not fully supported.
     pub fn suffix(&self) -> Result<&'static str> {
         match self {
             CompressionCodec::None => Ok(""),

crates/iceberg/src/io/object_cache.rs

@@ -192,6 +192,7 @@ mod tests {
 
     use super::*;
     use crate::TableIdent;
+    use crate::compression::CompressionCodec;
     use crate::io::{FileIO, OutputFile};
     use crate::spec::{
         DataContentType, DataFileBuilder, DataFileFormat, Literal, ManifestEntry,
@@ -272,6 +273,7 @@ mod tests {
                 None,
                 current_schema.clone(),
                 current_partition_spec.as_ref().clone(),
+                CompressionCodec::None,
             )
             .build_v2_data();
             writer
@@ -304,6 +306,7 @@ mod tests {
                 current_snapshot.snapshot_id(),
                 current_snapshot.parent_snapshot_id(),
                 current_snapshot.sequence_number(),
+                CompressionCodec::None,
             );
             manifest_list_write
                 .add_manifests(vec![data_file_manifest].into_iter())

crates/iceberg/src/puffin/mod.rs

@@ -66,9 +66,10 @@ mod tests {
         assert!(validate_puffin_compression(CompressionCodec::None).is_ok());
         assert!(validate_puffin_compression(CompressionCodec::Lz4).is_ok());
         assert!(validate_puffin_compression(CompressionCodec::zstd_default()).is_ok());
-        assert!(validate_puffin_compression(CompressionCodec::Zstd(5)).is_ok());
+        assert!(validate_puffin_compression(CompressionCodec::Zstd(3)).is_ok());
 
         // Unsupported codecs
         assert!(validate_puffin_compression(CompressionCodec::gzip_default()).is_err());
+        assert!(validate_puffin_compression(CompressionCodec::Snappy).is_err());
     }
 }

crates/iceberg/src/scan/mod.rs

@@ -565,7 +565,7 @@ pub mod tests {
     //! shared tests for the table scan API
     #![allow(missing_docs)]
 
-    use std::collections::HashMap;
+    use std::collections::{HashMap, HashSet};
     use std::fs;
     use std::fs::File;
     use std::sync::Arc;
@@ -586,6 +586,7 @@ pub mod tests {
 
     use crate::TableIdent;
     use crate::arrow::ArrowReaderBuilder;
+    use crate::compression::CompressionCodec;
     use crate::expr::{BoundPredicate, Reference};
     use crate::io::{FileIO, OutputFile};
     use crate::metadata_columns::RESERVED_COL_NAME_FILE;
@@ -756,6 +757,7 @@ pub mod tests {
                 None,
                 current_schema.clone(),
                 current_partition_spec.as_ref().clone(),
+                CompressionCodec::None,
             )
             .build_v2_data();
             writer
@@ -833,6 +835,7 @@ pub mod tests {
                 current_snapshot.snapshot_id(),
                 current_snapshot.parent_snapshot_id(),
                 current_snapshot.sequence_number(),
+                CompressionCodec::None,
             );
             manifest_list_write
                 .add_manifests(vec![data_file_manifest].into_iter())
@@ -980,6 +983,7 @@ pub mod tests {
                 None,
                 current_schema.clone(),
                 current_partition_spec.as_ref().clone(),
+                CompressionCodec::None,
             )
             .build_v2_data();
 
@@ -1064,6 +1068,7 @@ pub mod tests {
                 current_snapshot.snapshot_id(),
                 current_snapshot.parent_snapshot_id(),
                 current_snapshot.sequence_number(),
+                CompressionCodec::None,
             );
             manifest_list_write
                 .add_manifests(vec![data_file_manifest].into_iter())
@@ -1086,6 +1091,7 @@ pub mod tests {
                 None,
                 current_schema.clone(),
                 current_partition_spec.as_ref().clone(),
+                CompressionCodec::None,
             )
             .build_v2_data();
 
@@ -1121,6 +1127,7 @@ pub mod tests {
                 None,
                 current_schema.clone(),
                 current_partition_spec.as_ref().clone(),
+                CompressionCodec::None,
             )
             .build_v2_deletes();
 
@@ -1155,6 +1162,7 @@ pub mod tests {
                 current_snapshot.snapshot_id(),
                 current_snapshot.parent_snapshot_id(),
                 current_snapshot.sequence_number(),
+                CompressionCodec::None,
             );
             manifest_list_write
                 .add_manifests(vec![data_manifest, delete_manifest].into_iter())
@@ -1812,8 +1820,6 @@ pub mod tests {
 
     #[tokio::test]
     async fn test_select_with_file_column() {
-        use arrow_array::cast::AsArray;
-
         let mut fixture = TableTestFixture::new();
         fixture.setup_manifest_files().await;
 
@@ -1935,8 +1941,6 @@ pub mod tests {
 
     #[tokio::test]
     async fn test_file_column_with_multiple_files() {
-        use std::collections::HashSet;
-
         let mut fixture = TableTestFixture::new();
         fixture.setup_manifest_files().await;
 

crates/iceberg/src/spec/avro_util.rs

@@ -0,0 +1,153 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Utilities for working with Apache Avro in Iceberg.
+
+use apache_avro::{Codec, DeflateSettings, ZstandardSettings};
+use miniz_oxide::deflate::CompressionLevel;
+
+use crate::compression::CompressionCodec;
+use crate::{Error, ErrorKind, Result};
+
+/// Codec name for uncompressed (Avro-specific; maps to [`CompressionCodec::None`])
+const CODEC_UNCOMPRESSED: &str = "uncompressed";
+
+/// Default compression level for gzip in Avro (matches Java implementation)
+const DEFAULT_GZIP_LEVEL: u8 = 9;
+/// Default compression level for zstd in Avro (matches Java implementation)
+const DEFAULT_ZSTD_LEVEL: u8 = 1;
+/// Max supported level for ZSTD
+const MAX_ZSTD_LEVEL: u8 = 22;
+
+/// Parse a codec name and optional level into a [`CompressionCodec`].
+///
+/// The codec name is parsed via [`CompressionCodec`]'s standard deserialization.
+/// `"uncompressed"` (Avro-specific) is mapped to [`CompressionCodec::None`].
+/// Avro-specific defaults apply when `level` is `None`: gzip→9, zstd→1.
+pub(crate) fn parse_avro_codec(codec: Option<&str>, level: Option<u8>) -> Result<CompressionCodec> {
+    let Some(codec_str) = codec else {
+        return Ok(CompressionCodec::None);
+    };
+    let normalized = if codec_str.eq_ignore_ascii_case(CODEC_UNCOMPRESSED) {
+        "none"
+    } else {
+        codec_str
+    };
+    let parsed: CompressionCodec = serde_json::from_value(serde_json::Value::String(
+        normalized.to_string(),
+    ))
+    .map_err(|_| {
+        Error::new(
+            ErrorKind::DataInvalid,
+            format!("Unrecognized Avro compression codec: {codec_str}"),
+        )
+    })?;
+    match parsed {
+        CompressionCodec::None => Ok(CompressionCodec::None),
+        CompressionCodec::Snappy => Ok(CompressionCodec::Snappy),
+        CompressionCodec::Gzip(_) => {
+            Ok(CompressionCodec::Gzip(level.unwrap_or(DEFAULT_GZIP_LEVEL)))
+        }
+        CompressionCodec::Zstd(_) => {
+            Ok(CompressionCodec::Zstd(level.unwrap_or(DEFAULT_ZSTD_LEVEL)))
+        }
+        other => Err(Error::new(
+            ErrorKind::DataInvalid,
+            format!("Unsupported Avro compression codec: {}", other.name()),
+        )),
+    }
+}
+
+/// Convert a [`CompressionCodec`] to an [`apache_avro::Codec`] for use in Avro writers.
+pub(crate) fn to_avro_codec(codec: CompressionCodec) -> Codec {
+    match codec {
+        CompressionCodec::None => Codec::Null,
+        CompressionCodec::Snappy => Codec::Snappy,
+        CompressionCodec::Lz4 => Codec::Null,
+        CompressionCodec::Gzip(level) => {
+            let compression_level = match level {
+                0 => CompressionLevel::NoCompression,
+                1 => CompressionLevel::BestSpeed,
+                9 => CompressionLevel::BestCompression,
+                10 => CompressionLevel::UberCompression,
+                _ => CompressionLevel::DefaultLevel,
+            };
+            Codec::Deflate(DeflateSettings::new(compression_level))
+        }
+        CompressionCodec::Zstd(level) => {
+            Codec::Zstandard(ZstandardSettings::new(level.min(MAX_ZSTD_LEVEL)))
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use apache_avro::{Codec, DeflateSettings, ZstandardSettings};
+    use miniz_oxide::deflate::CompressionLevel;
+    use rstest::rstest;
+
+    use super::*;
+
+    #[rstest]
+    #[case::gzip_case_insensitive(Some("GZip"), Some(5), CompressionCodec::Gzip(5))]
+    #[case::gzip_avro_default_level(Some("gzip"), None, CompressionCodec::Gzip(DEFAULT_GZIP_LEVEL))]
+    #[case::zstd_explicit_level(Some("zstd"), Some(3), CompressionCodec::Zstd(3))]
+    #[case::zstd_avro_default_level(Some("zstd"), None, CompressionCodec::Zstd(DEFAULT_ZSTD_LEVEL))]
+    #[case::snappy(Some("snappy"), None, CompressionCodec::Snappy)]
+    #[case::uncompressed_avro_alias(Some("uncompressed"), None, CompressionCodec::None)]
+    #[case::no_codec(None, None, CompressionCodec::None)]
+    fn test_parse_avro_codec(
+        #[case] codec: Option<&str>,
+        #[case] level: Option<u8>,
+        #[case] expected: CompressionCodec,
+    ) {
+        assert_eq!(parse_avro_codec(codec, level).unwrap(), expected);
+    }
+
+    #[rstest]
+    #[case::unknown_codec(Some("unknown"), Some(1), "unknown")]
+    #[case::lz4_unsupported(Some("lz4"), None, "lz4")]
+    fn test_parse_avro_codec_error(
+        #[case] codec: Option<&str>,
+        #[case] level: Option<u8>,
+        #[case] expected_msg: &str,
+    ) {
+        let err = parse_avro_codec(codec, level).unwrap_err();
+        assert_eq!(err.kind(), ErrorKind::DataInvalid);
+        assert!(
+            err.to_string().contains(expected_msg),
+            "expected '{expected_msg}' in error: {err}"
+        );
+    }
+
+    #[rstest]
+    #[case::none(CompressionCodec::None, Codec::Null)]
+    #[case::snappy(CompressionCodec::Snappy, Codec::Snappy)]
+    #[case::gzip_best_compression(
+        CompressionCodec::Gzip(9),
+        Codec::Deflate(DeflateSettings::new(CompressionLevel::BestCompression))
+    )]
+    #[case::gzip_default_level(
+        CompressionCodec::Gzip(5),
+        Codec::Deflate(DeflateSettings::new(CompressionLevel::DefaultLevel))
+    )]
+    #[case::zstd(CompressionCodec::Zstd(3), Codec::Zstandard(ZstandardSettings::new(3)))]
+    #[case::zstd_level_clamped_to_max(CompressionCodec::Zstd(MAX_ZSTD_LEVEL + 1), Codec::Zstandard(ZstandardSettings::new(MAX_ZSTD_LEVEL)))]
+    fn test_to_avro_codec(#[case] input: CompressionCodec, #[case] expected: Codec) {
+        assert_eq!(to_avro_codec(input), expected);
+    }
+}

crates/iceberg/src/spec/manifest/mod.rs

@@ -165,6 +165,7 @@ mod tests {
     use tempfile::TempDir;
 
     use super::*;
+    use crate::compression::CompressionCodec;
     use crate::io::FileIO;
     use crate::spec::{Literal, NestedField, PrimitiveType, Struct, Transform, Type};
 
@@ -272,6 +273,7 @@ mod tests {
             None,
             metadata.schema.clone(),
             metadata.partition_spec.clone(),
+            CompressionCodec::None,
         )
         .build_v2_data();
         for entry in &entries {
@@ -457,6 +459,7 @@ mod tests {
             None,
             metadata.schema.clone(),
             metadata.partition_spec.clone(),
+            CompressionCodec::None,
         )
         .build_v2_data();
         for entry in &entries {
@@ -554,6 +557,7 @@ mod tests {
             None,
             metadata.schema.clone(),
             metadata.partition_spec.clone(),
+            CompressionCodec::None,
         )
         .build_v1();
         for entry in &entries {
@@ -663,6 +667,7 @@ mod tests {
             None,
             metadata.schema.clone(),
             metadata.partition_spec.clone(),
+            CompressionCodec::None,
         )
         .build_v1();
         for entry in &entries {
@@ -771,6 +776,7 @@ mod tests {
             None,
             metadata.schema.clone(),
             metadata.partition_spec.clone(),
+            CompressionCodec::None,
         )
         .build_v2_data();
         for entry in &entries {
@@ -1050,6 +1056,7 @@ mod tests {
             None,
             metadata.schema.clone(),
             metadata.partition_spec.clone(),
+            CompressionCodec::None,
         )
         .build_v2_data();
         for entry in &entries {