kszucs commented on code in PR #3889:
URL: https://github.com/apache/iceberg-python/pull/3889#discussion_r3907898761
##########
tests/io/test_pyarrow.py:
##########
@@ -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()
+
+
[email protected](
+ "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}",
Review Comment:
Keeping this as is: `f"file://{tmp_path}"` plus the `.replace("file://",
"")` read-back is the established pattern in this file (e.g. lines 3048 and
3116, and ~13 other sites), and the `windows-unit-test` job runs exactly this
test selection and passes on this PR. Switching a single test to
`tmp_path.as_uri()` would produce a differently-shaped URI from the rest of the
file for no behavioural gain; if the convention is worth changing, it should be
changed file-wide in a separate PR.
##########
tests/io/test_pyarrow.py:
##########
@@ -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()
+
+
[email protected](
+ "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://",
""))
Review Comment:
Keeping this as is: `f"file://{tmp_path}"` plus the `.replace("file://",
"")` read-back is the established pattern in this file (e.g. lines 3048 and
3116, and ~13 other sites), and the `windows-unit-test` job runs exactly this
test selection and passes on this PR. Switching a single test to
`tmp_path.as_uri()` would produce a differently-shaped URI from the rest of the
file for no behavioural gain; if the convention is worth changing, it should be
changed file-wide in a separate PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
