feat: add support for parquet content defined chunking options

Author: kszucs

datafusion/common/src/config.rs

@@ -845,6 +845,7 @@ config_namespace! {
         /// default parquet writer setting
         pub bloom_filter_ndv: Option<u64>, default = None
 
+
         /// (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
@@ -872,6 +873,27 @@ config_namespace! {
         /// writing out already in-memory data, such as from a cached
         /// data frame.
         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
     }
 }
 
@@ -1826,6 +1848,7 @@ config_field!(usize);
 config_field!(f64);
 config_field!(u64);
 config_field!(u32);
+config_field!(i64);
 
 impl ConfigField for u8 {
     fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {

datafusion/common/src/file_options/parquet_writer.rs

@@ -95,7 +95,7 @@ impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
             global,
             column_specific_options,
             key_value_metadata,
-            crypto: _,
+            ..
         } = table_parquet_options;
 
         let mut builder = global.into_writer_properties_builder()?;
@@ -191,6 +191,10 @@ 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,
 
             // not in WriterProperties
             enable_page_index: _,
@@ -247,6 +251,15 @@ impl ParquetOptions {
         if let Some(encoding) = encoding {
             builder = builder.set_encoding(parse_encoding_string(encoding)?);
         }
+        if *enable_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,
+                },
+            ));
+        }
 
         Ok(builder)
     }
@@ -460,6 +473,10 @@ 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,
         }
     }
 
@@ -576,6 +593,21 @@ 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),
             },
             column_specific_options,
             key_value_metadata,
@@ -786,6 +818,49 @@ 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.arrow_schema(&Arc::new(Schema::empty()));
+
+        let props = WriterPropertiesBuilder::try_from(&opts).unwrap().build();
+        let cdc = props.content_defined_chunking().expect("CDC should be set");
+        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_cdc_disabled_by_default() {
+        let mut opts = TableParquetOptions::default();
+        opts.arrow_schema(&Arc::new(Schema::empty()));
+
+        let props = WriterPropertiesBuilder::try_from(&opts).unwrap().build();
+        assert!(props.content_defined_chunking().is_none());
+    }
+
+    #[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.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);
+    }
+
     #[test]
     fn test_bloom_filter_set_ndv_only() {
         // the TableParquetOptions::default, with only ndv set

datafusion/datasource-parquet/src/file_format.rs

@@ -1370,7 +1370,11 @@ impl FileSink for ParquetSink {
 
         while let Some((path, mut rx)) = file_stream_rx.recv().await {
             let parquet_props = self.create_writer_props(&runtime, &path).await?;
-            if !parquet_opts.global.allow_single_file_parallelism {
+            // 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
+            {
                 let mut writer = self
                     .create_async_arrow_writer(
                         &path,

datafusion/proto-common/Cargo.toml

@@ -37,10 +37,12 @@ name = "datafusion_proto_common"
 [features]
 default = []
 json = ["serde", "pbjson"]
+parquet = ["datafusion-common/parquet", "dep:parquet"]
 
 [dependencies]
 arrow = { workspace = true }
 datafusion-common = { workspace = true }
+parquet = { workspace = true, optional = true }
 pbjson = { workspace = true, optional = true }
 prost = { workspace = true }
 serde = { version = "1.0", optional = true }

datafusion/proto-common/proto/datafusion_common.proto

@@ -603,6 +603,20 @@ message ParquetOptions {
   oneof max_predicate_cache_size_opt {
     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;
+  }
+
+  oneof cdc_norm_level_opt {
+    int32 cdc_norm_level = 38;
+  }
 }
 
 enum JoinSide {

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

@@ -1089,6 +1089,16 @@ 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),
         })
     }
 }
@@ -1151,17 +1161,17 @@ impl TryFrom<&protobuf::TableParquetOptions> for TableParquetOptions {
                 column_specific_options.insert(column_name.clone(), options.try_into()?);
             }
         }
-        Ok(TableParquetOptions {
+        let opts = TableParquetOptions {
             global: value
                 .global
                 .as_ref()
                 .map(|v| v.try_into())
                 .unwrap()
                 .unwrap(),
             column_specific_options,
-            key_value_metadata: Default::default(),
-            crypto: Default::default(),
-        })
+            ..Default::default()
+        };
+        Ok(opts)
     }
 }
 
@@ -1261,3 +1271,89 @@ pub(crate) fn csv_writer_options_from_proto(
         .with_null(writer_options.null_value.clone())
         .with_double_quote(writer_options.double_quote))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use datafusion_common::config::{ParquetOptions, TableParquetOptions};
+
+    fn parquet_options_proto_round_trip(opts: ParquetOptions) -> ParquetOptions {
+        let proto: crate::protobuf_common::ParquetOptions =
+            (&opts).try_into().expect("to_proto");
+        ParquetOptions::try_from(&proto).expect("from_proto")
+    }
+
+    fn table_parquet_options_proto_round_trip(
+        opts: TableParquetOptions,
+    ) -> TableParquetOptions {
+        let proto: crate::protobuf_common::TableParquetOptions =
+            (&opts).try_into().expect("to_proto");
+        TableParquetOptions::try_from(&proto).expect("from_proto")
+    }
+
+    #[test]
+    fn test_parquet_options_cdc_disabled_round_trip() {
+        let opts = ParquetOptions::default();
+        assert!(!opts.enable_content_defined_chunking);
+        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,
+            ..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);
+    }
+
+    #[test]
+    fn test_parquet_options_cdc_negative_norm_level_round_trip() {
+        let opts = ParquetOptions {
+            enable_content_defined_chunking: true,
+            cdc_norm_level: -3,
+            ..ParquetOptions::default()
+        };
+        let recovered = parquet_options_proto_round_trip(opts);
+        assert_eq!(recovered.cdc_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;
+
+        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);
+    }
+
+    #[test]
+    fn test_table_parquet_options_cdc_disabled_round_trip() {
+        let opts = TableParquetOptions::default();
+        assert!(!opts.global.enable_content_defined_chunking);
+        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
+        );
+    }
+}