Yicong-Huang opened a new pull request, #59419:
URL: https://github.com/apache/spark/pull/59419
### What changes were proposed in this pull request?
Continues the eval-type-handler migration (umbrella SPARK-59415), following
SPARK-60100. Moves three pandas aggregate/window eval types off the
`worker.py::read_udfs` if/elif chain onto handler classes in
`python/pyspark/eval_handlers/_pandas.py`:
- `PandasGroupedAggUDFHandler` for `SQL_GROUPED_AGG_PANDAS_UDF`.
- `PandasGroupedAggIterUDFHandler` for `SQL_GROUPED_AGG_PANDAS_ITER_UDF`,
keeping the drain of unread group batches and the existing
`prefers_large_types=False` on output.
- `PandasWindowAggUDFHandler` for `SQL_WINDOW_AGG_PANDAS_UDF`, keeping the
unbounded (repeat) and bounded (per-row `iloc` slice) frame paths.
Each is the pandas counterpart of the matching Arrow handler and calls
`ArrowToPandasConversion` / `PandasToArrowConversion` inline. The corresponding
branches, serializer-selection entries and now-unused imports are removed from
`worker.py`.
```
EvalTypeHandler
GroupedEvalTypeHandler
PandasGroupedAggUDFHandler (SQL_GROUPED_AGG_PANDAS_UDF, new)
PandasGroupedAggIterUDFHandler (SQL_GROUPED_AGG_PANDAS_ITER_UDF, new)
PandasWindowAggUDFHandler (SQL_WINDOW_AGG_PANDAS_UDF, new)
```
### Why are the changes needed?
Part of SPARK-59415: replace the `read_udfs` if/elif chain with a
per-eval-type handler registry so each eval type's logic lives in one
directly-testable class.
### Does this PR introduce _any_ user-facing change?
No. Internal refactor; UDF API, wire format and behavior are unchanged.
### How was this patch tested?
- New unit tests in
`python/pyspark/eval_handlers/tests/test_pandas_eval_type_handlers.py`
(registration, multiple UDFs, kwargs, empty groups, draining, large-var-types
handling, bounded/unbounded/mixed frames, invalid bound type).
- Existing suites pass:
`pyspark.sql.tests.pandas.test_pandas_udf_grouped_agg`,
`test_pandas_udf_window`.
- ASV `bench_eval_type`, master vs. this PR, one representative run:
```text
bench scenario udf before after
diff
GroupedAggPandasUDF few_groups_sm sum_udf 17.1+-0.2ms
16.7+-0.1ms -2%
GroupedAggPandasUDF many_groups_sm sum_udf 589+-4ms
582+-0.8ms -1%
GroupedAggPandasUDF many_groups_lg mean_multi_udf 190+-0.8ms 186+-1ms
-2%
GroupedAggPandasIter many_groups_sm sum_udf 573+-8ms 563+-5ms
-2%
GroupedAggPandasIter wide_cols mean_multi_udf 164+-1ms 162+-2ms
-1%
WindowAggPandasUDF few_groups_lg sum_udf 41.9+-0.4ms
41.8+-0.4ms 0%
WindowAggPandasUDF many_groups_sm sum_udf 668+-2ms 667+-2ms
0%
WindowAggPandasUDF many_groups_lg sum_udf 226+-1ms 229+-2ms
+1%
```
All other scenarios are also within +-2%.
### 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]
