Qing Fu created SPARK-59601:
-------------------------------
Summary: DiskRowQueue holds its output stream open until first
read after spilling
Key: SPARK-59601
URL: https://issues.apache.org/jira/browse/SPARK-59601
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Qing Fu
{{HybridQueue.spill()}} converts every in-memory queue but the last into a
disk-backed queue, draining the rows into it. The new {{DiskRowQueue}} is
complete at that point -- only the last queue is ever written to -- but its
output stream is left open.
{{DiskRowQueue}} opens its writer eagerly in the constructor:
{code:scala}
private var out = new DataOutputStream(serMgr.wrapForEncryption(
new BufferedOutputStream(new FileOutputStream(file.toString))))
{code}
and only closes it on the queue's *first read*, inside {{remove()}}. So after N
spills a task holds N open {{DataOutputStream}} -> {{BufferedOutputStream}} ->
{{FileOutputStream}} chains (plus the encryption wrapper when I/O encryption is
on), each with its 8 KB buffer, until those queues are read. That is a file
descriptor and a buffer per spilled queue held at exactly the moment memory is
scarce -- spilling happens because the task is under memory pressure.
It is also observable as unflushed data: rows written into a spilled queue sit
in the output stream's buffer rather than on disk until the first read.
This is a follow-up on SPARK-53481, which extracted the generic {{HybridQueue}}
/ {{Queue}} classes.
h3. Proposed fix
Add a {{closeOutputStream()}} method to the {{Queue}} trait, defaulting to a
no-op (in-memory queues hold no write-only resource), override it in
{{DiskRowQueue}} to close and clear {{out}}, and call it from the spill loop
once a queue has been drained.
One subtlety: {{DiskRowQueue.remove()}} currently uses {{if (out != null)}} as
its "first read" signal and opens the input stream inside that branch. Once the
output stream can already be closed before the first read, that branch no
longer fires and {{remove()}} NPEs on {{in.readInt()}}. The check has to be
keyed off {{in == null}} instead.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
