[python] Support writing shared-shredding MAPs in Parquet

[XiaoHongbo-Hope]

Purpose

Let PyPaimon writers honor the existing fields.<field>.map.storage-layout=shared-shredding option.

Changes

• Write Java-compatible dictionaries, value columns, overflow data, and adaptive column counts.
• Preserve logical statistics and row sidecars; support normal columns alongside BLOB/vector files.
• Scan keys to finalize metadata, coalesce small logical chunks within bounded row/byte windows, then stream physical batches to Parquet. Row groups do not follow individual input calls. This adds a key pass but avoids buffering the entire physical file or expanding empty slots into row dictionaries. Input buffers, dictionaries, and individual oversized rows are not hard-capped.
• Keep existing public configuration; no new write API. Reject CHAR keys, matching Java.
• Support Parquet writes. ORC is rejected because PyArrow cannot persist the required ORC user metadata.

Tests

• Full/selected-key reads, rolling, adaptive column counts, bounded physical batches, and failed-write cleanup.
• Non-nullable and nested MAP values, including Python 3.6 / PyArrow 6 subprocess regressions.
• Java reads Python-written Parquet; Python reads Java-written Parquet and ORC.
• Local 5ms RSS sampling, 5,000 rows with one key and 256 physical columns: peak about 282 → 176 MiB; 50,000 rows after the change: about 197 MiB. These are local measurements, not a memory guarantee.

[JingsongLi]

[P2] Coalesce small input chunks before writing Parquet row groups

Location: map_shared_shredding_writer.py:167–173.

data.to_batches(max_chunksize=batch_rows) preserves existing chunk boundaries; it does not combine small chunks. On the append-only path, each write_row() creates a one-row batch, and WriteBuffer.materialize() uses pa.concat_tables, which preserves those chunks. Consequently, this writer.write_table(physical) call creates a separate row group for every input row.

I reproduced this with Python 3.11 / PyArrow 18.1, an append-only Parquet table with id INT, m MAP<STRING, BIGINT>, shared shredding enabled for m, and the default 256 physical columns. For 1,000 rows containing one entry ('a', i) each:

Input method	Row groups	File size	Footer size
One write_arrow() call with a contiguous table	1	98,967 bytes	76,623 bytes
1,000 write_row() calls	1,000	32,456,977 bytes	16,716,965 bytes

This substantially increases storage, footer memory, and read overhead for a supported write API. Adaptive column counts only take effect after a file completes, so they do not protect the first file.

How Java handles this: ShreddingFormatWriter.addElement() converts one logical row and delegates to ParquetBulkWriter → ParquetWriter.write(). ColumnCompressionRecordWriter keeps column/page buffers across these calls and flushes a row group when the row-count limit or buffered-size threshold is reached; closing the file flushes the remainder. The default size target is 128 MiB. Thus, row-group boundaries are independent of individual input calls. Java also commits the final shredding dictionary metadata before closing the underlying writer, allowing a single pass over the rows.

Please coalesce small logical chunks within bounded row/byte windows before conversion, and keep row-group boundaries independent of input chunk boundaries. Increasing row_group_size alone will not combine separate write_table() calls. A regression test comparing one batch with repeated write_row() calls should verify that row-group count does not grow with the number of input calls.

[JingsongLi]

+1