andygrove opened a new pull request, #6125:
URL: https://github.com/apache/datafusion-comet/pull/6125

   ## Which issue does this PR close?
   
   No issue filed yet.
   
   ## Rationale for this change
   
   The native Parquet scan registers no memory reservation, so its working set 
is invisible to the task memory pool. With tracing on 
(`jemalloc,alloc-accounting`), a TPC-H SF100 run (2 executors x 8 cores, 16g 
off-heap) showed scan-only queries such as q1, q6 and q19 allocating 100 to 150 
MiB per executor natively while reserving almost nothing from the pool. The gap 
was flat for the whole query, ramping up as tasks started and back down as they 
finished. It is a fixed working set per running scan, not something that grows 
over time.
   
   That working set is, per partition, the compressed column chunks of the row 
group being decoded (arrow-rs fetches the projected column chunks of one row 
group before decoding it, and DataFusion reads one file per partition at a 
time), plus the decompressed pages and output batch the decoder builds. For q1 
on SF100 `lineitem`, the seven projected columns are 9.1 MiB compressed per row 
group, which matches the roughly 14 MiB per task that the trace showed 
unaccounted. The size is set by the file's layout (row-group size times 
projected columns), so wide tables with large row groups can hold far more per 
task than TPC-H does.
   
   ## What changes are included in this PR?
   
   All in Comet's `EagerPageIndexReaderFactory`, which already owns the scan's 
`AsyncFileReader`:
   
   - **Fetched data pages are charged exactly.** Each buffer returned by 
`get_byte_ranges` is wrapped in a `Bytes::from_owner` whose owner holds its 
share of a `MemoryReservation`. The reservation is released when arrow-rs drops 
the last reference to the buffer or any page sliced from it, so no change to 
the decoder is needed. One pool call is made per fetch, and the reservation is 
split per buffer.
   - **Decode buffers are charged as an estimate.** arrow-rs allocates 
decompressed pages and the output batch internally with no hook to observe 
them, so each open reader holds a fixed reservation of one default page (1 MiB) 
per projected leaf column plus one `batch_size` output batch. It is released 
when the reader is dropped.
   - **A refused reservation does not fail the scan.** Comet's unified pools 
panic on an infallible `grow` when Spark refuses memory, and a scan cannot 
spill, so the scan uses `try_grow` and keeps reading unaccounted when refused. 
The attempt still lets Spark ask the task's other consumers to spill first. 
Refused bytes are counted in a new `scan_io_unreserved_bytes` metric, surfaced 
as a Spark SQL metric and documented in the metrics guide.
   
   ## How are these changes tested?
   
   New unit tests in `eager_page_index_reader_factory.rs`:
   
   - the data-page reservation outlives the reader, follows slices, and returns 
to zero once every buffer is dropped;
   - a refused reservation returns the same data and counts the bytes in 
`scan_io_unreserved_bytes`;
   - the decode estimate counts leaf columns (including nested) and one batch;
   - the decode reservation is held exactly as long as the reader.
   
   Measured on TPC-H SF100, same commit with and without this change. The gap 
is the native allocated bytes minus the pool's reserved total at the same 
instant, per executor:
   
   | | Median gap | p95 gap |
   |---|---|---|
   | Before | 21.9 MiB | 140.2 MiB |
   | After | 7.9 MiB | 94.0 MiB |
   
   For scan-heavy queries the median gap drops from 71 to 127 MiB to 6 to 19 
MiB (q1 goes from 113 MiB to slightly over-reserved). What remains in the p95 
is spikes in joins and aggregations (q10, q13, q17, q18), which this change 
does not touch. Total run time was unchanged (208.7 s vs 208.8 s), and all 22 
result hashes match. Three alternating runs of q10 each way confirmed that an 
apparent q10 p95 increase in one run was noise.
   
   Runs under a reduced off-heap size, to exercise the refusal path and the 
effect on spilling, are in progress and will be added here.
   


-- 
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]

Reply via email to