viirya commented on code in PR #59282: URL: https://github.com/apache/spark/pull/59282#discussion_r4232927623
########## sql/core/src/main/scala/org/apache/spark/sql/execution/python/RowQueue.scala: ########## @@ -181,8 +181,13 @@ private[python] case class DiskRowQueue( * * HybridRowQueue could be safely appended in one thread, and pulled in another thread in the same * time. + * + * This is intentionally not a case class. As a memory consumer it must use identity equality: + * otherwise two queues created with the same arguments in one task are equal, so only one of + * them is tracked (and offered for spilling) by the [[TaskMemoryManager]], and each refuses to + * spill when the other one triggers it. */ -case class HybridRowQueue( +class HybridRowQueue( Review Comment: Thanks, confirmed. `close()` freed pages through `TaskMemoryManager.freePage`, which synchronizes on the TMM, while holding the queue's lock. That is the reverse of the spill path. In pipelined mode, one eval node's queue-close listener runs before the writer of an earlier-registered node is joined, so the two orders can overlap. Fixed in 4643c8f as suggested: `close()` now detaches the queue list under the lock and closes the detached queues after releasing it. I also added a regression test that uses latches and an overridden `freePage` to force the two lock orders to overlap. It times out without the fix. -- 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]
