`pyiceberg.io.pyarrow.write_file` does not take into account compression settings

[jonashaag]

Apache Iceberg version

main (development)

Please describe the bug 🐞

I expected table.append() to take my write.parquet.compression-codec etc settings into account but it looks like no settings are passed to ParquetWriter:

iceberg-python/pyiceberg/io/pyarrow.py

Line 1729 in 29db67f

with pq.ParquetWriter(fos, schema=file_schema, version="1.0", metadata_collector=collected_metrics) as writer:

For my data this means a 5x size increase compared to using zstd level 3

[Fokko]

@jonashaag Thanks for reporting this, and you're right. The settings are currently not taken into account. Are you interested in contributing this to the project?

We need to pass in the correct string to the ParquetWriter: https://arrow.apache.org/docs/python/generated/pyarrow.parquet.ParquetWriter.html And also set the compression level.

[jonashaag]

Yes happy to. Can you point me to some other places where the config is read and passed around etc?

[Fokko]

@jonashaag Sure thing! This is where we need to pass in the configuration:

iceberg-python/pyiceberg/io/pyarrow.py

Lines 1713 to 1758 in a4856bc

	def write_file(table: Table, tasks: Iterator[WriteTask]) -> Iterator[DataFile]:
	task = next(tasks)
	try:
	_ = next(tasks)
	# If there are more tasks, raise an exception
	raise NotImplementedError("Only unpartitioned writes are supported: https://github.com/apache/iceberg-python/issues/208")
	except StopIteration:
	pass
	file_path = f'{table.location()}/data/{task.generate_data_file_filename("parquet")}'
	file_schema = schema_to_pyarrow(table.schema())
	collected_metrics: List[pq.FileMetaData] = []
	fo = table.io.new_output(file_path)
	with fo.create(overwrite=True) as fos:
	with pq.ParquetWriter(fos, schema=file_schema, version="1.0", metadata_collector=collected_metrics) as writer:
	writer.write_table(task.df)
	data_file = DataFile(
	content=DataFileContent.DATA,
	file_path=file_path,
	file_format=FileFormat.PARQUET,
	partition=Record(),
	file_size_in_bytes=len(fo),
	# After this has been fixed:
	# https://github.com/apache/iceberg-python/issues/271
	# sort_order_id=task.sort_order_id,
	sort_order_id=None,
	# Just copy these from the table for now
	spec_id=table.spec().spec_id,
	equality_ids=None,
	key_metadata=None,
	)
	if len(collected_metrics) != 1:
	# One file has been written
	raise ValueError(f"Expected 1 entry, got: {collected_metrics}")
	fill_parquet_file_metadata(
	data_file=data_file,
	parquet_metadata=collected_metrics[0],
	stats_columns=compute_statistics_plan(table.schema(), table.properties),
	parquet_column_mapping=parquet_path_to_id_mapping(table.schema()),
	)
	return iter([data_file])

The table.properties contains the configuration option. Today we only support writing Parquet files, therefore we want to fetch the write.parquet.compression-codec property that ∈ {zstd, gzip, snappy} which corresponds with the properties in Arrow. Next, we also need to get write.parquet.compression-level to set the level.

I would just create a function to extract the relevant properties and then pass the into the writer. Let me know if this helps or if you have any other questions.

[jonashaag]

I think the REST catalog ignores the write.parquet.compression-codec option. No matter what options I set for the catalog, it always responds here

iceberg-python/pyiceberg/catalog/rest.py

Line 472 in a4856bc

table_response = TableResponse(**response.json())

with

'write.parquet.compression-codec': 'zstd'

even if I explicitly set write.parquet.compression-codec in the properties a few lines above.

I'll submit a patch that passes the right arguments to PyArrow but currently we can't support any other compression options for the REST catalog it seems.

[jonashaag]

Update: Hm, now it doesn't seem to be the case anymore. Not sure what happened there...

[Fokko]

@jonashaag The REST catalog sets the property as a default one, the other ones don't. If no property is set, it should default to zstd similar to Java.

[Fokko]

Fixed in #358