Roll streaming writes at the on-disk target file size

[azwanzuharimi]

Closes #3388

Rationale for this change

When you append a pa.RecordBatchReader, PyIceberg groups batches by their size in memory before it writes them. Parquet then compresses the data, so files on disk come out 3 to 10 times smaller than write.target-file-size-bytes. Memory also grows with the target, because a whole group waits in memory before the write starts.

This change writes each batch as it arrives and rolls to a new file when the file on disk reaches the target. It is the same idea as RollingFileWriter in Java.

What changes:

• FileFormatWriter gets a new abstract method, length(). It returns the estimated file size so far, like FileAppender.length() in Java. This is a breaking change for anyone who subclassed FileFormatWriter outside this repo. There are no subclasses in the repo.
• ParquetFormatWriter buffers rows and flushes a row group when the buffer reaches write.parquet.row-group-limit rows or write.parquet.row-group-size-bytes. Without this buffer every input batch would become its own row group.
• length() counts the bytes on disk plus the buffered rows. Buffered rows are scaled by the compression ratio seen on the last flush, so files land close to the target.
• The streaming branch of _dataframe_to_data_files writes one file at a time through the format writer and rolls when length() reaches the target.
• The pa.Table path does not change. It still calls write() once and close() once, so it writes the same row groups and the same bytes as before.
• bin_pack_record_batches stays, because it is public. Nothing in pyiceberg calls it now.
• write.parquet.row-group-size-bytes no longer warns as unsupported, because the streaming writer reads it. The configuration doc says it applies to streaming writes only.

Measured on macOS with pyarrow 25.0.1, 64 MiB target, default properties. The value is file size on disk divided by the target.

Data	Before	After
Random rows, 400 batches of 10,000	0.69, 0.69, 0.19	1.01, 0.56 (last file)
Compressible rows, 12,000 batches of 10,000	0.17 for each full file	1.00, 1.00, 0.81 (last file)

Memory during a streaming write is bounded by one row group plus one input batch and the writer's buffers. It no longer depends on the target.

Are these changes tested?

Yes.

• tests/io/test_format_writers.py: length() starts at 0, grows with each write, and equals the file size after close. Buffered rows count in full before the first flush and are scaled after it. A failed flush on the exception path closes the stream and keeps the original error.
• tests/io/test_pyarrow.py: the streaming path rolls at the on-disk target, lands within a few percent of the target on compressible data, writes one file for a large target, packs small batches into row groups, flushes row groups on bytes, yields the first file before it reads the rest of the reader, closes the stream when the reader raises, and stays fast with 5,000 one row batches. write_file writes the same bytes as a direct pq.ParquetWriter write for six settings of the row group properties.
• tests/catalog/test_catalog_behaviors.py: the existing RecordBatchReader tests pass unchanged.

make lint passes. tests/io, tests/catalog/test_catalog_behaviors.py and tests/table pass, with the S3, ADLS, GCS and integration markers deselected.

Are there any user-facing changes?

Yes.

• Streaming writes now produce files close to write.target-file-size-bytes on disk, and memory no longer grows with the target.
• FileFormatWriter has a new abstract method, length().
• write.parquet.row-group-size-bytes no longer warns as unsupported. It applies to streaming writes.

[azwanzuharimi]

@Fokko @kevinjqliu this picks up #3336 on the file format writer API from #3381. The two earlier PRs closed as stale without a review, so a first look on the design would help.