andygrove opened a new issue, #7088:
URL: https://github.com/apache/datafusion-comet/issues/7088
### Describe the bug
`CometFairMemoryPool` limits each memory consumer of a task to `pool_size /
registered consumers`, where `pool_size`
is `spark.memory.offHeap.size` times `spark.comet.exec.memoryPool.fraction`,
which is the executor's whole off-heap
pool. Spark limits each task to an even share of that pool among the running
tasks. With 8 tasks running, a task
with fewer than 8 consumers has a fair limit above its Spark share, so the
fair limit never binds, and the first
consumer to reserve can take the task's whole share. A consumer that needs
memory later is refused by Spark, and
nothing asks the first one to spill (#3873).
DataFusion's sorts are the common victims: `ExternalSorterMerge` cannot
spill, and a sort fails the task when it
cannot get the memory to merge.
TPC-H and TPC-DS at scale factor 100 on one machine, Spark 4.1.1 standalone,
two executors with 8 cores each, with
Comet's defaults (sort-merge joins):
| Run | Off-heap per executor, shuffle partitions | Failed
task attempts, all retried successfully |
| ------------------ | ----------------------------------------- |
---------------------------------------------- |
| TPC-H, 22 queries | 2 GiB, 32 | 12, in
q3, q5 and q10 |
| TPC-H, 22 queries | 1 GiB, 16 | 11, in q9
and q10 |
| TPC-DS, 99 queries | 2 GiB, 32 | 15, in
q4, q14, q78 and q97 |
Every one was a sort refused memory for its merge while another sort of the
same task held most of the task's
share, either in its buffer or, for a sort that fit in memory, in the merge
reservation that then holds its rows.
For example, TPC-H q9 at 1 GiB:
```
Additional allocation failed for ExternalSorterMerge[0] with top memory
consumers (across reservations) as:
ExternalSorter[0]#188(can spill: true) consumed 133.4 MB, peak 134.5 MB,
SMJStream[0]#195(can spill: false) consumed 0.0 B, peak 0.0 B,
ExternalSorterMerge[0]#192(can spill: false) consumed 0.0 B, peak 0.0 B,
ShuffleRepartitioner[0]#202(can spill: true) consumed 0.0 B, peak 0.0 B,
ExternalSorter[0]#191(can spill: true) consumed 0.0 B, peak 0.0 B,
ExternalSorterMerge[0]#189(can spill: false) consumed 0.0 B, peak 65.1 MB.
Error: Failed to acquire 65536 bytes plus 0 bytes overcommitted, only got 0
bytes. Reserved: 139885176 bytes
```
The retries succeed because a sort's need is bounded and a retried task
often finds a larger share. A consumer
that can spill and keeps whatever it is granted fails every retry instead.
The spilling shuffled hash join in #7057
is one: with it, TPC-H q21 at 1 GiB fails, because the join takes 124 MB of
the task's 130 MB for its build side
and a sort below its probe side then cannot get the 10 MiB it needs to merge.
### Steps to reproduce
1. Generate TPC-H at scale factor 100 as Parquet.
2. Start a Spark standalone cluster with two executors of 8 cores each.
3. Run q9 with `benchmarks/tpc/run.py` and an engine config that adds
`spark.memory.offHeap.size=1g` and
`spark.sql.shuffle.partitions=16` to Comet's defaults.
4. The driver log shows lost tasks failing with `Additional allocation
failed for ExternalSorterMerge[0]`, and Spark
retries them.
### Expected behavior
A consumer that can spill is held to a fair share of its task's memory, so
that a consumer that cannot spill, such
as a sort's merge, gets the memory it needs while consumers that can spill
hold memory they could give up. Tasks
should not fail for lack of memory in that case, on the first attempt or on
a retry.
### Additional context
Sizing the pool to one task's share alone, with
`spark.comet.exec.memoryPool.fraction=0.125` for 8 cores, makes the
fair limit bind, but it also limits consumers that cannot spill. At 1 GiB,
q16's broadcast hash join then fails at
a limit of 64 MiB, a sort's merge in q2 at 26.8 MB with 5 consumers
registered, and q17 takes 33 s instead of 26 s.
What does work is to do what DataFusion's `FairSpillPool` does, over the
task's share: limit only the consumers that
can spill, each to an even share of what the task's share leaves after the
consumers that cannot spill, and let the
consumers that cannot spill take what Spark grants. That second half is
#5465, which asks for evidence from a real
workload before changing the policy. A prototype of it, with the pool set to
one task's share through
`spark.comet.exec.memoryPool.fraction=0.125`, on TPC-H at scale factor 100
with 1 GiB per executor and 16 shuffle
partitions:
| Pool | Sort-merge join |
Forced, spilling hash join (#7057) |
| ------------------------------------- | ------------------------- |
---------------------------------- |
| `fair_unified` today | 272.1 s, 11 tasks retried | q21
fails |
| Task's share, fair over all consumers | q2 and q16 fail | q16
fails |
| Prototype | 277.6 s, no task retried | 248.1
s, no task retried |
Every query returned the same rows, and no task was retried at any of the
three sizes. The prototype costs about 2%
here, mostly in q17, which spills more and takes 30 s instead of 26 s. With
2 GiB and 32 partitions, the sort-merge
join takes 274.6 s instead of 268.0 s, with no task retried instead of 12,
and with 16 GiB and 200 partitions,
256.7 s instead of 255.5 s. #1369 is a reminder that fair-pool policy can
cost far more than that, so a real version
needs the same A/B on more workloads.
A real version would also size the share from the executor's task slots
instead of the deprecated fraction, and
might let a task use more while fewer tasks are running, as Spark does.
#3873 notes that a pool must not hold its
lock across a JNI call once Spark can ask it to shrink, and the prototype
keeps the existing locking.
<details>
<summary>Prototype</summary>
```diff
--- a/native/core/src/execution/memory_pools/fair_pool.rs
+++ b/native/core/src/execution/memory_pools/fair_pool.rs
@@ -39,6 +39,9 @@
struct CometFairPoolState {
used: usize,
+ /// Bytes held by consumers that cannot spill, and how many consumers
can spill.
+ unspillable: usize,
+ num_spill: usize,
/// Bytes held by each registered consumer, keyed by
[`MemoryConsumer::id`]. The sibling
/// reservations that `new_empty()`, `split()` and `take()` create
belong to the same consumer,
/// so they count against one share. The pool keeps these totals itself
because
@@ -75,6 +78,8 @@
pool_size,
state: Mutex::new(CometFairPoolState {
used: 0,
+ unspillable: 0,
+ num_spill: 0,
consumers: HashMap::new(),
}),
}
@@ -106,13 +111,20 @@
}
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() {
+ state.num_spill += 1;
+ }
}
fn unregister(&self, consumer: &MemoryConsumer) {
// DataFusion unregisters a consumer after its last reservation has
dropped and released
// its bytes. If a release panicked, this runs while unwinding, so
it must not panic too.
- self.state.lock().consumers.remove(&consumer.id());
+ let mut state = self.state.lock();
+ if state.consumers.remove(&consumer.id()).is_some() &&
consumer.can_spill() {
+ state.num_spill = state.num_spill.saturating_sub(1);
+ }
}
/// Records memory that already exists, so it must not fail and ignores
the fair and pool
@@ -125,6 +137,9 @@
let mut state = self.state.lock();
self.spark.acquire(additional);
state.used = state.used.saturating_add(additional);
+ if !reservation.consumer().can_spill() {
+ state.unspillable =
state.unspillable.saturating_add(additional);
+ }
let consumer_used = state.consumer_used(reservation);
*consumer_used = consumer_used.saturating_add(additional);
}
@@ -142,6 +157,9 @@
.release(subtractive)
.unwrap_or_else(|_| panic!("Failed to release {subtractive}
bytes"));
state.used = state.used.checked_sub(subtractive).unwrap();
+ if !reservation.consumer().can_spill() {
+ state.unspillable =
state.unspillable.saturating_sub(subtractive);
+ }
*state.consumer_used(reservation) -= subtractive;
}
}
@@ -153,27 +171,29 @@
) -> 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 used = state.used;
- if self.pool_size < used.saturating_add(additional) {
- return resources_err!(
- "Failed to acquire {additional} bytes where {used}
bytes already reserved ({} bytes overcommitted) and the pool limit is {} bytes",
- self.spark.overcommit(),
- self.pool_size
- );
+ let can_spill = reservation.consumer().can_spill();
+ // As DataFusion's FairSpillPool: consumers that can spill
share what the
+ // ones that cannot leave of the pool, and the ones that cannot
are limited by Spark only.
+ if can_spill {
+ let num = state.num_spill.max(1);
+ let limit =
self.pool_size.saturating_sub(state.unspillable) / num;
+ 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} spillable 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 used = state.used;
+ if self.pool_size < used.saturating_add(additional) {
+ return resources_err!(
+ "Failed to acquire {additional} bytes where {used}
bytes already reserved ({} bytes overcommitted) and the pool limit is {} bytes",
+ self.spark.overcommit(),
+ self.pool_size
+ );
+ }
}
// A partial grant is handed back and refused, which triggers
spilling in the caller.
@@ -190,6 +210,9 @@
.used
.checked_add(additional)
.expect("overflow in checked_add");
+ if !can_spill {
+ state.unspillable += additional;
+ }
*state.consumer_used(reservation) += additional;
}
Ok(())
```
</details>
Related: #5465 (account for spillable consumers), #3873 (Spark cannot ask
native operators to spill), #6187
(documentation of the fair pool and the fraction).
### Willingness to contribute
I can contribute a fix for this bug independently
--
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]
