HyukjinKwon commented on PR #57990:
URL: https://github.com/apache/spark/pull/57990#issuecomment-5335570627
## Code review (AI-assisted, self)
A correctness-focused pass over the analyzer rule, the Scala/Connect wiring,
and the Python module. Ranked findings below; each was verified against the
source.
### 1. Silent mis-count when one accumulator is referenced by two UDFs in a
single `select`/`withColumn`/`filter` (blocking)
`InjectObservedAccumulators.rewriteProject` funnels every extracted delta
into a **single** `CollectMetrics`, and `extractUdfs` names each metric
`__oa_metric_<accName>` (`sql/core/.../ObservedAccumulator.scala`,
`rewriteProject` + the metric alias in `extractUdfs`). Two UDFs that touch the
*same* accumulator in one projection therefore produce two aggregates aliased
identically. `CheckAnalysis.checkMetric` does not reject duplicate metric
names, and the harvest listener iterates `row.schema.fieldNames.foreach { fn =>
row.fieldIndex(fn) }` -- `fieldIndex` returns the **first** match both times,
so one UDF's delta is counted twice and the other is dropped. This applies to
both Scala and Python UDFs (shared JVM rule), and is **not covered by a test**
(the suite exercises multiple *distinct* accumulators in one UDF, and one
accumulator across *separate* queries -- not two UDFs / one accumulator / one
projection).
_Fix:_ dedupe by accumulator name when building `metrics` -- one
`Sum`/`CollectList` per name per `CollectMetrics`.
### 2. A fractional value added to an integer-typed accumulator is handled
three different ways (correctness / robustness)
`add()` stores `cur + term` verbatim, so `acc.add(1.5)` on an `int`
accumulator (`spark.accumulator(name, 0)`) produces a float delta, and then:
- **row-at-a-time scalar UDF** silently truncates via `int(d)`,
- **vectorized (pandas/Arrow) UDF** hard-crashes the query --
`pd.Series([1.5, ...], dtype="int64")` raises,
- **operator UDF** rounds the summed total via `int(round(...))` (the
operator delta column is always `DoubleType`).
Same misuse, three behaviors -- one of them a crash.
(`python/pyspark/sql/observed_accumulator.py`)
_Fix:_ pick one policy -- reject a non-integer `term` at `add()` with a
clear error, or coerce consistently everywhere and document it.
### 3. Scala integer accumulators are not exact past 2^53 (minor /
documented limitation)
`ObservedAccumulator.add(Long)` narrows to `Double` immediately
(`sql/api/.../ObservedAccumulator.scala`); `value = zero +
math.round(harvested)`. The exact-`Long` registry only fires on the Python path
(a `LongType` delta), so the Scala `add(Long)` / `value: Long` surface invites
an exactness assumption the implementation does not hold.
_Fix:_ carry a Long buffer on the Scala side too, or note the limit in the
scaladoc.
### Verified clean
Rule registration (Resolution batch, classic + Connect server), session-UUID
keying on the classic registry, the Connect client registry (a per-session
instance field -- no cross-session collision), the partials drain-and-fold
(intentional and cumulative, symmetric with classic), custom-merge folding, and
the JVM-client binary-compat excludes.
_Generated with assistance from Claude Code; findings hand-verified._
--
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]
