chore: update parquet CDC options and proto serialization for arrow-rs 58.1

Author: kszucs

Cargo.lock

@@ -687,6 +687,24 @@ config_namespace! {
     }
 }
 
+config_namespace! {
+    /// Options for content-defined chunking (CDC) when writing parquet files.
+    /// See [`ParquetOptions::use_content_defined_chunking`].
+    pub struct CdcOptions {
+        /// 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
+
+        /// Maximum chunk size in bytes. A split is forced when the accumulated
+        /// size exceeds this value. Default is 1 MiB.
+        pub max_chunk_size: usize, default = 1024 * 1024
+
+        /// Normalization level. Increasing this improves deduplication ratio
+        /// but increases fragmentation. Recommended range is [-3, 3], default is 0.
+        pub norm_level: i64, default = 0
+    }
+}
+
 config_namespace! {
     /// Options for reading and writing parquet files
     ///
@@ -875,25 +893,10 @@ config_namespace! {
         pub maximum_buffered_record_batches_per_stream: usize, default = 2
 
         /// (writing) EXPERIMENTAL: Enable content-defined chunking (CDC) when writing
-        /// parquet files. When true, the other `cdc_*` options control the chunking
-        /// behavior. When CDC is enabled, parallel writing is automatically disabled
-        /// since the chunker state must persist across row groups.
-        pub enable_content_defined_chunking: bool, default = false
-
-        /// (writing) Minimum chunk size in bytes for content-defined chunking.
-        /// The rolling hash will not be updated until this size is reached for each chunk.
-        /// Default is 256 KiB. Only used when `enable_content_defined_chunking` is true.
-        pub cdc_min_chunk_size: usize, default = 256 * 1024
-
-        /// (writing) Maximum chunk size in bytes for content-defined chunking.
-        /// The chunker will create a new chunk whenever the chunk size exceeds this value.
-        /// Default is 1 MiB. Only used when `enable_content_defined_chunking` is true.
-        pub cdc_max_chunk_size: usize, default = 1024 * 1024
-
-        /// (writing) Normalization level for content-defined chunking.
-        /// Increasing this improves deduplication ratio but increases fragmentation.
-        /// Recommended range is [-3, 3], default is 0. Only used when `enable_content_defined_chunking` is true.
-        pub cdc_norm_level: i64, default = 0
+        /// parquet files. When `Some`, CDC is enabled with the given options; when `None`
+        /// (the default), CDC is disabled. When CDC is enabled, parallel writing is
+        /// automatically disabled since the chunker state must persist across row groups.
+        pub use_content_defined_chunking: Option<CdcOptions>, default = None
     }
 }
 

datafusion/common/src/config.rs

@@ -191,10 +191,7 @@ impl ParquetOptions {
             bloom_filter_on_write,
             bloom_filter_fpp,
             bloom_filter_ndv,
-            enable_content_defined_chunking,
-            cdc_min_chunk_size,
-            cdc_max_chunk_size,
-            cdc_norm_level,
+            use_content_defined_chunking,
 
             // not in WriterProperties
             enable_page_index: _,
@@ -251,12 +248,12 @@ impl ParquetOptions {
         if let Some(encoding) = encoding {
             builder = builder.set_encoding(parse_encoding_string(encoding)?);
         }
-        if *enable_content_defined_chunking {
+        if let Some(cdc) = use_content_defined_chunking {
             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 as i32,
+                    min_chunk_size: cdc.min_chunk_size,
+                    max_chunk_size: cdc.max_chunk_size,
+                    norm_level: cdc.norm_level as i32,
                 },
             ));
         }
@@ -401,7 +398,9 @@ mod tests {
     use super::*;
     #[cfg(feature = "parquet_encryption")]
     use crate::config::ConfigFileEncryptionProperties;
-    use crate::config::{ParquetColumnOptions, ParquetEncryptionOptions, ParquetOptions};
+    use crate::config::{
+        CdcOptions, ParquetColumnOptions, ParquetEncryptionOptions, ParquetOptions,
+    };
     use crate::parquet_config::DFParquetWriterVersion;
     use parquet::basic::Compression;
     use parquet::file::properties::{
@@ -473,10 +472,7 @@ mod tests {
             skip_arrow_metadata: defaults.skip_arrow_metadata,
             coerce_int96: None,
             max_predicate_cache_size: defaults.max_predicate_cache_size,
-            enable_content_defined_chunking: defaults.enable_content_defined_chunking,
-            cdc_min_chunk_size: defaults.cdc_min_chunk_size,
-            cdc_max_chunk_size: defaults.cdc_max_chunk_size,
-            cdc_norm_level: defaults.cdc_norm_level,
+            use_content_defined_chunking: defaults.use_content_defined_chunking.clone(),
         }
     }
 
@@ -593,21 +589,13 @@ mod tests {
                 binary_as_string: global_options_defaults.binary_as_string,
                 skip_arrow_metadata: global_options_defaults.skip_arrow_metadata,
                 coerce_int96: None,
-                enable_content_defined_chunking: props
-                    .content_defined_chunking()
-                    .is_some(),
-                cdc_min_chunk_size: props
-                    .content_defined_chunking()
-                    .map(|c| c.min_chunk_size)
-                    .unwrap_or(global_options_defaults.cdc_min_chunk_size),
-                cdc_max_chunk_size: props
-                    .content_defined_chunking()
-                    .map(|c| c.max_chunk_size)
-                    .unwrap_or(global_options_defaults.cdc_max_chunk_size),
-                cdc_norm_level: props
-                    .content_defined_chunking()
-                    .map(|c| c.norm_level as i64)
-                    .unwrap_or(global_options_defaults.cdc_norm_level),
+                use_content_defined_chunking: props.content_defined_chunking().map(|c| {
+                    CdcOptions {
+                        min_chunk_size: c.min_chunk_size,
+                        max_chunk_size: c.max_chunk_size,
+                        norm_level: c.norm_level as i64,
+                    }
+                }),
             },
             column_specific_options,
             key_value_metadata,
@@ -821,10 +809,11 @@ mod tests {
     #[test]
     fn test_cdc_enabled_with_custom_options() {
         let mut opts = TableParquetOptions::default();
-        opts.global.enable_content_defined_chunking = true;
-        opts.global.cdc_min_chunk_size = 128 * 1024;
-        opts.global.cdc_max_chunk_size = 512 * 1024;
-        opts.global.cdc_norm_level = 2;
+        opts.global.use_content_defined_chunking = Some(CdcOptions {
+            min_chunk_size: 128 * 1024,
+            max_chunk_size: 512 * 1024,
+            norm_level: 2,
+        });
         opts.arrow_schema(&Arc::new(Schema::empty()));
 
         let props = WriterPropertiesBuilder::try_from(&opts).unwrap().build();
@@ -846,19 +835,20 @@ mod tests {
     #[test]
     fn test_cdc_round_trip_through_writer_props() {
         let mut opts = TableParquetOptions::default();
-        opts.global.enable_content_defined_chunking = true;
-        opts.global.cdc_min_chunk_size = 64 * 1024;
-        opts.global.cdc_max_chunk_size = 2 * 1024 * 1024;
-        opts.global.cdc_norm_level = -1;
+        opts.global.use_content_defined_chunking = Some(CdcOptions {
+            min_chunk_size: 64 * 1024,
+            max_chunk_size: 2 * 1024 * 1024,
+            norm_level: -1,
+        });
         opts.arrow_schema(&Arc::new(Schema::empty()));
 
         let props = WriterPropertiesBuilder::try_from(&opts).unwrap().build();
         let recovered = session_config_from_writer_props(&props);
 
-        assert_eq!(recovered.global.enable_content_defined_chunking, true);
-        assert_eq!(recovered.global.cdc_min_chunk_size, 64 * 1024);
-        assert_eq!(recovered.global.cdc_max_chunk_size, 2 * 1024 * 1024);
-        assert_eq!(recovered.global.cdc_norm_level, -1);
+        let cdc = recovered.global.use_content_defined_chunking.unwrap();
+        assert_eq!(cdc.min_chunk_size, 64 * 1024);
+        assert_eq!(cdc.max_chunk_size, 2 * 1024 * 1024);
+        assert_eq!(cdc.norm_level, -1);
     }
 
     #[test]

datafusion/common/src/file_options/parquet_writer.rs

@@ -1373,7 +1373,7 @@ impl FileSink for ParquetSink {
             // CDC requires the sequential writer: the chunker state lives in ArrowWriter
             // and persists across row groups. The parallel path bypasses ArrowWriter entirely.
             if !parquet_opts.global.allow_single_file_parallelism
-                || parquet_opts.global.enable_content_defined_chunking
+                || parquet_opts.global.use_content_defined_chunking.is_some()
             {
                 let mut writer = self
                     .create_async_arrow_writer(

datafusion/datasource-parquet/src/file_format.rs

@@ -604,19 +604,13 @@ message ParquetOptions {
     uint64 max_predicate_cache_size = 33;
   }
 
-  bool content_defined_chunking = 35; // default = false
-
-  oneof cdc_min_chunk_size_opt {
-    uint64 cdc_min_chunk_size = 36;
-  }
-
-  oneof cdc_max_chunk_size_opt {
-    uint64 cdc_max_chunk_size = 37;
-  }
+  CdcOptions content_defined_chunking = 35;
+}
 
-  oneof cdc_norm_level_opt {
-    int32 cdc_norm_level = 38;
-  }
+message CdcOptions {
+  uint64 min_chunk_size = 1;
+  uint64 max_chunk_size = 2;
+  int32 norm_level = 3;
 }
 
 enum JoinSide {

datafusion/proto-common/proto/datafusion_common.proto

@@ -40,7 +40,7 @@ use datafusion_common::{
     DataFusionError, JoinSide, ScalarValue, Statistics, TableReference,
     arrow_datafusion_err,
     config::{
-        CsvOptions, JsonOptions, ParquetColumnOptions, ParquetOptions,
+        CdcOptions, CsvOptions, JsonOptions, ParquetColumnOptions, ParquetOptions,
         TableParquetOptions,
     },
     file_options::{csv_writer::CsvWriterOptions, json_writer::JsonWriterOptions},
@@ -1090,16 +1090,14 @@ 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),
-            enable_content_defined_chunking: value.content_defined_chunking,
-            cdc_min_chunk_size: value.cdc_min_chunk_size_opt.map(|opt| match opt {
-                protobuf::parquet_options::CdcMinChunkSizeOpt::CdcMinChunkSize(v) => v as usize,
-            }).unwrap_or(ParquetOptions::default().cdc_min_chunk_size),
-            cdc_max_chunk_size: value.cdc_max_chunk_size_opt.map(|opt| match opt {
-                protobuf::parquet_options::CdcMaxChunkSizeOpt::CdcMaxChunkSize(v) => v as usize,
-            }).unwrap_or(ParquetOptions::default().cdc_max_chunk_size),
-            cdc_norm_level: value.cdc_norm_level_opt.map(|opt| match opt {
-                protobuf::parquet_options::CdcNormLevelOpt::CdcNormLevel(v) => v as i64,
-            }).unwrap_or(ParquetOptions::default().cdc_norm_level),
+            use_content_defined_chunking: value.content_defined_chunking.map(|cdc| {
+                let defaults = CdcOptions::default();
+                CdcOptions {
+                    min_chunk_size: if cdc.min_chunk_size != 0 { cdc.min_chunk_size as usize } else { defaults.min_chunk_size },
+                    max_chunk_size: if cdc.max_chunk_size != 0 { cdc.max_chunk_size as usize } else { defaults.max_chunk_size },
+                    norm_level: cdc.norm_level as i64,
+                }
+            }),
         })
     }
 }
@@ -1276,7 +1274,7 @@ pub(crate) fn csv_writer_options_from_proto(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use datafusion_common::config::{ParquetOptions, TableParquetOptions};
+    use datafusion_common::config::{CdcOptions, ParquetOptions, TableParquetOptions};
 
     fn parquet_options_proto_round_trip(opts: ParquetOptions) -> ParquetOptions {
         let proto: crate::protobuf_common::ParquetOptions =
@@ -1295,66 +1293,65 @@ mod tests {
     #[test]
     fn test_parquet_options_cdc_disabled_round_trip() {
         let opts = ParquetOptions::default();
-        assert!(!opts.enable_content_defined_chunking);
+        assert!(opts.use_content_defined_chunking.is_none());
         let recovered = parquet_options_proto_round_trip(opts.clone());
         assert_eq!(opts, recovered);
     }
 
     #[test]
     fn test_parquet_options_cdc_enabled_round_trip() {
         let opts = ParquetOptions {
-            enable_content_defined_chunking: true,
-            cdc_min_chunk_size: 128 * 1024,
-            cdc_max_chunk_size: 512 * 1024,
-            cdc_norm_level: 2,
+            use_content_defined_chunking: Some(CdcOptions {
+                min_chunk_size: 128 * 1024,
+                max_chunk_size: 512 * 1024,
+                norm_level: 2,
+            }),
             ..ParquetOptions::default()
         };
         let recovered = parquet_options_proto_round_trip(opts.clone());
-        assert_eq!(recovered.enable_content_defined_chunking, true);
-        assert_eq!(recovered.cdc_min_chunk_size, 128 * 1024);
-        assert_eq!(recovered.cdc_max_chunk_size, 512 * 1024);
-        assert_eq!(recovered.cdc_norm_level, 2);
+        let cdc = recovered.use_content_defined_chunking.unwrap();
+        assert_eq!(cdc.min_chunk_size, 128 * 1024);
+        assert_eq!(cdc.max_chunk_size, 512 * 1024);
+        assert_eq!(cdc.norm_level, 2);
     }
 
     #[test]
     fn test_parquet_options_cdc_negative_norm_level_round_trip() {
         let opts = ParquetOptions {
-            enable_content_defined_chunking: true,
-            cdc_norm_level: -3,
+            use_content_defined_chunking: Some(CdcOptions {
+                norm_level: -3,
+                ..CdcOptions::default()
+            }),
             ..ParquetOptions::default()
         };
         let recovered = parquet_options_proto_round_trip(opts);
-        assert_eq!(recovered.cdc_norm_level, -3);
+        assert_eq!(
+            recovered.use_content_defined_chunking.unwrap().norm_level,
+            -3
+        );
     }
 
     #[test]
     fn test_table_parquet_options_cdc_round_trip() {
         let mut opts = TableParquetOptions::default();
-        opts.global.enable_content_defined_chunking = true;
-        opts.global.cdc_min_chunk_size = 64 * 1024;
-        opts.global.cdc_max_chunk_size = 2 * 1024 * 1024;
-        opts.global.cdc_norm_level = -1;
+        opts.global.use_content_defined_chunking = Some(CdcOptions {
+            min_chunk_size: 64 * 1024,
+            max_chunk_size: 2 * 1024 * 1024,
+            norm_level: -1,
+        });
 
         let recovered = table_parquet_options_proto_round_trip(opts.clone());
-        assert_eq!(recovered.global.enable_content_defined_chunking, true);
-        assert_eq!(recovered.global.cdc_min_chunk_size, 64 * 1024);
-        assert_eq!(recovered.global.cdc_max_chunk_size, 2 * 1024 * 1024);
-        assert_eq!(recovered.global.cdc_norm_level, -1);
+        let cdc = recovered.global.use_content_defined_chunking.unwrap();
+        assert_eq!(cdc.min_chunk_size, 64 * 1024);
+        assert_eq!(cdc.max_chunk_size, 2 * 1024 * 1024);
+        assert_eq!(cdc.norm_level, -1);
     }
 
     #[test]
     fn test_table_parquet_options_cdc_disabled_round_trip() {
         let opts = TableParquetOptions::default();
-        assert!(!opts.global.enable_content_defined_chunking);
+        assert!(opts.global.use_content_defined_chunking.is_none());
         let recovered = table_parquet_options_proto_round_trip(opts.clone());
-        assert_eq!(recovered.global.enable_content_defined_chunking, false);
-        assert_eq!(
-            recovered.global.cdc_min_chunk_size,
-            ParquetOptions::default().cdc_min_chunk_size
-        );
-        assert_eq!(
-            recovered.global.cdc_max_chunk_size,
-            ParquetOptions::default().cdc_max_chunk_size
-        );
+        assert!(recovered.global.use_content_defined_chunking.is_none());
     }
 }