refactor: rename CDC options to ParquetCdcOptions and tidy mappings

Follow-up refinements to the parquet CDC options (all unreleased):

* Rename the config struct `CdcOptions` -> `ParquetCdcOptions` for explicitness
  and consistency with the other parquet sub-option structs
  (`ParquetColumnOptions`, `ParquetEncryptionOptions`), and to disambiguate from
  the unrelated "change data capture" meaning of CDC.
* Drop the chunk-size validation in the parquet writer path (parquet-rs enforces
  the bounds) and gate the writer call on the `enabled` flag.
* Add `From` conversions for the config <-> parquet-rs and config <-> proto
  `CdcOptions` mappings, replacing the inline field copies. parquet-rs has no
  `enabled` flag, so the conversion encodes enabled <-> Option presence.
* Rename the proto message to a top-level `ParquetCdcOptions` so config and proto
  names line up; field tags are unchanged, so the wire format is unaffected.

Regenerated prost/pbjson for proto-common and proto-models.

Author: kszucs

datafusion/common/src/config.rs

@@ -760,11 +760,11 @@ config_namespace! {
     /// Options for content-defined chunking (CDC) when writing parquet files.
     /// Mirrors `parquet::file::properties::CdcOptions`.
     ///
-    /// Carried as a [`CdcOptions`] in [`ParquetOptions::content_defined_chunking`]
+    /// Carried as a [`ParquetCdcOptions`] in [`ParquetOptions::content_defined_chunking`]
     /// with an explicit `enabled` flag, so it can be toggled with dotted config
     /// keys (`content_defined_chunking.enabled = true|false`) and the result is
     /// independent of the order in which the keys are set.
-    pub struct CdcOptions {
+    pub struct ParquetCdcOptions {
         /// (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.
@@ -784,20 +784,20 @@ config_namespace! {
     }
 }
 
-impl CdcOptions {
+impl ParquetCdcOptions {
     /// Returns enabled CDC options with the default chunking parameters.
     ///
-    /// Shorthand for `CdcOptions { enabled: true, ..Default::default() }`; combine
-    /// with struct-update syntax to override parameters, e.g.
-    /// `CdcOptions { min_chunk_size: 4096, ..CdcOptions::enabled() }`.
+    /// Shorthand for `ParquetCdcOptions { enabled: true, ..Default::default() }`;
+    /// combine with struct-update syntax to override parameters, e.g.
+    /// `ParquetCdcOptions { min_chunk_size: 4096, ..ParquetCdcOptions::enabled() }`.
     pub fn enabled() -> Self {
         Self {
             enabled: true,
             ..Default::default()
         }
     }
 
-    /// Returns disabled CDC options (equivalent to [`CdcOptions::default`]).
+    /// Returns disabled CDC options (equivalent to [`ParquetCdcOptions::default`]).
     pub fn disabled() -> Self {
         Self::default()
     }
@@ -1006,7 +1006,7 @@ config_namespace! {
         /// enabled, parallel writing is automatically disabled since the chunker state
         /// must persist across row groups. Mirrors
         /// `parquet::file::properties::WriterProperties::content_defined_chunking`.
-        pub content_defined_chunking: CdcOptions, default = Default::default()
+        pub content_defined_chunking: ParquetCdcOptions, default = Default::default()
     }
 }
 

datafusion/common/src/file_options/parquet_writer.rs

@@ -21,7 +21,7 @@ use std::sync::Arc;
 
 use crate::{
     _internal_datafusion_err, DataFusionError, Result,
-    config::{ParquetOptions, TableParquetOptions},
+    config::{ParquetCdcOptions, ParquetOptions, TableParquetOptions},
 };
 
 use arrow::datatypes::Schema;
@@ -166,6 +166,42 @@ impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
     }
 }
 
+/// Convert DataFusion's [`ParquetCdcOptions`] into parquet-rs's `Option<CdcOptions>`.
+///
+/// parquet-rs has no `enabled` flag; CDC is on when the option is `Some`. So a
+/// disabled [`ParquetCdcOptions`] maps to `None`, and an enabled one to `Some`
+/// with the chunking parameters.
+impl From<&ParquetCdcOptions> for Option<parquet::file::properties::CdcOptions> {
+    fn from(value: &ParquetCdcOptions) -> Self {
+        value
+            .enabled
+            .then_some(parquet::file::properties::CdcOptions {
+                min_chunk_size: value.min_chunk_size,
+                max_chunk_size: value.max_chunk_size,
+                norm_level: value.norm_level,
+            })
+    }
+}
+
+/// Convert parquet-rs's `Option<&CdcOptions>` back into DataFusion's
+/// [`ParquetCdcOptions`].
+///
+/// The presence of parquet-rs options means CDC was enabled, so `Some` maps to
+/// `enabled: true`; `None` yields the disabled default.
+impl From<Option<&parquet::file::properties::CdcOptions>> for ParquetCdcOptions {
+    fn from(value: Option<&parquet::file::properties::CdcOptions>) -> Self {
+        match value {
+            Some(cdc) => ParquetCdcOptions {
+                enabled: true,
+                min_chunk_size: cdc.min_chunk_size,
+                max_chunk_size: cdc.max_chunk_size,
+                norm_level: cdc.norm_level,
+            },
+            None => ParquetCdcOptions::default(),
+        }
+    }
+}
+
 impl ParquetOptions {
     /// Convert the global session options, [`ParquetOptions`], into a single write action's [`WriterPropertiesBuilder`].
     ///
@@ -249,27 +285,7 @@ impl ParquetOptions {
         if let Some(encoding) = encoding {
             builder = builder.set_encoding(parse_encoding_string(encoding)?);
         }
-        if content_defined_chunking.enabled {
-            let cdc = content_defined_chunking;
-            if cdc.min_chunk_size == 0 {
-                return Err(DataFusionError::Configuration(
-                    "CDC min_chunk_size must be greater than 0".to_string(),
-                ));
-            }
-            if cdc.max_chunk_size <= cdc.min_chunk_size {
-                return Err(DataFusionError::Configuration(format!(
-                    "CDC max_chunk_size ({}) must be greater than min_chunk_size ({})",
-                    cdc.max_chunk_size, cdc.min_chunk_size
-                )));
-            }
-            builder = builder.set_content_defined_chunking(Some(
-                parquet::file::properties::CdcOptions {
-                    min_chunk_size: cdc.min_chunk_size,
-                    max_chunk_size: cdc.max_chunk_size,
-                    norm_level: cdc.norm_level,
-                },
-            ));
-        }
+        builder = builder.set_content_defined_chunking(content_defined_chunking.into());
 
         Ok(builder)
     }
@@ -412,7 +428,7 @@ mod tests {
     #[cfg(feature = "parquet_encryption")]
     use crate::config::ConfigFileEncryptionProperties;
     use crate::config::{
-        CdcOptions, ParquetColumnOptions, ParquetEncryptionOptions, ParquetOptions,
+        ParquetCdcOptions, ParquetColumnOptions, ParquetEncryptionOptions, ParquetOptions,
     };
     use crate::parquet_config::DFParquetWriterVersion;
     use parquet::basic::Compression;
@@ -604,15 +620,7 @@ mod tests {
                 skip_arrow_metadata: global_options_defaults.skip_arrow_metadata,
                 coerce_int96: None,
                 coerce_int96_tz: None,
-                content_defined_chunking: props
-                    .content_defined_chunking()
-                    .map(|c| CdcOptions {
-                        enabled: true,
-                        min_chunk_size: c.min_chunk_size,
-                        max_chunk_size: c.max_chunk_size,
-                        norm_level: c.norm_level,
-                    })
-                    .unwrap_or_default(),
+                content_defined_chunking: props.content_defined_chunking().into(),
             },
             column_specific_options,
             key_value_metadata,
@@ -826,7 +834,7 @@ mod tests {
     #[test]
     fn test_cdc_enabled_with_custom_options() {
         let mut opts = TableParquetOptions::default();
-        opts.global.content_defined_chunking = CdcOptions {
+        opts.global.content_defined_chunking = ParquetCdcOptions {
             enabled: true,
             min_chunk_size: 128 * 1024,
             max_chunk_size: 512 * 1024,
@@ -854,7 +862,7 @@ mod tests {
     fn test_cdc_params_ignored_when_disabled() {
         // Parameters are customized but `enabled` is false, so CDC stays off.
         let mut opts = TableParquetOptions::default();
-        opts.global.content_defined_chunking = CdcOptions {
+        opts.global.content_defined_chunking = ParquetCdcOptions {
             enabled: false,
             min_chunk_size: 128 * 1024,
             max_chunk_size: 512 * 1024,
@@ -869,7 +877,7 @@ mod tests {
     #[test]
     fn test_cdc_round_trip_through_writer_props() {
         let mut opts = TableParquetOptions::default();
-        opts.global.content_defined_chunking = CdcOptions {
+        opts.global.content_defined_chunking = ParquetCdcOptions {
             enabled: true,
             min_chunk_size: 64 * 1024,
             max_chunk_size: 2 * 1024 * 1024,
@@ -887,31 +895,6 @@ mod tests {
         assert_eq!(cdc.norm_level, -1);
     }
 
-    #[test]
-    fn test_cdc_validation_zero_min_chunk_size() {
-        let mut opts = TableParquetOptions::default();
-        opts.global.content_defined_chunking = CdcOptions {
-            enabled: true,
-            min_chunk_size: 0,
-            ..CdcOptions::default()
-        };
-        opts.arrow_schema(&Arc::new(Schema::empty()));
-        assert!(WriterPropertiesBuilder::try_from(&opts).is_err());
-    }
-
-    #[test]
-    fn test_cdc_validation_max_not_greater_than_min() {
-        let mut opts = TableParquetOptions::default();
-        opts.global.content_defined_chunking = CdcOptions {
-            enabled: true,
-            min_chunk_size: 512 * 1024,
-            max_chunk_size: 256 * 1024,
-            ..CdcOptions::default()
-        };
-        opts.arrow_schema(&Arc::new(Schema::empty()));
-        assert!(WriterPropertiesBuilder::try_from(&opts).is_err());
-    }
-
     #[test]
     fn test_bloom_filter_set_ndv_only() {
         // the TableParquetOptions::default, with only ndv set

datafusion/core/tests/parquet/content_defined_chunking.rs

@@ -25,7 +25,7 @@ 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::{CdcOptions, TableParquetOptions};
+use datafusion_common::config::{ParquetCdcOptions, TableParquetOptions};
 use parquet::arrow::ArrowWriter;
 use parquet::arrow::arrow_reader::ArrowReaderMetadata;
 use parquet::file::properties::WriterProperties;
@@ -97,7 +97,7 @@ async fn cdc_data_round_trip() {
     let batch = make_test_batch(5000);
 
     let mut opts = TableParquetOptions::default();
-    opts.global.content_defined_chunking = CdcOptions::enabled();
+    opts.global.content_defined_chunking = ParquetCdcOptions::enabled();
     let props = writer_props(&mut opts, &batch.schema());
 
     let tmp = write_parquet_file(&batch, props);
@@ -145,7 +145,7 @@ async fn cdc_affects_page_boundaries() {
 
     // Write WITH CDC using small chunk sizes to maximize effect
     let mut cdc_opts = TableParquetOptions::default();
-    cdc_opts.global.content_defined_chunking = CdcOptions {
+    cdc_opts.global.content_defined_chunking = ParquetCdcOptions {
         enabled: true,
         min_chunk_size: 512,
         max_chunk_size: 2048,

datafusion/proto-common/proto/datafusion_common.proto

@@ -627,7 +627,7 @@ message ParquetOptions {
     uint64 max_predicate_cache_size = 33;
   }
 
-  CdcOptions content_defined_chunking = 35;
+  ParquetCdcOptions content_defined_chunking = 35;
 
   // Optional timezone applied to INT96-coerced timestamps when `coerce_int96`
   // is set. When `Some`, INT96 columns coerce to
@@ -638,7 +638,8 @@ message ParquetOptions {
   }
 }
 
-message CdcOptions {
+// Content-defined chunking (CDC) options for writing parquet files.
+message ParquetCdcOptions {
   bool enabled = 1;
   uint64 min_chunk_size = 2;
   uint64 max_chunk_size = 3;

datafusion/proto-common/src/from_proto/mod.rs

@@ -39,7 +39,7 @@ use datafusion_common::{
     DataFusionError, JoinSide, ScalarValue, Statistics, TableReference,
     arrow_datafusion_err,
     config::{
-        CdcOptions, CsvOptions, JsonOptions, ParquetColumnOptions, ParquetOptions,
+        CsvOptions, JsonOptions, ParquetCdcOptions, ParquetColumnOptions, ParquetOptions,
         TableParquetOptions,
     },
     file_options::{csv_writer::CsvWriterOptions, json_writer::JsonWriterOptions},
@@ -1130,16 +1130,22 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions {
             max_predicate_cache_size: value.max_predicate_cache_size_opt.map(|opt| match opt {
                 protobuf::parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize(v) => Some(v as usize),
             }).unwrap_or(None),
-            content_defined_chunking: value.content_defined_chunking.map(|cdc| CdcOptions {
-                enabled: cdc.enabled,
-                min_chunk_size: cdc.min_chunk_size as usize,
-                max_chunk_size: cdc.max_chunk_size as usize,
-                norm_level: cdc.norm_level,
-            }).unwrap_or_default(),
+            content_defined_chunking: value.content_defined_chunking.map(ParquetCdcOptions::from).unwrap_or_default(),
         })
     }
 }
 
+impl From<protobuf::ParquetCdcOptions> for ParquetCdcOptions {
+    fn from(value: protobuf::ParquetCdcOptions) -> Self {
+        ParquetCdcOptions {
+            enabled: value.enabled,
+            min_chunk_size: value.min_chunk_size as usize,
+            max_chunk_size: value.max_chunk_size as usize,
+            norm_level: value.norm_level,
+        }
+    }
+}
+
 impl TryFrom<&protobuf::ParquetColumnOptions> for ParquetColumnOptions {
     type Error = DataFusionError;
     fn try_from(
@@ -1324,7 +1330,9 @@ pub(crate) fn csv_writer_options_from_proto(
 
 #[cfg(test)]
 mod tests {
-    use datafusion_common::config::{CdcOptions, ParquetOptions, TableParquetOptions};
+    use datafusion_common::config::{
+        ParquetCdcOptions, ParquetOptions, TableParquetOptions,
+    };
 
     fn parquet_options_proto_round_trip(opts: ParquetOptions) -> ParquetOptions {
         let proto: crate::protobuf_common::ParquetOptions =
@@ -1384,7 +1392,7 @@ mod tests {
     #[test]
     fn test_parquet_options_cdc_enabled_round_trip() {
         let opts = ParquetOptions {
-            content_defined_chunking: CdcOptions {
+            content_defined_chunking: ParquetCdcOptions {
                 enabled: true,
                 min_chunk_size: 128 * 1024,
                 max_chunk_size: 512 * 1024,
@@ -1403,10 +1411,10 @@ mod tests {
     #[test]
     fn test_parquet_options_cdc_negative_norm_level_round_trip() {
         let opts = ParquetOptions {
-            content_defined_chunking: CdcOptions {
+            content_defined_chunking: ParquetCdcOptions {
                 enabled: true,
                 norm_level: -3,
-                ..CdcOptions::default()
+                ..ParquetCdcOptions::default()
             },
             ..ParquetOptions::default()
         };
@@ -1417,7 +1425,7 @@ mod tests {
     #[test]
     fn test_table_parquet_options_cdc_round_trip() {
         let mut opts = TableParquetOptions::default();
-        opts.global.content_defined_chunking = CdcOptions {
+        opts.global.content_defined_chunking = ParquetCdcOptions {
             enabled: true,
             min_chunk_size: 64 * 1024,
             max_chunk_size: 2 * 1024 * 1024,