From 7e19cdb0299b3db1f6c251b7d85c2f076917d790 Mon Sep 17 00:00:00 2001 From: Adegbite Ayoade Date: Mon, 19 Oct 2026 20:22:56 +0100 Subject: [PATCH] fix: reject invalid parquet writer options when they are set Parquet writer options that the parquet crate cannot accept (a zero write_batch_size, max_row_group_size, data_page_row_count_limit, truncate length or min_chunk_size, or a bloom_filter_fpp outside (0, 1)) were accepted by SET and then panicked or hung on the next write. They are now rejected with a configuration error when set. max_chunk_size > min_chunk_size compares two options, so it is checked when the writer properties are built. Protobuf decoding applies the same checks. A wire value of 0 for a non-optional field is read as absent and replaced by the default. Closes #24904 --- datafusion/common/src/config.rs | 136 +++++++++++++++--- .../common/src/file_options/parquet_writer.rs | 95 ++++++++---- datafusion/core/src/dataframe/parquet.rs | 3 +- .../tests/parquet/content_defined_chunking.rs | 6 +- .../datasource-parquet/src/file_format.rs | 16 +-- datafusion/proto-common/src/from_proto/mod.rs | 102 ++++++++++--- datafusion/proto-common/src/to_proto/mod.rs | 22 +-- datafusion/proto-models/src/from_proto.rs | 87 ++++++++--- .../proto/tests/cases/public_conversions.rs | 2 +- .../tests/cases/roundtrip_logical_plan.rs | 8 +- .../parquet_writer_options_validation.slt | 122 ++++++++++++++++ .../library-user-guide/upgrading/56.0.0.md | 46 ++++++ docs/source/user-guide/configs.md | 16 +-- 13 files changed, 537 insertions(+), 124 deletions(-) create mode 100644 datafusion/sqllogictest/test_files/parquet_writer_options_validation.slt diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 290baeb379f2e..37a9f27ffe994 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -666,6 +666,28 @@ impl Display for ConfigNonZeroUsize { } } +/// `ConfigField` for `Option`. A custom impl (rather than +/// the blanket `Option` one) so an invalid value is rejected without leaving +/// the option in an invalid intermediate state on error. +impl ConfigField for Option { + fn visit(&self, v: &mut V, key: &str, description: &'static str) { + match self { + Some(s) => v.some(key, s, description), + None => v.none(key, description), + } + } + + fn set(&mut self, _key: &str, value: &str) -> Result<()> { + *self = Some(ConfigNonZeroUsize::from_str(value)?); + Ok(()) + } + + fn reset(&mut self, _key: &str) -> Result<()> { + *self = None; + Ok(()) + } +} + /// A `usize` configuration value that rejects 0 and 1 when set from strings. /// /// Use this for options whose consumer divides the value in half to size an @@ -1236,11 +1258,13 @@ config_namespace! { pub enabled: bool, default = false /// Minimum chunk size in bytes. The rolling hash will not trigger a split - /// until this many bytes have been accumulated. Default is 256 KiB. - pub min_chunk_size: usize, default = 256 * 1024 + /// until this many bytes have been accumulated. Must be greater than 0. + /// Default is 256 KiB. + pub min_chunk_size: ConfigNonZeroUsize, default = non_zero_usize_default(256 * 1024) /// Maximum chunk size in bytes. A split is forced when the accumulated - /// size exceeds this value. Default is 1 MiB. + /// size exceeds this value. Must be greater than `min_chunk_size`. + /// Default is 1 MiB. pub max_chunk_size: usize, default = 1024 * 1024 /// Normalization level. Increasing this improves deduplication ratio @@ -1335,6 +1359,72 @@ impl ConfigField for Option { } } +/// Parquet bloom filter false positive probability. +/// +/// Wraps an `f64` so the "strictly between 0 and 1" constraint (arrow-rs +/// panics on any other value, including NaN) is validated when the config is +/// set, rather than when the writer properties are built. +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct BloomFilterFpp(f64); + +impl BloomFilterFpp { + /// Creates a `BloomFilterFpp`, rejecting values outside `(0.0, 1.0)`. + pub fn try_new(value: f64) -> Result { + // Written so that NaN, which fails every comparison, is rejected too + if !(value > 0.0 && value < 1.0) { + return Err(DataFusionError::Configuration(format!( + "bloom_filter_fpp must be between 0 and 1 (exclusive), got {value}" + ))); + } + Ok(Self(value)) + } + + /// Returns the configured false positive probability. + pub fn get(&self) -> f64 { + self.0 + } +} + +impl FromStr for BloomFilterFpp { + type Err = DataFusionError; + + fn from_str(s: &str) -> Result { + let value = s.parse::().map_err(|_| { + DataFusionError::Configuration(format!( + "Invalid bloom_filter_fpp: '{s}'. Expected a number between 0 and 1 (exclusive)." + )) + })?; + Self::try_new(value) + } +} + +impl Display for BloomFilterFpp { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +/// `ConfigField` for `Option`, for the same reason as the +/// `Option` impl above. +impl ConfigField for Option { + fn visit(&self, v: &mut V, key: &str, description: &'static str) { + match self { + Some(s) => v.some(key, s, description), + None => v.none(key, description), + } + } + + fn set(&mut self, _key: &str, value: &str) -> Result<()> { + *self = Some(BloomFilterFpp::from_str(value)?); + Ok(()) + } + + fn reset(&mut self, _key: &str) -> Result<()> { + *self = None; + Ok(()) + } +} + config_namespace! { /// Options for reading and writing parquet files /// @@ -1472,8 +1562,8 @@ config_namespace! { /// (writing) Sets best effort maximum size of data page in bytes pub data_pagesize_limit: usize, default = 1024 * 1024 - /// (writing) Sets write_batch_size in rows - pub write_batch_size: usize, default = 1024 + /// (writing) Sets write_batch_size in rows. Must be greater than 0. + pub write_batch_size: ConfigNonZeroUsize, default = non_zero_usize_default(1024) /// (writing) Sets parquet writer version /// valid values are "1.0" and "2.0" @@ -1512,8 +1602,8 @@ config_namespace! { /// rows). Writing larger row groups requires more memory to write, but /// can get better compression and be faster to read. When /// `max_row_group_bytes` is also set, the writer flushes a row group when - /// either limit is reached, whichever comes first. - pub max_row_group_size: usize, default = 1024 * 1024 + /// either limit is reached, whichever comes first. Must be greater than 0. + pub max_row_group_size: ConfigNonZeroUsize, default = non_zero_usize_default(1024 * 1024) /// (writing) Target maximum size of each row group in bytes. When set, /// the writer flushes whenever either this limit or `max_row_group_size` @@ -1528,15 +1618,17 @@ config_namespace! { /// (writing) Sets "created by" property pub created_by: String, default = concat!("datafusion version ", env!("CARGO_PKG_VERSION")).into() - /// (writing) Sets column index truncate length - pub column_index_truncate_length: Option, default = Some(64) + /// (writing) Sets column index truncate length. Must be greater than 0. + /// If NULL, column index values are not truncated + pub column_index_truncate_length: Option, default = Some(non_zero_usize_default(64)) - /// (writing) Sets statistics truncate length. If NULL, uses - /// default parquet writer setting - pub statistics_truncate_length: Option, default = Some(64) + /// (writing) Sets statistics truncate length. Must be greater than 0. + /// If NULL, uses default parquet writer setting + pub statistics_truncate_length: Option, default = Some(non_zero_usize_default(64)) - /// (writing) Sets best effort maximum number of rows in data page - pub data_page_row_count_limit: usize, default = 20_000 + /// (writing) Sets best effort maximum number of rows in data page. + /// Must be greater than 0. + pub data_page_row_count_limit: ConfigNonZeroUsize, default = non_zero_usize_default(20_000) /// (writing) Sets default encoding for any column. /// Valid values are: plain, plain_dictionary, rle, @@ -1549,9 +1641,9 @@ config_namespace! { /// (writing) Write bloom filters for all columns when creating parquet files pub bloom_filter_on_write: bool, default = false - /// (writing) Sets bloom filter false positive probability. If NULL, uses - /// default parquet writer setting - pub bloom_filter_fpp: Option, default = None + /// (writing) Sets bloom filter false positive probability. Must be + /// between 0 and 1 (exclusive). If NULL, uses default parquet writer setting + pub bloom_filter_fpp: Option, default = None /// (writing) Sets bloom filter number of distinct values. If NULL, uses /// default parquet writer setting @@ -3504,9 +3596,9 @@ config_namespace_with_hashmap! { /// default parquet options pub statistics_enabled: Option, default = None - /// Sets bloom filter false positive probability for the column path. If NULL, uses - /// default parquet options - pub bloom_filter_fpp: Option, default = None + /// Sets bloom filter false positive probability for the column path. Must be + /// between 0 and 1 (exclusive). If NULL, uses default parquet options + pub bloom_filter_fpp: Option, default = None /// Sets bloom filter number of distinct values. If NULL, uses /// default parquet options @@ -5101,7 +5193,7 @@ mod tests { .unwrap(); let cdc = &config.execution.parquet.content_defined_chunking; assert!(cdc.enabled); - assert_eq!(cdc.min_chunk_size, 256 * 1024); + assert_eq!(cdc.min_chunk_size.get(), 256 * 1024); assert_eq!(cdc.max_chunk_size, 1024 * 1024); assert_eq!(cdc.norm_level, 0); @@ -5132,7 +5224,7 @@ mod tests { .unwrap(); let cdc = &config.execution.parquet.content_defined_chunking; assert!(!cdc.enabled); - assert_eq!(cdc.min_chunk_size, 1024); + assert_eq!(cdc.min_chunk_size.get(), 1024); assert_eq!(cdc.max_chunk_size, 1024 * 1024); assert_eq!(cdc.norm_level, 0); } diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 3d904290b222c..a2f535b7e2ccf 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -21,7 +21,10 @@ use std::sync::Arc; use crate::{ _internal_datafusion_err, DataFusionError, Result, - config::{ParquetCdcOptions, ParquetOptions, TableParquetOptions}, + config::{ + ConfigNonZeroUsize, ParquetCdcOptions, ParquetOptions, TableParquetOptions, + }, + error::_config_err, parquet_config::{DFParquetCompression, DFParquetEncoding}, }; @@ -152,8 +155,8 @@ impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder { } if let Some(bloom_filter_fpp) = options.bloom_filter_fpp { - builder = - builder.set_column_bloom_filter_fpp(path.clone(), bloom_filter_fpp); + builder = builder + .set_column_bloom_filter_fpp(path.clone(), bloom_filter_fpp.get()); } if let Some(bloom_filter_ndv) = options.bloom_filter_ndv { @@ -176,7 +179,7 @@ impl From<&ParquetCdcOptions> for Option value .enabled .then_some(parquet::file::properties::CdcOptions { - min_chunk_size: value.min_chunk_size, + min_chunk_size: value.min_chunk_size.get(), max_chunk_size: value.max_chunk_size, norm_level: value.norm_level, }) @@ -193,7 +196,10 @@ impl From> for ParquetCdcOptions match value { Some(cdc) => ParquetCdcOptions { enabled: true, - min_chunk_size: cdc.min_chunk_size, + // parquet-rs rejects a zero `min_chunk_size` when the options + // are set, so the fallback is never taken in practice + min_chunk_size: ConfigNonZeroUsize::try_new(cdc.min_chunk_size) + .unwrap_or_else(|_| ParquetCdcOptions::default().min_chunk_size), max_chunk_size: cdc.max_chunk_size, norm_level: cdc.norm_level, }, @@ -254,9 +260,22 @@ impl ParquetOptions { row_group_range_assignment: _, } = self; + // Each chunk size is validated when it is set, but the relation between + // them can only be checked once both are known + if content_defined_chunking.enabled + && content_defined_chunking.max_chunk_size + <= content_defined_chunking.min_chunk_size.get() + { + return _config_err!( + "content_defined_chunking.max_chunk_size ({}) must be greater than content_defined_chunking.min_chunk_size ({})", + content_defined_chunking.max_chunk_size, + content_defined_chunking.min_chunk_size + ); + } + let mut builder = WriterProperties::builder() .set_data_page_size_limit(*data_pagesize_limit) - .set_write_batch_size(*write_batch_size) + .set_write_batch_size(write_batch_size.get()) .set_writer_version((*writer_version).into()) .set_dictionary_page_size_limit(*dictionary_page_size_limit) .set_statistics_enabled( @@ -264,19 +283,23 @@ impl ParquetOptions { .map(Into::into) .unwrap_or(DEFAULT_STATISTICS_ENABLED), ) - .set_max_row_group_row_count(Some(*max_row_group_size)) + .set_max_row_group_row_count(Some(max_row_group_size.get())) .set_max_row_group_bytes(max_row_group_bytes.as_ref().map(|v| v.get())) .set_created_by(created_by.clone()) - .set_column_index_truncate_length(*column_index_truncate_length) - .set_statistics_truncate_length(*statistics_truncate_length) - .set_data_page_row_count_limit(*data_page_row_count_limit) + .set_column_index_truncate_length( + column_index_truncate_length.map(ConfigNonZeroUsize::get), + ) + .set_statistics_truncate_length( + statistics_truncate_length.map(ConfigNonZeroUsize::get), + ) + .set_data_page_row_count_limit(data_page_row_count_limit.get()) .set_bloom_filter_enabled(*bloom_filter_on_write) .set_write_row_group_number_distinct_values( *write_row_group_number_distinct_values, ); if let Some(bloom_filter_fpp) = bloom_filter_fpp { - builder = builder.set_bloom_filter_fpp(*bloom_filter_fpp); + builder = builder.set_bloom_filter_fpp(bloom_filter_fpp.get()); } if let Some(bloom_filter_ndv) = bloom_filter_ndv { builder = builder.set_bloom_filter_max_ndv(*bloom_filter_ndv); @@ -333,7 +356,7 @@ mod tests { #[cfg(feature = "parquet_encryption")] use crate::config::ConfigFileEncryptionProperties; use crate::config::{ - MaxRowGroupBytes, ParquetCdcOptions, ParquetColumnOptions, + BloomFilterFpp, MaxRowGroupBytes, ParquetCdcOptions, ParquetColumnOptions, ParquetEncryptionOptions, ParquetOptions, }; use crate::parquet_config::{ @@ -349,6 +372,14 @@ mod tests { const COL_NAME: &str = "configured"; + fn nz(value: usize) -> ConfigNonZeroUsize { + ConfigNonZeroUsize::try_new(value).unwrap() + } + + fn fpp(value: f64) -> BloomFilterFpp { + BloomFilterFpp::try_new(value).unwrap() + } + /// Take the column defaults provided in [`ParquetOptions`], and generate a non-default col config. fn column_options_with_non_defaults( src_col_defaults: &ParquetOptions, @@ -359,7 +390,7 @@ mod tests { statistics_enabled: Some("none".into()), encoding: Some("RLE".into()), bloom_filter_enabled: Some(true), - bloom_filter_fpp: Some(0.72), + bloom_filter_fpp: Some(fpp(0.72)), bloom_filter_ndv: Some(72), } } @@ -375,21 +406,21 @@ mod tests { ParquetOptions { data_pagesize_limit: 42, - write_batch_size: 42, + write_batch_size: nz(42), writer_version, compression: Some(DFParquetCompression::Zstd(22)), dictionary_enabled: Some(!defaults.dictionary_enabled.unwrap_or(false)), dictionary_page_size_limit: 43, statistics_enabled: Some(DFParquetStatistics::Chunk), - max_row_group_size: 42, + max_row_group_size: nz(42), max_row_group_bytes: Some(MaxRowGroupBytes::try_new(42).unwrap()), created_by: "wordy".into(), - column_index_truncate_length: Some(42), - statistics_truncate_length: Some(42), - data_page_row_count_limit: 42, + column_index_truncate_length: Some(nz(42)), + statistics_truncate_length: Some(nz(42)), + data_page_row_count_limit: nz(42), encoding: Some(DFParquetEncoding::ByteStreamSplit), bloom_filter_on_write: !defaults.bloom_filter_on_write, - bloom_filter_fpp: Some(0.42), + bloom_filter_fpp: Some(fpp(0.42)), bloom_filter_ndv: Some(42), write_row_group_number_distinct_values: !defaults .write_row_group_number_distinct_values, @@ -445,7 +476,7 @@ mod tests { } .into(), ), - bloom_filter_fpp: bloom_filter_default_props.map(|p| p.fpp()), + bloom_filter_fpp: bloom_filter_default_props.map(|p| fpp(p.fpp())), bloom_filter_ndv: bloom_filter_default_props.map(|p| p.ndv()), } } @@ -491,19 +522,21 @@ mod tests { global: ParquetOptions { // global options data_pagesize_limit: props.data_page_size_limit(), - write_batch_size: props.write_batch_size(), + write_batch_size: nz(props.write_batch_size()), writer_version: props.writer_version().into(), dictionary_page_size_limit: props.dictionary_page_size_limit(), - max_row_group_size: props + max_row_group_size: nz(props .max_row_group_row_count() - .unwrap_or(DEFAULT_MAX_ROW_GROUP_ROW_COUNT), + .unwrap_or(DEFAULT_MAX_ROW_GROUP_ROW_COUNT)), max_row_group_bytes: props .max_row_group_bytes() .and_then(|v| MaxRowGroupBytes::try_new(v).ok()), created_by: props.created_by().to_string(), - column_index_truncate_length: props.column_index_truncate_length(), - statistics_truncate_length: props.statistics_truncate_length(), - data_page_row_count_limit: props.data_page_row_count_limit(), + column_index_truncate_length: props + .column_index_truncate_length() + .map(nz), + statistics_truncate_length: props.statistics_truncate_length().map(nz), + data_page_row_count_limit: nz(props.data_page_row_count_limit()), // global options which set the default column props encoding: default_col_props @@ -736,7 +769,7 @@ mod tests { // the TableParquetOptions::default, with only fpp set let mut default_table_writer_opts = TableParquetOptions::default(); default_table_writer_opts.global.bloom_filter_on_write = true; - default_table_writer_opts.global.bloom_filter_fpp = Some(0.42); + default_table_writer_opts.global.bloom_filter_fpp = Some(fpp(0.42)); default_table_writer_opts.arrow_schema(&Arc::new(Schema::empty())); // add the required arrow schema let from_datafusion_defaults = WriterPropertiesBuilder::try_from(&default_table_writer_opts) @@ -771,7 +804,7 @@ mod tests { let mut opts = TableParquetOptions::default(); opts.global.content_defined_chunking = ParquetCdcOptions { enabled: true, - min_chunk_size: 128 * 1024, + min_chunk_size: nz(128 * 1024), max_chunk_size: 512 * 1024, norm_level: 2, }; @@ -799,7 +832,7 @@ mod tests { let mut opts = TableParquetOptions::default(); opts.global.content_defined_chunking = ParquetCdcOptions { enabled: false, - min_chunk_size: 128 * 1024, + min_chunk_size: nz(128 * 1024), max_chunk_size: 512 * 1024, norm_level: 2, }; @@ -814,7 +847,7 @@ mod tests { let mut opts = TableParquetOptions::default(); opts.global.content_defined_chunking = ParquetCdcOptions { enabled: true, - min_chunk_size: 64 * 1024, + min_chunk_size: nz(64 * 1024), max_chunk_size: 2 * 1024 * 1024, norm_level: -1, }; @@ -825,7 +858,7 @@ mod tests { let cdc = recovered.global.content_defined_chunking; assert!(cdc.enabled); - assert_eq!(cdc.min_chunk_size, 64 * 1024); + assert_eq!(cdc.min_chunk_size.get(), 64 * 1024); assert_eq!(cdc.max_chunk_size, 2 * 1024 * 1024); assert_eq!(cdc.norm_level, -1); } diff --git a/datafusion/core/src/dataframe/parquet.rs b/datafusion/core/src/dataframe/parquet.rs index 175e8778761a3..0b4df13dd3bd9 100644 --- a/datafusion/core/src/dataframe/parquet.rs +++ b/datafusion/core/src/dataframe/parquet.rs @@ -115,6 +115,7 @@ mod tests { use crate::execution::options::ParquetReadOptions; use crate::test_util::{self, register_aggregate_csv}; + use datafusion_common::config::ConfigNonZeroUsize; use datafusion_common::file_options::parquet_writer::parse_compression_string; use datafusion_execution::config::SessionConfig; use datafusion_expr::{col, lit}; @@ -238,7 +239,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.max_row_group_size = rg_size; + options.global.max_row_group_size = ConfigNonZeroUsize::try_new(rg_size)?; options.global.allow_single_file_parallelism = true; df.write_parquet( output_path, diff --git a/datafusion/core/tests/parquet/content_defined_chunking.rs b/datafusion/core/tests/parquet/content_defined_chunking.rs index bd89b502bd272..2ef5ecdce02ef 100644 --- a/datafusion/core/tests/parquet/content_defined_chunking.rs +++ b/datafusion/core/tests/parquet/content_defined_chunking.rs @@ -25,7 +25,9 @@ use arrow::array::{AsArray, Int32Array, StringArray}; use arrow::datatypes::{DataType, Field, Int32Type, Int64Type, Schema}; use arrow::record_batch::RecordBatch; use datafusion::prelude::{ParquetReadOptions, SessionContext}; -use datafusion_common::config::{ParquetCdcOptions, TableParquetOptions}; +use datafusion_common::config::{ + ConfigNonZeroUsize, ParquetCdcOptions, TableParquetOptions, +}; use parquet::arrow::ArrowWriter; use parquet::arrow::arrow_reader::ArrowReaderMetadata; use parquet::file::properties::WriterProperties; @@ -147,7 +149,7 @@ async fn cdc_affects_page_boundaries() { let mut cdc_opts = TableParquetOptions::default(); cdc_opts.global.content_defined_chunking = ParquetCdcOptions { enabled: true, - min_chunk_size: 512, + min_chunk_size: ConfigNonZeroUsize::try_new(512).unwrap(), max_chunk_size: 2048, norm_level: 0, }; diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs index 50e76d2d211e1..38acee39d1719 100644 --- a/datafusion/datasource-parquet/src/file_format.rs +++ b/datafusion/datasource-parquet/src/file_format.rs @@ -714,7 +714,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions { reorder_filters: global_options.global.reorder_filters, force_filter_selections: global_options.global.force_filter_selections, data_pagesize_limit: global_options.global.data_pagesize_limit as u64, - write_batch_size: global_options.global.write_batch_size as u64, + write_batch_size: global_options.global.write_batch_size.get() as u64, writer_version: global_options.global.writer_version.to_string(), compression_opt: global_options.global.compression.map(|compression| { parquet_options::CompressionOpt::Compression(compression.to_string()) @@ -727,24 +727,24 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions { statistics_enabled_opt: global_options.global.statistics_enabled.map(|enabled| { parquet_options::StatisticsEnabledOpt::StatisticsEnabled(enabled.to_string()) }), - max_row_group_size: global_options.global.max_row_group_size as u64, + max_row_group_size: global_options.global.max_row_group_size.get() as u64, max_in_list_size: global_options.global.max_in_list_size as u64, row_group_range_assignment: global_options.global.row_group_range_assignment.to_string(), created_by: global_options.global.created_by.clone(), column_index_truncate_length_opt: global_options.global.column_index_truncate_length.map(|length| { - parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(length as u64) + parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(length.get() as u64) }), statistics_truncate_length_opt: global_options.global.statistics_truncate_length.map(|length| { - parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(length as u64) + parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(length.get() as u64) }), - data_page_row_count_limit: global_options.global.data_page_row_count_limit as u64, + data_page_row_count_limit: global_options.global.data_page_row_count_limit.get() as u64, encoding_opt: global_options.global.encoding.map(|encoding| { parquet_options::EncodingOpt::Encoding(encoding.to_string()) }), bloom_filter_on_read: global_options.global.bloom_filter_on_read, bloom_filter_on_write: global_options.global.bloom_filter_on_write, bloom_filter_fpp_opt: global_options.global.bloom_filter_fpp.map(|fpp| { - parquet_options::BloomFilterFppOpt::BloomFilterFpp(fpp) + parquet_options::BloomFilterFppOpt::BloomFilterFpp(fpp.get()) }), bloom_filter_ndv_opt: global_options.global.bloom_filter_ndv.map(|ndv| { parquet_options::BloomFilterNdvOpt::BloomFilterNdv(ndv) @@ -772,7 +772,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions { }), content_defined_chunking: Some(protobuf::ParquetCdcOptions { enabled: global_options.global.content_defined_chunking.enabled, - min_chunk_size: global_options.global.content_defined_chunking.min_chunk_size as u64, + min_chunk_size: global_options.global.content_defined_chunking.min_chunk_size.get() as u64, max_chunk_size: global_options.global.content_defined_chunking.max_chunk_size as u64, norm_level: global_options.global.content_defined_chunking.norm_level, }), @@ -797,7 +797,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions { parquet_column_options::StatisticsEnabledOpt::StatisticsEnabled(enabled) }), bloom_filter_fpp_opt: options.bloom_filter_fpp.map(|fpp| { - parquet_column_options::BloomFilterFppOpt::BloomFilterFpp(fpp) + parquet_column_options::BloomFilterFppOpt::BloomFilterFpp(fpp.get()) }), bloom_filter_ndv_opt: options.bloom_filter_ndv.map(|ndv| { parquet_column_options::BloomFilterNdvOpt::BloomFilterNdv(ndv) diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 24429f2fd1fad..450e429f4d213 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -40,8 +40,8 @@ use datafusion_common::{ DataFusionError, JoinSide, ScalarValue, Statistics, TableReference, arrow_datafusion_err, config::{ - CsvOptions, JsonOptions, MaxRowGroupBytes, ParquetCdcOptions, - ParquetColumnOptions, ParquetOptions, TableParquetOptions, + BloomFilterFpp, ConfigNonZeroUsize, CsvOptions, JsonOptions, MaxRowGroupBytes, + ParquetCdcOptions, ParquetColumnOptions, ParquetOptions, TableParquetOptions, }, file_options::{csv_writer::CsvWriterOptions, json_writer::JsonWriterOptions}, parquet_config::RowGroupRangeAssignment, @@ -1180,6 +1180,22 @@ impl TryFrom<&protobuf::CsvOptions> for CsvOptions { } } +/// Decodes a non-optional, must-be-positive `usize` option from the wire. +/// +/// Proto3 decodes an omitted scalar field as 0, which is never a valid value +/// for these options, so 0 is read as "absent" and replaced by `default`. +fn non_zero_usize_from_wire( + value: u64, + message: &str, + field: &str, + default: ConfigNonZeroUsize, +) -> datafusion_common::Result { + if value == 0 { + return Ok(default); + } + ConfigNonZeroUsize::try_new(usize_from_wire(value, message, field)?) +} + impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { type Error = DataFusionError; @@ -1188,6 +1204,10 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { ) -> datafusion_common::Result { let to_usize = |value: u64, field: &str| usize_from_wire(value, "ParquetOptions", field); + let defaults = ParquetOptions::default(); + let to_non_zero = |value: u64, field: &str, default: ConfigNonZeroUsize| { + non_zero_usize_from_wire(value, "ParquetOptions", field, default) + }; Ok(ParquetOptions { enable_page_index: value.enable_page_index, pruning: value.pruning, @@ -1207,7 +1227,11 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { value.data_pagesize_limit, "data_pagesize_limit", )?, - write_batch_size: to_usize(value.write_batch_size, "write_batch_size")?, + write_batch_size: to_non_zero( + value.write_batch_size, + "write_batch_size", + defaults.write_batch_size, + )?, writer_version: value.writer_version.parse().map_err(|e| { DataFusionError::Internal(format!("Failed to parse writer_version: {e}")) })?, @@ -1226,7 +1250,11 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { protobuf::parquet_options::StatisticsEnabledOpt::StatisticsEnabled(v) => v.parse(), }) .transpose()?, - max_row_group_size: to_usize(value.max_row_group_size, "max_row_group_size")?, + max_row_group_size: to_non_zero( + value.max_row_group_size, + "max_row_group_size", + defaults.max_row_group_size, + )?, max_in_list_size: to_usize(value.max_in_list_size, "max_in_list_size")?, row_group_range_assignment: RowGroupRangeAssignment::from_proto_str( &value.row_group_range_assignment, @@ -1236,7 +1264,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { .column_index_truncate_length_opt.as_ref() .map(|opt| match opt { protobuf::parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(v) => { - to_usize(*v, "column_index_truncate_length") + ConfigNonZeroUsize::try_new(to_usize(*v, "column_index_truncate_length")?) } }) .transpose()?, @@ -1244,13 +1272,14 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { .statistics_truncate_length_opt.as_ref() .map(|opt| match opt { protobuf::parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(v) => { - to_usize(*v, "statistics_truncate_length") + ConfigNonZeroUsize::try_new(to_usize(*v, "statistics_truncate_length")?) } }) .transpose()?, - data_page_row_count_limit: to_usize( + data_page_row_count_limit: to_non_zero( value.data_page_row_count_limit, "data_page_row_count_limit", + defaults.data_page_row_count_limit, )?, encoding: value .encoding_opt @@ -1264,9 +1293,9 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { bloom_filter_fpp: value.clone() .bloom_filter_fpp_opt .map(|opt| match opt { - protobuf::parquet_options::BloomFilterFppOpt::BloomFilterFpp(v) => Some(v), + protobuf::parquet_options::BloomFilterFppOpt::BloomFilterFpp(v) => BloomFilterFpp::try_new(v), }) - .unwrap_or(None), + .transpose()?, bloom_filter_ndv: value.clone() .bloom_filter_ndv_opt .map(|opt| match opt { @@ -1318,7 +1347,12 @@ impl TryFrom for ParquetCdcOptions { |value: u64, field: &str| usize_from_wire(value, "ParquetCdcOptions", field); Ok(ParquetCdcOptions { enabled: value.enabled, - min_chunk_size: to_usize(value.min_chunk_size, "min_chunk_size")?, + min_chunk_size: non_zero_usize_from_wire( + value.min_chunk_size, + "ParquetCdcOptions", + "min_chunk_size", + ParquetCdcOptions::default().min_chunk_size, + )?, max_chunk_size: to_usize(value.max_chunk_size, "max_chunk_size")?, norm_level: value.norm_level, }) @@ -1354,9 +1388,9 @@ impl TryFrom<&protobuf::ParquetColumnOptions> for ParquetColumnOptions { bloom_filter_fpp: value .bloom_filter_fpp_opt .map(|opt| match opt { - protobuf::parquet_column_options::BloomFilterFppOpt::BloomFilterFpp(v) => Some(v), + protobuf::parquet_column_options::BloomFilterFppOpt::BloomFilterFpp(v) => BloomFilterFpp::try_new(v), }) - .unwrap_or(None), + .transpose()?, bloom_filter_ndv: value .bloom_filter_ndv_opt .map(|opt| match opt { @@ -1452,7 +1486,8 @@ where mod tests { use datafusion_common::TableReference; use datafusion_common::config::{ - MaxRowGroupBytes, ParquetCdcOptions, ParquetOptions, TableParquetOptions, + ConfigNonZeroUsize, MaxRowGroupBytes, ParquetCdcOptions, ParquetOptions, + TableParquetOptions, }; use datafusion_common::parquet_config::{ DFParquetCompression, DFParquetEncoding, DFParquetStatistics, DFTimeUnit, @@ -1620,6 +1655,39 @@ mod tests { } } + #[test] + fn parquet_options_read_zero_non_zero_fields_as_absent() { + // Proto3 omits zero-valued scalars, so 0 means "not set" + let proto = crate::protobuf_common::ParquetOptions { + writer_version: "1.0".to_string(), + ..Default::default() + }; + let recovered = ParquetOptions::try_from(&proto).unwrap(); + let defaults = ParquetOptions::default(); + assert_eq!(recovered.write_batch_size, defaults.write_batch_size); + assert_eq!(recovered.max_row_group_size, defaults.max_row_group_size); + assert_eq!( + recovered.data_page_row_count_limit, + defaults.data_page_row_count_limit + ); + } + + #[test] + fn parquet_options_reject_explicit_zero_truncate_length() { + let proto = crate::protobuf_common::ParquetOptions { + writer_version: "1.0".to_string(), + statistics_truncate_length_opt: Some( + crate::protobuf_common::parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(0), + ), + ..Default::default() + }; + let err = ParquetOptions::try_from(&proto).unwrap_err(); + assert_eq!( + err.strip_backtrace(), + "Invalid or Unsupported Configuration: value must be greater than 0" + ); + } + #[test] fn table_parquet_options_propagates_nested_conversion_error() { let proto = crate::protobuf_common::TableParquetOptions { @@ -1846,7 +1914,7 @@ mod tests { let opts = ParquetOptions { content_defined_chunking: ParquetCdcOptions { enabled: true, - min_chunk_size: 128 * 1024, + min_chunk_size: ConfigNonZeroUsize::try_new(128 * 1024).unwrap(), max_chunk_size: 512 * 1024, norm_level: 2, }, @@ -1855,7 +1923,7 @@ mod tests { let recovered = parquet_options_proto_round_trip(opts.clone()); let cdc = recovered.content_defined_chunking; assert!(cdc.enabled); - assert_eq!(cdc.min_chunk_size, 128 * 1024); + assert_eq!(cdc.min_chunk_size.get(), 128 * 1024); assert_eq!(cdc.max_chunk_size, 512 * 1024); assert_eq!(cdc.norm_level, 2); } @@ -1879,7 +1947,7 @@ mod tests { let mut opts = TableParquetOptions::default(); opts.global.content_defined_chunking = ParquetCdcOptions { enabled: true, - min_chunk_size: 64 * 1024, + min_chunk_size: ConfigNonZeroUsize::try_new(64 * 1024).unwrap(), max_chunk_size: 2 * 1024 * 1024, norm_level: -1, }; @@ -1887,7 +1955,7 @@ mod tests { let recovered = table_parquet_options_proto_round_trip(opts.clone()); let cdc = recovered.global.content_defined_chunking; assert!(cdc.enabled); - assert_eq!(cdc.min_chunk_size, 64 * 1024); + assert_eq!(cdc.min_chunk_size.get(), 64 * 1024); assert_eq!(cdc.max_chunk_size, 2 * 1024 * 1024); assert_eq!(cdc.norm_level, -1); } diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs index 22b291843def5..4627ec13b926f 100644 --- a/datafusion/proto-common/src/to_proto/mod.rs +++ b/datafusion/proto-common/src/to_proto/mod.rs @@ -942,23 +942,23 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions { reorder_filters: value.reorder_filters, force_filter_selections: value.force_filter_selections, data_pagesize_limit: value.data_pagesize_limit as u64, - write_batch_size: value.write_batch_size as u64, + write_batch_size: value.write_batch_size.get() as u64, writer_version: value.writer_version.to_string(), 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())), - max_row_group_size: value.max_row_group_size as u64, + max_row_group_size: value.max_row_group_size.get() as u64, max_in_list_size: value.max_in_list_size as u64, row_group_range_assignment: value.row_group_range_assignment.to_string(), created_by: value.created_by.clone(), - column_index_truncate_length_opt: value.column_index_truncate_length.map(|v| protobuf::parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(v as u64)), - statistics_truncate_length_opt: value.statistics_truncate_length.map(|v| protobuf::parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(v as u64)), - data_page_row_count_limit: value.data_page_row_count_limit as u64, + column_index_truncate_length_opt: value.column_index_truncate_length.map(|v| protobuf::parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(v.get() as u64)), + statistics_truncate_length_opt: value.statistics_truncate_length.map(|v| protobuf::parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(v.get() as u64)), + data_page_row_count_limit: value.data_page_row_count_limit.get() as u64, encoding_opt: value.encoding.map(|encoding| protobuf::parquet_options::EncodingOpt::Encoding(encoding.to_string())), bloom_filter_on_read: value.bloom_filter_on_read, bloom_filter_on_write: value.bloom_filter_on_write, - bloom_filter_fpp_opt: value.bloom_filter_fpp.map(protobuf::parquet_options::BloomFilterFppOpt::BloomFilterFpp), + bloom_filter_fpp_opt: value.bloom_filter_fpp.map(|v| protobuf::parquet_options::BloomFilterFppOpt::BloomFilterFpp(v.get())), bloom_filter_ndv_opt: value.bloom_filter_ndv.map(protobuf::parquet_options::BloomFilterNdvOpt::BloomFilterNdv), allow_single_file_parallelism: value.allow_single_file_parallelism, maximum_parallel_row_group_writers: value.maximum_parallel_row_group_writers as u64, @@ -981,7 +981,7 @@ impl From<&ParquetCdcOptions> for protobuf::ParquetCdcOptions { fn from(value: &ParquetCdcOptions) -> Self { protobuf::ParquetCdcOptions { enabled: value.enabled, - min_chunk_size: value.min_chunk_size as u64, + min_chunk_size: value.min_chunk_size.get() as u64, max_chunk_size: value.max_chunk_size as u64, norm_level: value.norm_level, } @@ -1013,9 +1013,11 @@ impl TryFrom<&ParquetColumnOptions> for protobuf::ParquetColumnOptions { bloom_filter_enabled_opt: value .bloom_filter_enabled .map(protobuf::parquet_column_options::BloomFilterEnabledOpt::BloomFilterEnabled), - bloom_filter_fpp_opt: value - .bloom_filter_fpp - .map(protobuf::parquet_column_options::BloomFilterFppOpt::BloomFilterFpp), + bloom_filter_fpp_opt: value.bloom_filter_fpp.map(|v| { + protobuf::parquet_column_options::BloomFilterFppOpt::BloomFilterFpp( + v.get(), + ) + }), bloom_filter_ndv_opt: value .bloom_filter_ndv .map(protobuf::parquet_column_options::BloomFilterNdvOpt::BloomFilterNdv), diff --git a/datafusion/proto-models/src/from_proto.rs b/datafusion/proto-models/src/from_proto.rs index 6d3cadc577acb..10ed9c55a58c6 100644 --- a/datafusion/proto-models/src/from_proto.rs +++ b/datafusion/proto-models/src/from_proto.rs @@ -26,8 +26,8 @@ use std::sync::Arc; use datafusion_common::config::{ - CsvOptions, JsonOptions, MaxRowGroupBytes, ParquetCdcOptions, ParquetColumnOptions, - ParquetOptions, TableParquetOptions, + BloomFilterFpp, ConfigNonZeroUsize, CsvOptions, JsonOptions, MaxRowGroupBytes, + ParquetCdcOptions, ParquetColumnOptions, ParquetOptions, TableParquetOptions, }; use datafusion_common::display::{PlanType, StringifiedPlan}; use datafusion_common::parquet_config::RowGroupRangeAssignment; @@ -326,6 +326,23 @@ impl TryFrom<&JsonOptionsProto> for JsonOptions { } } +/// Decodes a non-optional, must-be-positive `usize` option from the wire. +/// +/// Proto3 decodes an omitted scalar field as 0, which is never a valid value +/// for these options, so 0 is read as "absent" and replaced by `default`, the +/// same way an empty `writer_version` is. +fn non_zero_usize_from_wire( + value: u64, + message: &str, + field: &str, + default: ConfigNonZeroUsize, +) -> datafusion_common::Result { + if value == 0 { + return Ok(default); + } + ConfigNonZeroUsize::try_new(usize_from_wire(value, message, field)?) +} + impl TryFrom for ParquetCdcOptions { type Error = datafusion_common::DataFusionError; @@ -336,7 +353,12 @@ impl TryFrom for ParquetCdcOptions { |value: u64, field: &str| usize_from_wire(value, "ParquetCdcOptions", field); Ok(ParquetCdcOptions { enabled: value.enabled, - min_chunk_size: to_usize(value.min_chunk_size, "min_chunk_size")?, + min_chunk_size: non_zero_usize_from_wire( + value.min_chunk_size, + "ParquetCdcOptions", + "min_chunk_size", + ParquetCdcOptions::default().min_chunk_size, + )?, max_chunk_size: to_usize(value.max_chunk_size, "max_chunk_size")?, norm_level: value.norm_level, }) @@ -358,6 +380,10 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { }; let to_usize = |value: u64, field: &str| usize_from_wire(value, "ParquetOptions", field); + let defaults = ParquetOptions::default(); + let to_non_zero = |value: u64, field: &str, default: ConfigNonZeroUsize| { + non_zero_usize_from_wire(value, "ParquetOptions", field, default) + }; Ok(ParquetOptions { enable_page_index: proto.enable_page_index, @@ -379,7 +405,11 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { proto.data_pagesize_limit, "data_pagesize_limit", )?, - write_batch_size: to_usize(proto.write_batch_size, "write_batch_size")?, + write_batch_size: to_non_zero( + proto.write_batch_size, + "write_batch_size", + defaults.write_batch_size, + )?, writer_version, compression: proto .compression_opt @@ -411,7 +441,11 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { ) => statistics.parse(), }) .transpose()?, - max_row_group_size: to_usize(proto.max_row_group_size, "max_row_group_size")?, + max_row_group_size: to_non_zero( + proto.max_row_group_size, + "max_row_group_size", + defaults.max_row_group_size, + )?, max_in_list_size: to_usize(proto.max_in_list_size, "max_in_list_size")?, row_group_range_assignment: RowGroupRangeAssignment::from_proto_str( &proto.row_group_range_assignment, @@ -422,7 +456,7 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { .as_ref() .map(|opt| match opt { parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(length) => { - to_usize(*length, "column_index_truncate_length") + ConfigNonZeroUsize::try_new(to_usize(*length, "column_index_truncate_length")?) } }) .transpose()?, @@ -431,13 +465,14 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { .as_ref() .map(|opt| match opt { parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(length) => { - to_usize(*length, "statistics_truncate_length") + ConfigNonZeroUsize::try_new(to_usize(*length, "statistics_truncate_length")?) } }) .transpose()?, - data_page_row_count_limit: to_usize( + data_page_row_count_limit: to_non_zero( proto.data_page_row_count_limit, "data_page_row_count_limit", + defaults.data_page_row_count_limit, )?, encoding: proto .encoding_opt @@ -452,8 +487,11 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { .bloom_filter_fpp_opt .as_ref() .map(|opt| match opt { - parquet_options::BloomFilterFppOpt::BloomFilterFpp(fpp) => *fpp, - }), + parquet_options::BloomFilterFppOpt::BloomFilterFpp(fpp) => { + BloomFilterFpp::try_new(*fpp) + } + }) + .transpose()?, bloom_filter_ndv: proto .bloom_filter_ndv_opt .as_ref() @@ -518,9 +556,13 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { } } -impl From for ParquetColumnOptions { - fn from(proto: ParquetColumnOptionsProto) -> Self { - ParquetColumnOptions { +impl TryFrom for ParquetColumnOptions { + type Error = datafusion_common::DataFusionError; + + fn try_from( + proto: ParquetColumnOptionsProto, + ) -> datafusion_common::Result { + Ok(ParquetColumnOptions { bloom_filter_enabled: proto.bloom_filter_enabled_opt.map( |parquet_column_options::BloomFilterEnabledOpt::BloomFilterEnabled(v)| v, ), @@ -538,11 +580,16 @@ impl From for ParquetColumnOptions { ), bloom_filter_fpp: proto .bloom_filter_fpp_opt - .map(|parquet_column_options::BloomFilterFppOpt::BloomFilterFpp(v)| v), + .map( + |parquet_column_options::BloomFilterFppOpt::BloomFilterFpp(v)| { + BloomFilterFpp::try_new(v) + }, + ) + .transpose()?, bloom_filter_ndv: proto .bloom_filter_ndv_opt .map(|parquet_column_options::BloomFilterNdvOpt::BloomFilterNdv(v)| v), - } + }) } } @@ -563,14 +610,14 @@ impl TryFrom<&TableParquetOptionsProto> for TableParquetOptions { .column_specific_options .iter() .map(|parquet_column_options| { - ( + Ok(( parquet_column_options.column_name.clone(), - ParquetColumnOptions::from( + ParquetColumnOptions::try_from( parquet_column_options.options.clone().unwrap_or_default(), - ), - ) + )?, + )) }) - .collect(), + .collect::>()?, key_value_metadata: proto .key_value_metadata .iter() diff --git a/datafusion/proto/tests/cases/public_conversions.rs b/datafusion/proto/tests/cases/public_conversions.rs index ec71960f643f3..3b2ea712fb5df 100644 --- a/datafusion/proto/tests/cases/public_conversions.rs +++ b/datafusion/proto/tests/cases/public_conversions.rs @@ -131,7 +131,7 @@ fn file_format_option_conversions_are_std_traits() { assert_datafusion_try_from::<&protobuf::CsvOptions, CsvOptions>(); assert_datafusion_try_from::<&protobuf::JsonOptions, JsonOptions>(); assert_datafusion_try_from::<&protobuf::ParquetOptions, ParquetOptions>(); - assert_from::(); + assert_datafusion_try_from::(); assert_datafusion_try_from::(); assert_datafusion_try_from::<&protobuf::TableParquetOptions, TableParquetOptions>(); assert_from::<&CsvFormatFactory, protobuf::CsvOptions>(); diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index d27aaf27a04f1..a8115e3cec7cc 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -70,7 +70,7 @@ use datafusion::functions_window::rank::rank_udwf; use datafusion::physical_expr::PhysicalExpr; use datafusion::prelude::*; use datafusion::test_util::{TestTableFactory, TestTableProvider}; -use datafusion_common::config::TableOptions; +use datafusion_common::config::{ConfigNonZeroUsize, TableOptions}; use datafusion_common::format::{ ExplainAnalyzeCategories, ExplainFormat, MetricCategory, MetricType, }; @@ -943,11 +943,11 @@ async fn roundtrip_logical_plan_copy_to_writer_options() -> Result<()> { parquet_format.global.bloom_filter_on_read = true; parquet_format.global.created_by = "DataFusion Test".to_string(); parquet_format.global.writer_version = DFParquetWriterVersion::V2_0; - parquet_format.global.write_batch_size = 111; + parquet_format.global.write_batch_size = ConfigNonZeroUsize::try_new(111)?; parquet_format.global.data_pagesize_limit = 222; - parquet_format.global.data_page_row_count_limit = 333; + parquet_format.global.data_page_row_count_limit = ConfigNonZeroUsize::try_new(333)?; parquet_format.global.dictionary_page_size_limit = 444; - parquet_format.global.max_row_group_size = 555; + parquet_format.global.max_row_group_size = ConfigNonZeroUsize::try_new(555)?; let file_type = format_as_file_type(Arc::new( ParquetFormatFactory::new_with_options(parquet_format), diff --git a/datafusion/sqllogictest/test_files/parquet_writer_options_validation.slt b/datafusion/sqllogictest/test_files/parquet_writer_options_validation.slt new file mode 100644 index 0000000000000..71822b1319811 --- /dev/null +++ b/datafusion/sqllogictest/test_files/parquet_writer_options_validation.slt @@ -0,0 +1,122 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at + +# http://www.apache.org/licenses/LICENSE-2.0 + +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# Parquet writer options that the parquet crate cannot accept are rejected +# with a configuration error instead of panicking (or hanging) on the next +# write. See https://github.com/apache/datafusion/issues/24904 + +statement ok +set datafusion.catalog.information_schema = true + +# Zero is rejected when the option is set + +statement error value must be greater than 0 +SET datafusion.execution.parquet.max_row_group_size = 0 + +statement error value must be greater than 0 +SET datafusion.execution.parquet.write_batch_size = 0 + +statement error value must be greater than 0 +SET datafusion.execution.parquet.data_page_row_count_limit = 0 + +statement error value must be greater than 0 +SET datafusion.execution.parquet.column_index_truncate_length = 0 + +statement error value must be greater than 0 +SET datafusion.execution.parquet.statistics_truncate_length = 0 + +statement error value must be greater than 0 +SET datafusion.execution.parquet.content_defined_chunking.min_chunk_size = 0 + +# A rejected value leaves the previous setting in place + +query T +SELECT value FROM information_schema.df_settings WHERE name = 'datafusion.execution.parquet.statistics_truncate_length' +---- +64 + +# The false positive probability must be strictly between 0 and 1 + +statement error bloom_filter_fpp must be between 0 and 1 \(exclusive\), got 1\.5 +SET datafusion.execution.parquet.bloom_filter_fpp = 1.5 + +statement error bloom_filter_fpp must be between 0 and 1 \(exclusive\), got 0 +SET datafusion.execution.parquet.bloom_filter_fpp = 0 + +statement error bloom_filter_fpp must be between 0 and 1 \(exclusive\), got 1 +SET datafusion.execution.parquet.bloom_filter_fpp = 1 + +statement error bloom_filter_fpp must be between 0 and 1 \(exclusive\), got -0\.1 +SET datafusion.execution.parquet.bloom_filter_fpp = -0.1 + +statement error bloom_filter_fpp must be between 0 and 1 \(exclusive\), got NaN +SET datafusion.execution.parquet.bloom_filter_fpp = 'NaN' + +statement error Invalid bloom_filter_fpp: 'abc' +SET datafusion.execution.parquet.bloom_filter_fpp = 'abc' + +# Valid values are still accepted and used for writing + +statement ok +SET datafusion.execution.parquet.bloom_filter_fpp = 0.05 + +query T +SELECT value FROM information_schema.df_settings WHERE name = 'datafusion.execution.parquet.bloom_filter_fpp' +---- +0.05 + +statement ok +SET datafusion.execution.parquet.max_row_group_size = 2 + +query I +COPY (VALUES (1), (2), (3)) TO 'test_files/scratch/parquet_writer_options_validation/valid.parquet' +---- +3 + +statement ok +RESET datafusion.execution.parquet.bloom_filter_fpp + +statement ok +RESET datafusion.execution.parquet.max_row_group_size + +# The same checks apply to options passed to COPY + +statement error value must be greater than 0 +COPY (VALUES (1)) TO 'test_files/scratch/parquet_writer_options_validation/zero_row_group.parquet' +OPTIONS ('format.max_row_group_size' 0) + +statement error bloom_filter_fpp must be between 0 and 1 \(exclusive\), got 1\.5 +COPY (VALUES (1)) TO 'test_files/scratch/parquet_writer_options_validation/global_fpp.parquet' +OPTIONS ('format.bloom_filter_fpp' 1.5) + +statement error bloom_filter_fpp must be between 0 and 1 \(exclusive\), got 1\.5 +COPY (VALUES (1)) TO 'test_files/scratch/parquet_writer_options_validation/column_fpp.parquet' +OPTIONS ('format.bloom_filter_fpp::column1' 1.5) + +# max_chunk_size must be greater than min_chunk_size. Each is valid on its own, +# so this is checked when the writer is built + +statement error content_defined_chunking\.max_chunk_size \(1024\) must be greater than content_defined_chunking\.min_chunk_size \(1024\) +COPY (VALUES (1)) TO 'test_files/scratch/parquet_writer_options_validation/cdc.parquet' +OPTIONS ( + 'format.content_defined_chunking.enabled' true, + 'format.content_defined_chunking.min_chunk_size' 1024, + 'format.content_defined_chunking.max_chunk_size' 1024 +) + +statement ok +RESET datafusion.catalog.information_schema diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 648071a2514ab..9e621b2d57ce2 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -403,6 +403,52 @@ options.execution.time_zone = Some("America/Denver".parse()?); let tz: Option<&str> = options.execution.time_zone.as_ref().map(|tz| tz.as_str()); ``` +### Parquet writer options are validated when they are set + +Parquet writer options that the `parquet` crate cannot accept are now rejected +when they are set, instead of panicking (or hanging) on the next write. The +following fields of `ParquetOptions` changed type: + +| Field | Before | After | +| ----------------------------------------- | --------------- | ---------------------------- | +| `write_batch_size` | `usize` | `ConfigNonZeroUsize` | +| `max_row_group_size` | `usize` | `ConfigNonZeroUsize` | +| `data_page_row_count_limit` | `usize` | `ConfigNonZeroUsize` | +| `column_index_truncate_length` | `Option` | `Option` | +| `statistics_truncate_length` | `Option` | `Option` | +| `bloom_filter_fpp` | `Option` | `Option` | +| `content_defined_chunking.min_chunk_size` | `usize` | `ConfigNonZeroUsize` | + +`ParquetColumnOptions::bloom_filter_fpp` changed the same way, and the protobuf +conversion `From for ParquetColumnOptions` is +now `TryFrom`, so an out-of-range value is rejected when a plan is decoded. + +**Who is affected:** + +- Code that reads or assigns these fields directly. +- Code that converts `protobuf::ParquetColumnOptions` with `From`/`into()`. +- Code that sets the values through SQL `SET`, `COPY ... OPTIONS` or + `ConfigOptions::set` needs no code changes, but invalid values are now + rejected immediately. + +**Migration guide:** + +```rust,ignore +// Before +options.global.max_row_group_size = 1024 * 1024; +options.global.bloom_filter_fpp = Some(0.01); +let rows: usize = options.global.max_row_group_size; +let column = ParquetColumnOptions::from(proto_column); + +// After +options.global.max_row_group_size = ConfigNonZeroUsize::try_new(1024 * 1024)?; +options.global.bloom_filter_fpp = Some(BloomFilterFpp::try_new(0.01)?); +let rows: usize = options.global.max_row_group_size.get(); +let column = ParquetColumnOptions::try_from(proto_column)?; +``` + +See [issue #24904](https://github.com/apache/datafusion/issues/24904) for details. + ### `ExecutionOptions` has a new `soft_max_bytes_per_output_file` field `ExecutionOptions` gained a public `soft_max_bytes_per_output_file: ConfigNonZeroUsize` diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 273b0e39cfedd..2e72d0fec907e 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -101,30 +101,30 @@ The following configuration settings are available: | datafusion.execution.parquet.enable_rle_to_dictionary | false | (reading) If true, top-level string and binary Parquet columns with dictionary pages are inferred and scanned as `Dictionary` / `Dictionary` instead of their plain value type. This applies only when DataFusion infers the table schema. Tables with a user-supplied schema are not promoted because the Parquet footer is not read at DDL time, so dictionary pages cannot be detected per column. See | | datafusion.execution.parquet.row_group_range_assignment | start_offset | (reading) Which byte range of a split file reads each row group. `start_offset` picks the range containing the row group's start. `midpoint` picks the range containing its midpoint, as Spark does, which spreads large row groups more evenly across ranges. | | datafusion.execution.parquet.data_pagesize_limit | 1048576 | (writing) Sets best effort maximum size of data page in bytes | -| datafusion.execution.parquet.write_batch_size | 1024 | (writing) Sets write_batch_size in rows | +| datafusion.execution.parquet.write_batch_size | 1024 | (writing) Sets write_batch_size in rows. Must be greater than 0. | | datafusion.execution.parquet.writer_version | 1.0 | (writing) Sets parquet writer version valid values are "1.0" and "2.0" | | datafusion.execution.parquet.skip_arrow_metadata | false | (writing) Skip encoding the embedded arrow metadata in the KV_meta This is analogous to the `ArrowWriterOptions::with_skip_arrow_metadata`. Refer to | | datafusion.execution.parquet.compression | zstd(3) | (writing) Sets default parquet compression codec. Valid values are: uncompressed, snappy, gzip(level), brotli(level), lz4, zstd(level), and lz4_raw. These values are not case sensitive. If NULL, uses default parquet writer setting Note that this default setting is not the same as the default parquet writer setting. | | datafusion.execution.parquet.dictionary_enabled | true | (writing) Sets if dictionary encoding is enabled. If NULL, uses default parquet writer setting | | datafusion.execution.parquet.dictionary_page_size_limit | 1048576 | (writing) Sets best effort maximum dictionary page size, in bytes | | datafusion.execution.parquet.statistics_enabled | page | (writing) Sets if statistics are enabled for any column Valid values are: "none", "chunk", and "page" These values are not case sensitive. If NULL, uses default parquet writer setting | -| datafusion.execution.parquet.max_row_group_size | 1048576 | (writing) Target maximum number of rows in each row group (defaults to 1M rows). Writing larger row groups requires more memory to write, but can get better compression and be faster to read. When `max_row_group_bytes` is also set, the writer flushes a row group when either limit is reached, whichever comes first. | +| datafusion.execution.parquet.max_row_group_size | 1048576 | (writing) Target maximum number of rows in each row group (defaults to 1M rows). Writing larger row groups requires more memory to write, but can get better compression and be faster to read. When `max_row_group_bytes` is also set, the writer flushes a row group when either limit is reached, whichever comes first. Must be greater than 0. | | datafusion.execution.parquet.max_row_group_bytes | NULL | (writing) Target maximum size of each row group in bytes. When set, the writer flushes whenever either this limit or `max_row_group_size` is reached, whichever comes first. Useful for bounding writer memory on wide schemas where a row-count limit can map to very different byte sizes. Matches the behavior of `parquet.block.size` in parquet-mr. If `None` (the default), only the row-count limit applies. Currently only honored when `allow_single_file_parallelism` is `false`; by default the parallel file writer ignores this limit. | | datafusion.execution.parquet.created_by | datafusion version 55.1.0 | (writing) Sets "created by" property | -| datafusion.execution.parquet.column_index_truncate_length | 64 | (writing) Sets column index truncate length | -| datafusion.execution.parquet.statistics_truncate_length | 64 | (writing) Sets statistics truncate length. If NULL, uses default parquet writer setting | -| datafusion.execution.parquet.data_page_row_count_limit | 20000 | (writing) Sets best effort maximum number of rows in data page | +| datafusion.execution.parquet.column_index_truncate_length | 64 | (writing) Sets column index truncate length. Must be greater than 0. If NULL, column index values are not truncated | +| datafusion.execution.parquet.statistics_truncate_length | 64 | (writing) Sets statistics truncate length. Must be greater than 0. If NULL, uses default parquet writer setting | +| datafusion.execution.parquet.data_page_row_count_limit | 20000 | (writing) Sets best effort maximum number of rows in data page. Must be greater than 0. | | datafusion.execution.parquet.encoding | NULL | (writing) Sets default encoding for any column. Valid values are: plain, plain_dictionary, rle, bit_packed, delta_binary_packed, delta_length_byte_array, delta_byte_array, rle_dictionary, and byte_stream_split. These values are not case sensitive. If NULL, uses default parquet writer setting | | datafusion.execution.parquet.bloom_filter_on_write | false | (writing) Write bloom filters for all columns when creating parquet files | -| datafusion.execution.parquet.bloom_filter_fpp | NULL | (writing) Sets bloom filter false positive probability. If NULL, uses default parquet writer setting | +| datafusion.execution.parquet.bloom_filter_fpp | NULL | (writing) Sets bloom filter false positive probability. Must be between 0 and 1 (exclusive). If NULL, uses default parquet writer setting | | datafusion.execution.parquet.bloom_filter_ndv | NULL | (writing) Sets bloom filter number of distinct values. If NULL, uses default parquet writer setting | | datafusion.execution.parquet.write_row_group_number_distinct_values | false | (writing) Write the number of distinct values (NDV) for each column in row group statistics when creating parquet files. Enabling this improves NDV-based query optimizations at the cost of hashing every non-null value during write. | | datafusion.execution.parquet.allow_single_file_parallelism | true | (writing) Controls whether DataFusion will attempt to speed up writing parquet files by serializing them in parallel. Each column in each row group in each output file are serialized in parallel leveraging a maximum possible core count of n_files\*n_row_groups\*n_columns. | | datafusion.execution.parquet.maximum_parallel_row_group_writers | 1 | (writing) By default parallel parquet writer is tuned for minimum memory usage in a streaming execution plan. You may see a performance benefit when writing large parquet files by increasing maximum_parallel_row_group_writers and maximum_buffered_record_batches_per_stream if your system has idle cores and can tolerate additional memory usage. Boosting these values is likely worthwhile when writing out already in-memory data, such as from a cached data frame. | | datafusion.execution.parquet.maximum_buffered_record_batches_per_stream | 2 | (writing) By default parallel parquet writer is tuned for minimum memory usage in a streaming execution plan. You may see a performance benefit when writing large parquet files by increasing maximum_parallel_row_group_writers and maximum_buffered_record_batches_per_stream if your system has idle cores and can tolerate additional memory usage. Boosting these values is likely worthwhile when writing out already in-memory data, such as from a cached data frame. | | datafusion.execution.parquet.content_defined_chunking.enabled | false | (writing) EXPERIMENTAL: Enable content-defined chunking (CDC) when writing parquet files. When enabled, parallel writing is automatically disabled since the chunker state must persist across row groups. | -| datafusion.execution.parquet.content_defined_chunking.min_chunk_size | 262144 | Minimum chunk size in bytes. The rolling hash will not trigger a split until this many bytes have been accumulated. Default is 256 KiB. | -| datafusion.execution.parquet.content_defined_chunking.max_chunk_size | 1048576 | Maximum chunk size in bytes. A split is forced when the accumulated size exceeds this value. Default is 1 MiB. | +| datafusion.execution.parquet.content_defined_chunking.min_chunk_size | 262144 | Minimum chunk size in bytes. The rolling hash will not trigger a split until this many bytes have been accumulated. Must be greater than 0. Default is 256 KiB. | +| datafusion.execution.parquet.content_defined_chunking.max_chunk_size | 1048576 | Maximum chunk size in bytes. A split is forced when the accumulated size exceeds this value. Must be greater than `min_chunk_size`. Default is 1 MiB. | | datafusion.execution.parquet.content_defined_chunking.norm_level | 0 | Normalization level. Increasing this improves deduplication ratio but increases fragmentation. Recommended range is [-3, 3], default is 0. | | datafusion.execution.planning_concurrency | 0 | Fan-out during initial physical planning. This is mostly use to plan `UNION` children in parallel. Defaults to the number of CPU cores on the system | | datafusion.execution.skip_physical_aggregate_schema_check | false | When set to true, skips verifying that the schema produced by planning the input of `LogicalPlan::Aggregate` exactly matches the schema of the input plan. When set to false, if the schema does not match exactly (including nullability and metadata), a planning error will be raised. This is used to workaround bugs in the planner that are now caught by the new schema verification step. |