Conversation
github-actions
Bot
added
documentation
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
20cc5e6 to
94e622b
Compare
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #24924 +/- ##
========================================
Coverage 81.93% 81.94%
========================================
Files 1136 1136
Lines 429152 429298 +146
Branches 429152 429298 +146
========================================
+ Hits 351633 351770 +137
+ Misses 56475 56472 -3
- Partials 21044 21056 +12 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
kosiew
left a comment
kosiew
left a comment
There was a problem hiding this comment.
@Tpt,
Thanks for working on this. I like the goal of making output file sizing more practical when row counts are a poor proxy for size, especially for blob-heavy datasets.
I found one issue that I think needs to be addressed before merging. The new byte limit is currently based on Arrow RecordBatch memory size rather than the number of serialized bytes written to the output file. For Parquet, compression can make those two values very different, so the current behavior can rotate files much earlier than the documented output-file byte target suggests.
I also left one small test coverage suggestion for the new config option.
| part_idx += 1; | ||
| } | ||
| row_counts[next_send_steam] += rb.num_rows(); | ||
| bytes_counts[next_send_steam] += rb.get_array_memory_size(); |
There was a problem hiding this comment.
I think this needs to measure serialized output bytes rather than the Arrow allocation size. get_array_memory_size() gives us the pre-serialization RecordBatch size, but Parquet defaults to zstd(3), so a highly compressible blob batch could be hundreds of MiB in memory and only a small fraction of that on disk.
In that case, the demuxer would rotate before the next batch and could create much smaller files than the configured soft_max_bytes_per_output_file suggests. That also seems particularly relevant to the blob use case this option is intended to help with.
Could we move the accounting to a point where serialized or emitted bytes are known, and feed that back to the demuxer if needed? If the intent is instead to use input memory size as an estimate, I think the option and docs should be narrowed to make that explicit.
It would also be useful to add a regression test using a large, highly compressible binary column with Parquet compression enabled. The current parameterized test uses UNCOMPRESSED, so it would not catch this difference.
There was a problem hiding this comment.
Sorry for the slow reply, got trapped by other things and thank you so much for the review
This is a great point. Done. I used a Arc<AtomicUsize> to record and report back the output size without blocking the code. An other approach might be to have an other channel giving back the written size but:
- either the demuxer block on it and we lose must of the concurrency and pipelining advantage (I guess very much a no go)
- we don't block and we get the same behavior of
Arc<AtomicUsize>with more overhead and code
The downside of Arc<AtomicUsize> is indeed that the reporting might lag quite a lot. I have updated the documentation to highlight this.
| /// number of rows written is not roughly divisible by the soft max | ||
| pub soft_max_rows_per_output_file: ConfigNonZeroUsize, default = non_zero_usize_default(50000000) | ||
|
|
||
| /// Target number of bytes in output files when writing multiple. |
There was a problem hiding this comment.
Could we also add a SET datafusion.execution.soft_max_bytes_per_output_file = 0 error case in set_variable.slt? ConfigNonZeroUsize already rejects zero, but adding coverage through the public SQL configuration path would match the existing row-limit coverage.
Allows to roughly limit the size of parquet files when the number of rows is a hard-to-use estimator
94e622b to
f9054bc
Compare
f9054bc to
cb03a02
Compare
kosiew
left a comment
kosiew
left a comment
There was a problem hiding this comment.
@Tpt,
Thanks for working on this. I like the direction of adding a byte-based soft limit, especially for workloads where row count is a poor proxy for file size.
I found one blocking issue with how the byte count is measured for Parquet. The current accounting is based on Arrow memory size rather than the encoded output size, which can cause highly compressible data to rotate files far too early. I also left one small suggestion to add SQL-path coverage for rejecting zero.
Once the byte accounting reflects serialized or emitted bytes, this should line up much better with the option's documented behavior.
| &write_id, | ||
| part_idx, | ||
| &file_extension, | ||
| single_file_output, |
There was a problem hiding this comment.
get_array_memory_size() measures the pre-serialization Arrow allocation rather than the number of bytes actually written to the output file. This is especially noticeable with Parquet, since the default compression is zstd(3). A highly compressible blob batch could occupy hundreds of MiB or even GiB in Arrow memory while producing a very small Parquet row group.
In that case, the demuxer can rotate before each following batch and create much smaller files than the configured soft_max_bytes_per_output_file target. That seems to conflict with the documented output-file-byte semantics and also affects the large-blob use case this option is intended to help with.
Could we move the accounting to a serialized or emitted-byte boundary, with feedback to the demuxer if needed? Alternatively, the option would need to be narrowed and renamed to make it clear that it is based on input memory size.
It would also be useful to add a regression test using a large, highly compressible Binary column with compressed Parquet. The current parameterized test forces UNCOMPRESSED, so it would not catch this difference.
| /// engines should set this to `false`: the coordinated fallback is then | ||
| /// disabled for left-emitting multi-partition joins, which instead fail | ||
| /// with a resource-exhaustion error under memory pressure rather than | ||
| /// deadlocking. Single-partition and non-left-emitting joins are |
There was a problem hiding this comment.
Could we also add the matching SET datafusion.execution.soft_max_bytes_per_output_file = 0 error case in set_variable.slt? ConfigNonZeroUsize already rejects zero, but adding coverage through the public SQL configuration path would keep this consistent with the sibling row-limit setting.
3 participants
Rationale for this change
Allows to roughly limit the size of parquet files when the number of rows is a hard-to-use estimator.
This is especially useful when a column contains potentially large blobs.
What changes are included in this PR?
datafusion.execution.soft_max_rows_per_output_file config parameter