This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new af6d5d50b71b [SPARK-56642][CORE][FOLLOWUP] Support Unix Domain Socket
mode in pipelined Python UDF data transfer
af6d5d50b71b is described below
commit af6d5d50b71b3493d49b88e84536687d090f52e4
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Mon Jul 6 12:27:05 2026 +0900
[SPARK-56642][CORE][FOLLOWUP] Support Unix Domain Socket mode in pipelined
Python UDF data transfer
### What changes were proposed in this pull request?
This makes the pipelined Python UDF read path work under Unix Domain Socket
(UDS) mode.
`BasePythonRunner.createPipelinedDataIn` (added in SPARK-56642 for
pipelined JVM↔Python UDF
data transfer) sets up the read side via
`worker.channel.socket().setSoTimeout(...)` and
`worker.channel.socket().getInputStream`. When Spark runs with Unix Domain
Sockets
(`spark.python.unix.domain.socket.enabled=true`), `worker.channel` is an
`AF_UNIX`
`SocketChannel`, whose `socket()` throws
`java.lang.UnsupportedOperationException: Not supported`
(a Unix-domain channel has no `java.net.Socket` adapter). Every pipelined
UDF task then 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)
```
This adds a UDS branch that reads straight from the channel via
`Channels.newInputStream(worker.channel)` (with no `SO_TIMEOUT`-based
idle-timeout detection, since
Unix-domain channels do not support `SO_TIMEOUT`). This mirrors how the
existing sync-mode server
loop already guards its `setSoTimeout` calls behind `!isUnixDomainSock` and
reads via
`Channels.newInputStream`. The default TCP path is unchanged (the new code
only runs when UDS is
enabled).
### Why are the changes needed?
`build_uds.yml` (the "Build / Unix Domain Socket" scheduled workflow, one
of the README CI badges)
is red: `pyspark.sql.tests.pandas.test_pipelined_udf` fails with `FAILED
(failures=1, errors=12)`
because pipelined UDFs cannot run at all under UDS mode.
Note: this is distinct from #56995 (SPARK-57931), which restores the
channel's blocking mode after
pipelined execution — a different bug in the same method. This PR fixes the
UDS `socket()` crash on
the read path.
### Does this PR introduce _any_ user-facing change?
No. It makes an existing feature (pipelined Python UDF) work under UDS
mode; behavior in the default
TCP mode is unchanged.
### How was this patch tested?
`pyspark.sql.tests.pandas.test_pipelined_udf` under UDS mode.
**Before (red — apache/spark `master`):**
- `build_uds` → `pyspark-sql` FAILED (`test_pipelined_udf`: `FAILED
(failures=1, errors=12)`):
https://github.com/apache/spark/actions/runs/28488010762
**After (green — with this change, fork verification):**
- `build_uds` (UDS mode) → `pyspark-sql` module PASSED:
https://github.com/HyukjinKwon/spark/actions/runs/28758332630
### Was this patch authored or co-authored using generative AI tooling?
Yes.
Co-authored-by: Isaac
Closes #57024 from HyukjinKwon/ci-fix/agent2-uds-pipelined-udf.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit 17b11e3dde1d3d48d6352cb2442a5217131087e7)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../scala/org/apache/spark/api/python/PythonRunner.scala | 13 +++++++++++++
1 file changed, 13 insertions(+)
diff --git a/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
b/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
index ee15bd5e46ef..923df591cc2e 100644
--- a/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
+++ b/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
@@ -222,6 +222,10 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
protected val faultHandlerEnabled: Boolean =
conf.get(PYTHON_WORKER_FAULTHANLDER_ENABLED)
protected val idleTimeoutSeconds: Long =
conf.get(PYTHON_WORKER_IDLE_TIMEOUT_SECONDS)
protected val killOnIdleTimeout: Boolean =
conf.get(PYTHON_WORKER_KILL_ON_IDLE_TIMEOUT)
+ // Unix domain socket channels have no java.net.Socket adapter, so
socket()-based APIs
+ // (setSoTimeout / getInputStream) are unavailable and SO_TIMEOUT-based idle
detection
+ // does not apply. See createPipelinedDataIn.
+ private val isUnixDomainSock: Boolean =
conf.get(PYTHON_UNIX_DOMAIN_SOCKET_ENABLED)
protected val tracebackDumpIntervalSeconds: Long =
conf.get(PYTHON_WORKER_TRACEBACK_DUMP_INTERVAL_SECONDS)
protected val killWorkerOnFlushFailure: Boolean =
@@ -441,6 +445,15 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
}
}
+ // Unix domain socket channels expose no java.net.Socket, so socket()
throws
+ // UnsupportedOperationException and SO_TIMEOUT-based idle detection is
unavailable.
+ // Fall back to reading straight from the channel (no idle timeout),
matching how sync
+ // mode also guards its setSoTimeout calls behind !isUnixDomainSock.
+ if (isUnixDomainSock) {
+ return new DataInputStream(
+ new BufferedInputStream(Channels.newInputStream(worker.channel),
bufferSize))
+ }
+
// Set socket read timeout for idle timeout detection in pipelined mode.
// Always set explicitly (including 0 = no timeout) because reused workers
may
// retain a stale SO_TIMEOUT from a previous task that had a different
setting.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
