[SPARK-58119][CORE][PYTHON] Introduce PythonWorkerHandle abstraction for the Python worker path#57247
Open
fapaul wants to merge 4 commits into
Open
[SPARK-58119][CORE][PYTHON] Introduce PythonWorkerHandle abstraction for the Python worker path#57247fapaul wants to merge 4 commits into
fapaul wants to merge 4 commits into
Conversation
added 2 commits
July 14, 2026 07:28
…for the Python worker path Replace direct use of java.lang.ProcessHandle in the Python worker path with a Spark-owned PythonWorkerHandle trait, so the runner depends only on the small set of operations it needs (isAlive, destroy, terminationDiagnostics) rather than on the sealed java.lang.ProcessHandle type, and the faulthandler-diagnostics logic lives behind one testable method instead of being scattered across the runners. - New PythonWorkerHandle trait + LocalPythonWorkerHandle (wraps java.lang.ProcessHandle) and PythonWorkerHandle.of factory. - SparkEnv.createPythonWorker, PythonWorkerFactory, PythonRunner, PythonUDFRunner, PythonArrowOutput, PythonPlannerRunner rewired to the trait. - Diagnostics flow through the handle; the runner supplies the (SQLConf-resolved) faulthandler flag, the handle locates its worker's log. The OS pid stays a factory-internal detail (daemon kill-by-pid), not part of the handle contract. - Adds PythonWorkerHandleSuite. Draft / design proposal for discussion on a personal fork; see SPARK-58119.
… state and simplify diagnostics closure - Fold the parallel daemonWorkers + daemonWorkerPids maps into a single daemonWorkers map valued by a DaemonWorker(handle, pid) record, so the worker handle and its kill-by-pid are one source of truth. - Simplify the injected faulthandler closure to () => Option[String]; the enabled gate now lives in terminationDiagnostics(faultHandlerEnabled), which the runner supplies from its per-session SQLConf value. - Add subprocess-backed PythonWorkerHandleSuite tests covering isAlive / destroy delegation and of() returning None for a dead pid.
Yicong-Huang
approved these changes
Jul 14, 2026
Yicong-Huang
left a comment
Yicong-Huang
left a comment
Contributor
There was a problem hiding this comment.
I had a prior pass on fapaul#1 already. This PR LGTM with a minor comment in test suite.
| val pid = proc.pid() | ||
| proc.destroyForcibly() | ||
| assert(proc.waitFor(10, TimeUnit.SECONDS)) | ||
| assert(PythonWorkerHandle.of(pid).isEmpty) |
Contributor
There was a problem hiding this comment.
Minor flake risk: the OS may reuse pid after the process is reaped, making of(pid) return a live handle. Idiomatic in Spark suites, but worth a note; alternatively assert on a handle taken before destroyForcibly.
Contributor
|
cc @HyukjinKwon @cloud-fan for a pass |
Member
|
cc @gaogaotiantian too |
Contributor
|
I'm a bit confused about the purpose and design. I have a few questions that are probably in different area.
|
…lake-free Capture the PythonWorkerHandle while the process is alive and assert it observes the death after the kill, instead of re-querying of(reapedPid), which could return a live handle if the OS recycled the reaped pid.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Introduce a Spark-owned
PythonWorkerHandletrait to replace direct use ofjava.lang.ProcessHandlein the Python worker path. The runner then depends only on the small set of operations it actually needs —isAlive(),destroy(),terminationDiagnostics(faultHandlerEnabled).PythonWorkerHandletrait +LocalPythonWorkerHandle(wrapsjava.lang.ProcessHandle) and aPythonWorkerHandle.of(pid, faultHandlerLog)factory (core/.../api/python/PythonWorkerHandle.scala).SparkEnv.createPythonWorker,PythonWorkerFactory,PythonRunner,PythonUDFRunner,PythonArrowOutput,PythonPlannerRunnerrewired fromOption[ProcessHandle]toOption[PythonWorkerHandle].handle.terminationDiagnostics(faultHandlerEnabled): the runner supplies the (SQLConf-resolved) faulthandler flag, and the handle locates its own worker's faulthandler log via an injected closure. The reader no longer needs to know about faulthandler log files.daemonWorkersmap (aDaemonWorker(handle, pid)record), keeping the pid a local-process detail.PythonWorkerHandleSuite.Behavior-preserving:
LocalPythonWorkerHandledelegates 1:1 to the JDK handle and the existing faulthandler read.Why are the changes needed?
java.lang.ProcessHandleis a sealed JDK type hardcoded across the Python worker path, and the faulthandler-log reading is duplicated inline across the runners. Introducing a small Spark-owned interface decouples the runner from the concrete JDK type and centralizes "what the runner needs from a worker handle" — including the diagnostics read — into a single, testable place.Does this PR introduce any user-facing change?
No. Internal refactor; no config, API, or behavior change.
How was this patch tested?
PythonWorkerHandleSuite(5 tests): the injected faulthandler-log reader is invoked only when enabled and its result returned verbatim; the default reader yieldsNone;ofbinds the pid into the reader;isAlive/destroydelegate to the underlying process (spawn a subprocess, assert alive → destroy → reaped → not alive); andofreturnsNonefor a pid with no live process.core(main + test) andsql/corecompile cleanly (Scala 2.13).Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude (Anthropic)