Skip to content

Commit 1267ad2

Browse files
kszucskszucs
committed
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.
1 parent 58749a3 commit 1267ad2

4 files changed

Lines changed: 118 additions & 9 deletions

File tree

‎mkdocs/docs/configuration.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,10 @@ Iceberg tables support table properties to configure table behavior.
8585
| `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 |
8686
| `write.parquet.page-row-limit` | Number of rows | 20000 | Set a target threshold for the maximum number of rows within a column chunk |
8787
| `write.parquet.dict-size-bytes` | Size in bytes | 2MB | Set the dictionary page size limit per row group |
88+
| `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`. |
89+
| `write.parquet.content-defined-chunking.min-chunk-size` | Size in bytes | 256KB | The minimum chunk size used for content-defined chunking |
90+
| `write.parquet.content-defined-chunking.max-chunk-size` | Size in bytes | 1MB | The maximum chunk size used for content-defined chunking |
91+
| `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 |
8892
| `write.metadata.previous-versions-max` | Integer | 100 | The max number of previous version metadata files to keep before deleting after commit. |
8993
| `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`. |
9094
| `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‎

Lines changed: 35 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -393,6 +393,13 @@ def to_input_file(self) -> PyArrowFile:
393393
return self
394394

395395

396+
def _require_pyarrow_version(min_version: str, feature: str) -> None:
397+
from packaging import version
398+
399+
if version.parse(pyarrow.__version__) < version.parse(min_version):
400+
raise ImportError(f"pyarrow version >= {min_version} required for {feature}, but found version {pyarrow.__version__}.")
401+
402+
396403
class PyArrowFileIO(FileIO):
397404
fs_by_scheme: Callable[[str, str | None], FileSystem]
398405

@@ -535,14 +542,7 @@ def _initialize_s3_fs(self, netloc: str | None) -> FileSystem:
535542

536543
def _initialize_azure_fs(self) -> FileSystem:
537544
# https://arrow.apache.org/docs/python/generated/pyarrow.fs.AzureFileSystem.html
538-
from packaging import version
539-
540-
MIN_PYARROW_VERSION_SUPPORTING_AZURE_FS = "20.0.0"
541-
if version.parse(pyarrow.__version__) < version.parse(MIN_PYARROW_VERSION_SUPPORTING_AZURE_FS):
542-
raise ImportError(
543-
f"pyarrow version >= {MIN_PYARROW_VERSION_SUPPORTING_AZURE_FS} required for AzureFileSystem support, "
544-
f"but found version {pyarrow.__version__}."
545-
)
545+
_require_pyarrow_version("20.0.0", "AzureFileSystem support")
546546

547547
from pyarrow.fs import AzureFileSystem
548548

@@ -2939,7 +2939,7 @@ def _get_parquet_writer_kwargs(table_properties: Properties) -> dict[str, Any]:
29392939
if compression_codec == ICEBERG_UNCOMPRESSED_CODEC:
29402940
compression_codec = PYARROW_UNCOMPRESSED_CODEC
29412941

2942-
return {
2942+
parquet_writer_kwargs = {
29432943
"compression": compression_codec,
29442944
"compression_level": compression_level,
29452945
"data_page_size": property_as_int(
@@ -2959,6 +2959,32 @@ def _get_parquet_writer_kwargs(table_properties: Properties) -> dict[str, Any]:
29592959
),
29602960
}
29612961

2962+
if property_as_bool(
2963+
properties=table_properties,
2964+
property_name=TableProperties.PARQUET_CDC_ENABLED,
2965+
default=TableProperties.PARQUET_CDC_ENABLED_DEFAULT,
2966+
):
2967+
_require_pyarrow_version("21.0.0", "Parquet content-defined chunking")
2968+
parquet_writer_kwargs["use_content_defined_chunking"] = {
2969+
"min_chunk_size": property_as_int(
2970+
properties=table_properties,
2971+
property_name=TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE,
2972+
default=TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT,
2973+
),
2974+
"max_chunk_size": property_as_int(
2975+
properties=table_properties,
2976+
property_name=TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE,
2977+
default=TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT,
2978+
),
2979+
"norm_level": property_as_int(
2980+
properties=table_properties,
2981+
property_name=TableProperties.PARQUET_CDC_NORM_LEVEL,
2982+
default=TableProperties.PARQUET_CDC_NORM_LEVEL_DEFAULT,
2983+
),
2984+
}
2985+
2986+
return parquet_writer_kwargs
2987+
29622988

29632989
def _dataframe_to_data_files(
29642990
table_metadata: TableMetadata,

‎pyiceberg/table/__init__.py‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,18 @@ class TableProperties:
161161

162162
PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX = "write.parquet.bloom-filter-enabled.column"
163163

164+
PARQUET_CDC_ENABLED = "write.parquet.content-defined-chunking.enabled"
165+
PARQUET_CDC_ENABLED_DEFAULT = False
166+
167+
PARQUET_CDC_MIN_CHUNK_SIZE = "write.parquet.content-defined-chunking.min-chunk-size"
168+
PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT = 256 * 1024 # 256 KiB
169+
170+
PARQUET_CDC_MAX_CHUNK_SIZE = "write.parquet.content-defined-chunking.max-chunk-size"
171+
PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT = 1024 * 1024 # 1 MiB
172+
173+
PARQUET_CDC_NORM_LEVEL = "write.parquet.content-defined-chunking.norm-level"
174+
PARQUET_CDC_NORM_LEVEL_DEFAULT = 0
175+
164176
WRITE_TARGET_FILE_SIZE_BYTES = "write.target-file-size-bytes"
165177
WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT = 512 * 1024 * 1024 # 512 MB
166178

‎tests/io/test_pyarrow.py‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,7 @@
7474
_check_pyarrow_schema_compatible,
7575
_ConvertToArrowSchema,
7676
_determine_partitions,
77+
_get_parquet_writer_kwargs,
7778
_primitive_to_physical,
7879
_read_deletes,
7980
_task_to_record_batches,
@@ -5462,3 +5463,69 @@ def test_dictionary_columns_produces_dict_encoded_output(tmpdir: str) -> None:
54625463

54635464
# Values must be identical
54645465
assert result_plain.column("label").to_pylist() == result_dict.column("label").to_pylist()
5466+
5467+
5468+
def test_get_parquet_writer_kwargs_cdc_disabled_by_default() -> None:
5469+
kwargs = _get_parquet_writer_kwargs({})
5470+
assert "use_content_defined_chunking" not in kwargs
5471+
5472+
5473+
def test_get_parquet_writer_kwargs_cdc_enabled_with_defaults() -> None:
5474+
kwargs = _get_parquet_writer_kwargs({TableProperties.PARQUET_CDC_ENABLED: "true"})
5475+
assert kwargs["use_content_defined_chunking"] == {
5476+
"min_chunk_size": TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT,
5477+
"max_chunk_size": TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT,
5478+
"norm_level": TableProperties.PARQUET_CDC_NORM_LEVEL_DEFAULT,
5479+
}
5480+
5481+
5482+
def test_get_parquet_writer_kwargs_cdc_enabled_with_custom_values() -> None:
5483+
kwargs = _get_parquet_writer_kwargs(
5484+
{
5485+
TableProperties.PARQUET_CDC_ENABLED: "true",
5486+
TableProperties.PARQUET_CDC_MIN_CHUNK_SIZE: "4096",
5487+
TableProperties.PARQUET_CDC_MAX_CHUNK_SIZE: "8192",
5488+
TableProperties.PARQUET_CDC_NORM_LEVEL: "2",
5489+
}
5490+
)
5491+
assert kwargs["use_content_defined_chunking"] == {
5492+
"min_chunk_size": 4096,
5493+
"max_chunk_size": 8192,
5494+
"norm_level": 2,
5495+
}
5496+
5497+
5498+
def test_get_parquet_writer_kwargs_cdc_enabled_unsupported_pyarrow_version(monkeypatch: pytest.MonkeyPatch) -> None:
5499+
monkeypatch.setattr(pyarrow, "__version__", "17.0.0")
5500+
with pytest.raises(ImportError, match="pyarrow version >= 21.0.0"):
5501+
_get_parquet_writer_kwargs({TableProperties.PARQUET_CDC_ENABLED: "true"})
5502+
5503+
5504+
def test_write_file_with_content_defined_chunking_enabled(tmp_path: Path) -> None:
5505+
"""Writing a table with CDC enabled should succeed and produce a readable Parquet file."""
5506+
from pyiceberg.table import WriteTask
5507+
5508+
table_schema = Schema(NestedField(1, "id", IntegerType(), required=False))
5509+
arrow_data = pa.table({"id": pa.array(range(1000), type=pa.int32())})
5510+
5511+
table_metadata = TableMetadataV2(
5512+
location=f"file://{tmp_path}",
5513+
last_column_id=1,
5514+
format_version=2,
5515+
schemas=[table_schema],
5516+
partition_specs=[PartitionSpec()],
5517+
properties={TableProperties.PARQUET_CDC_ENABLED: "true"},
5518+
)
5519+
5520+
task = WriteTask(
5521+
write_uuid=uuid.uuid4(),
5522+
task_id=0,
5523+
record_batches=arrow_data.to_batches(),
5524+
schema=table_schema,
5525+
)
5526+
5527+
data_files = list(write_file(io=PyArrowFileIO(), table_metadata=table_metadata, tasks=iter([task])))
5528+
assert len(data_files) == 1
5529+
5530+
written_table = pq.read_table(data_files[0].file_path.replace("file://", ""))
5531+
assert written_table.column("id").to_pylist() == list(range(1000))

0 commit comments

Comments
 (0)