misc: release parquet row group readers before prefetch - #805
Conversation
Release Parquet column reader state when a row group slides out of the prefetch window. Before loading the next current+prefetch window, the old row group's BufferedInput was erased, but PageReader state such as decompressed page buffers and dictionary string storage could remain attached to column readers until seekToRowGroup() replaced it later. That ordering can transiently overlap old row-group decode state with new DirectBufferedInput prefetch allocations. This is most visible for wide dictionary string columns where the dictionary page is held as both decompressed page data and dictionary string backing storage. Add an explicit recursive release that clears ScanState and resets ParquetData's PageReader before the new row-group window is loaded. This keeps the steady row-group window bounded without changing read semantics; seekToRowGroup() still constructs the reader for the next active row group. Add a regression test that writes multiple row groups with a large repeated string dictionary and reads them through a constrained memory pool. The test exercises row-group advancement and verifies the scan stays under the memory limit while reading all row groups. Co-authored-by: TRAE CLI <noreply@bytedance.com>
| namespace { | ||
| void releaseRowGroupReaderRecursive( | ||
| dwio::common::SelectiveColumnReader* reader) { | ||
| reader->scanState().clear(); |
There was a problem hiding this comment.
will this affect lazy column loading?
There was a problem hiding this comment.
The early release should not affect lazy column loading because of the scan/lazy-vector lifecycle.
A LazyVector produced by the Parquet reader is tied to the current reader pass. The downstream operator must consume it in that pass. If the column data is needed, it will be materialized through the lazy loader before the scan reader is advanced to produce another batch. In other words, the Parquet reader is not expected to keep old row-group PageReader state alive for lazy loads after the scan has moved forward.
This change releases the PageReader only when the scan is advancing from row group i - 1 to row group i:
if (currentGroup >= 1) {
inputs_.erase(rowGroupIds[currentGroup - 1]);
reader.releaseRowGroupReader();
}
At that point, row group i - 1 has already been consumed by the scan pipeline. Then seekToRowGroup(i) creates a new PageReader for the current row group before any data from row group i is read:
reader_ = std::make_unique<PageReader>(std::move(streams_[index]), ...);
So the release only drops page-reader state for the previous row group after its lifecycle is complete. It does not release the PageReader that will be used by lazy vectors produced for the current row group. The intent is only to avoid overlapping old row-group decode state with the next row-group prefetch window.
There was a problem hiding this comment.
Thank you for the explanation.
2 participants
What problem does this PR solve?
Release Parquet column reader state when a row group slides out of the prefetch window. Before loading the next current+prefetch window, the old row group's BufferedInput was erased, but PageReader state such as decompressed page buffers and dictionary string storage could remain attached to column readers until seekToRowGroup() replaced it later.
Type of Change
Description
That ordering can transiently overlap old row-group decode state with new DirectBufferedInput prefetch allocations. This is most visible for wide dictionary string columns where the dictionary page is held as both decompressed page data and dictionary string backing storage.
Add an explicit recursive release that clears ScanState and resets ParquetData's PageReader before the new row-group window is loaded. This keeps the steady row-group window bounded without changing read semantics; seekToRowGroup() still constructs the reader for the next active row group.
Add a regression test that writes multiple row groups with a large repeated string dictionary and reads them through a constrained memory pool. The test exercises row-group advancement and verifies the scan stays under the memory limit while reading all row groups.
Performance Impact
No Impact: This change does not affect the critical path (e.g., build system, doc, error handling).
Positive Impact: I have run benchmarks.
Click to view Benchmark Results
Negative Impact: Explained below (e.g., trade-off for correctness).
Release Note
Please describe the changes in this PR
Release Note:
Checklist (For Author)
Breaking Changes
No
Yes (Description: ...)
Click to view Breaking Changes