From d7f61aadfe4c2ee4ebfe41280ac0758159059101 Mon Sep 17 00:00:00 2001 From: Krisztian Szucs Date: Tue, 1 Sep 2026 21:53:18 +0200 Subject: [PATCH] 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. PyArrow validates the chunk sizes itself and raises a clear error, so pyiceberg doesn't duplicate those checks. Requesting CDC on an older PyArrow raises an ImportError from a shared _require_pyarrow_version helper, which also replaces the existing Azure filesystem guard. --- mkdocs/docs/configuration.md | 4 ++ pyiceberg/io/pyarrow.py | 49 ++++++++++++++++---- pyiceberg/table/__init__.py | 12 +++++ tests/io/test_pyarrow.py | 87 ++++++++++++++++++++++++++++++++++++ 4 files changed, 143 insertions(+), 9 deletions(-) diff --git a/mkdocs/docs/configuration.md b/mkdocs/docs/configuration.md index 54d33dd00e..8f65bda462 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`, and raises at write time on older versions. | +| `write.parquet.content-defined-chunking.min-chunk-size` | Size in bytes | 256KiB (262144) | The minimum chunk size used for content-defined chunking | +| `write.parquet.content-defined-chunking.max-chunk-size` | Size in bytes | 1MiB (1048576) | 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 c36f1639d9..f60ac7620b 100644 --- a/pyiceberg/io/pyarrow.py +++ b/pyiceberg/io/pyarrow.py @@ -393,6 +393,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] @@ -535,14 +542,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 @@ -2939,7 +2939,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( @@ -2959,6 +2959,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, diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 9624eac981..4be7dbc2e2 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -161,6 +161,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 b31c18949b..a0b24adc6c 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -74,6 +74,7 @@ _check_pyarrow_schema_compatible, _ConvertToArrowSchema, _determine_partitions, + _get_parquet_writer_kwargs, _primitive_to_physical, _read_deletes, _task_to_record_batches, @@ -5462,3 +5463,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(pa.ArrowIOError, match="max_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))