add data_page_compression_ratio_threshold to config

datafusion/common/src/config.rs

@@ -1218,6 +1218,10 @@ config_namespace! {
         /// the default parquet writer setting.
         pub compression: Option<String>, 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<f64>, default = None
+
         /// (writing) Sets if dictionary encoding is enabled. If NULL, uses
         /// default parquet writer setting
         pub dictionary_enabled: Option<bool>, default = Some(true)
@@ -3088,6 +3092,10 @@ config_namespace_with_hashmap! {
         /// default parquet options
         pub compression: Option<String>, 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<f64>, 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

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

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()

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,
             ),