fead: add Parquet content-defined chunking writer support

mkdocs/docs/configuration.md

@@ -85,6 +85,10 @@ Iceberg tables support table properties to configure table behavior.
 | `write.parquet.page-size-bytes`          | Size in bytes                      | 1MB                        | Set a target threshold for the approximate encoded size of data pages within a column chunk                                                          |
 | `write.parquet.page-row-limit`           | Number of rows                     | 20000                      | Set a target threshold for the maximum number of rows within a column chunk                                                                          |
 | `write.parquet.dict-size-bytes`          | Size in bytes                      | 2MB                        | Set the dictionary page size limit per row group                                                                                                     |
+| `write.parquet.content-defined-chunking.enabled` | Boolean                    | False                      | Enables content-defined chunking (CDC) for the Parquet writer, which produces stable page boundaries across appends. Requires `pyarrow>=21.0.0`, and raises at write time on older versions.     |
+| `write.parquet.content-defined-chunking.min-chunk-size` | Size in bytes       | 256KB                      | The minimum chunk size used for content-defined chunking                                                                                             |
+| `write.parquet.content-defined-chunking.max-chunk-size` | Size in bytes       | 1MB                        | The maximum chunk size used for content-defined chunking                                                                                             |
+| `write.parquet.content-defined-chunking.norm-level` | Integer                 | 0                          | The normalization level for content-defined chunking, controlling how tightly chunk sizes cluster around the average                                |
 | `write.metadata.previous-versions-max`   | Integer                            | 100                        | The max number of previous version metadata files to keep before deleting after commit.                                                              |
 | `write.metadata.delete-after-commit.enabled` | Boolean                        | False                      | Whether to automatically delete old *tracked* metadata files after each table commit. It will retain a number of the most recent metadata files, which can be set using property `write.metadata.previous-versions-max`. |
 | `write.object-storage.enabled`           | Boolean                            | False                      | Enables the [`ObjectStoreLocationProvider`](configuration.md#object-store-location-provider) that adds a hash component to file paths. |

pyiceberg/io/pyarrow.py

@@ -390,6 +390,13 @@ def to_input_file(self) -> PyArrowFile:
         return self
 
 
+def _require_pyarrow_version(min_version: str, feature: str) -> None:
+    from packaging import version
+
+    if version.parse(pyarrow.__version__) < version.parse(min_version):
+        raise ImportError(f"pyarrow version >= {min_version} required for {feature}, but found version {pyarrow.__version__}.")
+
+
 class PyArrowFileIO(FileIO):
     fs_by_scheme: Callable[[str, str | None], FileSystem]
 
@@ -525,14 +532,7 @@ def _initialize_s3_fs(self, netloc: str | None) -> FileSystem:
 
     def _initialize_azure_fs(self) -> FileSystem:
         # https://arrow.apache.org/docs/python/generated/pyarrow.fs.AzureFileSystem.html
-        from packaging import version
-
-        MIN_PYARROW_VERSION_SUPPORTING_AZURE_FS = "20.0.0"
-        if version.parse(pyarrow.__version__) < version.parse(MIN_PYARROW_VERSION_SUPPORTING_AZURE_FS):
-            raise ImportError(
-                f"pyarrow version >= {MIN_PYARROW_VERSION_SUPPORTING_AZURE_FS} required for AzureFileSystem support, "
-                f"but found version {pyarrow.__version__}."
-            )
+        _require_pyarrow_version("20.0.0", "AzureFileSystem support")
 
         from pyarrow.fs import AzureFileSystem
 
@@ -2845,7 +2845,7 @@ def _get_parquet_writer_kwargs(table_properties: Properties) -> dict[str, Any]:
     if compression_codec == ICEBERG_UNCOMPRESSED_CODEC:
         compression_codec = PYARROW_UNCOMPRESSED_CODEC
 
-    return {
+    parquet_writer_kwargs = {
         "compression": compression_codec,
         "compression_level": compression_level,
         "data_page_size": property_as_int(
@@ -2865,6 +2865,37 @@ def _get_parquet_writer_kwargs(table_properties: Properties) -> dict[str, Any]:
         ),
     }
 
+    # Unlike the properties above, which PyArrow's writer never supports and are safe to silently
+    # drop, CDC is a version-gated feature: silently ignoring it would produce a table that no longer
+    # has the content-defined chunk boundaries the user explicitly asked for, so this raises instead.
+    if property_as_bool(
+        properties=table_properties,
+        property_name=TableProperties.PARQUET_CDC_ENABLED,
+        default=TableProperties.PARQUET_CDC_ENABLED_DEFAULT,
+    ):
+        _require_pyarrow_version("21.0.0", "Parquet content-defined chunking")
+        # PyArrow itself validates these values (e.g. max-chunk-size > min-chunk-size) and raises a
+        # clear OSError, so there's no need to duplicate that validation here.
+        parquet_writer_kwargs["use_content_defined_chunking"] = {
+            "min_chunk_size": property_as_int(
+                properties=table_properties,
+                property_name=TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE,
+                default=TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT,
+            ),
+            "max_chunk_size": property_as_int(
+                properties=table_properties,
+                property_name=TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE,
+                default=TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT,
+            ),
+            "norm_level": property_as_int(
+                properties=table_properties,
+                property_name=TableProperties.PARQUET_CDC_NORM_LEVEL,
+                default=TableProperties.PARQUET_CDC_NORM_LEVEL_DEFAULT,
+            ),
+        }
+
+    return parquet_writer_kwargs
+
 
 def _dataframe_to_data_files(
     table_metadata: TableMetadata,

pyiceberg/table/__init__.py

@@ -154,6 +154,18 @@ class TableProperties:
 
     PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX = "write.parquet.bloom-filter-enabled.column"
 
+    PARQUET_CDC_ENABLED = "write.parquet.content-defined-chunking.enabled"
+    PARQUET_CDC_ENABLED_DEFAULT = False
+
+    PARQUET_CDC_MIN_CHUNK_SIZE = "write.parquet.content-defined-chunking.min-chunk-size"
+    PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT = 256 * 1024  # 256 KiB
+
+    PARQUET_CDC_MAX_CHUNK_SIZE = "write.parquet.content-defined-chunking.max-chunk-size"
+    PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT = 1024 * 1024  # 1 MiB
+
+    PARQUET_CDC_NORM_LEVEL = "write.parquet.content-defined-chunking.norm-level"
+    PARQUET_CDC_NORM_LEVEL_DEFAULT = 0
+
     WRITE_TARGET_FILE_SIZE_BYTES = "write.target-file-size-bytes"
     WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT = 512 * 1024 * 1024  # 512 MB
 

tests/io/test_pyarrow.py

@@ -72,6 +72,7 @@
     _check_pyarrow_schema_compatible,
     _ConvertToArrowSchema,
     _determine_partitions,
+    _get_parquet_writer_kwargs,
     _primitive_to_physical,
     _read_deletes,
     _task_to_record_batches,
@@ -5297,3 +5298,89 @@ def test_dictionary_columns_produces_dict_encoded_output(tmpdir: str) -> None:
 
     # Values must be identical
     assert result_plain.column("label").to_pylist() == result_dict.column("label").to_pylist()
+
+
+@pytest.mark.parametrize(
+    "table_properties,expected",
+    [
+        ({}, None),
+        (
+            {TableProperties.PARQUET_CDC_ENABLED: "true"},
+            {
+                "min_chunk_size": TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT,
+                "max_chunk_size": TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT,
+                "norm_level": TableProperties.PARQUET_CDC_NORM_LEVEL_DEFAULT,
+            },
+        ),
+        (
+            {
+                TableProperties.PARQUET_CDC_ENABLED: "true",
+                TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE: "4096",
+                TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE: "8192",
+                TableProperties.PARQUET_CDC_NORM_LEVEL: "2",
+            },
+            {"min_chunk_size": 4096, "max_chunk_size": 8192, "norm_level": 2},
+        ),
+    ],
+)
+def test_get_parquet_writer_kwargs_cdc(table_properties: dict[str, str], expected: dict[str, int] | None) -> None:
+    kwargs = _get_parquet_writer_kwargs(table_properties)
+    assert kwargs.get("use_content_defined_chunking") == expected
+
+
+def test_get_parquet_writer_kwargs_cdc_enabled_unsupported_pyarrow_version(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(pyarrow, "__version__", "17.0.0")
+    with pytest.raises(ImportError, match="pyarrow version >= 21.0.0"):
+        _get_parquet_writer_kwargs({TableProperties.PARQUET_CDC_ENABLED: "true"})
+
+
+def test_get_parquet_writer_kwargs_cdc_invalid_chunk_sizes_raises_from_pyarrow() -> None:
+    """PyArrow validates min/max chunk sizes itself; pyiceberg doesn't duplicate that check."""
+    kwargs = _get_parquet_writer_kwargs(
+        {
+            TableProperties.PARQUET_CDC_ENABLED: "true",
+            TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE: "8192",
+            TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE: "4096",
+        }
+    )
+    table = pa.table({"id": pa.array([1, 2, 3], type=pa.int32())})
+    with pytest.raises(OSError, match="max_chunk_size must be greater than min_chunk_size"):
+        with pq.ParquetWriter(pa.BufferOutputStream(), table.schema, **kwargs) as writer:
+            writer.write_table(table)
+
+
+def test_write_file_with_content_defined_chunking_enabled(tmp_path: Path) -> None:
+    """Writing a table with CDC enabled should forward use_content_defined_chunking to pq.ParquetWriter."""
+    from pyiceberg.table import WriteTask
+
+    table_schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+    arrow_data = pa.table({"id": pa.array(range(1000), type=pa.int32())})
+
+    table_metadata = TableMetadataV2(
+        location=f"file://{tmp_path}",
+        last_column_id=1,
+        format_version=2,
+        schemas=[table_schema],
+        partition_specs=[PartitionSpec()],
+        properties={TableProperties.PARQUET_CDC_ENABLED: "true"},
+    )
+
+    task = WriteTask(
+        write_uuid=uuid.uuid4(),
+        task_id=0,
+        record_batches=arrow_data.to_batches(),
+        schema=table_schema,
+    )
+
+    with patch("pyiceberg.io.pyarrow.pq.ParquetWriter", wraps=pq.ParquetWriter) as mock_writer:
+        data_files = list(write_file(io=PyArrowFileIO(), table_metadata=table_metadata, tasks=iter([task])))
+
+    assert mock_writer.call_args.kwargs["use_content_defined_chunking"] == {
+        "min_chunk_size": TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT,
+        "max_chunk_size": TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT,
+        "norm_level": TableProperties.PARQUET_CDC_NORM_LEVEL_DEFAULT,
+    }
+
+    assert len(data_files) == 1
+    written_table = pq.read_table(data_files[0].file_path.replace("file://", ""))
+    assert written_table.column("id").to_pylist() == list(range(1000))