From c158db2c9921826236fea4b2c8b027748072ceec Mon Sep 17 00:00:00 2001 From: Krisztian Szucs Date: Mon, 6 Jul 2026 18:22:46 +0200 Subject: [PATCH 1/3] Add Parquet content-defined chunking (CDC) writer support PyArrow's ParquetWriter has supported content-defined chunking natively since 21.0.0, producing stable page boundaries across appends for content-addressable storage. Wire this through as write.parquet.content-defined-chunking.* table properties, mirroring the property names and defaults already used by iceberg-rust. --- mkdocs/docs/configuration.md | 4 +++ pyiceberg/io/pyarrow.py | 44 ++++++++++++++++++----- pyiceberg/table/__init__.py | 12 +++++++ tests/io/test_pyarrow.py | 67 ++++++++++++++++++++++++++++++++++++ 4 files changed, 118 insertions(+), 9 deletions(-) diff --git a/mkdocs/docs/configuration.md b/mkdocs/docs/configuration.md index 44b5e395a9..6142eb0e2d 100644 --- a/mkdocs/docs/configuration.md +++ b/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`. | +| `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. | diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index 6259f311e9..958f558086 100644 --- a/pyiceberg/io/pyarrow.py +++ b/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,32 @@ def _get_parquet_writer_kwargs(table_properties: Properties) -> dict[str, Any]: ), } + 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") + 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, diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 63b87d290e..bc86a41e91 100644 --- a/pyiceberg/table/__init__.py +++ b/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 diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index 532311899d..b4d9275eab 100644 --- a/tests/io/test_pyarrow.py +++ b/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,69 @@ 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() + + +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, + } + + +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_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.""" + 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, + ) + + data_files = list(write_file(io=PyArrowFileIO(), table_metadata=table_metadata, tasks=iter([task]))) + 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)) From 9c9fd583348324e3e76c0c217f907344cbe699d7 Mon Sep 17 00:00:00 2001 From: Krisztian Szucs Date: Mon, 6 Jul 2026 21:43:50 +0200 Subject: [PATCH 2/3] 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. --- mkdocs/docs/configuration.md | 2 +- pyiceberg/io/pyarrow.py | 46 +++++++++++------ tests/io/test_pyarrow.py | 99 +++++++++++++++++++++++++----------- 3 files changed, 100 insertions(+), 47 deletions(-) diff --git a/mkdocs/docs/configuration.md b/mkdocs/docs/configuration.md index 6142eb0e2d..b2609ddca2 100644 --- a/mkdocs/docs/configuration.md +++ b/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 | diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index 958f558086..1166726d9e 100644 --- a/pyiceberg/io/pyarrow.py +++ b/pyiceberg/io/pyarrow.py @@ -2865,28 +2865,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 diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index b4d9275eab..f77ef7e05d 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -5300,34 +5300,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: @@ -5336,8 +5334,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)) @@ -5359,8 +5389,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)) From 466ca6a606099d2725676468b09692b4f704cae5 Mon Sep 17 00:00:00 2001 From: Krisztian Szucs Date: Mon, 6 Jul 2026 23:48:36 +0200 Subject: [PATCH 3/3] Remove redundant CDC chunk-size/norm-level validation PyArrow already validates these values itself (e.g. max_chunk_size must be greater than min_chunk_size) and raises a clear OSError, and iceberg-rust's equivalent ParquetWriterBuilder::from_table_properties doesn't duplicate this validation either -- defer to PyArrow instead of maintaining a second, slightly different copy of the same checks. --- pyiceberg/io/pyarrow.py | 45 +++++++++++++++------------------------- tests/io/test_pyarrow.py | 43 ++++++++++++-------------------------- 2 files changed, 30 insertions(+), 58 deletions(-) diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index 1166726d9e..b5192cc4a7 100644 --- a/pyiceberg/io/pyarrow.py +++ b/pyiceberg/io/pyarrow.py @@ -2874,35 +2874,24 @@ def _get_parquet_writer_kwargs(table_properties: Properties) -> dict[str, Any]: 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}") - + # 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": min_chunk_size, - "max_chunk_size": max_chunk_size, - "norm_level": norm_level, + "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 diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index f77ef7e05d..056a0dd3b8 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -5334,36 +5334,19 @@ 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_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: