HyukjinKwon commented on PR #57952: URL: https://github.com/apache/spark/pull/57952#issuecomment-5323531722
Thanks a lot for the thorough review, @dongjoon-hyun -- these were all real issues. Addressed in 5daa627 (pushed). Point by point: **Correctness** 1. **DISTINCT / FILTER** -- `checkUnsupportedAggregateClause` now also runs for `PythonAggregate` in `FunctionResolution`, so both are rejected at analysis instead of being silently dropped. 2. **Pivot** -- `ResolvePivot.checkValidAggregateExpression` now rejects `PythonAggregate` as well, so it no longer falls into the null-ignoring `If(...)` rewrite. 3. **Named arguments** -- rather than reject them, I threaded them through: the PARTIAL worker now appends the `kwargs_offsets` columns (in call order) after the positional ones into the `reduce` value tuple. Verified with both `agg(udaf(...)(v=col))` and SQL `my_agg(v => x)`. 4. **Floating-point grouping keys** -- the `PythonAggregate` planning branch now normalizes grouping keys via `NormalizeFloatingNumbers`, mirroring the SQL branch, so `0.0`/`-0.0` and NaN patterns collapse into one group. **Guards (clean errors instead of internal failures)** 6. **Window** -- `WindowResolution` now rejects `PythonAggregate` in a window with `UNSUPPORTED_EXPR_FOR_WINDOW` instead of misclassifying it as a SQL aggregate. 7. **Streaming** -- the streaming-aggregation guard in `SparkStrategies` now also matches `PythonAggregate`. 8. **Mixed aggregates** -- added a dedicated `INVALID_PYTHON_UDF_PLACEMENT` error so the message no longer wrongly blames "group aggregate pandas UDF". **Validation / design** 5. **Duplicate buffer field names** -- validated at `udaf()` creation now, raising `DUPLICATED_FIELD_NAME_IN_ARROW_STRUCT` up front. 10. **`bufferSchema`** -- promoted to a first-class `UserDefinedFunction` field (classic + Connect), so it survives `_wrapped()`, `asNondeterministic()`, and `spark.udf.register` without being a monkey-patched attribute. 9. **Map-side sort** -- the PARTIAL stage is now a hash-based combine: no `requiredChildOrdering`/pre-shuffle `SortExec`, and many groups per Arrow batch instead of one IPC stream per group. The worker keeps one running buffer per key and re-emits the keys; the FINAL stage re-groups authoritatively after the shuffle (kept sort-based), so the map-side combine only needs to be best-effort -- any keys it can't collapse (e.g. NaN) are merged downstream. All the arrow aggregator tests (classic + Connect parity) pass locally, and I added coverage for each of the above (DISTINCT/FILTER, pivot, window, mixed placement, named args, duplicate buffer names, float/NaN and struct grouping keys). One gap: I didn't add an automated streaming-rejection test (flaky in local runs) -- the guard is symmetric with the existing `PythonUDAF` one. Happy to add it if you'd prefer. PTAL, thanks again! -- 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]
