andygrove commented on PR #5214: URL: https://github.com/apache/datafusion-comet/pull/5214#issuecomment-5159170174
Follow-up to my review above: rather than reason about how tight the old cap was, I measured it. I temporarily instrumented `CometFairMemoryPool::register` to print the consumer count per task pool and ran two Comet queries with `spark.memory.offHeap.enabled=true`, `spark.memory.offHeap.size=512m`, `fair_unified`, on `local[5]` (instrumentation reverted afterwards). Consumers registering per task pool: | Plan | Peak `num` | Consumers | |---|---|---| | broadcast join + group-by + shuffle | 3 | `HashJoinInput[0]`, `GroupedHashAggregateStream[0]`, `ShuffleRepartitioner[0]` | | final sort stage | 2 | `ExternalSorter[0]`, `ExternalSorterMerge[0]` | | 2x SMJ + window + `countDistinct` | 7 | two join inputs, the multi-phase distinct-agg chain, `ShuffleRepartitioner[0]` | So `num` is 1-3 for ordinary stages and reaches ~7 on a wide one. Worth noting a single sort costs two consumers (`sorts/sort.rs:283,288`). That lets us say where the old cap actually binds. The old code caps a task's *total* native usage at `offHeap.size / num`, while Spark's `ExecutionMemoryPool` independently caps each active task at roughly `pool / N_active_tasks`. Whichever is smaller wins: | Scenario | old Comet cap | Spark's grant | binding constraint | |---|---|---|---| | 8 cores packed, num=3 | 33% of pool | ~12.5% | Spark - old cap invisible | | 8 cores packed, num=7 | 14% | ~12.5% | Spark, marginally | | 16 cores packed, num=7 | 14% | ~6% | Spark - old cap invisible | | 1-2 active tasks, num=7 | 14% | up to ~100% | **Comet, ~7x tighter** | | 2 cores, num=7 | 14% | 50% | Comet, ~3.5x tighter | The conclusion I'd draw: the defect is real, but on a busy executor Spark's own per-task fair share is tighter than the buggy cap, so it is masked. It bites when consumers-per-task exceeds active-tasks-per-executor - stage tails, AQE-coalesced partitions, low partition counts, small executors - where Spark would hand a lone task most of the pool and this check clamps it to `1/num`. The expected symptom is unnecessary spilling on long-tail tasks rather than a uniform throughput loss, which I think is a more defensible framing for the PR than "the default config can only use 10% of its off-heap share". Caveat on my own numbers: the two probe queries were small enough that the check never actually fired (zero `fair limit` errors in the run), so this measures `num` and the resulting arithmetic, not a performance delta. If you want a reproducer that demonstrates the fix rather than arguing for it, the low-concurrency case is the one to target: `local[1]` with `spark.sql.shuffle.partitions=1`, a modest `offHeap.size`, and a wide join + distinct query should hit "fair limit is M bytes, 7 registered" on main and get further after the fix. Happy to build that out if it would help the review. -- 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]
