HyukjinKwon commented on PR #57952:
URL: https://github.com/apache/spark/pull/57952#issuecomment-5274056640
Thanks for taking a look — happy to hold. Answering the three points:
**1. Physical operator — it already is one implementation.** Both stages run
through a single
`PythonIncrementalAggregateExecBase.doExecute`;
`PythonIncrementalAggregate{Partial,Final}Exec` are
thin parameterizations of it (which columns are sent to Python, the eval
type, the required child
distribution, the output attributes). That mirrors the normal path, where
the planner emits two
`HashAggregateExec` **nodes** (partial + final) around an `Exchange`, both
backed by one operator
class. If you'd prefer the shape where it's literally a single case class
parameterized by an
aggregate `mode` (rather than a shared base + two thin subclasses), that's a
mechanical change since
all the logic already lives in the base — glad to do it.
**2. What "blocks" collapsing the two eval types.** They encode two
genuinely different worker
computations with different I/O, not just a mode bit:
| stage | worker input | worker computation | worker output |
|-------|--------------|--------------------|---------------|
| PARTIAL | raw argument columns | `zero` then fold via `reduce` | one
buffer struct |
| FINAL | one buffer struct column | fold via `merge`, then `finish` | the
result value |
On the JVM the mode lives on `AggregateExpression.mode` and the operator
just reads it — there's no
boundary. But the worker sits across the serialization boundary, and
`PythonEvalType` is the
discriminator Spark uses there (as came up in the earlier thread —
"PythonEvalType decides internal
computation type"). Collapsing to one eval type doesn't remove the branch
(the worker still has to
know partial vs final); it just moves the discriminator into a **new
per-call `mode` field in the
Python UDF wire protocol**, which is more surface than two eval-type
constants. So it's a
protocol-surface tradeoff, not a fundamental blocker. If you'd rather have
one eval type + a mode
field, I'm happy to switch — just wanted to flag the cost.
**3. Multiple UDFs and mixing.**
- **Multiple `udaf`s in one aggregate**: supported. Each aggregator carries
its own `bufferSchema`;
PARTIAL emits one buffer struct column per aggregator (`_0.._n`), FINAL
reads each aggregator's own
buffer column positionally and runs `merge`+`finish` independently. Added
a test with two
aggregators (different buffer schemas) over the same input.
- **Mixing `udaf` with grouped-agg (iter) pandas/arrow UDFs, or with SQL
aggregates**: not supported
in this PR — it hits the same `INVALID_PANDAS_UDF_PLACEMENT` analysis
error that already forbids
mixing a grouped-agg pandas UDF with other aggregate functions (the
planner requires `forall` one
kind). I extended that error so it also names the incremental aggregators.
True cross-kind mixing
(incremental + whole-group in one `Aggregate`) would need a combined
physical operator and is
called out as a follow-up in the description.
Latest push (39170897) has the multi-UDF test and the error-message fix. Let
me know which shape you
prefer for (1)/(2) and I'll adjust.
--
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]
