L. C. Hsieh created SPARK-60414:
-----------------------------------
Summary: Make the state shared by the writer and task threads
thread-safe in pipelined Python UDF execution
Key: SPARK-60414
URL: https://issues.apache.org/jira/browse/SPARK-60414
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: L. C. Hsieh
In pipelined Python UDF execution (spark.python.udf.pipelined.enabled, added in
4.3.0 by SPARK-56642, default false), BasePythonRunner pulls the input iterator
on a writer thread while the task thread reads the results. SPARK-56642 made
the row queues lock-free for this split (HybridRowQueue with lockFree =
pipelined), but two other pieces of state are still shared by the two threads
without synchronization:
1. EvalPythonUDTFExec: count is a local var captured by the input iterator
(count += 1 runs on the writer thread) and by the output iterator (if (count >
0) { queue.remove(); count -= 1 } runs on the task thread). A lost update makes
the UDTF output of the last input row be joined with the previous input row
without any error, or makes the first terminate() batch call queue.remove() on
an exhausted queue.
2. ColumnarArrowEvalPythonEvaluatorFactory.evalArrowColumnar: passThruQueue is
a plain java.util.ArrayDeque that the writer thread adds to and the task thread
polls. A poll racing with an add that grows the deque can return null
(scala.MatchError) or return a batch twice, pairing a result with the wrong
pass-through columns. It needs more than 16 batches in flight.
count should be split into a writer-only counter and a task-only one (or made
an AtomicLong), and passThruQueue should be a ConcurrentLinkedQueue.
Found in the review of SPARK-60367 (https://github.com/apache/spark/pull/59616).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
