viirya opened a new pull request, #59282: URL: https://github.com/apache/spark/pull/59282
### What changes were proposed in this pull request? `HybridRowQueue` is changed from a case class to a regular class so that it uses identity equality. A companion `apply(memManager, tempDir, numFields, serMgr, lockFree = false)` keeps all existing construction sites source-compatible. No code uses `copy`, pattern matching or the case-class field accessors. ### Why are the changes needed? `HybridRowQueue` is a `MemoryConsumer` (via `HybridQueue`), but as a case class its `equals`/`hashCode` compare the constructor arguments. Two queues created in the same task with the same `TaskMemoryManager`, temp dir, number of fields, `SerializerManager` and `lockFree` flag are therefore equal, which breaks memory management: 1. `TaskMemoryManager` tracks consumers in a `HashSet`. A second queue that equals an already registered one is never added, so it is never offered for spilling when another consumer needs memory, and its memory is missing from the per-consumer breakdown reported on OOM. 2. `HybridQueue.spill(size, trigger)` returns 0 when `trigger == this`, which calls `equals`. A queue therefore refuses to spill when an equal but distinct queue triggers the spill. This happens in real queries. For example, with `coalesce`, `CoalescedRDD` evaluates several parent partitions in one task, and each one creates its own queue in `EvalPythonEvaluatorFactory` with the same width and local dir. These queues are only closed at task completion. Both sides of a sort-merge join with Python UDF eval nodes of the same input width are another example. This was found during the review of #58978. ### Does this PR introduce _any_ user-facing change? No, apart from the fix: the extra queues are now spilled under memory pressure and show up in the memory usage breakdown. ### How was this patch tested? Added two tests to `RowQueueSuite`, both of which fail without the fix: - `hybrid queues with the same fields are distinct memory consumers`: memory pressure from another consumer must be able to spill the second of two equal-argument queues (without the fix, the requesting consumer gets 0 bytes). - `hybrid queue spills for another queue with the same fields`: `spill` triggered by a distinct queue with the same arguments must release memory (without the fix, it returns 0). ``` build/sbt 'sql/testOnly *RowQueueSuite *PythonForeachWriterSuite' ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) -- 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]
