From 72de13f2c8aff7031b0dbce4aa3ca318518344bc Mon Sep 17 00:00:00 2001 From: Michal Piatkowski <291740709+MassivePizza@users.noreply.github.com> Date: Tue, 28 Jul 2026 20:44:42 +0200 Subject: [PATCH] add data_page_compression_ratio_threshold to config --- datafusion/common/src/config.rs | 8 ++++++++ .../common/src/file_options/parquet_writer.rs | 16 ++++++++++++++++ datafusion/proto-common/src/from_proto/mod.rs | 2 ++ .../proto/src/logical_plan/file_formats.rs | 2 ++ 4 files changed, 28 insertions(+) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 81f573fc2a23e..33a9da8b0f2f5 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1218,6 +1218,10 @@ config_namespace! { /// the default parquet writer setting. pub compression: Option, transform = str::to_lowercase, default = Some("zstd(3)".into()) + /// (writing) Sets the default compression ratio threshold at or above which a Data Page + /// v2's compressed values are discarded in favor of writing the values uncompressed. + pub data_page_compression_ratio_threshold: Option, default = None + /// (writing) Sets if dictionary encoding is enabled. If NULL, uses /// default parquet writer setting pub dictionary_enabled: Option, default = Some(true) @@ -3088,6 +3092,10 @@ config_namespace_with_hashmap! { /// default parquet options pub compression: Option, transform = str::to_lowercase, default = None + /// Sets the default compression ratio threshold at or above which a Data Page + /// v2's compressed values are discarded in favor of writing the values uncompressed. + pub data_page_compression_ratio_threshold: Option, default = None + /// Sets if statistics are enabled for the column /// Valid values are: "none", "chunk", and "page" /// These values are not case sensitive. If NULL, uses diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 320bfcf33e488..8e87602f0ded5 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -160,6 +160,13 @@ impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder { builder = builder .set_column_bloom_filter_max_ndv(path.clone(), bloom_filter_ndv); } + + if let Some(compression_threshold) = options.data_page_compression_ratio_threshold { + builder = builder.set_column_data_page_v2_compression_ratio_threshold( + path.clone(), + compression_threshold, + ); + } } Ok(builder) @@ -215,6 +222,7 @@ impl ParquetOptions { write_batch_size, writer_version, compression, + data_page_compression_ratio_threshold: compression_threshold, dictionary_enabled, dictionary_page_size_limit, statistics_enabled, @@ -284,6 +292,10 @@ impl ParquetOptions { if let Some(compression) = compression { builder = builder.set_compression(parse_compression_string(compression)?); } + if let Some(compression_threshold) = compression_threshold { + builder = builder + .set_data_page_v2_compression_ratio_threshold(*compression_threshold); + } if let Some(encoding) = encoding { builder = builder.set_encoding(parse_encoding_string(encoding)?); } @@ -449,6 +461,7 @@ mod tests { ) -> ParquetColumnOptions { ParquetColumnOptions { compression: Some("zstd(22)".into()), + data_page_compression_ratio_threshold: Some(0.9), dictionary_enabled: src_col_defaults.dictionary_enabled.map(|v| !v), statistics_enabled: Some("none".into()), encoding: Some("RLE".into()), @@ -472,6 +485,7 @@ mod tests { write_batch_size: 42, writer_version, compression: Some("zstd(22)".into()), + data_page_compression_ratio_threshold: Some(0.9), dictionary_enabled: Some(!defaults.dictionary_enabled.unwrap_or(false)), dictionary_page_size_limit: 42, statistics_enabled: Some("chunk".into()), @@ -526,6 +540,7 @@ mod tests { } _ => None, }, + data_page_compression_ratio_threshold: Some(props.data_page_v2_compression_ratio_threshold()), statistics_enabled: Some( match props.statistics_enabled(&col) { EnabledStatistics::None => "none", @@ -597,6 +612,7 @@ mod tests { // global options which set the default column props encoding: default_col_props.encoding, compression: default_col_props.compression, + data_page_compression_ratio_threshold: default_col_props.data_page_compression_ratio_threshold, dictionary_enabled: default_col_props.dictionary_enabled, statistics_enabled: default_col_props.statistics_enabled, bloom_filter_on_write: default_col_props diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 97cc9af230105..c4b4ac31d5fd0 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -1071,6 +1071,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { compression: value.compression_opt.clone().map(|opt| match opt { protobuf::parquet_options::CompressionOpt::Compression(v) => Some(v), }).unwrap_or(None), + data_page_compression_ratio_threshold: None, 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: value.dictionary_page_size_limit as usize, @@ -1158,6 +1159,7 @@ impl TryFrom<&protobuf::ParquetColumnOptions> for ParquetColumnOptions { compression: value.compression_opt.clone().map(|opt| match opt { protobuf::parquet_column_options::CompressionOpt::Compression(v) => Some(v), }).unwrap_or(None), + data_page_compression_ratio_threshold: None, dictionary_enabled: value.dictionary_enabled_opt.as_ref().map(|protobuf::parquet_column_options::DictionaryEnabledOpt::DictionaryEnabled(v)| *v), statistics_enabled: value .statistics_enabled_opt.clone() diff --git a/datafusion/proto/src/logical_plan/file_formats.rs b/datafusion/proto/src/logical_plan/file_formats.rs index 8940b16bf83f5..4db9b24bac9cb 100644 --- a/datafusion/proto/src/logical_plan/file_formats.rs +++ b/datafusion/proto/src/logical_plan/file_formats.rs @@ -554,6 +554,7 @@ mod parquet { compression.clone() } }), + data_page_compression_ratio_threshold: None, dictionary_enabled: proto.dictionary_enabled_opt.as_ref().map(|opt| { match opt { parquet_options::DictionaryEnabledOpt::DictionaryEnabled( @@ -665,6 +666,7 @@ mod parquet { compression: proto .compression_opt .map(|parquet_column_options::CompressionOpt::Compression(v)| v), + data_page_compression_ratio_threshold: None, statistics_enabled: proto.statistics_enabled_opt.map( |parquet_column_options::StatisticsEnabledOpt::StatisticsEnabled(v)| v, ),