This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 44432cbfdd9c [SPARK-58241][4.2][PYTHON] Fix the columnar Arrow Python
UDF runner writing the input schema in the legacy protocol position
44432cbfdd9c is described below
commit 44432cbfdd9cae414cdef1bdea247edb28c713b7
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Wed Jul 22 01:34:05 2026 -0700
[SPARK-58241][4.2][PYTHON] Fix the columnar Arrow Python UDF runner writing
the input schema in the legacy protocol position
### What changes were proposed in this pull request?
Backport of the Scala fix from #57403 (master commit e6d66abda0a) to
branch-4.2.
`ColumnarArrowPythonWithNamedArgumentRunner` still wrote `schema.json` as a
UTF string at the head of the UDF command section for `SQL_ARROW_BATCHED_UDF`.
SPARK-56340 -- which is in branch-4.2 -- moved that schema into `evalConf`
under the `"input_type"` key (branch-4.2's worker reads it from there, and
branch-4.2's row-based `ArrowPythonWithNamedArgumentRunner` already writes it
there), so the worker parses the command section as a UDF count followed by UDF
entries. The stray schema [...]
Scope differences from the master PR, both verified against branch-4.2:
- The Python-side fail-fast hardening is not applicable: the
message-receiver files (`pyspark/messages/`, `worker_message.py`) do not exist
on branch-4.2 (the message-based protocol refactor is 4.3+).
- The end-to-end test is not portable: it drives the columnar runner
through the Arrow-cached scan (SPARK-57268), which is 4.3+; branch-4.2 has no
in-tree producer of Arrow-backed columnar input, so the bug there is reachable
only via Arrow-backed DSv2 connectors.
### Why are the changes needed?
Both origins of the bug are in branch-4.2: SPARK-56340 (protocol migration,
reader side) and SPARK-56350 (the columnar runner with the legacy write). Any
Python UDF whose input plan produces Arrow-backed `ColumnarBatch` (e.g. an
Arrow-backed DSv2 connector) deadlocks on 4.2 exactly as on master.
### Does this PR introduce _any_ user-facing change?
No. It fixes a hang; the wire format now matches what the branch-4.2 worker
expects.
### How was this patch tested?
The fix mirrors the `evalConf` pattern already shipping in branch-4.2's
row-based runner. Compile and scalastyle pass on branch-4.2, and
`pyspark.sql.tests.arrow.test_arrow_python_udf` passes against the patched
build. The scenario-level end-to-end coverage lives in the master PR (#57403),
where the Arrow-cached scan provides an in-tree columnar Arrow input;
branch-4.2 has no in-tree way to construct that input.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
This pull request and its description were written by Claude Code.
Closes #57419 from viirya/SPARK-58241-branch-4.2.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../sql/execution/python/ColumnarArrowPythonRunner.scala | 16 ++++++++++++++--
1 file changed, 14 insertions(+), 2 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowPythonRunner.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowPythonRunner.scala
index dcc915d5021a..f08ee6ab7367 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowPythonRunner.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowPythonRunner.scala
@@ -51,10 +51,22 @@ private[python] class
ColumnarArrowPythonWithNamedArgumentRunner(
override protected def runnerConf: Map[String, String] = super.runnerConf ++
pythonRunnerConf
- override protected def writeUDF(dataOut: DataOutputStream): Unit = {
+ // The input schema travels in evalConf (key "input_type"), exactly like
+ // ArrowPythonWithNamedArgumentRunner. It must NOT be written into the UDF
command section:
+ // the worker reads that section as a UDF count followed by UDF entries, so
an extra UTF
+ // string there desyncs the whole command parse and the worker blocks
waiting for bytes that
+ // never arrive, hanging the task forever.
+ override protected def evalConf: Map[String, String] = {
if (evalType == PythonEvalType.SQL_ARROW_BATCHED_UDF) {
- PythonWorkerUtils.writeUTF(schema.json, dataOut)
+ super.evalConf ++ Map(
+ "input_type" -> schema.json
+ )
+ } else {
+ super.evalConf
}
+ }
+
+ override protected def writeUDF(dataOut: DataOutputStream): Unit = {
PythonUDFRunner.writeUDFs(dataOut, funcs, argMetas)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
