Yicong-Huang opened a new pull request, #58956: URL: https://github.com/apache/spark/pull/58956
### 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. -- 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]
