Satyr09 opened a new pull request, #25041:
URL: https://github.com/apache/datafusion/pull/25041
# fix: honor max_row_group_bytes in the parallel Parquet writer
## Which issue does this PR close?
Closes #22982.
## Rationale for this change
`max_row_group_bytes` currently has no effect with the default
`allow_single_file_parallelism = true`. Users must disable parallel writing to
apply the byte target introduced in #22649.
This change applies the target while retaining parallel column encoding. It
uses completed column estimates to reproduce the single-threaded
`ArrowWriter`'s predictive row-group boundaries for the same input batches and
writer properties.
Either the row-count limit or the estimated byte target can finish a group.
## What changes are included in this PR?
- Column workers report encoded-size estimates and root-row progress. When a
byte target is configured, the dispatcher waits for the columns to finish the
current slice before using their combined estimate to size the next slice.
- Batch splitting follows the serial writer's row-first, predictive byte
policy, including reconsidering a remaining suffix after a row-group boundary.
The first slice of each group is accepted subject to the row limit. Estimates
can overshoot
the target.
- Worker errors propagate to the writer; cancellation releases workers and
memory reservations. Row-group ordinals remain correct for encrypted output.
- Feedback channels and per-slice synchronization are enabled only when a
byte target is set. Configuration documentation describes the throughput
tradeoff and distinguishes the estimated target from a hard byte or memory
limit.
Synchronized feedback gives reproducible boundaries and a direct
serial-writer reference for tests. I also compared a version that projects
partially completed column estimates: its boundaries varied with scheduling,
and the measurements did
not establish a consistent throughput advantage sufficient to justify that
approximation.
## What is the testing strategy for this PR?
Added SQL coverage in `parquet_max_row_group_bytes.slt` and writer tests in
`sink/tests.rs`. They check readback and serial boundary parity across row/byte
limits, nested columns, small fractional estimates, changing row widths,
dictionary fallback and compression, empty input, exact final boundaries,
immediate flushing, worker failures, cancellation, and encrypted row groups.
Benchmarks used DataFusion's existing SQL `benchmark_runner`, with twelve
COPY cases covering four input shapes and three writing modes, plus one
unchanged Parquet row-filter-skip scan case. The [successful comparison
run](https://github.com/Satyr09/datafusion/actions/runs/34101568243) contains
raw measurements and output layouts; the [fork-only validation
source](https://github.com/Satyr09/datafusion/tree/701433ce6e0a9e945c79ec86041fb7506f39e8c5/benchmark-validation)
contains the workloads and driver.
The comparison used base `5bf6aef8e37c91d882c0f75b772201d62733c88f`, a
single Linux runner with four logical CPUs, `release-nonlto`, 131,072 rows in
2,048-row batches, and two concurrent row-group writers. Byte targets were 256
KiB for
integers/Booleans and 4 MiB for strings. Two rounds reversed revision order;
each had seven iterations with the first excluded. These are pooled medians in
ms:
| Input | Original parallel, target ignored |
This PR parallel, target applied | This PR serial, target applied |
| ------------------------------------ | --------------------------------: |
-------------------------------: | -----------------------------: |
| Four integer columns | 20.82 |
16.65 | 32.00 |
| Four wide string columns | 276.86 |
292.99 | 643.81 |
| String widths rotate between columns | 99.77 |
111.78 | 226.39 |
| 64 Boolean columns | 21.37 |
24.57 | 39.41 |
The original parallel writer ignores the byte target, so its comparison
includes the cost of producing different row groups. The proposed writer
matched serial per-group row counts and compressed/uncompressed sizes for all
four inputs in
both rounds. All 108 retained outputs across revisions, modes and rounds
passed row-count and decoded-content checks.
With the option unset, three controls were within 0.5% of the original
pooled medians; the wide-string control was noisy. Hosted-runner timings varied
between rounds, so these results do not establish a universal performance
guarantee.
One open question before we need to resolve is memory use on rotating
strings: peak tracked reservation increased from 14.43 MiB in the original
parallel writer to 40.22 MiB with this change, in both rounds, while integer
and wide-string peaks decreased.
The original produced one group and this change produced six; the
contribution of concurrent groups and fresh encoder allocations has not been
isolated. These are DataFusion reservation peaks, not process RSS.
The byte target does not bound total writer memory.
## Are there any user-facing changes?
Yes. Setting `max_row_group_bytes` now affects parallel Parquet output.
Columns within a slice still encode concurrently, but synchronization between
slices can reduce write throughput. The default remains `None`, using only the
row-count
limit. The configuration reference and SQL format-options guide document
this behavior and the best-effort nature of the byte target.
--
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]
