peter-toth opened a new pull request, #59639: URL: https://github.com/apache/spark/pull/59639
### What changes were proposed in this pull request? The late-materialization Parquet reader (SPARK-59620) buffers a row group's surviving key values and row ranges before it emits that row group, up to `spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes` per reader. Today the buffer is outside Spark's memory manager. This PR accounts it to the task's execution memory through a `MemoryConsumer`. It was on the follow-up list of #58895, and viirya suggested it there after the merge ([comment](https://github.com/apache/spark/pull/58895#issuecomment-6064730845)). - **Reserving:** phase 1 reserves the buffer as it grows. Each request asks for at least what is already reserved, so a growing buffer asks only a few times. - A denied reservation gives splicing up, as passing the cap does, and phase 2 reads the key columns again. - If the row ranges are denied too, the filter is given up for that row group. - **Releasing:** each batch's share is released when the next batch is read. The rest goes with the row group, or at close. - **Spilling:** a spill from another consumer frees the queued survivors of the row group being emitted. The next batch reads the keys of the remaining rows again, while the non-key column readers carry on. - A spill acts only on the reading thread, where it can come only between two batches. A spill from another thread returns 0. So the reader takes no lock and cannot deadlock with the task memory manager. Locking against such a spill is what deadlocked `UnsafeExternalSorter` with it twice (SPARK-27338, SPARK-39283). - Only a row group that phase 2 reads in part can be spilled. The key re-read needs the keys' offset indexes. Parquet builds a row group's column index store once, over the columns requested then. So the store is built over the whole projection just before phase 2, which reads it anyway. For a row group read whole, that would be an extra read. A damaged index there would also fail the re-read. - **Off heap:** the consumer is `ON_HEAP` whatever the reader's memory mode, as `WindowSegmentTree`'s spiller is (SPARK-56546). Off-heap column vectors allocate native memory outside the Tungsten off-heap pool, so an `OFF_HEAP` consumer would charge that pool for memory it never held. With off-heap vectors only the row ranges are accounted, and a spill frees nothing. - `VectorizedParquetRecordReader` gets `replaceColumnReader`, which points one top-level slot at a new column reader in the middle of a row group. ### Why are the changes needed? Each reader buffers up to 64 MB by default, which the memory manager cannot see. Today a task holds one per scan at a time, so an executor holds about one per running task. Under a sorted merge over many files, as storage-partitioned joins do once storage filters reach them, one task would hold many at once. Accounted, other consumers spill earlier, and a reader that is denied memory gives the buffer up instead. ### Does this PR introduce _any_ user-facing change? No. The feature is unreleased and off by default. With it on, the buffer now counts against the task's execution memory. ### How was this patch tested? New tests in `ParquetStorageFilterSuite`, under a task whose memory manager the test controls: - the buffer is held while a spliced row group is emitted, and released batch by batch and at the end. On heap that is the key values and the row ranges, off heap the row ranges only. A reader closed after its first batch releases what it holds; - memory denied for the keys gives splicing up, for the same bytes as a cap of that size, and memory denied for the ranges gives the filter up; - a buffer that fits in the free memory spills no other consumer; - a spill from another consumer between two batches frees the queued survivors, and the rows stay right. A spill from another thread leaves the buffer alone; - a spill leaves alone a row group that phase 2 reads whole, here one whose offset index does not parse; - in a task, a row group that phase 2 reads whole or not at all reads no offset index. Each was checked by mutation. These all make at least one of them fail: - ignoring a denied reservation; - not reading the keys again after a spill; - spilling from any thread; - not releasing a batch's share, or not releasing at close; - accounting the key values off heap; - not building the column index store over the whole projection; - a spill that frees no memory; - spilling a row group that phase 2 reads whole; - building the store for every row group before phase 1; - asking for a fixed 256 KB at a time. `ParquetStorageFilterSuite` passes. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code with Claude Opus 5.5 -- 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]
