[SPARK-59704][PYTHON] Migrate Arrow aggregate and window UDF eval types to the EvalTypeHandler pipeline

[Yicong-Huang]

What changes were proposed in this pull request?

This PR migrates the three non-incremental Arrow-native aggregate/window UDF eval types off the hand-written if/elif chain in worker.py's read_udfs and onto the EvalTypeHandler pipeline introduced in SPARK-59425 and extended in SPARK-59622. The migrated types are SQL_GROUPED_AGG_ARROW_UDF (252), SQL_GROUPED_AGG_ARROW_ITER_UDF (254), and SQL_WINDOW_AGG_ARROW_UDF (253).

Each becomes a GroupedEvalTypeHandler subclass in python/pyspark/eval_handlers/_arrow.py (ArrowGroupedAggUDFHandler, ArrowGroupedAggIterUDFHandler, ArrowWindowAggUDFHandler), declaring its eval_type so it self-registers and is dispatched via get_eval_type_handler. The handlers follow the existing lazy-pyarrow convention: string forward-ref bases, require_minimum_pyarrow_version() in __init__, and import pyarrow as pa inside run. The window handler keeps the unbounded (compute once, repeat per row) vs bounded (per-row frame slicing) dispatch verbatim. The repeated pre-19.0.0 concat_batches fallback is lifted into a shared _concat_group_batches helper. The corresponding branches and the migrated types' entries in the serializer-selection lists are removed from worker.py; the incremental variants are out of scope and stay in worker.py.

This is a pure code move with no behavior change: the UDF logic, output coercion, and stream draining are identical to the removed branches.

Why are the changes needed?

Consolidating the Arrow eval types onto one registry-dispatched pipeline removes the growing per-eval-type if/elif chain in read_udfs, so the execution model for each type lives in one cohesive handler class rather than being split between a serializer-selection block and a function-construction block. Follows the umbrella SPARK-59415.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing per-eval-type tests, classic and Spark Connect: pyspark.sql.tests.arrow.test_arrow_udf_grouped_agg, pyspark.sql.tests.arrow.test_arrow_udf_window, and their connect.arrow.test_parity_* counterparts all pass. Worker microbenchmarks for the three eval types (including the window bounded and unbounded frames) show no regression versus the pre-refactor baseline.

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

Generated-by: Isaac

This pull request and its description were written by Isaac.

[Yicong-Huang]

Merge Summary:

• merged into master de8be6d
• merged into branch-4.x e889528

Posted by merge_spark_pr.py