HyukjinKwon commented on PR #57990:
URL: https://github.com/apache/spark/pull/57990#issuecomment-5288761266
**Code review — commit `7a87d3b`** (observe-backed accumulator). Reviewed
the committed PR head only; a few local uncommitted edits to these files were
in flight at review time and are not covered here. Three points, roughly in
order of severity.
### 1. The classic JVM registry is process-global keyed by name — same-named
accumulators collide across sessions
`ObservedAccumulatorRegistry` is a singleton `object`, and its store is
keyed by accumulator name alone (`sql/core/.../ObservedAccumulator.scala:49`,
read via `registryValue(name)` at `:55`). The harvest listener is registered
per session, but every session's listener bumps the *same* global
`registry[name]`. So two sessions that each do `spark.accumulator("bad")` write
into one shared slot, and `acc.value` returns the summed contributions of both
— silently wrong.
The `checkSession` guard (`sql/api/.../ObservedAccumulator.scala:146`) does
**not** prevent this: it only rejects a read while a *different* session is
active. Reading `accA.value` with session A active passes the guard and still
returns A+B, and even purely sequential use (A runs, B runs, A reads) is
contaminated. Named accumulators (`"bad"`, `"errors"`, …) are exactly the
common case, and `newSession()` on one context hits it immediately.
Note this diverges from the Connect path, which already scopes state to the
session instance (`sql/connect/common/.../SparkSession.scala:105`,
`observedAccumulatorRegistry` on the session). The classic side should do the
same — key the registry by session identity (or hang it off the session like
Connect does) rather than by bare name. The existing `enabledSessions` set
already tracks per-session identity, so the plumbing is there. Worth a test
with two same-named accumulators on `spark` and `spark.newSession()`.
### 2. `accumulator(...)` positional args differ between Scala and Python
Scala: `def accumulator(name: String)`
(`sql/api/.../SparkSession.scala:191`). Python: `def accumulator(self, zero=0,
name=None, merge=None)` (`python/pyspark/sql/session.py:1075`). So
`spark.accumulator("bad")` binds `name` in Scala but `zero` in Python — in
Python it produces an auto-named accumulator whose zero is the string `"bad"`,
which then blows up on the numeric path (`"bad" + float`) when you read
`.value`. The Scala doc/examples and the Python examples use different first
args, so a user porting the Scala snippet gets a silently broken accumulator.
Since the description says the API shape is open, worth settling on one arg
order across languages (e.g. `name` first in both, or `zero` first in both).
### 3. (Likely already in progress) Scala UDF inside a higher-order function
isn't rejected
`rejectAccumulatorInHigherOrderFunction` only matches `PythonUDF` markers
(`sql/core/.../ObservedAccumulator.scala:251`), so a `ScalaUDF` capturing an
accumulator inside a `transform`/`filter` lambda is neither rewritten nor
rejected — it just returns 0, rather than failing fast as the description
promises for HOFs. There's a Python test for the fail-fast but no Scala
equivalent. (Flagging for completeness — the in-flight local changes to this
file appear to already be adding `ScalaUDF` handling here.)
<sub>Reviewed with Claude Code.</sub>
--
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]
