HyukjinKwon opened a new pull request, #57023: URL: https://github.com/apache/spark/pull/57023
### What changes were proposed in this pull request? [SPARK-56642](https://issues.apache.org/jira/browse/SPARK-56642) added pipelined JVM->Python UDF data transfer via `BasePythonRunner.createPipelinedDataIn` (`core/.../api/python/PythonRunner.scala`). That code reads the worker through the legacy blocking-socket API: ```scala worker.channel.socket().setSoTimeout(...) // idle-timeout detection worker.channel.socket().getInputStream // read side of the pipeline ``` For a **Unix domain** `SocketChannel`, `SocketChannel.socket()` throws `java.lang.UnsupportedOperationException` — a UDS channel has no `java.net.Socket` adapter, and `SO_TIMEOUT` does not apply to it. So whenever `spark.python.unix.domain.socket.enabled=true` is combined with pipelined execution (`spark.python.udf.pipelined.enabled=true`), every pipelined Arrow/Python-UDF evaluation fails: ``` java.lang.UnsupportedOperationException: Not supported at java.base/sun.nio.ch.SocketChannelImpl.socket(SocketChannelImpl.java:228) at org.apache.spark.api.python.BasePythonRunner.createPipelinedDataIn(PythonRunner.scala:447) at org.apache.spark.api.python.BasePythonRunner.compute(PythonRunner.scala:390) at ...ArrowEvalPythonEvaluatorFactory.evaluate(ArrowEvalPythonExec.scala:209) ``` This PR makes the pipelined read path UDS-aware, mirroring the existing `if (!isUnixDomainSock)` handling already used elsewhere in `PythonRunner` (the sync-mode `ReaderInputStream` reads the channel directly and does idle-timeout via selector waits, not `SO_TIMEOUT`): - In UDS mode, read straight from the channel with `Channels.newInputStream(worker.channel)` and skip the `SO_TIMEOUT`-based idle-timeout wrapper (a blocking channel stream never raises `SocketTimeoutException`, so the idle-timeout branch is simply never taken — matching sync mode's behavior for UDS). - In TCP mode the behavior is unchanged. ### Why are the changes needed? The scheduled `Build / Unix Domain Socket (JDK 17, Scala 2.13)` (`build_uds`) master lane is red: `PYSPARK_UDS_MODE=true` turns on Unix domain sockets, and the pipelined UDF path then throws `UnsupportedOperationException` for every Arrow/Python UDF, failing the `pyspark-sql` module (12 errors + 1 downstream assertion failure) and the docs generation. Unix domain sockets are a supported transport, so this is a real regression, not a flake. ### Does this PR introduce _any_ user-facing change? No — internal JVM<->Python transport only. It restores correct behavior when Unix domain sockets and pipelined UDF execution are both enabled. ### How was this patch tested? `pyspark.sql.tests.pandas.test_pipelined_udf` (forces `spark.python.udf.pipelined.enabled=true`) and `pyspark.sql.tests.arrow.test_arrow_python_udf` were run with `PYSPARK_UDS_MODE=true` — i.e. exactly the pipelined + Unix-domain-socket combination that was broken — on GitHub Actions: - **Before (apache/spark `master`, `build_uds`):** `pyspark-sql` FAILED with `UnsupportedOperationException: Not supported` at `createPipelinedDataIn(PythonRunner.scala:447)` — https://github.com/apache/spark/actions/runs/28488010762 - **After (this change, fork validation):** PASS — https://github.com/HyukjinKwon/spark/actions/runs/28762213817 — `test_pipelined_udf` (pipelined mode asserted active) + `test_arrow_python_udf` both PASS under `PYSPARK_UDS_MODE=true` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
