Repository navigation
Conversation
|
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Rationale for this change
The asynchronous Parquet record batch generator decoded an entire row group into memory before slicing it into record batches. For files containing large row groups, this caused high peak memory usage even when a smaller
batch_sizewas configured.What changes are included in this PR?
Are these changes tested?
Yes. Tests verify that:
Are there any user-facing changes?
There are no public API changes.
For pre-buffered Parquet dataset scans, the existing batch_size option now also limits the number of rows decoded per batch instead of only controlling how an already-decoded row group is sliced. This reduces peak memory usage for large row groups.
Was AI used for this PR?
AI was used to help trace the relevant call chain, review concurrency behavior, suggest tests, and draft portions of the implementation and PR description. The changes were reviewed before submission.
Benchmark
Workload: one 262,144-row group with 128
int64columns and pre-buffering enabled. Results are medians of five runs using Google Benchmark wall-clock time.The existing
use_threadsoption remains supported. At a 32,768-row batch size, threaded column decoding improved incremental throughput from 61.1M to 70.3M rows/s in this workload.These results show that incremental decoding turns the existing
batch_sizesetting into a direct memory-throughput tuning knob. Smaller batches reduce peak decoded memory, while larger batches recover throughput.With the eager path,
batch_sizecontrolled only the emitted batches; the complete row group was still decoded first, so peak decoded memory remained determined by row-group size. The incremental path lets users choose anappropriate point in this tradeoff instead of always accepting a row-group-sized decoded allocation.
A 1,481,181-row, 650-column Parquet file with one row group was scanned in a fresh process for each case.
The incremental path reduced Arrow peak memory by 89.4% at a 65,536-row batch size and by 79.0% at 131,072 rows. The eager path's peak was unchanged across batch sizes, while incremental memory scaled with the configured batch size.
Associated Arrow Issue Links
#36778
#20473
Downstream relevance
We encountered this behavior while investigating high memory usage in PyIceberg’s
to_arrow_batch_reader(). Tracing the call path into PyArrow showed that the asynchronous Parquet scanner decoded an entire row group before applying the configured batch_size. This Arrow-level behavior can amplify memory usage when Parquet files contain large row groups. The change in this PR makes decoding incremental for the pre-buffered scanner path. This will also help address concerns similar to ours for decoding large row groups using PyIceberg’sto_arrow_batch_readerPR code and description written by:
Reviewed before submission by: