HyukjinKwon commented on PR #57952: URL: https://github.com/apache/spark/pull/57952#issuecomment-5277521758
## Code review (self, head `1d6c9391`) **Verdict: approve with minor cleanups.** No correctness or design blockers. The two-stage design is coherent, both earlier blocking comments (stream batches into buffers; emit `finish(zero)` for empty global input) are genuinely fixed, and Connect parity + SQL registration are wired through. Findings below are all minor. ### Findings **1. FINAL worker handler builds output arrays without an explicit Arrow type** — *robustness* In `python/pyspark/worker.py`, the FINAL handler does: ```python result_arrays = [pa.array([r]) for r in results] ``` whereas the PARTIAL handler correctly passes `type=return_schema.field(i).type`. Relying purely on `enforce_schema` to coerce an inferred type is fragile for non-trivial `outputType`s (decimal, timestamp, nested struct) or an all-`None` column. Suggest aligning FINAL with PARTIAL: ```python result_arrays = [pa.array([r], type=return_schema.field(i).type) for i, r in enumerate(results)] ``` **2. Loop-invariant `field_names` recomputation in both worker handlers** — *minor perf* `field_names = [f.name for f in agg.bufferSchema.fields]` is recomputed per *(group × batch × aggregator)* in the FINAL handler and per *(group × aggregator)* in the PARTIAL handler, though it depends only on the fixed i-th aggregator. Precompute a `field_names_by_udf` list once where `grouped_func` is defined and index by `i`. **3. Unqualified Scaladoc link** — *doc nit* In `sql/catalyst/.../expressions/PythonUDF.scala`, `[[PythonIncrementalAggregateExec]]` cannot resolve from `sql/catalyst` (the class lives in `sql/core`). Fully-qualify it as `[[org.apache.spark.sql.execution.python.PythonIncrementalAggregateExec]]` (as the reverse-direction reference already does) or use plain text. **4. `aggregator.py` docstring polish** — *optional* - The `reduce` example `(buffer[0] + v, buffer[1] + 1)` raises on a null input value; consider showing null handling so the example isn't copied as a fragile pattern. - `udaf`'s `Raises:` lists only `PySparkImportError`, but it also raises `PySparkTypeError` for a non-`Aggregator` arg or a non-`StructType` `bufferSchema`. ### Test portfolio Strong coverage: builtin mean/sum equivalence, no-group, empty-global input, custom buffer, multiple aggregators, partition-count independence, SQL registration, and a Connect parity mirror. Two gaps worth a follow-up: no test with a non-trivial output type (decimal/timestamp) — which would exercise finding #1 — and no test of `reduce` receiving a null input value. -- 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]
