Gengliang Wang created SPARK-61195:
--------------------------------------
Summary: Stop the threads that read a task's input on their own
before its readers are closed
Key: SPARK-61195
URL: https://issues.apache.org/jira/browse/SPARK-61195
Project: Spark
Issue Type: Improvement
Components: SQL
Affects Versions: 5.0.0
Reporter: Gengliang Wang
Some operators read their input on a thread of their own, which the task does
not wait for: the writer thread of script transformation ({{TRANSFORM}}), the
stdin writer thread of {{PipedRDD}}, and the writer thread of pipelined Python
UDFs ({{spark.python.udf.pipelined.enabled}}). Such a thread can still read a
columnar batch after its task completed, while the completion listener of
{{DataSourceRDD}} closes the reader that returned the batch, for example under
a {{LIMIT}} or when the script exits early.
The readers of native data sources read the buffers of a batch at their
addresses (SPARK-60753). So a reader that is closed before the end of its
stream cannot free the buffers of its last batch at once: it leaves them to a
Cleaner, which frees them once the reader and the batch are unreachable, at a
garbage collection that this off-heap memory does not trigger. With
{{spark.sql.columnVector.offheap.enabled}}, the vectorized Parquet reader
instead frees its {{OffHeapColumnVector}}s at close, so such a thread can read
freed memory there.
Script transformation also does not destroy its process when its task
completes, unlike {{PipedRDD}}. Under a {{LIMIT}}, a script that keeps writing
output that nobody reads any more blocks once the pipe is full, and so does the
writer thread, which then keeps the thread, the process and the batch alive.
Proposal: have these operators stop their thread, and wait for it to end, from
a task completion listener that they register when they start the thread.
Listeners run in reverse order of registration, so this one runs before
{{DataSourceRDD}} closes the readers. Script transformation should destroy its
process without interrupting its writer thread: the interrupt makes
{{proc.waitFor()}} in the {{finally}} block of the writer throw
{{InterruptedException}}, which {{NonFatal}} does not catch, so that
{{SparkUncaughtExceptionHandler}} exits the executor. The failures that the
writer gets once its task completed should not be logged as errors.
{{NativePartitionReader.close()}} can then free the buffers of its last batch
at once, without the Cleaner.
Discussion: https://github.com/apache/spark/pull/59961#discussion_r4488238438,
https://github.com/apache/spark/pull/59961#discussion_r4500198812 and
https://github.com/apache/spark/pull/59961#discussion_r4500198806
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
