comphead commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4338820605
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -153,22 +219,32 @@ impl MemoryPool for CometFairMemoryPool {
) -> Result<(), DataFusionError> {
if additional > 0 {
let mut state = self.state.lock();
- let num = state.consumers.len();
- let limit = self
- .pool_size
- .checked_div(num)
- .expect("overflow in checked_div");
- let consumer_used = *state.consumer_used(reservation);
- if limit < consumer_used.saturating_add(additional) {
- return resources_err!(
- "Failed to acquire {additional} bytes where this consumer
already holds {consumer_used} bytes and the fair limit is {limit} bytes, {num}
registered ({} bytes overcommitted)",
- self.spark.overcommit()
- );
- }
- // The shares alone do not bound the pool's total, because a
consumer keeps what it
- // reserved before another consumer registered.
+ let can_spill = reservation.consumer().can_spill();
let used = state.used;
- if self.pool_size < used.saturating_add(additional) {
+ if can_spill {
Review Comment:
DataFusion 55.2 registers `OrderedFinalAggregateStream[{partition}]` with
`with_can_spill(true)` even when its input is sorted on every grouping key.
`ordered_final_stream.rs` line 289 calls this a workaround for
apache/datafusion#17334. That stream builds a spill context only for
`InputOrderMode::PartiallySorted` (line 337), so in the fully sorted case a
refused `try_resize` is returned as the query error (line 459). Comet reaches
it through `CometSortAggregateExec` when the native sort runs in the same plan,
which is what the `orderedByGroupingKeys` comment in `operators.scala` relies
on. `SpillReplayPool` does not step in either, because the table is the
consumer's only reservation.
On main its limit was `pool_size` divided by all registered consumers. With
this change it is `(task_share - unspillable) / num_spill`, shared with the
sort below it and any shuffle writer. So a consumer with no spill path now gets
the limit meant for consumers that can spill, and a refusal fails the task. A
consumer that cannot spill would instead get Spark's grant. Could the pool
treat `OrderedFinalAggregateStream[` as unable to spill? `is_final_aggregate`
in `spill_replay.rs` already matches that name. The partially sorted variant
would then still spill when Spark refuses it. If not, the 1.2.0 migration note,
which now names the sort and the hash join build side, could name this
aggregate too. I have not run a fully sorted native sort aggregate under a
small share to confirm the refusal.
--
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]
