HyukjinKwon opened a new pull request, #58142: URL: https://github.com/apache/spark/pull/58142
### What changes were proposed in this pull request? A follow-up to [SPARK-58736](https://issues.apache.org/jira/browse/SPARK-58736) (incremental Python `Aggregator` / `udaf`). This lets an incremental aggregator be used as a **window function** -- `df.withColumn("m", udaf(agg)(col).over(window))` -- which previously failed analysis with `UNSUPPORTED_EXPR_FOR_WINDOW`. A window has no shuffle, so it reuses neither the map-side PARTIAL nor the post-shuffle FINAL two-stage eval type. A new eval type `SQL_WINDOW_AGG_ARROW_INCREMENTAL_UDF` (257) routes the aggregator through the **existing Arrow window operator** (`ArrowWindowPythonExec`), which sends each frame to the Python worker -- the whole partition for an unbounded frame, or per-row `[begin, end)` slices for a bounded one. The worker folds the frame's rows with `reduce` from a fresh `zero` and produces the value with `finish` (one output value per input row); `merge` is not used on the window path. Concretely: - Route `PythonAggregate` to the Python window path: `PythonUDF.isWindowPandasUDF` and `WindowFunctionType.pythonEvalType` now recognize it, and the `WindowResolution` reject guard is removed. - Generalize `ArrowWindowPythonEvaluatorFactory` from `asInstanceOf[PythonUDAF]` to `PythonFuncExpression` (both `PythonUDAF` and `PythonAggregate` extend it) and add the new eval type to `ArrowWindowPythonExec`'s supported set. All the frame-boundary machinery is reused. - Add the worker handler for the new eval type in `worker.py`. No Spark Connect-specific change is needed: a window-over-aggregator resolves through the same server-side Catalyst rules. ### Why are the changes needed? Parity. Grouped-agg pandas/arrow UDFs already work as window functions, and so does the Scala `Aggregator` (via the JVM SQL window path). The incremental Python aggregator is presented as the analog of the Scala `Aggregator`, so rejecting it over a window was an inconsistent limitation. ### Does this PR introduce _any_ user-facing change? Yes. `udaf(...).over(window)` now works; previously it raised `UNSUPPORTED_EXPR_FOR_WINDOW` at analysis. No other behavior changes. This is relative to the unreleased incremental-aggregator feature (SPARK-58736); no released behavior changes. ### How was this patch tested? New tests in `python/pyspark/sql/tests/arrow/test_arrow_python_aggregator.py`, run for both classic and Spark Connect (via the parity suite): - `test_window_unbounded` -- unbounded partition frame. - `test_window_running_frame` -- ordered growing frame (unbounded preceding .. current row). - `test_window_sliding_frame` -- sliding frame (1 preceding .. 1 following) with a custom single-field buffer aggregator. Each cross-checks the aggregator's result against the equivalent SQL window aggregate. Ran locally: classic 26/26 and Spark Connect parity 24/24 pass. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) -- 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]
