[SPARK-59425][PYTHON] Introduce EvalTypeHandler pipeline and migrate scalar Arrow UDF

[Yicong-Huang]

What changes were proposed in this pull request?

This introduces an extensible execution model for the Arrow/Pandas UDF eval types in the PySpark worker and migrates the first eval type onto it. It is the foundation for the umbrella refactor SPARK-59415 (design: "1DD: Python UDF Eval Handler API Design").

read_udfs in python/pyspark/worker.py has grown into one large if/elif dispatcher over ~30 eval types. This PR adds a pyspark.sql.eval_handlers package where each eval type is a self-contained EvalTypeHandler subclass:

• EvalTypeHandler[InputBatch, OutputBatch] declares a serializer and implements run(split_index, data), which consumes the input stream and yields the output stream.
• Three typed category bases fix the input stream type and the default serializer: BatchEvalTypeHandler (Iterator[pa.RecordBatch]), GroupedEvalTypeHandler (Iterator[GroupedBatch]), and CoGroupedEvalTypeHandler (Iterator[CoGroupedBatch]).
• A handler declares its eval_type and is auto-registered in EVAL_TYPE_HANDLERS via __init_subclass__; read_udfs looks it up and delegates, so a new eval type attaches without editing a central branch.
• ArrowScalarUDFHandler migrates SQL_SCALAR_ARROW_UDF as the first handler.

Two supporting moves keep the package self-contained (no behavior change): the shared result-verification helpers move to pyspark.sql.eval_handlers.verification, and RunnerConf / EvalConf move next to their Conf base in pyspark.worker_util.

The remaining Arrow/Pandas eval types stay on the existing if/elif path and will be migrated incrementally, each independently revertible.

Why are the changes needed?

To make each eval type's execution explicit, self-contained, and unit-testable, and to remove the central if/elif dispatch so subsequent eval types can be migrated one at a time.

Does this PR introduce any user-facing change?

No. This is an internal worker refactor: no change to any UDF API, the on-the-wire format, or JVM-side code, and the migrated SQL_SCALAR_ARROW_UDF path is behavior-identical.

How was this patch tested?

New pyspark.sql.tests.test_eval_type_handlers unit suite (handler registration, per-category serializers and abstractness, run, and end-to-end ArrowScalarUDFHandler output including schema coercion), plus the existing Arrow UDF suites that exercise the migrated path.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Isaac

This pull request and its description were written by Isaac.

[HyukjinKwon]

I checked the migrated path against the removed code and it's identical:

• Serializer — the old SQL_SCALAR_ARROW_UDF fell through to the else branch (ArrowStreamSerializer(write_start_stream=True)), which is exactly what BatchEvalTypeHandler.serializer returns.
• udfs construction — the dispatch prologue's list comprehension is identical to the shared one further down (read_single_udf(pickleSer, udf_info, eval_type, runner_conf, udf_index=...)).
• run body — a faithful copy of the old inline func, with col_names / combined_arrow_schema still hoisted into __init__ (once per read_udfs, as before).

Import integrity of the moves also checks out: worker_util.py already imports Any/Optional/Union, and the PR adds import json + the pyspark.sql.types import that RunnerConf/EvalConf need; worker.py has no remaining bare Conf reference after the import swap; and to_arrow_schema is still used elsewhere in worker.py, so nothing becomes a dead import. unittest.TestCase is the right base for these handler-level tests (no SparkSession/SparkContext needed), and the new module is wired into modules.py + both setup.py files.

Two optional, non-blocking nits:

1. Import-cycle fragility in the package. eval_handlers/__init__.py imports _arrow at the bottom while _arrow imports BatchEvalTypeHandler back from the package. That's fine as long as the package is always imported first (which the worker and tests do), but importing pyspark.sql.eval_handlers._arrow directly first would raise ImportError because ArrowScalarUDFHandler isn't defined yet when __init__ re-enters it. Moving the base classes into a small leaf module (e.g. _base.py) that both __init__ and _arrow import would remove the cycle entirely. Low priority since _arrow is private.

2. assert for duplicate-registration detection in __init_subclass__ is stripped under python -O. A plain if eval_type in EVAL_TYPE_HANDLERS: raise ... would keep the guard (and test_duplicate_eval_type_rejected) working under optimized mode. Minor — it matches the existing assert style in the worker.

Nice foundation for migrating the remaining eval types one at a time.