From ca7d5ba577ec3b2d473d911638ecdcb837c9a9b5 Mon Sep 17 00:00:00 2001 From: Xander Date: Fri, 24 Jul 2026 13:20:56 +0100 Subject: [PATCH 1/3] feat(paquet): wire in additional parquet writer settings --- crates/iceberg/public-api.txt | 19 +- crates/iceberg/src/spec/table_properties.rs | 172 +++++++++++++ .../src/writer/file_writer/parquet_writer.rs | 228 +++++++++++++++++- .../datafusion/src/physical_plan/write.rs | 1 + 4 files changed, 407 insertions(+), 13 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index f5a54304df..1d5ef2f8c7 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -2750,6 +2750,12 @@ pub iceberg::spec::TableProperties::max_ref_age_ms: i64 pub iceberg::spec::TableProperties::max_snapshot_age_ms: i64 pub iceberg::spec::TableProperties::metadata_compression_codec: iceberg::compression::CompressionCodec pub iceberg::spec::TableProperties::min_snapshots_to_keep: usize +pub iceberg::spec::TableProperties::parquet_compression_codec: alloc::string::String +pub iceberg::spec::TableProperties::parquet_compression_level: core::option::Option +pub iceberg::spec::TableProperties::parquet_dict_size_bytes: usize +pub iceberg::spec::TableProperties::parquet_page_row_limit: usize +pub iceberg::spec::TableProperties::parquet_page_size_bytes: usize +pub iceberg::spec::TableProperties::parquet_row_group_size_bytes: usize pub iceberg::spec::TableProperties::write_datafusion_fanout_enabled: bool pub iceberg::spec::TableProperties::write_format_default: alloc::string::String pub iceberg::spec::TableProperties::write_metadata_path: core::option::Option @@ -2798,6 +2804,17 @@ pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE: & pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT: usize pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL: &str pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT: i32 +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC: &str +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT: &str +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL: &str +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES: &str +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT: usize +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT: &str +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT: usize +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES: &str +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT: usize +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES: &str +pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT: usize pub const iceberg::spec::TableProperties::PROPERTY_SNAPSHOT_COUNT: &str pub const iceberg::spec::TableProperties::PROPERTY_UUID: &str pub const iceberg::spec::TableProperties::PROPERTY_WRITE_METADATA_PATH: &str @@ -3270,7 +3287,7 @@ pub async fn iceberg::writer::file_writer::ParquetWriter::close(self) -> iceberg pub async fn iceberg::writer::file_writer::ParquetWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<()> pub struct iceberg::writer::file_writer::ParquetWriterBuilder impl iceberg::writer::file_writer::ParquetWriterBuilder -pub fn iceberg::writer::file_writer::ParquetWriterBuilder::from_table_properties(table_props: &iceberg::spec::TableProperties, schema: iceberg::spec::SchemaRef) -> Self +pub fn iceberg::writer::file_writer::ParquetWriterBuilder::from_table_properties(table_props: &iceberg::spec::TableProperties, schema: iceberg::spec::SchemaRef) -> iceberg::Result pub fn iceberg::writer::file_writer::ParquetWriterBuilder::new(props: parquet::file::properties::WriterProperties, schema: iceberg::spec::SchemaRef) -> Self pub fn iceberg::writer::file_writer::ParquetWriterBuilder::new_with_match_mode(props: parquet::file::properties::WriterProperties, schema: iceberg::spec::SchemaRef, match_mode: iceberg::arrow::FieldMatchMode) -> Self pub fn iceberg::writer::file_writer::ParquetWriterBuilder::with_match_mode(self, match_mode: iceberg::arrow::FieldMatchMode) -> Self diff --git a/crates/iceberg/src/spec/table_properties.rs b/crates/iceberg/src/spec/table_properties.rs index 379feee5c1..f7915916e6 100644 --- a/crates/iceberg/src/spec/table_properties.rs +++ b/crates/iceberg/src/spec/table_properties.rs @@ -40,6 +40,28 @@ where }) } +/// Parse an optional property, returning `None` when the key is absent and an +/// error when the value is present but fails to parse. +fn parse_optional_property( + properties: &HashMap, + key: &str, +) -> Result> +where + ::Err: Display, +{ + properties + .get(key) + .map(|value| { + value.parse::().map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid value for {key}: {e}"), + ) + }) + }) + .transpose() +} + /// Strips trailing slashes from a location, preserving a bare URI scheme root fn strip_trailing_slash(path: &str) -> &str { let mut path = path; @@ -168,6 +190,20 @@ pub struct TableProperties { pub cdc_max_chunk_size: usize, /// Content-defined chunking normalization level (gearhash bit adjustment). pub cdc_norm_level: i32, + /// Parquet compression codec name (e.g. `zstd`, `gzip`). Validated when the + /// writer is built, not when properties are parsed. + pub parquet_compression_codec: String, + /// Parquet compression level for codecs that accept one. `None` uses the + /// codec's default level. + pub parquet_compression_level: Option, + /// Approximate maximum Parquet row group size in bytes. + pub parquet_row_group_size_bytes: usize, + /// Approximate maximum Parquet data page size in bytes. + pub parquet_page_size_bytes: usize, + /// Maximum number of rows per Parquet data page. + pub parquet_page_row_limit: usize, + /// Approximate maximum Parquet dictionary page size in bytes. + pub parquet_dict_size_bytes: usize, /// The master key id used to encrypt this table's manifest list and data /// files. `None` if `encryption.key-id` is not set. pub encryption_key_id: Option, @@ -317,6 +353,36 @@ impl TableProperties { /// Default matches `parquet::file::properties::DEFAULT_CDC_NORM_LEVEL`. pub const PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT: i32 = 0; + /// Compression codec for Parquet data files (e.g. `zstd`, `gzip`, `snappy`, + /// `lz4`, `brotli`, `uncompressed`). The codec name is validated when the + /// writer is built, not when properties are parsed. + pub const PROPERTY_PARQUET_COMPRESSION_CODEC: &str = "write.parquet.compression-codec"; + /// Default Parquet compression codec. + pub const PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT: &str = "zstd"; + /// Compression level for Parquet data files, for codecs that take one + /// (`gzip`, `zstd`, `brotli`). When unset, the codec's default level is used. + pub const PROPERTY_PARQUET_COMPRESSION_LEVEL: &str = "write.parquet.compression-level"; + + /// Approximate maximum size of a Parquet row group in bytes. + pub const PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES: &str = "write.parquet.row-group-size-bytes"; + /// Default Parquet row group size in bytes. + pub const PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT: usize = 128 * 1024 * 1024; + + /// Approximate maximum size of a Parquet data page in bytes. + pub const PROPERTY_PARQUET_PAGE_SIZE_BYTES: &str = "write.parquet.page-size-bytes"; + /// Default Parquet page size in bytes. + pub const PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT: usize = 1024 * 1024; + + /// Maximum number of rows per Parquet data page. + pub const PROPERTY_PARQUET_PAGE_ROW_LIMIT: &str = "write.parquet.page-row-limit"; + /// Default Parquet page row limit. + pub const PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT: usize = 20000; + + /// Approximate maximum size of the Parquet dictionary page in bytes. + pub const PROPERTY_PARQUET_DICT_SIZE_BYTES: &str = "write.parquet.dict-size-bytes"; + /// Default Parquet dictionary page size in bytes. + pub const PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT: usize = 2 * 1024 * 1024; + /// Property key for the master key id used to encrypt the table's manifest /// list and data files as defined in https://iceberg.apache.org/docs/nightly/encryption/. pub const PROPERTY_ENCRYPTION_KEY_ID: &str = "encryption.key-id"; @@ -413,6 +479,35 @@ impl TryFrom<&HashMap> for TableProperties { TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL, TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT, )?, + parquet_compression_codec: parse_property( + props, + TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC, + TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT.to_string(), + )?, + parquet_compression_level: parse_optional_property( + props, + TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL, + )?, + parquet_row_group_size_bytes: parse_property( + props, + TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES, + TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT, + )?, + parquet_page_size_bytes: parse_property( + props, + TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES, + TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT, + )?, + parquet_page_row_limit: parse_property( + props, + TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT, + TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT, + )?, + parquet_dict_size_bytes: parse_property( + props, + TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES, + TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT, + )?, encryption_key_id: props .get(TableProperties::PROPERTY_ENCRYPTION_KEY_ID) .cloned(), @@ -947,4 +1042,81 @@ mod tests { let tp = TableProperties::try_from(&props).unwrap(); assert!(!tp.cdc_enabled); } + + #[test] + fn test_parquet_sizing_defaults() { + let tp = TableProperties::try_from(&HashMap::new()).unwrap(); + assert_eq!( + tp.parquet_compression_codec, + TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT + ); + assert_eq!(tp.parquet_compression_level, None); + assert_eq!( + tp.parquet_row_group_size_bytes, + TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT + ); + assert_eq!( + tp.parquet_page_size_bytes, + TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT + ); + assert_eq!( + tp.parquet_page_row_limit, + TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT + ); + assert_eq!( + tp.parquet_dict_size_bytes, + TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT + ); + } + + #[test] + fn test_parquet_sizing_overrides() { + let props = HashMap::from([ + ( + TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC.to_string(), + "gzip".to_string(), + ), + ( + TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL.to_string(), + "4".to_string(), + ), + ( + TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES.to_string(), + "1048576".to_string(), + ), + ( + TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES.to_string(), + "65536".to_string(), + ), + ( + TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT.to_string(), + "5000".to_string(), + ), + ( + TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES.to_string(), + "131072".to_string(), + ), + ]); + let tp = TableProperties::try_from(&props).unwrap(); + assert_eq!(tp.parquet_compression_codec, "gzip"); + assert_eq!(tp.parquet_compression_level, Some(4)); + assert_eq!(tp.parquet_row_group_size_bytes, 1048576); + assert_eq!(tp.parquet_page_size_bytes, 65536); + assert_eq!(tp.parquet_page_row_limit, 5000); + assert_eq!(tp.parquet_dict_size_bytes, 131072); + } + + #[test] + fn test_parquet_invalid_sizing_rejected() { + let props = HashMap::from([( + TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES.to_string(), + "not_a_number".to_string(), + )]); + let err = TableProperties::try_from(&props).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.to_string() + .contains(TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES) + ); + } } diff --git a/crates/iceberg/src/writer/file_writer/parquet_writer.rs b/crates/iceberg/src/writer/file_writer/parquet_writer.rs index db9f170938..a1f583d17b 100644 --- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs +++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs @@ -27,6 +27,7 @@ use itertools::Itertools; use parquet::arrow::AsyncArrowWriter; use parquet::arrow::async_reader::AsyncFileReader; use parquet::arrow::async_writer::AsyncFileWriter as ArrowAsyncFileWriter; +use parquet::basic::{BrotliLevel, Compression, GzipLevel, ZstdLevel}; use parquet::file::metadata::ParquetMetaData; use parquet::file::properties::{CdcOptions, WriterProperties}; use parquet::file::statistics::Statistics; @@ -82,22 +83,26 @@ impl ParquetWriterBuilder { /// schema, translating `write.parquet.*` settings into `WriterProperties` /// instead of using parquet-rs defaults. /// - /// Currently translates the content-defined-chunking keys - /// (`write.parquet.content-defined-chunking.*`); other keys fall back to /// parquet-rs defaults. - pub fn from_table_properties(table_props: &TableProperties, schema: SchemaRef) -> Self { + pub fn from_table_properties(table_props: &TableProperties, schema: SchemaRef) -> Result { let cdc = table_props.cdc_enabled.then_some(CdcOptions { min_chunk_size: table_props.cdc_min_chunk_size, max_chunk_size: table_props.cdc_max_chunk_size, norm_level: table_props.cdc_norm_level, }); - // TODO: translate the remaining write.parquet.* keys (e.g. compression-codec, - // row-group-size-bytes, page-size-bytes). - // This constructor is intended to be the single place that maps them. + let compression = parquet_compression( + &table_props.parquet_compression_codec, + table_props.parquet_compression_level, + )?; let props = WriterProperties::builder() .set_content_defined_chunking(cdc) + .set_compression(compression) + .set_max_row_group_bytes(Some(table_props.parquet_row_group_size_bytes)) + .set_data_page_size_limit(table_props.parquet_page_size_bytes) + .set_data_page_row_count_limit(table_props.parquet_page_row_limit) + .set_dictionary_page_size_limit(table_props.parquet_dict_size_bytes) .build(); - Self::new_with_match_mode(props, schema, FieldMatchMode::Id) + Ok(Self::new_with_match_mode(props, schema, FieldMatchMode::Id)) } /// Set the field match mode used to map Arrow fields to Iceberg fields. @@ -110,6 +115,63 @@ impl ParquetWriterBuilder { } } +fn parquet_compression(codec: &str, level: Option) -> Result { + let compression = match codec.to_lowercase().as_str() { + "uncompressed" | "none" => Compression::UNCOMPRESSED, + "snappy" => Compression::SNAPPY, + "lzo" => Compression::LZO, + "lz4" => Compression::LZ4, + "lz4_raw" => Compression::LZ4_RAW, + "gzip" => { + let level = match level { + Some(l) => GzipLevel::try_new(level_as_u32("gzip", l)?) + .map_err(|e| invalid_level_error("gzip", e))?, + None => GzipLevel::default(), + }; + Compression::GZIP(level) + } + "brotli" => { + let level = match level { + Some(l) => BrotliLevel::try_new(level_as_u32("brotli", l)?) + .map_err(|e| invalid_level_error("brotli", e))?, + None => BrotliLevel::default(), + }; + Compression::BROTLI(level) + } + "zstd" => { + let level = match level { + Some(l) => ZstdLevel::try_new(l).map_err(|e| invalid_level_error("zstd", e))?, + None => ZstdLevel::default(), + }; + Compression::ZSTD(level) + } + other => { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Unsupported Parquet compression codec: {other}. Supported codecs: \ + uncompressed, snappy, gzip, lzo, brotli, lz4, lz4_raw, zstd" + ), + )); + } + }; + Ok(compression) +} + +fn invalid_level_error(codec: &str, source: impl Into) -> Error { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid {codec} compression level"), + ) + .with_source(source) +} + +/// Resolve a `u32` compression level for the gzip/brotli codecs, whose +/// parquet-rs levels are unsigned. A negative level is invalid. +fn level_as_u32(codec: &str, level: i32) -> Result { + u32::try_from(level).map_err(|e| invalid_level_error(codec, e)) +} + impl FileWriterBuilder for ParquetWriterBuilder { type R = ParquetWriter; @@ -664,6 +726,7 @@ mod tests { use arrow_select::concat::concat_batches; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; use parquet::file::statistics::ValueStatistics; + use parquet::schema::types::ColumnPath; use tempfile::TempDir; use uuid::Uuid; @@ -2341,10 +2404,6 @@ mod tests { assert_eq!(upper_bounds, HashMap::from([(0, Datum::int(i32::MAX))])); } - // ----------------------------------------------------------------- - // ParquetWriterBuilder::from_table_properties - // ----------------------------------------------------------------- - fn cdc_test_schema() -> SchemaRef { Arc::new( Schema::builder() @@ -2366,7 +2425,7 @@ mod tests { #[test] fn test_from_table_properties_no_cdc_by_default() { let tp = table_props(HashMap::new()); - let builder = ParquetWriterBuilder::from_table_properties(&tp, cdc_test_schema()); + let builder = ParquetWriterBuilder::from_table_properties(&tp, cdc_test_schema()).unwrap(); assert!(builder.props.content_defined_chunking().is_none()); } @@ -2404,6 +2463,7 @@ mod tests { .new_output(format!("{}/cdc.parquet", tmp.path().to_str().unwrap())) .unwrap(); let writer = ParquetWriterBuilder::from_table_properties(&tp, cdc_test_schema()) + .unwrap() .build(output) .await .unwrap(); @@ -2417,4 +2477,148 @@ mod tests { assert_eq!(cdc.max_chunk_size, 8192); assert_eq!(cdc.norm_level, 2); } + + #[test] + fn test_from_table_properties_sizing_defaults() { + // With no properties set, the writer must use Iceberg's defaults (which + // differ from parquet-rs's own defaults), not parquet-rs's. + let tp = table_props(HashMap::new()); + let props = ParquetWriterBuilder::from_table_properties(&tp, cdc_test_schema()) + .unwrap() + .props; + + assert_eq!( + props.max_row_group_bytes(), + Some(TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT) + ); + assert_eq!( + props.data_page_size_limit(), + TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT + ); + assert_eq!( + props.data_page_row_count_limit(), + TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT + ); + assert_eq!( + props.dictionary_page_size_limit(), + TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT + ); + // Default codec is zstd at parquet-rs's default level. + assert_eq!( + props.compression(&ColumnPath::from("id")), + Compression::ZSTD(ZstdLevel::default()) + ); + } + + #[test] + fn test_from_table_properties_sizing_and_compression_overrides() { + let tp = table_props(HashMap::from([ + ( + TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES.to_string(), + "1048576".to_string(), + ), + ( + TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES.to_string(), + "65536".to_string(), + ), + ( + TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT.to_string(), + "5000".to_string(), + ), + ( + TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES.to_string(), + "131072".to_string(), + ), + ( + TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC.to_string(), + "gzip".to_string(), + ), + ( + TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL.to_string(), + "9".to_string(), + ), + ])); + let props = ParquetWriterBuilder::from_table_properties(&tp, cdc_test_schema()) + .unwrap() + .props; + + assert_eq!(props.max_row_group_bytes(), Some(1048576)); + assert_eq!(props.data_page_size_limit(), 65536); + assert_eq!(props.data_page_row_count_limit(), 5000); + assert_eq!(props.dictionary_page_size_limit(), 131072); + assert_eq!( + props.compression(&ColumnPath::from("id")), + Compression::GZIP(GzipLevel::try_new(9).unwrap()) + ); + } + + #[test] + fn test_from_table_properties_invalid_codec_errors() { + let tp = table_props(HashMap::from([( + TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC.to_string(), + "bogus".to_string(), + )])); + let err = ParquetWriterBuilder::from_table_properties(&tp, cdc_test_schema()).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.to_string().contains("bogus")); + } + + #[test] + fn test_parquet_compression_codec_mapping() { + // Codecs without a level. + assert_eq!( + parquet_compression("uncompressed", None).unwrap(), + Compression::UNCOMPRESSED + ); + assert_eq!( + parquet_compression("snappy", None).unwrap(), + Compression::SNAPPY + ); + assert_eq!(parquet_compression("lz4", None).unwrap(), Compression::LZ4); + assert_eq!( + parquet_compression("lz4_raw", None).unwrap(), + Compression::LZ4_RAW + ); + assert_eq!(parquet_compression("lzo", None).unwrap(), Compression::LZO); + + // Case-insensitive codec names. + assert_eq!( + parquet_compression("ZSTD", None).unwrap(), + Compression::ZSTD(ZstdLevel::default()) + ); + + // Level-taking codecs use their default level when none is supplied. + assert_eq!( + parquet_compression("gzip", None).unwrap(), + Compression::GZIP(GzipLevel::default()) + ); + assert_eq!( + parquet_compression("brotli", None).unwrap(), + Compression::BROTLI(BrotliLevel::default()) + ); + + // Explicit levels are honored. + assert_eq!( + parquet_compression("zstd", Some(10)).unwrap(), + Compression::ZSTD(ZstdLevel::try_new(10).unwrap()) + ); + } + + #[test] + fn test_parquet_compression_invalid_inputs() { + // Unknown codec. + assert_eq!( + parquet_compression("bogus", None).unwrap_err().kind(), + ErrorKind::DataInvalid + ); + // Level out of range for zstd (valid range is 1..=22). + assert_eq!( + parquet_compression("zstd", Some(99)).unwrap_err().kind(), + ErrorKind::DataInvalid + ); + // Negative level for a u32-based codec. + let err = parquet_compression("gzip", Some(-1)).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.to_string().contains("gzip")); + } } diff --git a/crates/integrations/datafusion/src/physical_plan/write.rs b/crates/integrations/datafusion/src/physical_plan/write.rs index a7d771ec1b..8f48b9d4cc 100644 --- a/crates/integrations/datafusion/src/physical_plan/write.rs +++ b/crates/integrations/datafusion/src/physical_plan/write.rs @@ -226,6 +226,7 @@ impl ExecutionPlan for IcebergWriteExec { &table_props, self.table.metadata().current_schema().clone(), ) + .map_err(to_datafusion_error)? .with_match_mode(FieldMatchMode::Name); let target_file_size = table_props.write_target_file_size_bytes; From 67c28d845f55d82f04f216413ff6d2e03ed3ff06 Mon Sep 17 00:00:00 2001 From: Xander Date: Fri, 31 Jul 2026 00:45:41 +0100 Subject: [PATCH 2/3] fix compression level --- .../src/writer/file_writer/parquet_writer.rs | 44 +++++++++++-------- 1 file changed, 26 insertions(+), 18 deletions(-) diff --git a/crates/iceberg/src/writer/file_writer/parquet_writer.rs b/crates/iceberg/src/writer/file_writer/parquet_writer.rs index a1f583d17b..a2e442d799 100644 --- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs +++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs @@ -47,6 +47,16 @@ use crate::transform::create_transform_function; use crate::writer::{CurrentFileStatus, DataFile}; use crate::{Error, ErrorKind, Result}; +/// Default compression levels, pinned to parquet-java's defaults so files are +/// comparable across implementations rather than tracking the parquet-rs +/// defaults, which may differ and could change without us noticing. +/// Default zstd level (parquet-rs uses 1). +const DEFAULT_ZSTD_COMPRESSION_LEVEL: i32 = 3; +/// Default gzip level. +const DEFAULT_GZIP_COMPRESSION_LEVEL: u32 = 6; +/// Default brotli level. +const DEFAULT_BROTLI_COMPRESSION_LEVEL: u32 = 1; + /// ParquetWriterBuilder is used to builder a [`ParquetWriter`] #[derive(Clone, Debug)] pub struct ParquetWriterBuilder { @@ -124,25 +134,23 @@ fn parquet_compression(codec: &str, level: Option) -> Result { "lz4_raw" => Compression::LZ4_RAW, "gzip" => { let level = match level { - Some(l) => GzipLevel::try_new(level_as_u32("gzip", l)?) - .map_err(|e| invalid_level_error("gzip", e))?, - None => GzipLevel::default(), + Some(l) => level_as_u32("gzip", l)?, + None => DEFAULT_GZIP_COMPRESSION_LEVEL, }; + let level = GzipLevel::try_new(level).map_err(|e| invalid_level_error("gzip", e))?; Compression::GZIP(level) } "brotli" => { let level = match level { - Some(l) => BrotliLevel::try_new(level_as_u32("brotli", l)?) - .map_err(|e| invalid_level_error("brotli", e))?, - None => BrotliLevel::default(), + Some(l) => level_as_u32("brotli", l)?, + None => DEFAULT_BROTLI_COMPRESSION_LEVEL, }; + let level = BrotliLevel::try_new(level).map_err(|e| invalid_level_error("brotli", e))?; Compression::BROTLI(level) } "zstd" => { - let level = match level { - Some(l) => ZstdLevel::try_new(l).map_err(|e| invalid_level_error("zstd", e))?, - None => ZstdLevel::default(), - }; + let level = level.unwrap_or(DEFAULT_ZSTD_COMPRESSION_LEVEL); + let level = ZstdLevel::try_new(level).map_err(|e| invalid_level_error("zstd", e))?; Compression::ZSTD(level) } other => { @@ -2503,10 +2511,10 @@ mod tests { props.dictionary_page_size_limit(), TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT ); - // Default codec is zstd at parquet-rs's default level. + // Default codec is zstd at the Java-aligned default level (3). assert_eq!( props.compression(&ColumnPath::from("id")), - Compression::ZSTD(ZstdLevel::default()) + Compression::ZSTD(ZstdLevel::try_new(DEFAULT_ZSTD_COMPRESSION_LEVEL).unwrap()) ); } @@ -2581,20 +2589,20 @@ mod tests { ); assert_eq!(parquet_compression("lzo", None).unwrap(), Compression::LZO); - // Case-insensitive codec names. + // Case-insensitive codec names. With no level, each codec uses its + // parquet-java-aligned default, pinned explicitly rather than inherited + // from the parquet-rs. assert_eq!( parquet_compression("ZSTD", None).unwrap(), - Compression::ZSTD(ZstdLevel::default()) + Compression::ZSTD(ZstdLevel::try_new(DEFAULT_ZSTD_COMPRESSION_LEVEL).unwrap()) ); - - // Level-taking codecs use their default level when none is supplied. assert_eq!( parquet_compression("gzip", None).unwrap(), - Compression::GZIP(GzipLevel::default()) + Compression::GZIP(GzipLevel::try_new(DEFAULT_GZIP_COMPRESSION_LEVEL).unwrap()) ); assert_eq!( parquet_compression("brotli", None).unwrap(), - Compression::BROTLI(BrotliLevel::default()) + Compression::BROTLI(BrotliLevel::try_new(DEFAULT_BROTLI_COMPRESSION_LEVEL).unwrap()) ); // Explicit levels are honored. From da6cb46307f16aabcbac851f465ecdced405e7e8 Mon Sep 17 00:00:00 2001 From: Xander Date: Fri, 31 Jul 2026 00:55:35 +0100 Subject: [PATCH 3/3] fmt --- crates/iceberg/src/writer/file_writer/parquet_writer.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/iceberg/src/writer/file_writer/parquet_writer.rs b/crates/iceberg/src/writer/file_writer/parquet_writer.rs index a2e442d799..bab6c9cd38 100644 --- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs +++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs @@ -145,7 +145,8 @@ fn parquet_compression(codec: &str, level: Option) -> Result { Some(l) => level_as_u32("brotli", l)?, None => DEFAULT_BROTLI_COMPRESSION_LEVEL, }; - let level = BrotliLevel::try_new(level).map_err(|e| invalid_level_error("brotli", e))?; + let level = + BrotliLevel::try_new(level).map_err(|e| invalid_level_error("brotli", e))?; Compression::BROTLI(level) } "zstd" => {