feat(writer): honor write.parquet.* via ParquetWriterBuilder::from_table_properties (#2561)

## Which issue does this PR close?

N/A — no tracking issue.

## What changes are included in this PR?

Only the DataFusion `INSERT INTO` path honored `write.parquet.*` table
properties, and only the content-defined-chunking (CDC) keys, via a
hand-rolled translation inlined in `IcebergWriteExec`. Anything writing
through the writer stack directly (`DataFileWriter` →
`ParquetWriterBuilder`) silently fell back to parquet-rs defaults.

This PR moves that translation behind a reusable constructor:

- Add `ParquetWriterBuilder::from_table_properties(&TableProperties,
schema)`, the single place that maps `write.parquet.*` into
`WriterProperties`. It currently translates the CDC keys
(`write.parquet.content-defined-chunking.*`); other keys still fall back
to parquet-rs defaults and can be added here later.
- Add a chainable `with_match_mode` setter so the field match mode can
be overridden — DataFusion needs name-based matching since its Arrow
batches carry no field-id metadata.
- Refactor the DataFusion `insert_into` writer to build via
`from_table_properties`, reusing the `TableProperties` it already parses
and dropping the inline CDC translation.

Additive only: `new` and `new_with_match_mode` are unchanged; no
breaking changes.

## Are these changes tested?

- Unit tests in `parquet_writer.rs`: CDC is off by default, and CDC
properties propagate through `build()` to the writer's
`WriterProperties` — asserted on the getter rather than by re-reading a
written file, so future `write.parquet.*` options only need an assertion
on their corresponding getter.
- Existing `test_insert_into*` DataFusion integration tests cover the
refactored path, which is behaviorally unchanged.

Author: kszucs

crates/iceberg/public-api.txt

@@ -3205,8 +3205,10 @@ pub async fn iceberg::writer::file_writer::ParquetWriter::close(self) -> iceberg
 pub async fn iceberg::writer::file_writer::ParquetWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<()>
 pub struct iceberg::writer::file_writer::ParquetWriterBuilder
 impl iceberg::writer::file_writer::ParquetWriterBuilder
+pub fn iceberg::writer::file_writer::ParquetWriterBuilder::from_table_properties(table_props: &iceberg::spec::TableProperties, schema: iceberg::spec::SchemaRef) -> Self
 pub fn iceberg::writer::file_writer::ParquetWriterBuilder::new(props: parquet::file::properties::WriterProperties, schema: iceberg::spec::SchemaRef) -> Self
 pub fn iceberg::writer::file_writer::ParquetWriterBuilder::new_with_match_mode(props: parquet::file::properties::WriterProperties, schema: iceberg::spec::SchemaRef, match_mode: iceberg::arrow::FieldMatchMode) -> Self
+pub fn iceberg::writer::file_writer::ParquetWriterBuilder::with_match_mode(self, match_mode: iceberg::arrow::FieldMatchMode) -> Self
 impl core::clone::Clone for iceberg::writer::file_writer::ParquetWriterBuilder
 pub fn iceberg::writer::file_writer::ParquetWriterBuilder::clone(&self) -> iceberg::writer::file_writer::ParquetWriterBuilder
 impl core::fmt::Debug for iceberg::writer::file_writer::ParquetWriterBuilder

crates/iceberg/src/writer/file_writer/parquet_writer.rs

@@ -28,7 +28,7 @@ use parquet::arrow::AsyncArrowWriter;
 use parquet::arrow::async_reader::AsyncFileReader;
 use parquet::arrow::async_writer::AsyncFileWriter as ArrowAsyncFileWriter;
 use parquet::file::metadata::ParquetMetaData;
-use parquet::file::properties::WriterProperties;
+use parquet::file::properties::{CdcOptions, WriterProperties};
 use parquet::file::statistics::Statistics;
 
 use super::{FileWriter, FileWriterBuilder};
@@ -40,7 +40,7 @@ use crate::io::{FileIO, FileWrite, OutputFile};
 use crate::spec::{
     DataContentType, DataFileBuilder, DataFileFormat, Datum, ListType, Literal, MapType,
     NestedFieldRef, PartitionSpec, PrimitiveType, Schema, SchemaRef, SchemaVisitor, Struct,
-    StructType, TableMetadata, Type, visit_schema,
+    StructType, TableMetadata, TableProperties, Type, visit_schema,
 };
 use crate::transform::create_transform_function;
 use crate::writer::{CurrentFileStatus, DataFile};
@@ -57,6 +57,10 @@ pub struct ParquetWriterBuilder {
 impl ParquetWriterBuilder {
     /// Create a new `ParquetWriterBuilder`
     /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+    ///
+    /// When writing into an existing Iceberg table, prefer
+    /// [`Self::from_table_properties`], which derives `WriterProperties` from
+    /// the table's `write.parquet.*` properties.
     pub fn new(props: WriterProperties, schema: SchemaRef) -> Self {
         Self::new_with_match_mode(props, schema, FieldMatchMode::Id)
     }
@@ -73,6 +77,37 @@ impl ParquetWriterBuilder {
             match_mode,
         }
     }
+
+    /// Build a `ParquetWriterBuilder` from Iceberg table properties and a
+    /// schema, translating `write.parquet.*` settings into `WriterProperties`
+    /// instead of using parquet-rs defaults.
+    ///
+    /// Currently translates the content-defined-chunking keys
+    /// (`write.parquet.content-defined-chunking.*`); other keys fall back to
+    /// parquet-rs defaults.
+    pub fn from_table_properties(table_props: &TableProperties, schema: SchemaRef) -> Self {
+        let cdc = table_props.cdc_enabled.then_some(CdcOptions {
+            min_chunk_size: table_props.cdc_min_chunk_size,
+            max_chunk_size: table_props.cdc_max_chunk_size,
+            norm_level: table_props.cdc_norm_level,
+        });
+        // TODO: translate the remaining write.parquet.* keys (e.g. compression-codec,
+        // row-group-size-bytes, page-size-bytes).
+        // This constructor is intended to be the single place that maps them.
+        let props = WriterProperties::builder()
+            .set_content_defined_chunking(cdc)
+            .build();
+        Self::new_with_match_mode(props, schema, FieldMatchMode::Id)
+    }
+
+    /// Set the field match mode used to map Arrow fields to Iceberg fields.
+    ///
+    /// Defaults to [`FieldMatchMode::Id`]. Use [`FieldMatchMode::Name`] when the
+    /// incoming Arrow schema does not carry Iceberg field-id metadata.
+    pub fn with_match_mode(mut self, match_mode: FieldMatchMode) -> Self {
+        self.match_mode = match_mode;
+        self
+    }
 }
 
 impl FileWriterBuilder for ParquetWriterBuilder {
@@ -2279,4 +2314,81 @@ mod tests {
         assert_eq!(lower_bounds, HashMap::from([(0, Datum::int(i32::MIN))]));
         assert_eq!(upper_bounds, HashMap::from([(0, Datum::int(i32::MAX))]));
     }
+
+    // -----------------------------------------------------------------
+    // ParquetWriterBuilder::from_table_properties
+    // -----------------------------------------------------------------
+
+    fn cdc_test_schema() -> SchemaRef {
+        Arc::new(
+            Schema::builder()
+                .with_schema_id(1)
+                .with_fields(vec![
+                    NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
+                    NestedField::required(2, "payload", Type::Primitive(PrimitiveType::String))
+                        .into(),
+                ])
+                .build()
+                .unwrap(),
+        )
+    }
+
+    fn table_props(entries: HashMap<String, String>) -> TableProperties {
+        TableProperties::try_from(&entries).unwrap()
+    }
+
+    #[test]
+    fn test_from_table_properties_no_cdc_by_default() {
+        let tp = table_props(HashMap::new());
+        let builder = ParquetWriterBuilder::from_table_properties(&tp, cdc_test_schema());
+        assert!(builder.props.content_defined_chunking().is_none());
+    }
+
+    #[tokio::test]
+    async fn test_from_table_properties_propagate_to_writer() {
+        // `build()` must carry the translated `WriterProperties` through to the
+        // `ParquetWriter` unchanged — otherwise the `write.parquet.*` settings
+        // derived in `from_table_properties` would never reach parquet-rs.
+        //
+        // Asserting on the writer's `WriterProperties` (rather than re-reading a
+        // written file) keeps this a direct propagation check: every future
+        // `write.parquet.*` option just adds an assertion on its corresponding
+        // `WriterProperties` getter here.
+        let tp = table_props(HashMap::from([
+            (
+                TableProperties::PROPERTY_PARQUET_CDC_ENABLED.to_string(),
+                "true".to_string(),
+            ),
+            (
+                TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE.to_string(),
+                "4096".to_string(),
+            ),
+            (
+                TableProperties::PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE.to_string(),
+                "8192".to_string(),
+            ),
+            (
+                TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL.to_string(),
+                "2".to_string(),
+            ),
+        ]));
+
+        let tmp = TempDir::new().unwrap();
+        let output = FileIO::new_with_fs()
+            .new_output(format!("{}/cdc.parquet", tmp.path().to_str().unwrap()))
+            .unwrap();
+        let writer = ParquetWriterBuilder::from_table_properties(&tp, cdc_test_schema())
+            .build(output)
+            .await
+            .unwrap();
+
+        let cdc = writer
+            .writer_properties
+            .content_defined_chunking()
+            .copied()
+            .expect("CDC should be enabled on the built writer");
+        assert_eq!(cdc.min_chunk_size, 4096);
+        assert_eq!(cdc.max_chunk_size, 8192);
+        assert_eq!(cdc.norm_level, 2);
+    }
 }

crates/integrations/datafusion/src/physical_plan/write.rs

@@ -45,7 +45,6 @@ use iceberg::writer::file_writer::location_generator::{
 };
 use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder;
 use iceberg::{Error, ErrorKind};
-use parquet::file::properties::{CdcOptions, WriterPropertiesBuilder};
 use uuid::Uuid;
 
 use crate::physical_plan::DATA_FILES_COL_NAME;
@@ -225,21 +224,14 @@ impl ExecutionPlan for IcebergWriteExec {
             )));
         }
 
-        // Create data file writer builder
-        let cdc_options = table_props.cdc_enabled.then_some(CdcOptions {
-            min_chunk_size: table_props.cdc_min_chunk_size,
-            max_chunk_size: table_props.cdc_max_chunk_size,
-            norm_level: table_props.cdc_norm_level,
-        });
-        let writer_properties = WriterPropertiesBuilder::default()
-            .set_content_defined_chunking(cdc_options)
-            .build();
-
-        let parquet_file_writer_builder = ParquetWriterBuilder::new_with_match_mode(
-            writer_properties,
+        // Build the writer from the already-parsed table properties so it honors
+        // `write.parquet.*` settings (e.g. CDC). Arrow batches flowing through
+        // DataFusion carry no field-id metadata, so match fields by name.
+        let parquet_file_writer_builder = ParquetWriterBuilder::from_table_properties(
+            &table_props,
             self.table.metadata().current_schema().clone(),
-            FieldMatchMode::Name,
-        );
+        )
+        .with_match_mode(FieldMatchMode::Name);
         let target_file_size = table_props.write_target_file_size_bytes;
 
         let file_io = self.table.file_io().clone();