Yicong-Huang opened a new pull request, #59312:
URL: https://github.com/apache/spark/pull/59312
### What changes were proposed in this pull request?
Continues the eval-type-handler migration (umbrella SPARK-59415), following
`PandasScalarUDFHandler` (SPARK-59911) and SPARK-59959. Moves three pandas
grouped/cogrouped map eval types off the `worker.py::read_udfs` if/elif chain
onto handler classes in `python/pyspark/eval_handlers/_pandas.py`:
- `PandasGroupedMapUDFHandler` for `SQL_GROUPED_MAP_PANDAS_UDF`
(applyInPandas), including the output byte-cap `resize_batches`.
- `PandasGroupedMapIterUDFHandler` for `SQL_GROUPED_MAP_PANDAS_ITER_UDF`,
keeping lazy per-batch conversion and draining unread group batches.
- `PandasCoGroupedMapUDFHandler` for `SQL_COGROUPED_MAP_PANDAS_UDF`, keeping
the left-then-right grouping-key lookup.
Each is the pandas counterpart of the matching Arrow handler and calls
`ArrowToPandasConversion` / `PandasToArrowConversion` inline, as
`PandasScalarUDFHandler` does.
```
EvalTypeHandler
GroupedEvalTypeHandler
PandasGroupedMapUDFHandler (SQL_GROUPED_MAP_PANDAS_UDF, new)
PandasGroupedMapIterUDFHandler (SQL_GROUPED_MAP_PANDAS_ITER_UDF, new)
CoGroupedEvalTypeHandler
PandasCoGroupedMapUDFHandler (SQL_COGROUPED_MAP_PANDAS_UDF, new)
```
The corresponding branches and serializer-selection entries are removed from
`read_udfs`.
### 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, per-group mapping, grouping key, output coercion, byte-cap
resizing, draining unread batches, empty-left key fallback, non-DataFrame
rejection).
- Existing suites pass: `pyspark.sql.tests.pandas.test_pandas_grouped_map`,
`test_pandas_cogrouped_map`, `test_pandas_cogrouped_map_misc`.
- ASV `bench_eval_type` (`GroupedMapPandasUDFTimeBench`,
`GroupedMapPandasIterUDFTimeBench`, `CogroupedMapPandasUDFTimeBench`), median
of 5 runs, master vs. this PR: cogrouped and grouped-iter within about +-2%;
grouped mostly within +-2-6%, with `lg_grp_many_col` / `sort_udf` about +4% in
an interleaved rerun (master 282-293ms, PR 301-306ms). The per-group code path
is unchanged.
### 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]
