Yicong-Huang opened a new pull request, #59459:
URL: https://github.com/apache/spark/pull/59459
### What changes were proposed in this pull request?
Moves the five element-wise UDF eval types (Python UDFs lifted out of
higher-order function lambdas, e.g. `transform(arr, x -> udf(x))`) from
`read_udfs` in `worker.py` to handlers in a new
`pyspark/eval_handlers/_elementwise.py`:
```
BatchEvalTypeHandler
├── ArrowElementwiseUDFHandler SQL_ARROW_ELEMENTWISE_UDF
(102)
├── _ScalarElementwiseUDFHandler (abstract) flatten -> invoke -> verify
-> re-nest
│ ├── PandasScalarElementwiseUDFHandler
SQL_SCALAR_PANDAS_ELEMENTWISE_UDF (103)
│ └── ArrowScalarElementwiseUDFHandler
SQL_SCALAR_ARROW_ELEMENTWISE_UDF (105)
└── _ScalarIterElementwiseUDFHandler (abstract) FIFO re-nest of streamed
results
├── PandasScalarIterElementwiseUDFHandler
SQL_SCALAR_PANDAS_ITER_ELEMENTWISE_UDF (104)
└── ArrowScalarIterElementwiseUDFHandler
SQL_SCALAR_ARROW_ITER_ELEMENTWISE_UDF (106)
```
The shared bases keep the existing flatten / re-nest logic as is; the
per-flavor input conversion, result checks and output conversion become hook
methods (the Arrow and pandas conversions live in `_ArrowElementwiseConversion`
/ `_PandasElementwiseConversion` mixins shared by the scalar and iterator
handlers), replacing the `is_pandas` branches. The flatten / re-nest helpers
move from `worker.py` unchanged, and the five types are removed from
`read_udfs`.
### Why are the changes needed?
Part of SPARK-59415, moving each eval type's worker logic into a
self-contained handler.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New `pyspark.eval_handlers.tests.test_elementwise_eval_type_handlers`,
covering both flavors with nesting depth > 1, multiple differently shaped UDFs
fused in one operator, empty/null arrays and the error paths. Existing
`test_udf_in_higher_order_function` and its Connect parity suite pass.
Worker microbenchmark (median of 15 runs, ms, +- stdev; 20 batches of 2000
`array<long>` rows, depth 2 wraps them in another array):
| Eval type | depth | before | after |
|---|---|---|---|
| 102 | 1 | 22.1 +- 0.5 | 21.4 +- 0.1 |
| 103 | 1 | 15.8 +- 2.7 | 16.1 +- 2.8 |
| 104 | 1 | 15.9 +- 0.3 | 16.0 +- 0.3 |
| 105 | 1 | 12.1 +- 0.3 | 12.3 +- 0.3 |
| 106 | 1 | 12.1 +- 0.1 | 12.1 +- 0.1 |
| 102 | 2 | 22.9 +- 0.1 | 22.4 +- 0.1 |
| 103 | 2 | 19.5 +- 0.4 | 19.8 +- 0.6 |
| 104 | 2 | 19.7 +- 0.3 | 20.5 +- 0.3 |
| 105 | 2 | 16.0 +- 0.3 | 15.5 +- 0.1 |
| 106 | 2 | 15.6 +- 0.1 | 16.1 +- 0.8 |
An interleaved rerun of 104 / 106 at depth 2 (30 runs each) showed no
difference (104: 19.6 / 19.9 before vs 19.7 / 19.4 after).
### 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]
