refactor: convert CdcOptions <-> parquet-rs via From impls

Replace the inline field-by-field mapping in the parquet writer path with two
`From` impls that encapsulate the enabled-flag <-> Option-presence translation:

* `From<&CdcOptions> for Option<parquet::file::properties::CdcOptions>` — a
  disabled `CdcOptions` maps to `None`, an enabled one to `Some(..)`.
* `From<Option<&parquet::file::properties::CdcOptions>> for CdcOptions` — `Some`
  implies `enabled: true`; `None` yields the disabled default.

The writer set/read sites collapse to `content_defined_chunking.into()` and
`props.content_defined_chunking().into()` respectively.

Author: kszucs

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::{CdcOptions, ParquetOptions, TableParquetOptions},
 };
 
 use arrow::datatypes::Schema;
@@ -166,6 +166,41 @@ impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
     }
 }
 
+/// Convert DataFusion's [`CdcOptions`] into parquet-rs's `Option<CdcOptions>`.
+///
+/// parquet-rs has no `enabled` flag; CDC is on when the option is `Some`. So a
+/// disabled [`CdcOptions`] maps to `None`, and an enabled one to `Some` with the
+/// chunking parameters.
+impl From<&CdcOptions> for Option<parquet::file::properties::CdcOptions> {
+    fn from(value: &CdcOptions) -> 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 [`CdcOptions`].
+///
+/// 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 CdcOptions {
+    fn from(value: Option<&parquet::file::properties::CdcOptions>) -> Self {
+        match value {
+            Some(cdc) => CdcOptions {
+                enabled: true,
+                min_chunk_size: cdc.min_chunk_size,
+                max_chunk_size: cdc.max_chunk_size,
+                norm_level: cdc.norm_level,
+            },
+            None => CdcOptions::default(),
+        }
+    }
+}
+
 impl ParquetOptions {
     /// Convert the global session options, [`ParquetOptions`], into a single write action's [`WriterPropertiesBuilder`].
     ///
@@ -249,15 +284,7 @@ impl ParquetOptions {
         if let Some(encoding) = encoding {
             builder = builder.set_encoding(parse_encoding_string(encoding)?);
         }
-        if content_defined_chunking.enabled {
-            builder = builder.set_content_defined_chunking(Some(
-                parquet::file::properties::CdcOptions {
-                    min_chunk_size: content_defined_chunking.min_chunk_size,
-                    max_chunk_size: content_defined_chunking.max_chunk_size,
-                    norm_level: content_defined_chunking.norm_level,
-                },
-            ));
-        }
+        builder = builder.set_content_defined_chunking(content_defined_chunking.into());
 
         Ok(builder)
     }
@@ -592,15 +619,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,