dongjoon-hyun commented on PR #57952:
URL: https://github.com/apache/spark/pull/57952#issuecomment-5329944181
### Review findings
Overall this looks well-designed and well-tested. I found a few issues worth
addressing before merge — two functional, two error-handling, and some
duplication/efficiency notes.
#### Functional
1. **Profiler breaks incremental aggregators** — `python/pyspark/sql/udf.py`
(~L605): the profiler exclusion list in `UserDefinedFunction.__call__` was not
extended with `SQL_GROUPED_AGG_ARROW_INCREMENTAL_FINAL_UDF`. With
`spark.python.profile=true`, `self.func` (the `Aggregator` object) is replaced
by a plain wrapper function before pickling, so the worker fails with
`AttributeError: 'function' object has no attribute 'bufferSchema'`. With
`spark.python.profile.memory=true` it fails even earlier on the driver at
`inspect.getsourcelines(f.__code__)`, since an `Aggregator` instance has no
`__code__`.
2. **`NormalizePlan.normalizeExprIds` not extended** —
`sql/catalyst/.../plans/NormalizePlan.scala` (~L112): `resultId` is normalized
for `PythonUDF`/`PythonUDAF` but not for the new `PythonAggregate`. Any
`comparePlans`-based test on plans containing a `PythonAggregate` will fail
spuriously, and `HybridAnalyzer`'s dual-run comparison would report a mismatch
for semantically identical plans.
#### Error handling
3. **Error message asserts a restriction the implementation doesn't have** —
`error-conditions.json` (`INVALID_PYTHON_UDF_PLACEMENT`): "Each such function
must be the only aggregate expression in its aggregation" contradicts the
supported (and tested, `test_multiple_incremental_aggregators`) case of
multiple incremental aggregators in one aggregation. Only mixing with *other
kinds* of aggregate functions is invalid.
4. **Bare `require` instead of a classed error** —
`UserDefinedPythonFunction.builder` (~L113): `require(bufferType != null, ...)`
+ unchecked `asInstanceOf[StructType]`. A Connect proto with `eval_type=256`
and a missing (or non-struct) `buffer_type` surfaces as a raw
`IllegalArgumentException`/`ClassCastException` (INTERNAL_ERROR) rather than
following the planner's `InvalidPlanInput` convention. Also reachable
classically via `UserDefinedFunction(f, evalType=256)` with no `bufferSchema`.
5. **Mixed pandas + incremental diagnostic drops names** —
`SparkStrategies.scala` (~L825): when both a grouped-agg pandas UDAF and an
incremental aggregator are mixed with other aggregates, the new fallthrough
names only the `PythonAggregate` functions; the co-offending pandas UDAF names
(which the replaced code listed) are omitted. This mixed case has no test.
6. *(Minor, pre-existing pattern)* **Connect `is_distinct` silently
ignored** — `SparkConnectPlanner.scala` (~L2204): `agg.toAggregateExpression()`
hardcodes `isDistinct = false`, so a client sending `is_distinct=true` silently
gets non-distinct results — while this PR's own `FunctionResolution` change
explicitly rejects DISTINCT on the SQL path ("reject rather than silently drop
the clause"). Same hole exists for `PythonUDAF`, but this PR is where the
reject-don't-drop rule was introduced.
#### Efficiency / duplication
7. **Unbounded map-side combine** — `worker.py` PARTIAL handler (~L2294):
all per-group buffers are held in one dict for the whole partition and emitted
as a *single* `RecordBatch` (no `maxRecordsPerBatch` chunking, no cap/flush).
High-cardinality keys — exactly where partial aggregation degenerates — will
OOM the Python worker where JVM hash aggregation would spill. Since the FINAL
stage re-merges duplicate keys authoritatively, capping the map and flushing
early (and emitting the end-of-partition output in bounded chunks) is safe and
cheap. I saw spill is listed as a follow-up, but a simple size cap + chunked
emission seems worth doing in this PR.
8. **`doExecute` duplication has already drifted** —
`PythonIncrementalAggregateExecBase.doExecute` is a near-verbatim copy of
`ArrowAggregatePythonExec.doExecute`, and the copy hard-codes `lockFree =
false` (~L162) where the original picks it from
`PYTHON_UDF_PIPELINED_EXECUTION` — so the FINAL stage silently misses the
pipelined-queue optimization. Extracting the shared grouped-input/queue/join
scaffolding (the `argMetas` dedup loop alone now exists twice within this PR)
would prevent further drift.
9. **Redundant Arrow→Python conversions** — `worker.py` (~L2299): the same
column is `to_pylist()`-ed once per referencing UDF plus again for grouping
keys (N aggregators sharing a column ⇒ N+1 conversions per batch). One
`{offset: pylist}` memo per batch would do.
10. **Duplicated Py4J constructor call** — `udf.py` `_create_judf` (~L535):
the 8-line constructor invocation is duplicated in an if/else differing only in
the trailing `bufferType`, with a JVM auxiliary constructor existing solely for
the arity. Py4J maps `None` → `null` and the Scala param already defaults to
`null`, so a single call passing `None` would remove both the else branch and
the auxiliary constructor.
--
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]
