You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/24 19:22:49 UTC
(datafusion) branch gh-readonly-queue/main/pr-25626-b4f813b14fca7ab556c119dcf23a87d473d56465 created (now db20a1ebe5)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25626-b4f813b14fca7ab556c119dcf23a87d473d56465
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at db20a1ebe5 fix: validate parquet.compression when it is set (#25626)
This branch includes the following new commits:
new db20a1ebe5 fix: validate parquet.compression when it is set (#25626)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: validate parquet.compression when it is set (#25626)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25626-b4f813b14fca7ab556c119dcf23a87d473d56465
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit db20a1ebe50fda78f30d7f5d8a418cd7c4c83eff
Author: akashjainn <80...@users.noreply.github.com>
AuthorDate: Thu Sep 24 19:22:36 2026 +0000
fix: validate parquet.compression when it is set (#25626)
## Which issue does this PR close?
- Part of #17498. This covers `datafusion.execution.parquet.compression`
only.
## Rationale for this change
The option is stored as a plain string, so `SET
datafusion.execution.parquet.compression = 'zstdd(3)'` succeeds, `SHOW`
reports the bad value, and the error only appears later when a Parquet
file is written. The same goes for a codec without its level (`zstd`), a
level on a codec that takes none (`snappy(2)`), and a level out of range
(`zstd(23)`). The issue is about catching these at `SET` time, and this
option was converted in the first attempt (#17697) and planned in #19498
but never landed.
## What changes are included in this PR?
- `DFParquetCompression` in `parquet_config.rs`, an enum of the seven
supported codecs with the level carried by `Gzip`, `Brotli` and `Zstd`.
`FromStr` does the validation, `Display` produces the same strings the
option stored before (`zstd(3)`, `lz4_raw`), and the `ConfigField` impls
follow `DFParquetStatistics`, including the `Option` handling that
leaves an unset value unset when a bad value is rejected.
- `ParquetOptions.compression` becomes `Option<DFParquetCompression>`,
with the same default of `zstd(3)`. The per-column
`ParquetColumnOptions.compression` stays a string, matching the scope of
#24642.
- `parse_compression_string` is now a two-line wrapper over the new
type, so there is one parser. Its error messages are unchanged.
- Proto conversion parses on the way in, so an invalid value is rejected
there too.
- The level ranges are mirrored from the parquet crate as constants,
with a test that fails if they ever differ from what `GzipLevel`,
`BrotliLevel` and `ZstdLevel` accept. That test caught my first copy of
the zstd range, which came from parquet 59; the workspace uses parquet
60, where `zstd(0)` is accepted, and it still is.
Behavior notes:
- A malformed closing parenthesis such as `zstd(3x` used to be read as
`zstd(3)` and is now rejected.
- An out-of-range level gets a DataFusion message naming the valid range
instead of the parquet crate's message.
- The error for a bad value set through `COPY ... OPTIONS` now arrives
when the option is set, wrapped in "Error setting config ...", which one
existing test in `listing/table.rs` asserted on and now expects.
This changes the type of a public field, so it should carry the `api
change` label. Two call sites in this repository that assigned a string
needed to parse instead.
## What is the testing strategy for this PR?
- `parquet_config.rs`: parsing and display for every codec, nine
rejected inputs with their messages, and the range-drift test above.
- `config.rs`: `test_parquet_compression_validation`, mirroring the
statistics test, including that a rejected value leaves the previous one
in place and that an unset value stays unset.
- `proto-common`: a round trip and a rejected value from proto.
- `set_variable.slt`: set, show, three rejected values with their full
error text, and that the setting is unchanged afterwards.
- Run locally: `datafusion-common` with all features (628 unit tests)
and without the parquet feature, `proto-common` and `proto-models`,
`datasource-parquet` (271), the `set_variable` and `information_schema`
sqllogictest files, the core listing, parquet and dataframe unit tests
(114) and `parquet_integration` (242). `cargo clippy --workspace
--all-targets --all-features -- -D warnings`, `cargo fmt --check`, typos
and `dev/update_config_docs.sh` are clean; the generated docs did not
change because the default displays the same.
## Are there any user-facing changes?
Yes. An invalid `datafusion.execution.parquet.compression` value is
rejected by `SET` and by `COPY ... OPTIONS` with a message, instead of
being accepted and failing at write time. The field type in
`ParquetOptions` changes from `Option<String>` to
`Option<DFParquetCompression>`.
---
datafusion/common/src/config.rs | 85 +++++-
.../common/src/file_options/parquet_writer.rs | 104 +-------
datafusion/common/src/parquet_config.rs | 294 +++++++++++++++++++++
datafusion/core/src/dataframe/parquet.rs | 2 +-
datafusion/core/src/datasource/listing/table.rs | 3 +-
datafusion/core/src/execution/context/parquet.rs | 4 +-
datafusion/datasource-parquet/src/file_format.rs | 2 +-
datafusion/proto-common/src/from_proto/mod.rs | 37 ++-
datafusion/proto-common/src/to_proto/mod.rs | 2 +-
datafusion/proto-models/src/from_proto.rs | 14 +-
.../sqllogictest/test_files/set_variable.slt | 42 +++
11 files changed, 482 insertions(+), 107 deletions(-)
diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs
index e36fa80f53..02ab4f5b45 100644
--- a/datafusion/common/src/config.rs
+++ b/datafusion/common/src/config.rs
@@ -23,7 +23,9 @@ use arrow_ipc::CompressionType;
use crate::encryption::{FileDecryptionProperties, FileEncryptionProperties};
use crate::error::{_config_datafusion_err, _config_err};
use crate::format::{ExplainAnalyzeCategories, ExplainFormat, MetricType};
-use crate::parquet_config::{DFParquetStatistics, DFParquetWriterVersion};
+use crate::parquet_config::{
+ DFParquetCompression, DFParquetStatistics, DFParquetWriterVersion,
+};
use crate::parsers::{CompressionTypeVariant, CsvQuoteStyle};
use crate::utils::get_available_parallelism;
use crate::{DataFusionError, Result};
@@ -1440,7 +1442,7 @@ config_namespace! {
///
/// Note that this default setting is not the same as
/// the default parquet writer setting.
- pub compression: Option<String>, transform = str::to_lowercase, default = Some("zstd(3)".into())
+ pub compression: Option<DFParquetCompression>, default = Some(DFParquetCompression::Zstd(3))
/// (writing) Sets if dictionary encoding is enabled. If NULL, uses
/// default parquet writer setting
@@ -4684,6 +4686,85 @@ mod tests {
assert_eq!(scalar, DFParquetStatistics::Page);
}
+ #[cfg(feature = "parquet")]
+ #[test]
+ fn test_parquet_compression_validation() {
+ use crate::{config::ConfigOptions, parquet_config::DFParquetCompression};
+
+ let mut config = ConfigOptions::default();
+ assert_eq!(
+ config.execution.parquet.compression,
+ Some(DFParquetCompression::Zstd(3))
+ );
+
+ for (value, expected) in [
+ ("snappy", DFParquetCompression::Snappy),
+ ("GZIP(6)", DFParquetCompression::Gzip(6)),
+ ("'zstd(22)'", DFParquetCompression::Zstd(22)),
+ ] {
+ config
+ .set("datafusion.execution.parquet.compression", value)
+ .unwrap();
+ assert_eq!(config.execution.parquet.compression, Some(expected));
+ }
+
+ for (value, message) in [
+ (
+ "zstdd(3)",
+ "Unknown or unsupported parquet compression: zstdd(3)",
+ ),
+ (
+ "zstd",
+ "zstd compression requires specifying a level such as zstd(4)",
+ ),
+ (
+ "snappy(2)",
+ "Compression snappy does not support specifying a level",
+ ),
+ ("zstd(23)", "Invalid compression level 23 for zstd"),
+ ] {
+ let err = config
+ .set("datafusion.execution.parquet.compression", value)
+ .unwrap_err();
+ assert_contains!(err.to_string(), message);
+ // A rejected value leaves the previous one in place.
+ assert_eq!(
+ config.execution.parquet.compression,
+ Some(DFParquetCompression::Zstd(22))
+ );
+ }
+
+ // An unset value can arise from deserialization. An invalid update must
+ // leave that state unchanged rather than inserting the default.
+ config.execution.parquet.compression = None;
+ assert!(
+ config
+ .set("datafusion.execution.parquet.compression", "zstd")
+ .is_err()
+ );
+ assert_eq!(config.execution.parquet.compression, None);
+
+ config.execution.parquet.compression = Some(DFParquetCompression::Lz4);
+ assert!(
+ config
+ .set("datafusion.execution.parquet.compression.typo", "snappy")
+ .is_err()
+ );
+ assert!(
+ config
+ .reset("datafusion.execution.parquet.compression.typo")
+ .is_err()
+ );
+ assert_eq!(
+ config.execution.parquet.compression,
+ Some(DFParquetCompression::Lz4)
+ );
+
+ let mut scalar = DFParquetCompression::Lz4;
+ assert!(ConfigField::set(&mut scalar, "typo", "snappy").is_err());
+ assert_eq!(scalar, DFParquetCompression::Lz4);
+ }
+
#[cfg(feature = "parquet")]
#[test]
fn set_cdc_enabled_flag() {
diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs
index c50bb42a38..af4554c4ee 100644
--- a/datafusion/common/src/file_options/parquet_writer.rs
+++ b/datafusion/common/src/file_options/parquet_writer.rs
@@ -22,13 +22,13 @@ use std::sync::Arc;
use crate::{
_internal_datafusion_err, DataFusionError, Result,
config::{ParquetCdcOptions, ParquetOptions, TableParquetOptions},
+ parquet_config::DFParquetCompression,
};
use arrow::datatypes::Schema;
use parquet::arrow::encode_arrow_schema;
use parquet::{
arrow::ARROW_SCHEMA_META_KEY,
- basic::{BrotliLevel, GzipLevel, ZstdLevel},
file::{
metadata::KeyValue,
properties::{
@@ -282,7 +282,7 @@ impl ParquetOptions {
// We do not have access to default ColumnProperties set in Arrow.
// Therefore, only overwrite if these settings exist.
if let Some(compression) = compression {
- builder = builder.set_compression(parse_compression_string(compression)?);
+ builder = builder.set_compression((*compression).into());
}
if let Some(encoding) = encoding {
builder = builder.set_encoding(parse_encoding_string(encoding)?);
@@ -320,94 +320,11 @@ pub(crate) fn parse_encoding_string(
}
}
-/// Splits compression string into compression codec and optional compression_level
-/// I.e. gzip(2) -> gzip, 2
-fn split_compression_string(str_setting: &str) -> Result<(String, Option<u32>)> {
- // ignore string literal chars passed from sqlparser i.e. remove single quotes
- let str_setting = str_setting.replace('\'', "");
- let split_setting = str_setting.split_once('(');
-
- match split_setting {
- Some((codec, rh)) => {
- let level = &rh[..rh.len() - 1].parse::<u32>().map_err(|_| {
- DataFusionError::Configuration(format!(
- "Could not parse compression string. \
- Got codec: {codec} and unknown level from {str_setting}"
- ))
- })?;
- Ok((codec.to_owned(), Some(*level)))
- }
- None => Ok((str_setting.to_owned(), None)),
- }
-}
-
-/// Helper to ensure compression codecs which don't support levels
-/// don't have one set. E.g. snappy(2) is invalid.
-fn check_level_is_none(codec: &str, level: Option<&u32>) -> Result<()> {
- if level.is_some() {
- return Err(DataFusionError::Configuration(format!(
- "Compression {codec} does not support specifying a level"
- )));
- }
- Ok(())
-}
-
-/// Helper to ensure compression codecs which require a level
-/// do have one set. E.g. zstd is invalid, zstd(3) is valid
-fn require_level(codec: &str, level: Option<u32>) -> Result<u32> {
- level.ok_or(DataFusionError::Configuration(format!(
- "{codec} compression requires specifying a level such as {codec}(4)"
- )))
-}
-
/// Parses datafusion.execution.parquet.compression String to a parquet::basic::Compression
pub fn parse_compression_string(
str_setting: &str,
) -> Result<parquet::basic::Compression> {
- let str_setting_lower: &str = &str_setting.to_lowercase();
- let (codec, level) = split_compression_string(str_setting_lower)?;
- let codec = codec.as_str();
- match codec {
- "uncompressed" => {
- check_level_is_none(codec, level.as_ref())?;
- Ok(parquet::basic::Compression::UNCOMPRESSED)
- }
- "snappy" => {
- check_level_is_none(codec, level.as_ref())?;
- Ok(parquet::basic::Compression::SNAPPY)
- }
- "gzip" => {
- let level = require_level(codec, level)?;
- Ok(parquet::basic::Compression::GZIP(GzipLevel::try_new(
- level,
- )?))
- }
- "brotli" => {
- let level = require_level(codec, level)?;
- Ok(parquet::basic::Compression::BROTLI(BrotliLevel::try_new(
- level,
- )?))
- }
- "lz4" => {
- check_level_is_none(codec, level.as_ref())?;
- Ok(parquet::basic::Compression::LZ4)
- }
- "zstd" => {
- let level = require_level(codec, level)?;
- Ok(parquet::basic::Compression::ZSTD(ZstdLevel::try_new(
- level as i32,
- )?))
- }
- "lz4_raw" => {
- check_level_is_none(codec, level.as_ref())?;
- Ok(parquet::basic::Compression::LZ4_RAW)
- }
- _ => Err(DataFusionError::Configuration(format!(
- "Unknown or unsupported parquet compression: \
- {str_setting}. Valid values are: uncompressed, snappy, gzip(level), \
- brotli(level), lz4, zstd(level), and lz4_raw."
- ))),
- }
+ Ok(str_setting.parse::<DFParquetCompression>()?.into())
}
pub(crate) fn parse_statistics_string(str_setting: &str) -> Result<EnabledStatistics> {
@@ -433,7 +350,9 @@ mod tests {
MaxRowGroupBytes, ParquetCdcOptions, ParquetColumnOptions,
ParquetEncryptionOptions, ParquetOptions,
};
- use crate::parquet_config::{DFParquetStatistics, DFParquetWriterVersion};
+ use crate::parquet_config::{
+ DFParquetCompression, DFParquetStatistics, DFParquetWriterVersion,
+ };
use parquet::basic::Compression;
use parquet::file::properties::{
BloomFilterProperties, DEFAULT_BLOOM_FILTER_FPP, DEFAULT_BLOOM_FILTER_NDV,
@@ -471,7 +390,7 @@ mod tests {
data_pagesize_limit: 42,
write_batch_size: 42,
writer_version,
- compression: Some("zstd(22)".into()),
+ compression: Some(DFParquetCompression::Zstd(22)),
dictionary_enabled: Some(!defaults.dictionary_enabled.unwrap_or(false)),
dictionary_page_size_limit: 43,
statistics_enabled: Some(DFParquetStatistics::Chunk),
@@ -597,7 +516,12 @@ mod tests {
// global options which set the default column props
encoding: default_col_props.encoding,
- compression: default_col_props.compression,
+ compression: match props.compression(&default_col) {
+ Compression::ZSTD(level) => {
+ Some(DFParquetCompression::Zstd(level.compression_level() as u32))
+ }
+ _ => None,
+ },
dictionary_enabled: default_col_props.dictionary_enabled,
statistics_enabled: Some(props.statistics_enabled(&default_col).into()),
bloom_filter_on_write: default_col_props
@@ -770,7 +694,7 @@ mod tests {
let mut from_extern_parquet =
session_config_from_writer_props(&default_writer_props);
from_extern_parquet.global.created_by = same_created_by;
- from_extern_parquet.global.compression = Some("zstd(3)".into());
+ from_extern_parquet.global.compression = Some(DFParquetCompression::Zstd(3));
from_extern_parquet.global.skip_arrow_metadata = true;
assert_eq!(
diff --git a/datafusion/common/src/parquet_config.rs b/datafusion/common/src/parquet_config.rs
index b727664e77..c9710188e0 100644
--- a/datafusion/common/src/parquet_config.rs
+++ b/datafusion/common/src/parquet_config.rs
@@ -221,3 +221,297 @@ impl From<parquet::file::properties::EnabledStatistics> for DFParquetStatistics
}
}
}
+
+/// Parquet compression codec, with a level for the codecs that take one
+///
+/// This enum validates compression settings at configuration time. An unknown
+/// codec, a level on a codec that does not take one, a missing level on a codec
+/// that requires one, or a level out of range is rejected by `SET` and by
+/// deserialization instead of when the first file is written.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum DFParquetCompression {
+ /// No compression
+ Uncompressed,
+ /// Snappy
+ Snappy,
+ /// Gzip with a level from [`GZIP_LEVELS`]
+ Gzip(u32),
+ /// Brotli with a level from [`BROTLI_LEVELS`]
+ Brotli(u32),
+ /// LZ4 (Hadoop framed)
+ Lz4,
+ /// Zstd with a level from [`ZSTD_LEVELS`]
+ Zstd(u32),
+ /// LZ4 raw block format
+ Lz4Raw,
+}
+
+/// Valid gzip levels, the same range as `parquet::basic::GzipLevel`
+pub const GZIP_LEVELS: std::ops::RangeInclusive<u32> = 0..=9;
+/// Valid brotli levels, the same range as `parquet::basic::BrotliLevel`
+pub const BROTLI_LEVELS: std::ops::RangeInclusive<u32> = 0..=11;
+/// Valid zstd levels. `parquet::basic::ZstdLevel` also accepts negative levels,
+/// which the string form of this setting has never allowed.
+pub const ZSTD_LEVELS: std::ops::RangeInclusive<u32> = 0..=22;
+
+impl DFParquetCompression {
+ /// Splits `gzip(2)` into `("gzip", Some(2))` and `snappy` into `("snappy", None)`
+ fn split(setting: &str) -> Result<(&str, Option<u32>)> {
+ let Some((codec, rest)) = setting.split_once('(') else {
+ return Ok((setting, None));
+ };
+ let level = rest
+ .strip_suffix(')')
+ .and_then(|level| level.parse::<u32>().ok())
+ .ok_or_else(|| {
+ DataFusionError::Configuration(format!(
+ "Could not parse compression string. \
+ Got codec: {codec} and unknown level from {setting}"
+ ))
+ })?;
+ Ok((codec, Some(level)))
+ }
+
+ fn without_level(codec: &str, level: Option<u32>) -> Result<()> {
+ if level.is_some() {
+ return Err(DataFusionError::Configuration(format!(
+ "Compression {codec} does not support specifying a level"
+ )));
+ }
+ Ok(())
+ }
+
+ fn level_in(
+ codec: &str,
+ level: Option<u32>,
+ valid: std::ops::RangeInclusive<u32>,
+ ) -> Result<u32> {
+ let level = level.ok_or_else(|| {
+ DataFusionError::Configuration(format!(
+ "{codec} compression requires specifying a level such as {codec}(4)"
+ ))
+ })?;
+ if !valid.contains(&level) {
+ return Err(DataFusionError::Configuration(format!(
+ "Invalid compression level {level} for {codec}. \
+ Expected a level between {} and {}",
+ valid.start(),
+ valid.end()
+ )));
+ }
+ Ok(level)
+ }
+}
+
+impl FromStr for DFParquetCompression {
+ type Err = DataFusionError;
+
+ fn from_str(s: &str) -> Result<Self, Self::Err> {
+ // ignore string literal chars passed from sqlparser i.e. remove single quotes
+ let setting = s.replace('\'', "").to_lowercase();
+ let (codec, level) = Self::split(&setting)?;
+ match codec {
+ "uncompressed" => {
+ Self::without_level(codec, level)?;
+ Ok(Self::Uncompressed)
+ }
+ "snappy" => {
+ Self::without_level(codec, level)?;
+ Ok(Self::Snappy)
+ }
+ "gzip" => Ok(Self::Gzip(Self::level_in(codec, level, GZIP_LEVELS)?)),
+ "brotli" => Ok(Self::Brotli(Self::level_in(codec, level, BROTLI_LEVELS)?)),
+ "lz4" => {
+ Self::without_level(codec, level)?;
+ Ok(Self::Lz4)
+ }
+ "zstd" => Ok(Self::Zstd(Self::level_in(codec, level, ZSTD_LEVELS)?)),
+ "lz4_raw" => {
+ Self::without_level(codec, level)?;
+ Ok(Self::Lz4Raw)
+ }
+ _ => Err(DataFusionError::Configuration(format!(
+ "Unknown or unsupported parquet compression: \
+ {s}. Valid values are: uncompressed, snappy, gzip(level), \
+ brotli(level), lz4, zstd(level), and lz4_raw."
+ ))),
+ }
+ }
+}
+
+impl Display for DFParquetCompression {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::Uncompressed => f.write_str("uncompressed"),
+ Self::Snappy => f.write_str("snappy"),
+ Self::Gzip(level) => write!(f, "gzip({level})"),
+ Self::Brotli(level) => write!(f, "brotli({level})"),
+ Self::Lz4 => f.write_str("lz4"),
+ Self::Zstd(level) => write!(f, "zstd({level})"),
+ Self::Lz4Raw => f.write_str("lz4_raw"),
+ }
+ }
+}
+
+impl ConfigField for DFParquetCompression {
+ fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+ v.some(key, self, description)
+ }
+
+ fn set(&mut self, key: &str, value: &str) -> Result<()> {
+ if !key.is_empty() {
+ return crate::error::_config_err!(
+ "Config field parquet.compression is a scalar DFParquetCompression and does not have nested field \"{}\"",
+ key
+ );
+ }
+
+ *self = Self::from_str(value)?;
+ Ok(())
+ }
+}
+
+/// `ConfigField` for `Option<DFParquetCompression>` parses before assigning so
+/// an invalid value does not turn an unset option into the default.
+impl ConfigField for Option<DFParquetCompression> {
+ fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+ match self {
+ Some(compression) => compression.visit(v, key, description),
+ None => v.none(key, description),
+ }
+ }
+
+ fn set(&mut self, key: &str, value: &str) -> Result<()> {
+ if !key.is_empty() {
+ return crate::error::_config_err!(
+ "Config field parquet.compression is a scalar Option<DFParquetCompression> and does not have nested field \"{}\"",
+ key
+ );
+ }
+
+ *self = Some(DFParquetCompression::from_str(value)?);
+ Ok(())
+ }
+
+ fn reset(&mut self, key: &str) -> Result<()> {
+ if key.is_empty() {
+ *self = None;
+ Ok(())
+ } else {
+ crate::error::_config_err!(
+ "Config field parquet.compression is a scalar Option<DFParquetCompression> and does not have nested field \"{}\"",
+ key
+ )
+ }
+ }
+}
+
+#[cfg(feature = "parquet")]
+impl From<DFParquetCompression> for parquet::basic::Compression {
+ fn from(value: DFParquetCompression) -> Self {
+ use parquet::basic::{BrotliLevel, GzipLevel, ZstdLevel};
+ // The levels were checked against the same ranges when the value was parsed.
+ match value {
+ DFParquetCompression::Uncompressed => Self::UNCOMPRESSED,
+ DFParquetCompression::Snappy => Self::SNAPPY,
+ DFParquetCompression::Gzip(level) => {
+ Self::GZIP(GzipLevel::try_new(level).expect("gzip level was validated"))
+ }
+ DFParquetCompression::Brotli(level) => Self::BROTLI(
+ BrotliLevel::try_new(level).expect("brotli level was validated"),
+ ),
+ DFParquetCompression::Lz4 => Self::LZ4,
+ DFParquetCompression::Zstd(level) => Self::ZSTD(
+ ZstdLevel::try_new(level as i32).expect("zstd level was validated"),
+ ),
+ DFParquetCompression::Lz4Raw => Self::LZ4_RAW,
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn parquet_compression_parses_and_displays() {
+ for (input, expected, displayed) in [
+ (
+ "uncompressed",
+ DFParquetCompression::Uncompressed,
+ "uncompressed",
+ ),
+ ("SNAPPY", DFParquetCompression::Snappy, "snappy"),
+ ("gzip(6)", DFParquetCompression::Gzip(6), "gzip(6)"),
+ ("Brotli(11)", DFParquetCompression::Brotli(11), "brotli(11)"),
+ ("lz4", DFParquetCompression::Lz4, "lz4"),
+ ("'zstd(3)'", DFParquetCompression::Zstd(3), "zstd(3)"),
+ ("lz4_raw", DFParquetCompression::Lz4Raw, "lz4_raw"),
+ ("zstd(0)", DFParquetCompression::Zstd(0), "zstd(0)"),
+ ] {
+ let parsed: DFParquetCompression = input.parse().unwrap();
+ assert_eq!(parsed, expected, "input: {input}");
+ assert_eq!(parsed.to_string(), displayed, "input: {input}");
+ assert_eq!(displayed.parse::<DFParquetCompression>().unwrap(), parsed);
+ }
+ }
+
+ #[test]
+ fn parquet_compression_rejects_invalid_settings() {
+ for (input, message) in [
+ (
+ "zstdd(3)",
+ "Unknown or unsupported parquet compression: zstdd(3)",
+ ),
+ (
+ "zstd",
+ "zstd compression requires specifying a level such as zstd(4)",
+ ),
+ (
+ "snappy(2)",
+ "Compression snappy does not support specifying a level",
+ ),
+ ("zstd(x)", "Could not parse compression string"),
+ ("zstd(3", "Could not parse compression string"),
+ (
+ "zstd(23)",
+ "Invalid compression level 23 for zstd. Expected a level between 0 and 22",
+ ),
+ (
+ "gzip(10)",
+ "Invalid compression level 10 for gzip. Expected a level between 0 and 9",
+ ),
+ ("brotli(12)", "Invalid compression level 12 for brotli"),
+ ("", "Unknown or unsupported parquet compression: "),
+ ] {
+ let err = input
+ .parse::<DFParquetCompression>()
+ .unwrap_err()
+ .to_string();
+ assert!(err.contains(message), "input {input:?}: {err}");
+ }
+ }
+
+ /// The level ranges above must stay equal to the ones parquet enforces, or
+ /// the infallible conversion would panic on a level parquet rejects.
+ #[cfg(feature = "parquet")]
+ #[test]
+ fn parquet_compression_level_ranges_match_parquet() {
+ use parquet::basic::{BrotliLevel, GzipLevel, ZstdLevel};
+
+ for level in GZIP_LEVELS {
+ assert!(GzipLevel::try_new(level).is_ok(), "gzip({level})");
+ }
+ assert!(GzipLevel::try_new(GZIP_LEVELS.end() + 1).is_err());
+
+ for level in BROTLI_LEVELS {
+ assert!(BrotliLevel::try_new(level).is_ok(), "brotli({level})");
+ }
+ assert!(BrotliLevel::try_new(BROTLI_LEVELS.end() + 1).is_err());
+
+ for level in ZSTD_LEVELS {
+ assert!(ZstdLevel::try_new(level as i32).is_ok(), "zstd({level})");
+ }
+ assert!(ZstdLevel::try_new(*ZSTD_LEVELS.end() as i32 + 1).is_err());
+ }
+}
diff --git a/datafusion/core/src/dataframe/parquet.rs b/datafusion/core/src/dataframe/parquet.rs
index 1685dff23d..dd751eb977 100644
--- a/datafusion/core/src/dataframe/parquet.rs
+++ b/datafusion/core/src/dataframe/parquet.rs
@@ -189,7 +189,7 @@ mod tests {
let ctx = &test_df.session_state;
ctx.runtime_env().register_object_store(&local_url, local);
let mut options = TableParquetOptions::default();
- options.global.compression = Some(compression.to_string());
+ options.global.compression = Some(compression.parse()?);
df.write_parquet(
output_path,
DataFrameWriteOptions::new().with_single_file_output(true),
diff --git a/datafusion/core/src/datasource/listing/table.rs b/datafusion/core/src/datasource/listing/table.rs
index 982766dc88..a620ff2f08 100644
--- a/datafusion/core/src/datasource/listing/table.rs
+++ b/datafusion/core/src/datasource/listing/table.rs
@@ -808,9 +808,10 @@ mod tests {
)
.await
.expect_err("Example should fail!");
+ // The value is rejected when the option is set, before any file is written
assert_eq!(
e.strip_backtrace(),
- "Invalid or Unsupported Configuration: zstd compression requires specifying a level such as zstd(4)"
+ "Error setting config datafusion.execution.parquet.compression\ncaused by\nInvalid or Unsupported Configuration: zstd compression requires specifying a level such as zstd(4)"
);
Ok(())
diff --git a/datafusion/core/src/execution/context/parquet.rs b/datafusion/core/src/execution/context/parquet.rs
index aad8518ba4..c015b99c0d 100644
--- a/datafusion/core/src/execution/context/parquet.rs
+++ b/datafusion/core/src/execution/context/parquet.rs
@@ -103,11 +103,11 @@ mod tests {
use crate::arrow::datatypes::{DataType, Field, Schema};
use crate::arrow::record_batch::RecordBatch;
use crate::dataframe::DataFrameWriteOptions;
- use crate::parquet::basic::Compression;
use crate::test_util::parquet_test_data;
use arrow::util::pretty::pretty_format_batches;
use datafusion_common::config::TableParquetOptions;
+ use datafusion_common::parquet_config::DFParquetCompression;
use datafusion_common::{assert_batches_sorted_eq, assert_contains};
use datafusion_execution::config::SessionConfig;
@@ -271,7 +271,7 @@ mod tests {
std::fs::create_dir(dir).expect("create dir failed");
let mut options = TableParquetOptions::default();
- options.global.compression = Some(Compression::SNAPPY.to_string());
+ options.global.compression = Some(DFParquetCompression::Snappy);
// Write the dataframe to a parquet file named 'output1.parquet'
write_df
diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs
index 65ea0416b0..dc0f9a7be4 100644
--- a/datafusion/datasource-parquet/src/file_format.rs
+++ b/datafusion/datasource-parquet/src/file_format.rs
@@ -670,7 +670,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions {
write_batch_size: global_options.global.write_batch_size as u64,
writer_version: global_options.global.writer_version.to_string(),
compression_opt: global_options.global.compression.map(|compression| {
- parquet_options::CompressionOpt::Compression(compression)
+ parquet_options::CompressionOpt::Compression(compression.to_string())
}),
dictionary_enabled_opt: global_options.global.dictionary_enabled.map(|enabled| {
parquet_options::DictionaryEnabledOpt::DictionaryEnabled(enabled)
diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs
index 62b16664f1..0bde696c7c 100644
--- a/datafusion/proto-common/src/from_proto/mod.rs
+++ b/datafusion/proto-common/src/from_proto/mod.rs
@@ -1202,9 +1202,9 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions {
writer_version: value.writer_version.parse().map_err(|e| {
DataFusionError::Internal(format!("Failed to parse writer_version: {e}"))
})?,
- compression: value.compression_opt.clone().map(|opt| match opt {
- protobuf::parquet_options::CompressionOpt::Compression(v) => Some(v),
- }).unwrap_or(None),
+ compression: value.compression_opt.as_ref().map(|opt| match opt {
+ protobuf::parquet_options::CompressionOpt::Compression(v) => v.parse(),
+ }).transpose()?,
dictionary_enabled: value.dictionary_enabled_opt.as_ref().map(|protobuf::parquet_options::DictionaryEnabledOpt::DictionaryEnabled(v)| *v),
// Continuing from where we left off in the TryFrom implementation
dictionary_page_size_limit: to_usize(
@@ -1438,7 +1438,7 @@ mod tests {
use datafusion_common::config::{
MaxRowGroupBytes, ParquetCdcOptions, ParquetOptions, TableParquetOptions,
};
- use datafusion_common::parquet_config::DFParquetStatistics;
+ use datafusion_common::parquet_config::{DFParquetCompression, DFParquetStatistics};
#[test]
fn constraint_requires_mode() {
@@ -1561,6 +1561,35 @@ mod tests {
);
}
+ #[test]
+ fn test_parquet_compression_round_trip() {
+ let opts = ParquetOptions {
+ compression: Some(DFParquetCompression::Gzip(6)),
+ ..ParquetOptions::default()
+ };
+ let recovered = parquet_options_proto_round_trip(opts);
+ assert_eq!(recovered.compression, Some(DFParquetCompression::Gzip(6)));
+ }
+
+ #[test]
+ fn test_invalid_parquet_compression_rejected_from_proto() {
+ let opts = ParquetOptions::default();
+ let mut proto: crate::protobuf_common::ParquetOptions =
+ (&opts).try_into().expect("to_proto");
+ proto.compression_opt = Some(
+ crate::protobuf_common::parquet_options::CompressionOpt::Compression(
+ "zstd".to_string(),
+ ),
+ );
+
+ let err = ParquetOptions::try_from(&proto).unwrap_err();
+ assert!(
+ err.to_string()
+ .contains("zstd compression requires specifying a level such as zstd(4)"),
+ "unexpected error: {err}"
+ );
+ }
+
#[test]
fn test_invalid_parquet_statistics_rejected_from_proto() {
let opts = ParquetOptions::default();
diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs
index f9a881e76d..b8eb46f2c8 100644
--- a/datafusion/proto-common/src/to_proto/mod.rs
+++ b/datafusion/proto-common/src/to_proto/mod.rs
@@ -942,7 +942,7 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions {
data_pagesize_limit: value.data_pagesize_limit as u64,
write_batch_size: value.write_batch_size as u64,
writer_version: value.writer_version.to_string(),
- compression_opt: value.compression.clone().map(protobuf::parquet_options::CompressionOpt::Compression),
+ compression_opt: value.compression.map(|v| protobuf::parquet_options::CompressionOpt::Compression(v.to_string())),
dictionary_enabled_opt: value.dictionary_enabled.map(protobuf::parquet_options::DictionaryEnabledOpt::DictionaryEnabled),
dictionary_page_size_limit: value.dictionary_page_size_limit as u64,
statistics_enabled_opt: value.statistics_enabled.map(|v| protobuf::parquet_options::StatisticsEnabledOpt::StatisticsEnabled(v.to_string())),
diff --git a/datafusion/proto-models/src/from_proto.rs b/datafusion/proto-models/src/from_proto.rs
index b2340a0967..fdf3239647 100644
--- a/datafusion/proto-models/src/from_proto.rs
+++ b/datafusion/proto-models/src/from_proto.rs
@@ -380,11 +380,15 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions {
)?,
write_batch_size: to_usize(proto.write_batch_size, "write_batch_size")?,
writer_version,
- compression: proto.compression_opt.as_ref().map(|opt| match opt {
- parquet_options::CompressionOpt::Compression(compression) => {
- compression.clone()
- }
- }),
+ compression: proto
+ .compression_opt
+ .as_ref()
+ .map(|opt| match opt {
+ parquet_options::CompressionOpt::Compression(compression) => {
+ compression.parse()
+ }
+ })
+ .transpose()?,
dictionary_enabled: proto.dictionary_enabled_opt.as_ref().map(|opt| {
match opt {
parquet_options::DictionaryEnabledOpt::DictionaryEnabled(
diff --git a/datafusion/sqllogictest/test_files/set_variable.slt b/datafusion/sqllogictest/test_files/set_variable.slt
index 5d9cc3bd4a..bf9145fe44 100644
--- a/datafusion/sqllogictest/test_files/set_variable.slt
+++ b/datafusion/sqllogictest/test_files/set_variable.slt
@@ -751,6 +751,48 @@ Invalid or Unsupported Configuration: Invalid parquet statistics setting: invali
statement ok
RESET datafusion.execution.parquet.statistics_enabled
+# parquet.compression is validated when it is set, including the level
+statement ok
+SET datafusion.execution.parquet.compression = 'GZIP(6)'
+
+query TT
+SHOW datafusion.execution.parquet.compression
+----
+datafusion.execution.parquet.compression gzip(6)
+
+statement error
+SET datafusion.execution.parquet.compression = 'zstdd(3)'
+----
+DataFusion error: Error setting config datafusion.execution.parquet.compression
+caused by
+Invalid or Unsupported Configuration: Unknown or unsupported parquet compression: zstdd(3). Valid values are: uncompressed, snappy, gzip(level), brotli(level), lz4, zstd(level), and lz4_raw.
+
+
+statement error
+SET datafusion.execution.parquet.compression = 'zstd'
+----
+DataFusion error: Error setting config datafusion.execution.parquet.compression
+caused by
+Invalid or Unsupported Configuration: zstd compression requires specifying a level such as zstd(4)
+
+
+statement error
+SET datafusion.execution.parquet.compression = 'zstd(23)'
+----
+DataFusion error: Error setting config datafusion.execution.parquet.compression
+caused by
+Invalid or Unsupported Configuration: Invalid compression level 23 for zstd. Expected a level between 0 and 22
+
+
+# The rejected values did not change the setting
+query TT
+SHOW datafusion.execution.parquet.compression
+----
+datafusion.execution.parquet.compression gzip(6)
+
+statement ok
+RESET datafusion.execution.parquet.compression
+
statement error DataFusion error: Error setting config datafusion\.execution\.batch_size\ncaused by\nInvalid or Unsupported Configuration: value must be greater than 0
SET datafusion.execution.batch_size = 0
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org