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: [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]
