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

Cargo.toml

@@ -40,7 +40,7 @@ rust-version = "1.87"
 
 [workspace.dependencies]
 anyhow = "1.0.72"
-apache-avro = { version = "0.20", features = ["zstandard"] }
+apache-avro = { version = "0.20", features = ["zstandard", "snappy"] }
 array-init = "2"
 arrow-arith = "56.2"
 arrow-array = "56.2"

crates/iceberg/Cargo.toml

@@ -67,6 +67,8 @@ flate2 = { workspace = true }
 fnv = { workspace = true }
 futures = { workspace = true }
 itertools = { workspace = true }
+log = { workspace = true }
+miniz_oxide = "0.8"
 moka = { version = "0.12.10", features = ["future"] }
 murmur3 = { workspace = true }
 num-bigint = { workspace = true }

crates/iceberg/src/io/object_cache.rs

@@ -194,8 +194,9 @@ mod tests {
     use crate::TableIdent;
     use crate::io::{FileIO, OutputFile};
     use crate::spec::{
-        DataContentType, DataFileBuilder, DataFileFormat, Literal, ManifestEntry,
-        ManifestListWriter, ManifestStatus, ManifestWriterBuilder, Struct, TableMetadata,
+        CompressionSettings, DataContentType, DataFileBuilder, DataFileFormat, Literal,
+        ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, Struct,
+        TableMetadata,
     };
     use crate::table::Table;
 
@@ -275,6 +276,7 @@ mod tests {
                 None,
                 current_schema.clone(),
                 current_partition_spec.as_ref().clone(),
+                CompressionSettings::default(),
             )
             .build_v2_data();
             writer
@@ -307,6 +309,7 @@ mod tests {
                 current_snapshot.snapshot_id(),
                 current_snapshot.parent_snapshot_id(),
                 current_snapshot.sequence_number(),
+                CompressionSettings::default(),
             );
             manifest_list_write
                 .add_manifests(vec![data_file_manifest].into_iter())

crates/iceberg/src/scan/mod.rs

@@ -577,9 +577,9 @@ pub mod tests {
     use crate::io::{FileIO, OutputFile};
     use crate::scan::FileScanTask;
     use crate::spec::{
-        DataContentType, DataFileBuilder, DataFileFormat, Datum, Literal, ManifestEntry,
-        ManifestListWriter, ManifestStatus, ManifestWriterBuilder, NestedField, PartitionSpec,
-        PrimitiveType, Schema, Struct, StructType, TableMetadata, Type,
+        CompressionSettings, DataContentType, DataFileBuilder, DataFileFormat, Datum, Literal,
+        ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, NestedField,
+        PartitionSpec, PrimitiveType, Schema, Struct, StructType, TableMetadata, Type,
     };
     use crate::table::Table;
 
@@ -749,6 +749,7 @@ pub mod tests {
                 None,
                 current_schema.clone(),
                 current_partition_spec.as_ref().clone(),
+                CompressionSettings::default(),
             )
             .build_v2_data();
             writer
@@ -826,6 +827,7 @@ pub mod tests {
                 current_snapshot.snapshot_id(),
                 current_snapshot.parent_snapshot_id(),
                 current_snapshot.sequence_number(),
+                CompressionSettings::default(),
             );
             manifest_list_write
                 .add_manifests(vec![data_file_manifest].into_iter())
@@ -961,6 +963,7 @@ pub mod tests {
                 None,
                 current_schema.clone(),
                 current_partition_spec.as_ref().clone(),
+                CompressionSettings::default(),
             )
             .build_v2_data();
 
@@ -1045,6 +1048,7 @@ pub mod tests {
                 current_snapshot.snapshot_id(),
                 current_snapshot.parent_snapshot_id(),
                 current_snapshot.sequence_number(),
+                CompressionSettings::default(),
             );
             manifest_list_write
                 .add_manifests(vec![data_file_manifest].into_iter())