Yicong-Huang opened a new pull request, #58990: URL: https://github.com/apache/spark/pull/58990
### What changes were proposed in this pull request? This PR migrates the three Arrow-native incremental aggregate/window UDF eval types off the `if/elif` chain in `worker.py`'s `read_udfs` onto the `EvalTypeHandler` pipeline (SPARK-59425, extended in SPARK-59622 / SPARK-59704): `SQL_GROUPED_AGG_ARROW_INCREMENTAL_PARTIAL_UDF` (255), `SQL_GROUPED_AGG_ARROW_INCREMENTAL_FINAL_UDF` (256) and `SQL_WINDOW_AGG_ARROW_INCREMENTAL_UDF` (257). - `ArrowGroupedAggIncrementalPartialUDFHandler` (`BatchEvalTypeHandler`, plain `ArrowStreamSerializer`): map-side hash-combine via `reduce`, with the `maxRecordsPerBatch` buffer cap. - `ArrowGroupedAggIncrementalFinalUDFHandler` (`GroupedEvalTypeHandler`): `merge` + `finish` per group, skipping null partial buffers. - `ArrowWindowAggIncrementalUDFHandler` (`GroupedEvalTypeHandler`): unbounded/bounded dispatch, extending the running buffer when consecutive frames only grow on the right. `_hashable_grouping_key` moves to `pyspark.eval_handlers.utils` as `hashable_grouping_key`. The corresponding branches and serializer-selection entries are removed from `worker.py`. This is a pure code move with no behavior change. ### Why are the changes needed? Completes moving the Arrow-native eval types onto one registry-dispatched pipeline. Part of SPARK-59415. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing tests pass: `pyspark.sql.tests.arrow.test_arrow_python_aggregator` and `pyspark.sql.tests.connect.arrow.test_parity_arrow_python_aggregator`. Added handler unit tests in `pyspark.eval_handlers.tests.test_arrow_eval_type_handlers` (buffer-cap flush boundary, NaN/complex key canonicalization, null buffer skipping, running vs sliding frame folding). Worker microbenchmark (`worker_main` over the mock wire protocol, median of 10 runs, ms): | Scenario | Before | After | |---|---|---| | partial, 1k keys | 343 +- 7 | 345 +- 8 | | partial, 150k keys (cap flush) | 741 +- 23 | 753 +- 29 | | final, 2k groups | 417 +- 8 | 408 +- 6 | | window, unbounded | 94 +- 2 | 92 +- 5 | | window, running frame | 199 +- 1 | 199 +- 13 | | window, sliding frame | 662 +- 15 | 657 +- 9 | ### 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]
