rich7420 commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4363132257
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -84,39 +126,74 @@ impl CometFairMemoryPool {
pub(super) fn overcommit(&self) -> usize {
self.spark.overcommit()
}
+
+ /// The task's share of the off-heap pool and the number of tasks running
native plans that it
+ /// is an even share among, as Spark limits each task holding execution
memory to an even share
+ /// of its pool. `running_tasks` counts this pool's task once it holds the
pool.
+ fn task_share(&self) -> (usize, usize) {
+ let tasks = self.running_tasks.count().max(1);
+ ((self.off_heap_size / tasks).min(self.pool_size), tasks)
+ }
}
impl Display for CometFairMemoryPool {
fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
let state = self.state.lock();
write!(
f,
- "CometFairMemoryPool(pool_size={}, used={}, num={},
overcommit={})",
+ "CometFairMemoryPool(pool_size={}, off_heap_size={},
running_tasks={}, used={}, unspillable={}, num={}, num_spill={},
overcommit={})",
self.pool_size,
+ self.off_heap_size,
+ self.running_tasks.count(),
state.used,
+ state.unspillable,
state.consumers.len(),
+ state.num_spill,
self.spark.overcommit()
)
}
}
+/// Whether a request from `consumer` is held to its fair limit and the task's
share.
+///
+/// That is the case for every consumer that can spill, with one exception.
DataFusion 55.2
+/// registers `OrderedFinalAggregateStream` as able to spill even when its
input is sorted on every
+/// grouping key, to work around apache/datafusion#17334, but it builds a
spill path only for
+/// partially sorted input. Fully sorted, as in a `CometSortAggregateExec`
over a native sort, a
+/// refused request fails the query. So its requests are checked only against
the pool, like those
+/// of a consumer that cannot spill. The partially sorted stream still spills
when Spark refuses it.
+///
+/// Only the check differs. The aggregate still counts as a consumer that can
spill everywhere
+/// else, so it is one of `num_spill` and its bytes are not `unspillable`. A
sort below it then
+/// stays limited to its fair share and leaves the aggregate the rest.
+fn limited_to_its_share(consumer: &MemoryConsumer) -> bool {
+ consumer.can_spill() &&
!consumer.name().starts_with(ORDERED_FINAL_AGGREGATE)
+}
+
impl MemoryPool for CometFairMemoryPool {
fn name(&self) -> &str {
"CometFairMemoryPool"
}
fn register(&self, consumer: &MemoryConsumer) {
- self.state.lock().consumers.insert(consumer.id(), 0);
+ let mut state = self.state.lock();
+ state.consumers.insert(consumer.id(), 0);
+ if consumer.can_spill() {
Review Comment:
Could a fully ordered `OrderedPartialAggregateStream` (registered with
`can_spill=false`) also get a `num_spill` slot, like the final one, so a sort
below it leaves it some memory?
##########
docs/source/user-guide/latest/migration-guide.md:
##########
@@ -107,6 +107,32 @@ that still sets it to `true` keeps the split plan, with
iceberg-java writing the
on Spark 3.5+ it now also plans merge-on-read writes as the split plan. Unset
it to plan Spark's
own operator.
+### Memory Pool Limits
+
+The `fair_unified` memory pool, which is the default for
`spark.comet.exec.memoryPool`, limited each
+of a task's operators to a share of all of `spark.memory.offHeap.size`, while
Spark limits each task
+to an even share of it among the running tasks. Whenever more than one task
was running, the share
+was the looser limit, so one operator could take all of its task's memory, and
another operator
+that cannot spill, such as the final merge of a sort, then failed the task
+([#7088](https://github.com/apache/datafusion-comet/issues/7088)). Spark
retried such tasks, and the
+retries usually succeeded. Comet 1.2.0 divides `spark.memory.offHeap.size`
among the tasks running
+native plans, as Spark does, and within a task limits only the operators that
can spill, each to an
+even share of what the operators that cannot spill leave. So sorts,
aggregations and the shuffle
+writer can spill sooner when several of them run in one task, and the
operators that cannot spill
Review Comment:
#6279's window docs and OOM hint assume `fair_unified` limits the window to
an equal share, but here only Spark limits it. Could the PR that lands second
update them?
--
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]
