viirya opened a new pull request, #59413: URL: https://github.com/apache/spark/pull/59413
### What changes were proposed in this pull request? Backport of #59282 to branch-4.1. `HybridQueue` now defines final identity `equals`/`hashCode` and `spill` checks `trigger eq this`. `HybridRowQueue` is a regular class instead of a case class, with a `toString` showing its width and identity hash, and a companion `apply` whose parameter names match the constructor. `HybridQueue.close()` detaches its queue list under the queue lock and frees the pages after releasing it. The cherry-pick needed two adaptations: - `HybridRowQueue` has no `lockFree` parameter on this branch, so `toString`, the companion `apply` and the `toString` assertion in the test omit it. - `RowQueueSuite` had an import conflict, because the exact-fit partial page test from master is not on this branch. ### Why are the changes needed? `HybridRowQueue` is a `MemoryConsumer`, but as a case class its `equals`/`hashCode` compared the constructor arguments, so two queues created in one task with the same arguments (for example several partitions evaluated by a Python UDF node in one task under `coalesce`) were equal: 1. `TaskMemoryManager` tracks consumers in a `HashSet`, so the second queue was never offered for spilling when another consumer needed memory, and its memory was missing from the per-consumer breakdown reported on OOM. 2. `spill(size, trigger)` returned 0 when `trigger == this`, so a queue refused to spill when an equal but distinct queue triggered the spill. Once equal-argument queues can spill for each other, they are also exposed to a lock-order inversion between `close()` (which freed pages through `TaskMemoryManager.freePage` while holding the queue lock) and the spill path (which holds the `TaskMemoryManager` monitor and then takes the queue lock), so the PR also fixes `close()`. ### 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, as `HybridRowQueue(numFields=...)@<identity hash>`. ### How was this patch tested? The three tests added in #59282 to `RowQueueSuite` pass on this branch, as do the existing `RowQueueSuite` and `PythonForeachWriterSuite` tests. ``` 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]
