You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/10/17 17:34:34 UTC
[PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
andygrove opened a new pull request, #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096
## Which issue does this PR close?
Closes #7088.
Closes #5465.
#5465 asked for this accounting of the consumers that can spill once a real workload showed that it was needed, and #7088 is that workload.
## Rationale for this change
`fair_unified` limited each memory consumer to `pool_size / registered consumers`, where `pool_size` is the executor's whole off-heap pool. Spark limits each task to an even share of that pool among the running tasks, so whenever more than one task was running, a consumer's share was more than Spark would ever grant its task, and it never bound. The first consumer to reserve could take the task's whole share. A consumer that cannot spill, such as a sort's merge, was then refused by Spark, and nothing asks the first one to spill (#3873). At SF100 this showed up as failed task attempts that Spark retried, in TPC-H q3, q5, q9 and q10 and TPC-DS q4, q14, q78 and q97, and with the spilling hash join from #7057 as a failure of TPC-H q21 on every attempt. #7088 has the details, and a prototype of this fix measured on TPC-H SF100.
## What changes are included in this PR?
`CometFairMemoryPool` now does what DataFusion's `FairSpillPool` does, over the task's share of the off-heap pool:
- A consumer that can spill is limited to an even share of what the consumers that cannot spill leave of the task's share, and the task's total to the task's share. These are the two checks the pool already made, against the task's share instead of the whole pool, and with only the consumers that can spill dividing it.
- A consumer that cannot spill takes what Spark grants. The pool refuses it only when the task's total would pass `pool_size`, which Spark's own limit on the task is at least as tight as unless the deprecated `spark.comet.exec.memoryPool.fraction` is below `1 / N`.
- The task's share is `spark.memory.offHeap.size` divided by the number of tasks on the executor that are running native plans, or `pool_size` if that is smaller. A new `RunningTasks` count in `task_shared.rs` counts a task for as long as it holds its task-shared pool, of either type. Like Spark's own divisor it follows tasks as they start and finish, so a task that runs alone still has the whole pool, and with every core busy the share is the one the prototype in #7088 got from `spark.comet.exec.memoryPool.fraction=0.125` on 8 cores. `CometExecIterator.getMemoryConfig` now passes `spark.memory.offHeap.size` to `createPlan` next to `memory_limit`.
Dividing `pool_size` itself by the running tasks would have been simpler, but then the fraction would cap the executor instead of each task, and every suite that sets it to force a spill would get a pool that depends on how many tasks happen to overlap. One of them failed that way: five tasks left a sort a share smaller than one of its batches. With the cap on each task, those suites keep the pool they ask for, and the ones that use the fraction to run a native writer out of memory still do. One test needed a smaller fraction: in `CometExecIteratorLifecycleSuite`, the sort's merge reservation no longer halves the sort's share, so the sort no longer spilled in the 4 MiB pool it was given. It now gets 2 MiB.
Docs: the fair pool section of the memory management guide, the tuning guide, the fraction's config doc, a 1.2.0 upgrade note, and the `review-comet-memory-pr` skill, which described the old checks.
`iceberg_write.rs` and `delta_dv.rs` registered one consumer per task rather than one per file so that a widening write would not shrink every other consumer's share. Their consumers cannot spill, so that reason no longer holds, and the comments no longer give it. The Delta test's pool mirrored the old admission check, and now only counts registrations, which is what the test checks.
## How are these changes tested?
New unit tests, each of which fails without the change:
- `consumers_that_can_spill_share_their_tasks_share_of_the_pool` is the starvation from the issue. With eight tasks running and Spark granting the task 130 bytes, the first of two sorts is refused at 65 bytes, so the other sort's merge gets the rest. Before, the first sort could take all 130.
- `a_consumer_that_cannot_spill_is_limited_only_by_the_pool`, `consumers_that_can_spill_share_what_the_others_leave` and `memory_recorded_for_a_consumer_that_cannot_spill_comes_out_of_the_shares` cover the `FairSpillPool` accounting through `try_grow`, `grow` and `shrink`.
- `a_tasks_share_follows_the_number_of_running_tasks`, `a_tasks_share_is_at_most_the_pool`, `a_task_shares_the_pool_with_the_other_tasks_running_native_plans`, `the_fraction_limits_a_task_that_shares_the_off_heap_pool` and `a_task_runs_until_its_last_plan_releases_the_pool` cover the task's share, the count behind it, and how `create_pool` wires them. `fair_unified_is_sized_from_the_limit_and_the_off_heap_size` and a new `CometExecIteratorLifecycleSuite` test cover the off-heap size on its way from `getMemoryConfig` to the pool.
The existing fair pool tests now use consumers that can spill, since they test the shares, and so do the `SpillReplayPool` tests, as DataFusion's final aggregates do.
The suites that force native spills or run a native writer out of memory pass on the default Spark 4.1 profile: `CometAggregateSuite`, `CometTaskMetricsSuite`, `CometParquetWriterSuite`, `CometTaskMemoryManagerSuite`, `CometExecIteratorLifecycleSuite`, `CometNativeShuffleSuite`, `CometExecSuite` and `CometIcebergWriteActionSuite`, 672 tests. `cargo test -p datafusion-comet --lib` passes, and so does clippy with Rust 1.99.1, the CI's stable.
I ran TPC-H SF100 against main (a0cbb0fa0) with the setup from #7088: Spark 4.1.1 standalone with two executors of 8 cores, one `spark-submit` per query, sort-merge joins and Comet's defaults otherwise. Main and this PR alternated, with two rounds at 1 GiB. Every failed task attempt on main was a sort's merge refused memory, `Additional allocation failed for ExternalSorterMerge`, and each query returned the same rows in every run.
| Off-heap per executor, shuffle partitions | main | this PR |
| --- | --- | --- |
| 1 GiB, 16, first round | 276.5 s, 26 task attempts failed and retried | 278.4 s, none |
| 1 GiB, 16, second round | q9 failed: one of its tasks failed all 4 attempts | 278.1 s, none |
| 2 GiB, 32 | 270.7 s, 8 task attempts failed and retried | 275.9 s, none |
| 16 GiB, 200 | 257.1 s, none | 261.0 s, none |
A sort is now held to its share of the task's memory where it used to take everything Spark granted, so where the shares bind it spills more: 90 to 93 GiB against main's 87 GiB at 1 GiB, 56 GiB against 39 GiB at 2 GiB, and 1.4 GiB against none at 16 GiB. That costs 1 to 2% here. Most of it is q17, 30 s instead of 26 s at 1 GiB, as the prototype in #7088 measured, and q8 at 16 GiB, 20 s instead of 18 s.
The same suite on a 28-core Mac in `local[16,4]`, which is one executor of 16 cores, with 2 GiB of off-heap memory and 16 partitions, so each task has the share it has at 1 GiB above:
| Round | main | this PR |
| --- | --- | --- |
| first | 250.2 s, 60 task attempts failed and retried | 217.0 s, none |
| second | 246.7 s, 24 task attempts failed and retried | 216.0 s, none |
There the retries cost main more than the extra spilling costs this PR, mostly in q9, 42 s instead of 25 s, and q10.
At low concurrency, which #5465 asked about, the change makes no difference that I could measure. TPC-H SF10 in `local[2,4]` with 2 GiB of off-heap memory and 20 partitions, as #6205 measured it, took 94.4 s and 97.3 s on main and 94.8 s and 89.7 s with this PR. Every run spilled 1.5 GiB, and no task attempt failed.
TPC-DS SF100 with the same setup at 2 GiB and 32 partitions took 978.4 s on main, with 17 task attempts failed and retried in q4, q11, q14 and q78, and 990.3 s with this PR, with none. It spilled 61 GiB against 42 GiB. Every query returned the same rows except q39a and q39b, whose floating point columns differ in the last bits, by at most 4.4e-16 relative. Two runs of main differ in the same way, by up to 2.2e-16.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4338747669
##########
docs/source/contributor-guide/memory_management.md:
##########
@@ -311,34 +312,50 @@ JNI, which goes through Spark's ordinary `TaskMemoryManager`. That means:
is refused unless Spark can cover both, so operators spill until the debt is repaid. Both pools'
`Display` output and their `try_grow` errors report the current overcommit.
-`CometFairMemoryPool` additionally applies two local checks before it asks Spark, and refuses the
+`CometFairMemoryPool` exists because Spark cannot ask a native operator to spill. Spark limits each
+task to an even share of its off-heap pool among the tasks that hold memory from it. Once one
+operator holds its task's share, Spark refuses the task's other operators, and one that cannot
+spill, such as a sort's merge or the build side of a hash join, fails the task. The pool keeps
+memory for those operators by limiting the ones that can spill to a share of the task's, as
+DataFusion's `FairSpillPool` does over its whole pool
+([issue #7088](https://github.com/apache/datafusion-comet/issues/7088)).
+
+The task's share is `off_heap_size` divided by the number of tasks on the executor that are running
+native plans, that is, that hold a task-shared pool of either type, or `pool_size` if that is
+smaller, which it is only while the deprecated fraction is below `1 / N`. Like Spark's own divisor,
+the number follows tasks as they start and finish, so a task that runs alone has the whole pool.
+Spark's count differs in two ways: it includes tasks that run no native plan, and it counts a task
Review Comment:
Done in 6a1c4dffca. The guide now says which way each difference pushes: tasks that run no native plan but hold off-heap memory make Comet divide by fewer tasks than Spark, so #7088 can still happen in mixed stages, and tasks that hold a native pool but no memory make Comet divide by more, so sorts spill earlier than Spark requires. It also covers the off-heap storage region that dwsmith1983 pointed out.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#issuecomment-6360340063
Thanks @comphead. Both review-body items are in 6303b8e39:
- `ORDERED_FINAL_AGGREGATE` in `spill_replay.rs` is the one copy of the prefix, used by `is_final_aggregate` and the fair pool. The module doc's list of DataFusion names to re-check on an upgrade now includes `OrderedFinalAggregateStream` registering as spillable, with a note to drop the fair pool's exception once apache/datafusion#17334 is fixed.
- `spillable(name, pool)` lives once in `memory_pools/testing.rs`, and the `fair_pool.rs`, `spill_replay.rs`, `mod.rs` and `iceberg_write.rs` tests use it.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4338748585
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -31,14 +32,31 @@ use parking_lot::Mutex;
/// A DataFusion fair `MemoryPool` implementation for Comet. Internally this is
/// implemented via delegating calls to [`crate::jvm_bridge::CometTaskMemoryManager`].
+///
+/// Spark limits each task to an even share of its off-heap pool among the tasks that hold memory
+/// from it, and it cannot ask a native consumer to spill. So, as DataFusion's `FairSpillPool` does
+/// over its whole pool, this pool limits each consumer that can spill to an even share of what the
Review Comment:
Done in 6a1c4dffca. The doc comment on `CometFairMemoryPool` and the guide both give `task_share` and `limit` as a formula block, with the prose reduced to defining the terms.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4357393616
##########
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:
Agreed, the exception went one step too far. In 6303b8e39 it applies only where `try_grow` picks its check (`limited_to_its_share`). `register`, `unregister`, `grow`, `shrink` and the `unspillable` bookkeeping go back to `can_spill()`, so the aggregate counts in `num_spill`, its bytes are not `unspillable`, and a sort below it stays at half of the task's share. I added your sort-first case as `a_sort_below_an_ordered_final_aggregate_leaves_it_a_share`: with `pool(&fake, 100)` the sort is refused at 51, then the sort's 50 and the aggregate's 50 both succeed. It fails with the bookkeeping at d79cb838b. `an_ordered_final_aggregate_is_limited_as_a_consumer_that_cannot_spill` passes either way, as you said. `CometAggregateSuite` passes on 4.1.
##########
native/core/src/execution/memory_pools/mod.rs:
##########
@@ -145,6 +179,71 @@ mod tests {
}
}
+ #[test]
+ fn a_task_shares_the_pool_with_the_other_tasks_running_native_plans() {
+ let tasks = RunningTasks::default();
+ let fake = spark_memory::fake::FakeSpark::with(usize::MAX);
+ let fair = MemoryPoolConfig::new(MemoryPoolType::FairUnified, 100);
+ let greedy = MemoryPoolConfig::new(MemoryPoolType::GreedyUnified, 0);
+ let pool = create_pool(&fair, -3004, &tasks, || fake.memory());
+ let other_task = create_pool(&greedy, -3005, &tasks, || fake.memory());
+ let sort = MemoryConsumer::new("ExternalSorter[0]")
+ .with_can_spill(true)
+ .register(&pool);
+
+ // A task with the other pool type counts too.
+ sort.try_grow(50).unwrap();
+ assert!(sort.try_grow(1).is_err());
+
+ // Once it finishes, the task has the pool to itself.
+ drop(other_task);
+ sort.try_grow(50).unwrap();
+ assert!(sort.try_grow(1).is_err());
+ }
+
+ /// The tasks of an executor share one count, which `create_memory_pool` takes from
+ /// [`RunningTasks::of_executor`]. A pool created with a count of its own would see only its
+ /// own task and give it the whole pool. No other test counts tasks in the executor's count.
+ #[test]
+ fn the_tasks_of_an_executor_share_one_count() {
Review Comment:
Done in 6303b8e39. `the_tasks_of_an_executor_share_one_count` now creates the other task with the `greedy_unified` config, so it covers both the executor-wide count and the mixed pool types, and `a_task_shares_the_pool_with_the_other_tasks_running_native_plans` is gone.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4321652975
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -153,22 +209,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 {
+ let (task_share, tasks) = self.task_share();
+ // `num_spill` counts this consumer.
+ let limit = task_share.saturating_sub(state.unspillable) / state.num_spill.max(1);
Review Comment:
This ceiling, `(task_share - unspillable) / num_spill`, can be well below what Spark would grant the task. DataFusion 55.2's `ExternalSorter` fails instead of spilling in a few spots:
- `reserve_memory_for_batch_and_maybe_spill` when `in_mem_batches` is empty
- the same function's retry after a spill
- the `try_resize` in `sort_batch_stream`
Each returns "Not enough memory to continue external sort". The description mentions hitting exactly this in a suite ("five tasks left a sort a share smaller than one of its batches"). Keeping the fraction as a per-task cap fixed the suite, but the same thing can happen in production. Take 16 cores, 1 GiB of off-heap and three or four spillable operators in a task. Each one gets tens of MiB at most, and one batch of wide string rows can be bigger than that. On main, that sort could use the task's whole Spark grant.
Is this a trade-off you have weighed? If it is accepted, could the 1.2.0 migration note say that a sort can now fail when a single batch exceeds its share, and point to `greedy_unified` or a larger off-heap size as the way out?
##########
native/core/src/execution/operators/iceberg_write.rs:
##########
@@ -851,8 +851,7 @@ impl ExecutionPlan for IcebergWriteExec {
context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
let metrics = WriteMetrics::new(&self.metrics, partition);
- // One consumer for the whole task, however many files it opens: a consumer per file
- // would shrink every other consumer's share of the fair pool as a fanout write widened.
+ // One consumer for the whole task, however many files it opens.
Review Comment:
The fanout path does respond to a refusal. `reserve()` writes out and closes the partitions holding the most until the resize fits. With this change an unspillable consumer is held to no share, so a fanout write grows until Spark refuses, and every byte it holds is subtracted from the shares of the task's spillable operators. In a plan that sorts and writes in the same task, the sort ends up spilling where the writer could have closed files instead.
Should the fanout writer register `with_can_spill(true)`? The unpartitioned and clustered writers would stay unspillable, since a refusal is fatal for them. Either way, the new sentence at `memory_management.md:454` ("They cannot spill") contradicts the paragraph right above it, which describes the fanout writer closing files on refusal.
##########
spark/src/main/scala/org/apache/spark/Plugins.scala:
##########
@@ -322,7 +322,7 @@ object CometDriverPlugin extends Logging {
// spark.comet.exec.memoryPool.fraction was documented as holding back part of the off-heap pool
// for the native memory that Comet does not reserve. It cannot: Spark hands out the whole pool
- // to the tasks that ask for it, and the fraction only caps each task's consumers under
+ // to the tasks that ask for it, and the fraction only limits the operators that can spill under
Review Comment:
"the fraction only limits the operators that can spill" is not quite accurate. `pool_size` also caps the task's total for unspillable consumers, through the `else if self.pool_size < used.saturating_add(additional)` branch in `try_grow`. Something like "the fraction only caps each task's total under fair_unified" would match the code.
##########
docs/source/contributor-guide/memory_management.md:
##########
@@ -311,34 +312,50 @@ JNI, which goes through Spark's ordinary `TaskMemoryManager`. That means:
is refused unless Spark can cover both, so operators spill until the debt is repaid. Both pools'
`Display` output and their `try_grow` errors report the current overcommit.
-`CometFairMemoryPool` additionally applies two local checks before it asks Spark, and refuses the
+`CometFairMemoryPool` exists because Spark cannot ask a native operator to spill. Spark limits each
+task to an even share of its off-heap pool among the tasks that hold memory from it. Once one
+operator holds its task's share, Spark refuses the task's other operators, and one that cannot
+spill, such as a sort's merge or the build side of a hash join, fails the task. The pool keeps
+memory for those operators by limiting the ones that can spill to a share of the task's, as
+DataFusion's `FairSpillPool` does over its whole pool
+([issue #7088](https://github.com/apache/datafusion-comet/issues/7088)).
+
+The task's share is `off_heap_size` divided by the number of tasks on the executor that are running
+native plans, that is, that hold a task-shared pool of either type, or `pool_size` if that is
+smaller, which it is only while the deprecated fraction is below `1 / N`. Like Spark's own divisor,
+the number follows tasks as they start and finish, so a task that runs alone has the whole pool.
+Spark's count differs in two ways: it includes tasks that run no native plan, and it counts a task
Review Comment:
Could this also say what each difference does? Spark's `ExecutionMemoryPool` counts a task from its first acquire until its balance returns to zero (`memoryForTask.remove`).
- When tasks that run no native plan hold off-heap memory, Comet divides by fewer tasks than Spark does. The spillable shares then add up to more than Spark will grant, so #7088 can still happen in mixed JVM and native stages.
- When tasks hold a native pool but no memory, for example a native scan feeding JVM shuffle, Comet divides by more. Sorts then spill earlier than Spark requires.
That would tell the next reader which way to look when a mixed workload misbehaves.
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -31,14 +32,31 @@ use parking_lot::Mutex;
/// A DataFusion fair `MemoryPool` implementation for Comet. Internally this is
/// implemented via delegating calls to [`crate::jvm_bridge::CometTaskMemoryManager`].
+///
+/// Spark limits each task to an even share of its off-heap pool among the tasks that hold memory
+/// from it, and it cannot ask a native consumer to spill. So, as DataFusion's `FairSpillPool` does
+/// over its whole pool, this pool limits each consumer that can spill to an even share of what the
Review Comment:
Could this doc comment, and the matching paragraph in the guide, state the limit as a formula as well? For example:
```text
task_share = min(off_heap_size / running_tasks, pool_size)
limit = (task_share - unspillable) / num_spill
```
The prose version nests "an even share of what ... leave of the task's share, which is an even share of ..." and takes a few reads to parse. The guide already uses a formula block for `memory_limit`.
##########
native/core/src/execution/delta_dv.rs:
##########
@@ -1837,21 +1808,16 @@ mod tests {
}
/// One `attach_access_plans` call must add exactly one consumer to the task's pool no
- /// matter how many DV'd files it attaches. `CometFairMemoryPool` divides the pool by the
- /// number of registered consumers, and the reservations live in the returned files until
- /// the task ends, so one consumer per file would lower every other native operator's fair
- /// limit for the whole task even when the DV bytes themselves are tiny. Three DV'd files
- /// must leave a later consumer (a hash join build, say) its half of the pool. Each file's
- /// bytes must still return to the pool when that file alone drops, while the shared
- /// registration stays until the last file is gone.
+ /// matter how many DV'd files it attaches. Each file's bytes must still return to the pool
Review Comment:
Now that the fair-share reason is gone, is there still a reason to pin one consumer per call rather than one per file? Keeping `TrackConsumersPool`'s top-10 list readable, or keeping the pool's per-consumer map small, might be one. If so, could this comment and the one on `attach_access_plans` say so? Otherwise the test asserts "not one per file" without anything explaining why that matters.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4338746703
##########
native/core/src/execution/operators/iceberg_write.rs:
##########
@@ -851,8 +851,7 @@ impl ExecutionPlan for IcebergWriteExec {
context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
let metrics = WriteMetrics::new(&self.metrics, partition);
- // One consumer for the whole task, however many files it opens: a consumer per file
- // would shrink every other consumer's share of the fair pool as a fanout write widened.
+ // One consumer for the whole task, however many files it opens.
Review Comment:
Agreed, and thanks for the catch. The fanout writer now registers with `with_can_spill(true)` in 6a1c4dffca, through a small `write_consumer` helper, so it is held to a share and closes files rather than growing until Spark refuses. The unpartitioned and clustered writers stay unspillable. A new test, `only_a_fanout_write_is_held_to_a_share_of_the_pool`, checks all three modes against a sort in the same task, and the guide now describes the two cases separately, which removes the contradiction.
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -153,22 +209,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 {
+ let (task_share, tasks) = self.task_share();
+ // `num_spill` counts this consumer.
+ let limit = task_share.saturating_sub(state.unspillable) / state.num_spill.max(1);
Review Comment:
This is a trade-off I accepted rather than weighed carefully enough, so thank you for spelling out the paths. I kept the policy as benchmarked and documented it instead: the 1.2.0 migration note, the tuning guide and the contributor guide now say a sort can fail with "Not enough memory to continue external sort" when one batch is larger than its share, and point to more off-heap memory, a smaller `spark.comet.batchSize` or `greedy_unified`. I did not add an exemption for a consumer that holds nothing, since that changes behavior the benchmarks did not measure. Would you want it in this PR, or as a follow-up once there is a repro?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4354915418
##########
native/core/src/execution/memory_pools/mod.rs:
##########
@@ -145,6 +179,71 @@ mod tests {
}
}
+ #[test]
+ fn a_task_shares_the_pool_with_the_other_tasks_running_native_plans() {
+ let tasks = RunningTasks::default();
+ let fake = spark_memory::fake::FakeSpark::with(usize::MAX);
+ let fair = MemoryPoolConfig::new(MemoryPoolType::FairUnified, 100);
+ let greedy = MemoryPoolConfig::new(MemoryPoolType::GreedyUnified, 0);
+ let pool = create_pool(&fair, -3004, &tasks, || fake.memory());
+ let other_task = create_pool(&greedy, -3005, &tasks, || fake.memory());
+ let sort = MemoryConsumer::new("ExternalSorter[0]")
+ .with_can_spill(true)
+ .register(&pool);
+
+ // A task with the other pool type counts too.
+ sort.try_grow(50).unwrap();
+ assert!(sort.try_grow(1).is_err());
+
+ // Once it finishes, the task has the pool to itself.
+ drop(other_task);
+ sort.try_grow(50).unwrap();
+ assert!(sort.try_grow(1).is_err());
+ }
+
+ /// The tasks of an executor share one count, which `create_memory_pool` takes from
+ /// [`RunningTasks::of_executor`]. A pool created with a count of its own would see only its
+ /// own task and give it the whole pool. No other test counts tasks in the executor's count.
+ #[test]
+ fn the_tasks_of_an_executor_share_one_count() {
Review Comment:
This test and `a_task_shares_the_pool_with_the_other_tasks_running_native_plans` above have the same body: a 100-byte pool, a second task, the sort refused at 50, then 50 more once the other task drops. If `other_task` here were created with the `greedy_unified` config, one test would cover both the executor-wide count and the mixed pool types, and the one above could go.
##########
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:
Thanks for the quick turn. Looking at `d79cb838b`, I think my suggestion went one step too far. Taking the aggregate out of `num_spill` also lifts the cap on the sort below it. In a `CometSortAggregateExec` over a native sort, the sort reads all of its input before the aggregate's first request. At `6a1c4dffca` the sort was held to half of what the merge leaves, so the aggregate kept the other half. Now the sort can hold `task_share` less its merge reservation. When it fits in memory it keeps holding that while it emits (`in_mem_sort_stream` in DataFusion 55.2). The aggregate's headroom is then about the `sort_spill_reservation_bytes` that `sort()` frees before emitting, plus whatever Spark grants past the share. That is the #17334 shape that DataFusion's flag works around.
Could the exception apply only where `try_grow` picks its branch, while `register`, `unregister`, `grow`, `shrink` and the `unspillable` bookkeeping keep `can_spill()`? The aggregate would still count in `num_spill`, so the sort leaves it a share, but it would not be refused at that share. It would also be one call site instead of five, and your new test passes either way.
A test that grows the sort first would pin it. With `pool(&fake, 100)`, the proposed version refuses `sort.try_grow(51)`, and `aggregate.try_grow(50)` succeeds after `sort.try_grow(50)`. At the head, `sort.try_grow(90)` succeeds and the `pool_size` check then refuses `aggregate.try_grow(11)`. I worked these through `try_grow` by hand and did not run 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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4345560907
##########
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:
Good catch. I checked DataFusion 55.2 and it is as you describe. `OrderedFinalAggregateStream` registers with `with_can_spill(true)` as the #17334 workaround, builds a spill context only for `PartiallySorted` input, and returns the refusal as the error otherwise.
The pool now treats it as a consumer that cannot spill. `limited_as_spillable` in `fair_pool.rs` is what `register`, `unregister`, `grow`, `shrink` and `try_grow` now ask instead of `can_spill()`. It returns false for `OrderedFinalAggregateStream[`, so that stream takes Spark's grant, and its bytes come out of the shares of the consumers that can spill. The partially sorted stream still spills when Spark refuses it.
`an_ordered_final_aggregate_is_limited_as_a_consumer_that_cannot_spill` has a sort and the aggregate in one task. The aggregate takes 60 of the task's 100 bytes, and the sort is then limited to the 40 left. With `limited_as_spillable` reduced to `can_spill()`, the aggregate is refused at 50 and the test fails. The contributor guide names the exception next to the paragraph on consumers that cannot spill. (d79cb838b)
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -203,15 +282,213 @@ impl MemoryPool for CometFairMemoryPool {
#[cfg(test)]
mod tests {
use super::super::spark_memory::fake::FakeSpark;
+ use super::super::task_shared::RunningTask;
use super::*;
use std::sync::Arc;
+ /// A pool of `pool_size` bytes, all of the off-heap pool as with the default fraction, which
+ /// `tasks` share, connected to `fake`.
+ fn shared_pool(
+ fake: &Arc<FakeSpark>,
+ pool_size: usize,
+ tasks: &RunningTasks,
+ ) -> Arc<dyn MemoryPool> {
+ Arc::new(CometFairMemoryPool::with_spark(
+ fake.memory(),
+ pool_size,
+ pool_size,
+ tasks.clone(),
+ ))
+ }
+
+ /// A pool of `pool_size` bytes for a task that runs alone, connected to `fake`.
+ fn pool(fake: &Arc<FakeSpark>, pool_size: usize) -> Arc<dyn MemoryPool> {
+ shared_pool(fake, pool_size, &RunningTasks::default())
+ }
+
+ /// Starts `n` tasks, which run until the returned guards drop.
+ fn start(tasks: &RunningTasks, n: usize) -> Vec<RunningTask> {
+ (0..n).map(|_| tasks.start()).collect()
+ }
+
+ /// A consumer that can spill, like a sort's buffer.
+ fn spillable(name: &str, pool: &Arc<dyn MemoryPool>) -> MemoryReservation {
+ MemoryConsumer::new(name)
+ .with_can_spill(true)
+ .register(pool)
+ }
+
+ #[test]
+ fn consumers_that_can_spill_share_their_tasks_share_of_the_pool() {
+ // Eight running tasks share a 1040-byte pool, and Spark grants this one its 130 bytes.
+ let fake = FakeSpark::with(130);
+ let tasks = RunningTasks::default();
+ let _running = start(&tasks, 8);
+ let pool = shared_pool(&fake, 1040, &tasks);
+ let sort = spillable("ExternalSorter[0]", &pool);
+ let _other_sort = spillable("ExternalSorter[1]", &pool);
+ let other_merge = MemoryConsumer::new("ExternalSorterMerge[1]").register(&pool);
+
+ // The two sorts share the task's 130 bytes, so the first is refused, and spills, at 65.
+ sort.try_grow(65).unwrap();
+ assert!(sort.try_grow(1).is_err());
+ // That leaves the rest of what Spark grants the task for the other sort's merge, which
+ // cannot spill.
+ other_merge.try_grow(65).unwrap();
+ assert_eq!(fake.held(), 130);
+ }
+
+ #[test]
+ fn a_consumer_that_cannot_spill_is_limited_only_by_the_pool() {
+ let fake = FakeSpark::with(usize::MAX);
+ let tasks = RunningTasks::default();
+ let _running = start(&tasks, 4);
+ let pool = shared_pool(&fake, 400, &tasks);
+ let _sort = spillable("ExternalSorter[0]", &pool);
+ let _other_sort = spillable("ExternalSorter[1]", &pool);
+ let merge = MemoryConsumer::new("ExternalSorterMerge[0]").register(&pool);
+
+ // Neither a share of the task's 100 bytes nor the task's share itself limits the merge.
+ merge.try_grow(400).unwrap();
+ // The pool does, but only binds before Spark does when the deprecated
+ // `spark.comet.exec.memoryPool.fraction` is below 1.0.
+ assert!(merge.try_grow(1).is_err());
+ assert_eq!(fake.held(), 400);
+ }
+
+ #[test]
+ fn consumers_that_can_spill_share_what_the_others_leave() {
+ let fake = FakeSpark::with(usize::MAX);
+ let pool = pool(&fake, 100);
+ let first = spillable("ExternalSorter[0]", &pool);
+ let second = spillable("ExternalSorter[1]", &pool);
+ let merge = MemoryConsumer::new("ExternalSorterMerge[0]").register(&pool);
+ merge.try_grow(40).unwrap();
+
+ // The two sorts share the 60 bytes that the merge leaves.
+ first.try_grow(30).unwrap();
+ assert!(first.try_grow(1).is_err());
+
+ // Their shares grow as the merge releases memory.
+ merge.shrink(40);
+ first.try_grow(20).unwrap();
+ assert!(first.try_grow(1).is_err());
+ second.try_grow(50).unwrap();
+ assert!(second.try_grow(1).is_err());
+ assert_eq!(pool.reserved(), 100);
+ }
+
+ #[test]
+ fn memory_recorded_for_a_consumer_that_cannot_spill_comes_out_of_the_shares() {
+ let fake = FakeSpark::with(usize::MAX);
+ let pool = pool(&fake, 100);
+ let sort = spillable("ExternalSorter[0]", &pool);
+ let _other_sort = spillable("ExternalSorter[1]", &pool);
+ // Like a batch read back from a spill file, which already exists.
+ let reader = MemoryConsumer::new("spill reader").register(&pool);
+ reader.grow(40);
+
+ // The two sorts share the 60 bytes left.
+ sort.try_grow(30).unwrap();
+ assert!(sort.try_grow(1).is_err());
+ }
+
+ #[test]
+ fn a_tasks_share_follows_the_number_of_running_tasks() {
+ let fake = FakeSpark::with(usize::MAX);
+ let tasks = RunningTasks::default();
+ let mut running = start(&tasks, 4);
+ let pool = shared_pool(&fake, 400, &tasks);
+ let sort = spillable("ExternalSorter[0]", &pool);
+
+ // Four running tasks share the 400-byte pool.
+ sort.try_grow(100).unwrap();
+ assert!(sort.try_grow(1).is_err());
+
+ // Once two of them finish, the other two share it.
+ running.truncate(2);
+ sort.try_grow(100).unwrap();
+ assert!(sort.try_grow(1).is_err());
+
+ // A task that starts takes part of the share back, so the sort spills before it grows.
+ running.push(tasks.start());
+ assert!(sort.try_grow(1).is_err());
+ sort.shrink(150);
+ sort.try_grow(83).unwrap();
+ assert!(sort.try_grow(1).is_err());
+ }
+
+ #[test]
+ fn the_tasks_total_is_checked_against_the_tasks_share_not_the_pool() {
+ // One task has the whole 200-byte pool, and its sort takes half.
+ let fake = FakeSpark::with(usize::MAX);
+ let tasks = RunningTasks::default();
+ let _running = tasks.start();
+ let pool = shared_pool(&fake, 200, &tasks);
+ let first = spillable("ExternalSorter[0]", &pool);
+ first.try_grow(100).unwrap();
+
+ // Another task starts, which halves the task's share to the 100 bytes that the sort holds.
+ // A second sort is under its own limit of 50 bytes, but the task's total is at its share.
+ let _other_task = tasks.start();
+ let second = spillable("ExternalSorter[1]", &pool);
+ let error = second.try_grow(50).unwrap_err().to_string();
+ assert!(error.contains("the task's share is 100 bytes"), "{error}");
+
+ // A consumer that cannot spill is limited by the pool, not by the task's share.
+ let merge = MemoryConsumer::new("ExternalSorterMerge[1]").register(&pool);
+ merge.try_grow(100).unwrap();
+ assert_eq!(pool.reserved(), 200);
+ }
+
+ /// The trade-off that the migration guide describes: a consumer that cannot spill can take
+ /// what Spark grants, which can be the task's whole share, and then a consumer that can spill
+ /// is refused its first request, so a sort fails where it used to run.
+ #[test]
+ fn a_consumer_that_can_spill_is_refused_once_the_others_hold_the_tasks_share() {
+ let fake = FakeSpark::with(usize::MAX);
+ let tasks = RunningTasks::default();
+ let _running = start(&tasks, 4);
+ let pool = shared_pool(&fake, 400, &tasks);
+ // Like a shuffled hash join's build side, which Spark lets grow past `off_heap_size / 4`
+ // while the other tasks hold nothing yet.
+ let build = MemoryConsumer::new("HashJoinInput[0]").register(&pool);
+ build.try_grow(100).unwrap();
+ let sort = spillable("ExternalSorter[0]", &pool);
+
+ let error = sort.try_grow(1).unwrap_err().to_string();
+ assert!(error.contains("the fair limit is 0 bytes"), "{error}");
+
+ // The sort can reserve again once the build side gives memory back.
+ build.shrink(60);
+ sort.try_grow(60).unwrap();
+ assert!(sort.try_grow(1).is_err());
+ }
+
+ #[test]
+ fn a_tasks_share_is_at_most_the_pool() {
Review Comment:
It is, thanks. Dropped. The `mod.rs` test covers the same case through `create_pool`. (d79cb838b)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#issuecomment-6335120601
Thanks @comphead. Both threads are addressed in d79cb838b, which also merges main and resolves the conflict in the migration guide. The pool now limits a fully ordered `OrderedFinalAggregateStream` as a consumer that cannot spill, and the duplicate test is gone.
@viirya and @comphead, I added `run-iceberg-tests` and `run-delta-tests` so the Iceberg Spark SQL and Delta contrib suites report before the merge queue.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4362552466
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -153,22 +209,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 {
+ let (task_share, tasks) = self.task_share();
+ // `num_spill` counts this consumer.
+ let limit = task_share.saturating_sub(state.unspillable) / state.num_spill.max(1);
Review Comment:
A follow-up is fine with me. The guides now name the failure and the workarounds, and `a_consumer_that_can_spill_is_refused_once_the_others_hold_the_tasks_share` pins the shape. Could you or @dwsmith1983 open the tracking issue now rather than waiting for a repro, and link it from the tuning paragraph in `memory_management.md`? Otherwise it only gets filed if someone happens to hit it, and the next person who sees "Not enough memory to continue external sort" under `fair_unified` has nowhere to land.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#issuecomment-6319895582
Thanks @dwsmith1983 and @viirya. The review feedback is in 6a1c4dffca.
For dwsmith1983's points:
- **A sort refused while another consumer holds the share.** I added `a_consumer_that_can_spill_is_refused_once_the_others_hold_the_tasks_share`, which pins the shape you describe, and named it in the migration note, the tuning guide and the contributor guide. I did not let a consumer that holds nothing fall through to Spark, because that changes behavior the benchmarks did not measure. I am happy to add it if you or viirya think it should go in here.
- **The executor-wide count.** `create_memory_pool` now goes through `create_executor_pool`, and `the_tasks_of_an_executor_share_one_count` fails if that passes a fresh `RunningTasks`.
- **The task total against `pool_size`.** `the_tasks_total_is_checked_against_the_tasks_share_not_the_pool` follows your sketch and fails if the check uses `pool_size`.
- **The churn test.** It now shares one count across the eight threads and asserts it is 0 afterwards. I confirmed it fails when the drop skips its `fetch_sub`.
- **The off-heap storage region.** It is in the guide next to the two task-count differences.
- **The review skill.** It now says `getMemoryConfig` also passes `off_heap_size`.
I checked the new tests by breaking the line each one guards. `cargo test` for `iceberg_write`, `delta_dv` and `memory_pools` passes, clippy is clean, and `CometIcebergWriteActionSuite` passes 111 of 111 on Spark 4.1.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4367296384
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -153,22 +209,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 {
+ let (task_share, tasks) = self.task_share();
+ // `num_spill` counts this consumer.
+ let limit = task_share.saturating_sub(state.unspillable) / state.num_spill.max(1);
Review Comment:
> Could you or @dwsmith1983 open the tracking issue now rather than waiting for a repro, and link it from the tuning paragraph in `memory_management.md`?
Opened #7197. #5613 links it from the tuning paragraph (1a5b53d22), since it edits that guide anyway.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#issuecomment-6371277821
Thanks @viirya
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4338749022
##########
native/core/src/execution/delta_dv.rs:
##########
@@ -1837,21 +1808,16 @@ mod tests {
}
/// One `attach_access_plans` call must add exactly one consumer to the task's pool no
- /// matter how many DV'd files it attaches. `CometFairMemoryPool` divides the pool by the
- /// number of registered consumers, and the reservations live in the returned files until
- /// the task ends, so one consumer per file would lower every other native operator's fair
- /// limit for the whole task even when the DV bytes themselves are tiny. Three DV'd files
- /// must leave a later consumer (a hash join build, say) its half of the pool. Each file's
- /// bytes must still return to the pool when that file alone drops, while the shared
- /// registration stays until the last file is gone.
+ /// matter how many DV'd files it attaches. Each file's bytes must still return to the pool
Review Comment:
Yes, there is still one. The pool and `TrackConsumersPool` each keep an entry per registered consumer, and its out-of-memory report lists the ten consumers holding the most. A partition can have thousands of DV'd files, so one consumer per file would grow both maps with the file count and fill that report with this one operator. The comments on `attach_access_plans` and on the test say so now (6a1c4dffca).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4338820622
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -203,15 +282,213 @@ impl MemoryPool for CometFairMemoryPool {
#[cfg(test)]
mod tests {
use super::super::spark_memory::fake::FakeSpark;
+ use super::super::task_shared::RunningTask;
use super::*;
use std::sync::Arc;
+ /// A pool of `pool_size` bytes, all of the off-heap pool as with the default fraction, which
+ /// `tasks` share, connected to `fake`.
+ fn shared_pool(
+ fake: &Arc<FakeSpark>,
+ pool_size: usize,
+ tasks: &RunningTasks,
+ ) -> Arc<dyn MemoryPool> {
+ Arc::new(CometFairMemoryPool::with_spark(
+ fake.memory(),
+ pool_size,
+ pool_size,
+ tasks.clone(),
+ ))
+ }
+
+ /// A pool of `pool_size` bytes for a task that runs alone, connected to `fake`.
+ fn pool(fake: &Arc<FakeSpark>, pool_size: usize) -> Arc<dyn MemoryPool> {
+ shared_pool(fake, pool_size, &RunningTasks::default())
+ }
+
+ /// Starts `n` tasks, which run until the returned guards drop.
+ fn start(tasks: &RunningTasks, n: usize) -> Vec<RunningTask> {
+ (0..n).map(|_| tasks.start()).collect()
+ }
+
+ /// A consumer that can spill, like a sort's buffer.
+ fn spillable(name: &str, pool: &Arc<dyn MemoryPool>) -> MemoryReservation {
+ MemoryConsumer::new(name)
+ .with_can_spill(true)
+ .register(pool)
+ }
+
+ #[test]
+ fn consumers_that_can_spill_share_their_tasks_share_of_the_pool() {
+ // Eight running tasks share a 1040-byte pool, and Spark grants this one its 130 bytes.
+ let fake = FakeSpark::with(130);
+ let tasks = RunningTasks::default();
+ let _running = start(&tasks, 8);
+ let pool = shared_pool(&fake, 1040, &tasks);
+ let sort = spillable("ExternalSorter[0]", &pool);
+ let _other_sort = spillable("ExternalSorter[1]", &pool);
+ let other_merge = MemoryConsumer::new("ExternalSorterMerge[1]").register(&pool);
+
+ // The two sorts share the task's 130 bytes, so the first is refused, and spills, at 65.
+ sort.try_grow(65).unwrap();
+ assert!(sort.try_grow(1).is_err());
+ // That leaves the rest of what Spark grants the task for the other sort's merge, which
+ // cannot spill.
+ other_merge.try_grow(65).unwrap();
+ assert_eq!(fake.held(), 130);
+ }
+
+ #[test]
+ fn a_consumer_that_cannot_spill_is_limited_only_by_the_pool() {
+ let fake = FakeSpark::with(usize::MAX);
+ let tasks = RunningTasks::default();
+ let _running = start(&tasks, 4);
+ let pool = shared_pool(&fake, 400, &tasks);
+ let _sort = spillable("ExternalSorter[0]", &pool);
+ let _other_sort = spillable("ExternalSorter[1]", &pool);
+ let merge = MemoryConsumer::new("ExternalSorterMerge[0]").register(&pool);
+
+ // Neither a share of the task's 100 bytes nor the task's share itself limits the merge.
+ merge.try_grow(400).unwrap();
+ // The pool does, but only binds before Spark does when the deprecated
+ // `spark.comet.exec.memoryPool.fraction` is below 1.0.
+ assert!(merge.try_grow(1).is_err());
+ assert_eq!(fake.held(), 400);
+ }
+
+ #[test]
+ fn consumers_that_can_spill_share_what_the_others_leave() {
+ let fake = FakeSpark::with(usize::MAX);
+ let pool = pool(&fake, 100);
+ let first = spillable("ExternalSorter[0]", &pool);
+ let second = spillable("ExternalSorter[1]", &pool);
+ let merge = MemoryConsumer::new("ExternalSorterMerge[0]").register(&pool);
+ merge.try_grow(40).unwrap();
+
+ // The two sorts share the 60 bytes that the merge leaves.
+ first.try_grow(30).unwrap();
+ assert!(first.try_grow(1).is_err());
+
+ // Their shares grow as the merge releases memory.
+ merge.shrink(40);
+ first.try_grow(20).unwrap();
+ assert!(first.try_grow(1).is_err());
+ second.try_grow(50).unwrap();
+ assert!(second.try_grow(1).is_err());
+ assert_eq!(pool.reserved(), 100);
+ }
+
+ #[test]
+ fn memory_recorded_for_a_consumer_that_cannot_spill_comes_out_of_the_shares() {
+ let fake = FakeSpark::with(usize::MAX);
+ let pool = pool(&fake, 100);
+ let sort = spillable("ExternalSorter[0]", &pool);
+ let _other_sort = spillable("ExternalSorter[1]", &pool);
+ // Like a batch read back from a spill file, which already exists.
+ let reader = MemoryConsumer::new("spill reader").register(&pool);
+ reader.grow(40);
+
+ // The two sorts share the 60 bytes left.
+ sort.try_grow(30).unwrap();
+ assert!(sort.try_grow(1).is_err());
+ }
+
+ #[test]
+ fn a_tasks_share_follows_the_number_of_running_tasks() {
+ let fake = FakeSpark::with(usize::MAX);
+ let tasks = RunningTasks::default();
+ let mut running = start(&tasks, 4);
+ let pool = shared_pool(&fake, 400, &tasks);
+ let sort = spillable("ExternalSorter[0]", &pool);
+
+ // Four running tasks share the 400-byte pool.
+ sort.try_grow(100).unwrap();
+ assert!(sort.try_grow(1).is_err());
+
+ // Once two of them finish, the other two share it.
+ running.truncate(2);
+ sort.try_grow(100).unwrap();
+ assert!(sort.try_grow(1).is_err());
+
+ // A task that starts takes part of the share back, so the sort spills before it grows.
+ running.push(tasks.start());
+ assert!(sort.try_grow(1).is_err());
+ sort.shrink(150);
+ sort.try_grow(83).unwrap();
+ assert!(sort.try_grow(1).is_err());
+ }
+
+ #[test]
+ fn the_tasks_total_is_checked_against_the_tasks_share_not_the_pool() {
+ // One task has the whole 200-byte pool, and its sort takes half.
+ let fake = FakeSpark::with(usize::MAX);
+ let tasks = RunningTasks::default();
+ let _running = tasks.start();
+ let pool = shared_pool(&fake, 200, &tasks);
+ let first = spillable("ExternalSorter[0]", &pool);
+ first.try_grow(100).unwrap();
+
+ // Another task starts, which halves the task's share to the 100 bytes that the sort holds.
+ // A second sort is under its own limit of 50 bytes, but the task's total is at its share.
+ let _other_task = tasks.start();
+ let second = spillable("ExternalSorter[1]", &pool);
+ let error = second.try_grow(50).unwrap_err().to_string();
+ assert!(error.contains("the task's share is 100 bytes"), "{error}");
+
+ // A consumer that cannot spill is limited by the pool, not by the task's share.
+ let merge = MemoryConsumer::new("ExternalSorterMerge[1]").register(&pool);
+ merge.try_grow(100).unwrap();
+ assert_eq!(pool.reserved(), 200);
+ }
+
+ /// The trade-off that the migration guide describes: a consumer that cannot spill can take
+ /// what Spark grants, which can be the task's whole share, and then a consumer that can spill
+ /// is refused its first request, so a sort fails where it used to run.
+ #[test]
+ fn a_consumer_that_can_spill_is_refused_once_the_others_hold_the_tasks_share() {
+ let fake = FakeSpark::with(usize::MAX);
+ let tasks = RunningTasks::default();
+ let _running = start(&tasks, 4);
+ let pool = shared_pool(&fake, 400, &tasks);
+ // Like a shuffled hash join's build side, which Spark lets grow past `off_heap_size / 4`
+ // while the other tasks hold nothing yet.
+ let build = MemoryConsumer::new("HashJoinInput[0]").register(&pool);
+ build.try_grow(100).unwrap();
+ let sort = spillable("ExternalSorter[0]", &pool);
+
+ let error = sort.try_grow(1).unwrap_err().to_string();
+ assert!(error.contains("the fair limit is 0 bytes"), "{error}");
+
+ // The sort can reserve again once the build side gives memory back.
+ build.shrink(60);
+ sort.try_grow(60).unwrap();
+ assert!(sort.try_grow(1).is_err());
+ }
+
+ #[test]
+ fn a_tasks_share_is_at_most_the_pool() {
Review Comment:
This test looks like a copy of `the_fraction_limits_a_task_that_shares_the_off_heap_pool` in `mod.rs`: 100-byte pool, 1000-byte off-heap size, two running tasks, and the same `try_grow(100)` then refused `try_grow(1)` on a sort. The `mod.rs` version goes through `create_pool`, so it also checks the argument order and the consumer that cannot spill. Could this one go?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4338748172
##########
spark/src/main/scala/org/apache/spark/Plugins.scala:
##########
@@ -322,7 +322,7 @@ object CometDriverPlugin extends Logging {
// spark.comet.exec.memoryPool.fraction was documented as holding back part of the off-heap pool
// for the native memory that Comet does not reserve. It cannot: Spark hands out the whole pool
- // to the tasks that ask for it, and the fraction only caps each task's consumers under
+ // to the tasks that ask for it, and the fraction only limits the operators that can spill under
Review Comment:
Fixed in 6a1c4dffca, using your wording: the comment now says the fraction only caps each task's total under fair_unified.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: make fair_unified's per-consumer limit bind when tasks run concurrently [datafusion-comet]
Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.
dwsmith1983 commented on code in PR #7096:
URL: https://github.com/apache/datafusion-comet/pull/7096#discussion_r4338798415
##########
native/core/src/execution/memory_pools/fair_pool.rs:
##########
@@ -153,22 +209,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 {
+ let (task_share, tasks) = self.task_share();
+ // `num_spill` counts this consumer.
+ let limit = task_share.saturating_sub(state.unspillable) / state.num_spill.max(1);
Review Comment:
> Would you want it in this PR, or as a follow-up once there is a repro?
A follow-up. The guides now name the failure and the workarounds, and the new test pins the shape, so this can land as benchmarked. When I rebase #5613 onto it I'll try the shape under the TPC-H memory pressure setup from that PR, and open an issue if a sort fails there.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org