[
https://issues.apache.org/jira/browse/SPARK-60414?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
L. C. Hsieh reassigned SPARK-60414:
-----------------------------------
Assignee: L. C. Hsieh
> 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: 4.3.0, 4.4.0
> Reporter: L. C. Hsieh
> Assignee: L. C. Hsieh
> Priority: Major
>
> 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
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
