Validate CDC chunk-size/norm-level properties and strengthen tests

Raise a clear ValueError for invalid content-defined-chunking config
(min-chunk-size <= 0, max-chunk-size <= min-chunk-size, negative
norm-level) instead of letting PyArrow's opaque internal error surface.
Also parametrize the near-duplicate kwargs tests, add coverage for the
new validation, and make the write-path test assert
use_content_defined_chunking actually reaches pq.ParquetWriter instead
of only checking a round-trip that would pass even if the kwarg were
silently dropped.

Author: kszucs

mkdocs/docs/configuration.md

@@ -85,7 +85,7 @@ 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`.     |
+| `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                                |

pyiceberg/io/pyarrow.py

@@ -2959,28 +2959,44 @@ 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")
+        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,
+        )
+        if min_chunk_size is not None and min_chunk_size <= 0:
+            raise ValueError(f"{TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE} must be greater than 0, got {min_chunk_size}")
+        if max_chunk_size is not None and min_chunk_size is not None and max_chunk_size <= min_chunk_size:
+            raise ValueError(
+                f"{TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE} ({max_chunk_size}) must be greater than "
+                f"{TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE} ({min_chunk_size})"
+            )
+        if norm_level is not None and norm_level < 0:
+            raise ValueError(f"{TableProperties.PARQUET_CDC_NORM_LEVEL} must be greater than or equal to 0, got {norm_level}")
+
         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,
-            ),
+            "min_chunk_size": min_chunk_size,
+            "max_chunk_size": max_chunk_size,
+            "norm_level": norm_level,
         }
 
     return parquet_writer_kwargs

tests/io/test_pyarrow.py

@@ -5465,34 +5465,32 @@ def test_dictionary_columns_produces_dict_encoded_output(tmpdir: str) -> None:
     assert result_plain.column("label").to_pylist() == result_dict.column("label").to_pylist()
 
 
-def test_get_parquet_writer_kwargs_cdc_disabled_by_default() -> None:
-    kwargs = _get_parquet_writer_kwargs({})
-    assert "use_content_defined_chunking" not in kwargs
-
-
-def test_get_parquet_writer_kwargs_cdc_enabled_with_defaults() -> None:
-    kwargs = _get_parquet_writer_kwargs({TableProperties.PARQUET_CDC_ENABLED: "true"})
-    assert 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,
-    }
-
-
-def test_get_parquet_writer_kwargs_cdc_enabled_with_custom_values() -> None:
-    kwargs = _get_parquet_writer_kwargs(
-        {
-            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",
-        }
-    )
-    assert kwargs["use_content_defined_chunking"] == {
-        "min_chunk_size": 4096,
-        "max_chunk_size": 8192,
-        "norm_level": 2,
-    }
+@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:
@@ -5501,8 +5499,40 @@ def test_get_parquet_writer_kwargs_cdc_enabled_unsupported_pyarrow_version(monke
         _get_parquet_writer_kwargs({TableProperties.PARQUET_CDC_ENABLED: "true"})
 
 
+@pytest.mark.parametrize(
+    "table_properties,match",
+    [
+        (
+            {
+                TableProperties.PARQUET_CDC_ENABLED: "true",
+                TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE: "0",
+            },
+            "min-chunk-size must be greater than 0",
+        ),
+        (
+            {
+                TableProperties.PARQUET_CDC_ENABLED: "true",
+                TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE: "8192",
+                TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE: "4096",
+            },
+            "max-chunk-size .* must be greater than .*min-chunk-size",
+        ),
+        (
+            {
+                TableProperties.PARQUET_CDC_ENABLED: "true",
+                TableProperties.PARQUET_CDC_NORM_LEVEL: "-1",
+            },
+            "norm-level must be greater than or equal to 0",
+        ),
+    ],
+)
+def test_get_parquet_writer_kwargs_cdc_invalid_properties(table_properties: dict[str, str], match: str) -> None:
+    with pytest.raises(ValueError, match=match):
+        _get_parquet_writer_kwargs(table_properties)
+
+
 def test_write_file_with_content_defined_chunking_enabled(tmp_path: Path) -> None:
-    """Writing a table with CDC enabled should succeed and produce a readable Parquet file."""
+    """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))
@@ -5524,8 +5554,15 @@ def test_write_file_with_content_defined_chunking_enabled(tmp_path: Path) -> Non
         schema=table_schema,
     )
 
-    data_files = list(write_file(io=PyArrowFileIO(), table_metadata=table_metadata, tasks=iter([task])))
-    assert len(data_files) == 1
+    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))