diff --git a/mkdocs/docs/configuration.md b/mkdocs/docs/configuration.md index 44b5e395a9..b2609ddca2 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 | 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..b5192cc4a7 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,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 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..056a0dd3b8 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,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))