Yicong-Huang opened a new pull request, #59860:
URL: https://github.com/apache/spark/pull/59860
### What changes were proposed in this pull request?
The Python worker (`worker.py`) now sends an init response once it has
finished setting up a task (eval type, conf, UDF/UDTF/command), before any
output, and `BasePythonRunner.ReaderIterator` waits for it before reading
output.
- Frame: `START_OF_INIT_RESPONSE` (-9) followed by a length-prefixed JSON
object, mirroring the init message (`START_OF_INIT_MESSAGE`) and the JSON
metrics report. It is `{}` on success, or `{"error": {"message", "traceback",
"errorClass", "isPayloadError"}}` if initialization failed, in which case no
output follows.
- `isPayloadError` is set when the worker failed to load the UDF payload
(unpickling, UDF/UDTF/handler construction) rather than to set up the task
(e.g. Python version mismatch).
- Failures after the init response keep using `PYTHON_EXCEPTION_THROWN`.
- On the JVM side, an init error surfaces as the same `PythonException`
(`PYTHON_EXCEPTION`, same message and traceback) as before, so user-facing
errors are unchanged.
### Why are the changes needed?
Part of SPARK-55278. The language-agnostic gRPC UDF protocol replies to
`Init` with exactly one `InitResponse` (success or error) before any data. The
socket protocol had no init acknowledgement, so init failures only surfaced
when the JVM read output and were indistinguishable from data-phase failures.
This aligns the two protocols and gives the JVM an explicit initialized point.
### Does this PR introduce _any_ user-facing change?
No. The JVM-to-worker protocol is internal; the JVM and `worker.py` ship
together.
### How was this patch tested?
- New `pyspark.tests.test_worker_init_response`: drives `worker.main` over
in-memory streams and checks the init response on success, on a payload error,
on a setup error, and that a data-phase error still follows a successful init
response.
- New `BasePythonRunnerSuite` cases for decoding the init response and
rejecting malformed ones.
- Ran `pyspark.tests.test_worker`, `pyspark.sql.tests.test_udtf` and
`pyspark.sql.tests.arrow.test_arrow_python_udf`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
This pull request and its description were written by Isaac.
--
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]
