dwsmith1983 commented on PR #6125:
URL:
https://github.com/apache/datafusion-comet/pull/6125#issuecomment-6115269080
> @dwsmith1983 you changed how native scan partitions are read in #6753, so
you know this area. If you have time, would you be able to review the memory
accounting here? Thanks.
Thanks for the ping. The points @comphead, @viirya and @sunchao raised still
apply at this head: the swallowed try_grow error, a consumer per reader, the
file-independent decode estimate, the stale memory_management.md (lines 394 and
480 still describe the scan as uncharged), the Scala metric list, and a way to
turn it off. A few things changed on main since the last commit that affect the
design.
grow rather than try_grow. The description gives the reason for try_grow as
the unified pools panicking on an infallible grow. Since #6128 they no longer
do: grow records the bytes and carries what Spark declines as overcommit, and
the next try_grow from a consumer that can spill pays it back. The fetched
pages already exist when they are charged, so this is not admission control.
With grow they would be charged, the pressure would land on consumers that can
spill, and both the error classification question and scan_io_unreserved_bytes
would go away. The trade off is that grow ignores the fair limit.
A sketch for reserve_buffers after merging main (on this branch's base grow
still unwraps try_grow), which also gets the fetch down to one grow and one
shrink by sharing the reservation instead of splitting it per buffer.
ReservedBytes::_reservation becomes an Arc<MemoryReservation>, the call site
drops the metrics argument, and the bytes are released when the last buffer of
the fetch drops rather than buffer by buffer, so
fetched_data_pages_hold_a_reservation_until_dropped and
refused_reservation_returns_data_unaccounted would change:
fn reserve_buffers(reservation: &MemoryReservation, buffers: Vec<Bytes>) ->
Vec<Bytes> {
let fetch = reservation.new_empty();
// Recorded even when Spark declines part of it; the shortfall is
carried as overcommit.
fetch.grow(buffers.iter().map(Bytes::len).sum());
let fetch = Arc::new(fetch);
buffers
.into_iter()
.map(|bytes| {
Bytes::from_owner(ReservedBytes {
bytes,
_reservation: Arc::clone(&fetch),
})
})
.collect()
}
If try_grow stays, only a refusal should let the scan keep reading. A spill
failure, a kill interrupt or a JNI failure arrives as External, and today both
get_byte_ranges and create_reader return Ok when the pool fails that way (a
pool whose try_grow returns DataFusionError::External shows it). Classifying
the error keeps the existing tests passing:
fn reserve_buffers(
reservation: &MemoryReservation,
buffers: Vec<Bytes>,
scan_io_metrics: &ScanIoMetrics,
) -> DFResult<Vec<Bytes>> {
let total = buffers.iter().map(Bytes::len).sum();
let fetch = reservation.new_empty();
match fetch.try_grow(total) {
Ok(()) => {}
Err(e) if matches!(e.find_root(),
DataFusionError::ResourcesExhausted(_)) => {
scan_io_metrics.unreserved_bytes.add(total);
return Ok(buffers);
}
Err(e) => return Err(e),
}
// split per buffer as now
}
// get_byte_ranges
Some(reservation) => reserve_buffers(reservation, bytes, &scan_io_metrics)
.map_err(|e| ParquetError::External(Box::new(e)))?,
The decode reservation in create_reader takes the same match inside its
closure and ends with .transpose()?.
Overcommit debt. If try_grow stays, each scan call also asks Spark for the
tas, since SparkMemory::try_acquire asks for additional plus the debt. The
scanbecomes the most frequent caller and the only one that cannot act on a
refusal, so under pressure it keeps asking for the whole debt, gets a short
grant, hands it back and reads anyway: a JNI round trip per row group, and
possibly a JVM spill that buys nothing.
One consumer per scan. #6247 registers the writers once per task because
everyers the share of every other one in the task. The same applies here.
Oneconsumer on the factory with new_empty() per reader would also keep the
count steady while DataFusion holds the second reader for Bloom filter pruning.
With a pool that counts registrations, three readers register three consumers
at this head. Storing the registration instead of the pool (memory_pool:
Option<Arc<dyn MemoryPool>> becomes reservation: Option<MemoryReservation>,
which is Send + Sync) makes it one:
// with_memory_pool: register once for the scan
self.reservation = Some(MemoryConsumer::new("ParquetScan").register(&pool));
// create_reader: every reader charges the same consumer
let reservation =
self.reservation.as_ref().map(MemoryReservation::new_empty);
Sizing the decode estimate. The factory already loads the footer and offset
index, so the estimate could come from the file: per projected leaf, the
smaller of one page and the largest selected column chunk's uncompressed size,
plus min(batch_size, num_rows) * row_width. Charging it at the first
get_byte_ranges rather than in create_reader keeps the JNI call out of the
planning step, and a file that is pruned entirely never pays it. Capping by
file size would not bound it, since decompressed pages can be larger than the
file.
Shrinks per page. With the page index and a row selection, parquet 59.3
fetches one range per selected page (scan_ranges_from_selectors), and on the
local path each comes back as its
own buffer. split per buffer then gives every page its own reservation and
itsrow group drops. One reservation shared by the whole fetch, as in the
sketchabove, keeps it at one grow and one shrink.
Release from the FFI callback. If a page buffer ever ends up aliased by an
expnk runs from the release callback, and a panic there cannot unwind across
extern "C", so it aborts the executor rather than failing the task. A Utf8View
hint in the file's ARROW:schema is kept unless the cast to the Spark type
copies, so that copy is what the
invariant depends on. Worth recording in the comment.
Two of the earlier points are mostly settled on main: partial grants log at
DEp since #6269, and since #6261 a blocked acquire hands the worker's other
tasksto another thread, though that thread still waits in Spark.
For the rebase, the conflicts are in eager_page_index_reader_factory.rs,
parquet_exec.rs and metrics.md, and the scan metric list in CometExecSuite is
now around line 2538.
--
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]
