(spark) branch master updated: [MINOR][PYTHON] Fix stale comments in worker.py

[yi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

Yicong-Huang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 41f1c80892e0 [MINOR][PYTHON] Fix stale comments in worker.py
41f1c80892e0 is described below

commit 41f1c80892e0a039c8430ef1f1587fe72136de26
Author: Yicong Huang <17...@users.noreply.github.com>
AuthorDate: Fri Oct 23 09:10:09 2026 +0000

    [MINOR][PYTHON] Fix stale comments in worker.py
    
    ### What changes were proposed in this pull request?
    
    Fix stale comments and docstrings in `python/pyspark/worker.py`:
    
    - Module docstring: drop "receives input from Piped RDD" (`PipedRDD` is `rdd.pipe()`, unrelated). Describe `invoke_udf` as runnable against any `SparkMessageReceiver`, `main` as the socket entry used by `PythonRunner` / `pyspark.daemon`, and the NON_UDF vs eval-type-handler split.
    - `invoke_udf` docstring: it reads from `message_receiver` and writes to `outfile`, not specifically "from the JVM".
    - `read_single_udtf`: remove a comment about an `accepts_analyze_result` flag that no longer exists, and describe the signature-based constructor choice instead of the old "try with AnalyzeResult, fall back on failure".
    - Profiler comment in `invoke_udf`: UDTFs are not profiled; only UDFs fold profiling into `func`.
    
    ### Why are the changes needed?
    
    The comments no longer match the code after the eval-handler migration and earlier UDTF changes.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    Comment-only change; no behavior change.
    
    ### 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.
    
    Closes #59742 from Yicong-Huang/worker-py-docstring-refresh.
    
    Authored-by: Yicong Huang <17...@users.noreply.github.com>
    Signed-off-by: Yicong-Huang <17...@users.noreply.github.com>
---
 python/pyspark/worker.py | 22 ++++++++++++----------
 1 file changed, 12 insertions(+), 10 deletions(-)

diff --git a/python/pyspark/worker.py b/python/pyspark/worker.py
index 931c4a390a57..6105b65f1aac 100644
--- a/python/pyspark/worker.py
+++ b/python/pyspark/worker.py
@@ -16,7 +16,11 @@
 #
 
 """
-Worker that receives input from Piped RDD.
+Python worker that runs one task at a time. ``invoke_udf`` runs a task against any
+``SparkMessageReceiver``, so it does not depend on how the worker was started; ``main`` drives
+it over the socket of a worker launched by the JVM's PythonRunner, directly or forked by
+pyspark.daemon. A task runs either a classic RDD command (eval type NON_UDF) or SQL UDFs/UDTFs
+through the handler registered for their eval type in pyspark.eval_handlers.
 """
 
 import dataclasses
@@ -220,8 +224,6 @@ def read_single_udtf(pickleSer, udtf_info, handler_cls):
         pickled_analyze_result = pickleSer.loads(udtf_info.pickled_analyze_result)
     else:
         pickled_analyze_result = None
-    # Initially we assume that the UDTF __init__ method accepts the pickled AnalyzeResult,
-    # although we may set this to false later if we find otherwise.
     handler = read_command(pickleSer, udtf_info.handler)
     if not isinstance(handler, type):
         raise PySparkRuntimeError(
@@ -235,10 +237,9 @@ def read_single_udtf(pickleSer, udtf_info, handler_cls):
             f"The return type of a UDTF must be a struct type, but got {type(return_type)}."
         )
 
-    # Update the handler that creates a new UDTF instance to first try calling the UDTF constructor
-    # with one argument containing the previous AnalyzeResult. If that fails, then try a constructor
-    # with no arguments. In this way each UDTF class instance can decide if it wants to inspect the
-    # AnalyzeResult.
+    # Pick how to construct the UDTF from the count of named positional __init__ parameters
+    # after self. With an AnalyzeResult, one parameter receives a copy of it, zero means the
+    # class is called with no arguments, and more are an error; without one, any is an error.
     udtf_init_args = inspect.getfullargspec(handler)
     if pickled_analyze_result is not None:
         if len(udtf_init_args.args) > 2:
@@ -301,7 +302,8 @@ def read_single_udtf(pickleSer, udtf_info, handler_cls):
 def invoke_udf(message_receiver: SparkMessageReceiver, outfile: BinaryIO):
     """
     This function is the main processing function for worker.py.
-    It receives messages from the JVM, processes the data, and sends back results.
+    It reads the task's messages from message_receiver, processes the data, and writes the
+    results to outfile.
     This method goes through three phases:
 
     Initialization -> Processing -> Finish/Cleanup
@@ -337,8 +339,8 @@ def invoke_udf(message_receiver: SparkMessageReceiver, outfile: BinaryIO):
         eval_type = init_info.eval_type
         runner_conf = RunnerConf(init_info.runner_conf)
         eval_conf = EvalConf(init_info.eval_conf)
-        # UDF and UDTF runners fold profiling into func at construction time (see
-        # read_single_udf); only the classic RDD command carries a profiler of its own.
+        # Only the classic RDD command carries a profiler of its own. UDF runners fold profiling
+        # into func at construction time (see read_single_udf); UDTF runners are not profiled.
         profiler = None
         if eval_type == PythonEvalType.NON_UDF:
             assert isinstance(init_info.udf_info, (bytes, memoryview))


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org