HyukjinKwon commented on PR #57952:
URL: https://github.com/apache/spark/pull/57952#issuecomment-5335385019
Thanks again @dongjoon-hyun, and @cloud-fan for the AI pass -- these
overlapped, so this covers both. All addressed in 56fd4be (pushed). Classic
(24) + Connect parity (22) tests pass locally.
**Correctness / functional**
1. **Profiler breaks incremental aggregators** --
`UserDefinedFunction.__call__` now short-circuits when the eval type is the
incremental aggregator: it warns ("Profiling incremental Python aggregators is
not supported.") and takes the non-profiled path, so neither the CPU-profiler
wrapper (which would drop the `Aggregator`'s `zero`/`reduce`/`bufferSchema`)
nor the memory profiler's `inspect.getsourcelines(f.__code__)` runs. Added CPU
+ memory profiler fallback tests.
2. **`NormalizePlan.normalizeExprIds`** -- now resets
`PythonAggregate.resultId` alongside `PythonUDF`/`PythonUDAF`.
3. **`INVALID_PYTHON_UDF_PLACEMENT` message** -- reworded; multiple
incremental aggregators together are supported (and tested), only mixing with
other *kinds* of aggregate is invalid.
4. **Missing/non-struct buffer schema** -- `builder` now returns a classed
`INVALID_PYTHON_AGGREGATOR_BUFFER_SCHEMA` (via `QueryCompilationErrors`)
instead of a bare `require`/`ClassCastException`, covering the
malformed-Connect-proto and direct-`UserDefinedFunction` paths. Added a test.
5. **Mixed diagnostic drops pandas names** -- the fallthrough error now
names *both* the incremental aggregators and any co-offending grouped-agg
pandas/arrow UDAFs. Added a mixed pandas+incremental test asserting both names
appear.
6. **Connect `is_distinct`** -- the Connect planner now rejects
`is_distinct` for `PythonAggregate` with the same `DISTINCT`-unsupported error
the SQL path raises (a resolved Connect aggregate bypasses
`FunctionResolution`, so it was silently dropped before).
**Efficiency / duplication**
7. **Unbounded map-side combine** -- the PARTIAL worker now caps the per-key
buffer at `maxRecordsPerBatch`, flushes early when the cap is reached, and
emits end-of-partition buffers in bounded chunks. This is safe because FINAL
re-merges any duplicate keys the early flushes produce. Added a test with
`maxRecordsPerBatch=2` to exercise flushing.
8. **`doExecute` duplication + `lockFree`** -- the FINAL stage now reads
`PYTHON_UDF_PIPELINED_EXECUTION` for the queue's `lockFree` flag (no longer
hard-coded `false`), and the `argMetas` dedup is extracted into a shared
`buildArgMetas` used by both stages.
9. **Redundant Arrow->Python conversions** -- the PARTIAL worker now
converts each needed input column at most once per batch (memoized by offset),
even when several aggregators or a grouping key share it.
10. **Duplicated Py4J constructor** -- collapsed into a single call passing
`None` (Py4J maps it to the `null` the Scala `bufferType` already defaults to),
and removed the now-unused auxiliary JVM constructor.
Two notes:
- **#6**: the programmatic `is_distinct=true` proto path isn't reachable
from the PySpark client API, so there's no dedicated client test for it; the
user-facing SQL `DISTINCT` rejection is covered for both classic and Connect by
`test_distinct_and_filter_rejected`.
- **#2**: symmetric one-liner with `PythonUDAF`; exercised via the
plan-comparison path rather than a new standalone test.
PTAL, thanks!
--
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]
