This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new ac67095dcd9b [SPARK-60570][CORE] Add AccumulatorContext.getOrNull to
avoid an Option per driver-side accumulator lookup
ac67095dcd9b is described below
commit ac67095dcd9be8250865a2f67966f4a45ab8bd06
Author: Shujing Yang <[email protected]>
AuthorDate: Tue Oct 27 16:29:38 2026 +0000
[SPARK-60570][CORE] Add AccumulatorContext.getOrNull to avoid an Option per
driver-side accumulator lookup
### What changes were proposed in this pull request?
This PR adds `AccumulatorContext.getOrNull`. It does the same lookup as
`AccumulatorContext.get`, but returns `null` instead of `None` when the
accumulator is not registered or has been garbage collected. `get` now
delegates to `getOrNull` and wraps the result in `Option`.
`DAGScheduler.updateAccumulators` now calls `getOrNull` and null-checks the
result.
### Why are the changes needed?
`DAGScheduler.updateAccumulators` runs on the driver for every finished
task, and it looks up every accumulator update the task carries. A task can
carry hundreds of updates when SQL metrics are involved.
`AccumulatorContext.get` returns an `Option`, so each lookup allocates a `Some`
that is dropped right away. `getOrNull` removes that allocation from the
per-task path.
### Does this PR introduce _any_ user-facing change?
No. `get` returns the same results as before, including `None` for a
garbage-collected accumulator. The garbage-collected warning is still logged
inside the lookup. A missing accumulator in `updateAccumulators` still throws
`accessNonExistentAccumulatorError`. `AccumulatorContext` is `private[spark]`,
so no public API changes.
### How was this patch tested?
No new test, because the changed methods return, log, and throw exactly as
before. Existing suites cover the changed paths and pass locally:
```
build/sbt "core/testOnly org.apache.spark.AccumulatorSuite
org.apache.spark.util.AccumulatorV2Suite
org.apache.spark.scheduler.DAGSchedulerSuite"
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
This pull request and its description were written by Isaac.
Closes #59801 from shujingyang-db/accum-get-or-null.
Authored-by: Shujing Yang <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 293dcf9af9e3acd20779d622b535657a52aae9c3)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../scala/org/apache/spark/scheduler/DAGScheduler.scala | 8 +++++---
.../main/scala/org/apache/spark/util/AccumulatorV2.scala | 13 +++++++++++--
2 files changed, 16 insertions(+), 5 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 45a9948a069b..d4e32f93d64c 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -3517,10 +3517,12 @@ private[spark] class DAGScheduler(
val id = updates.id
try {
// Find the corresponding accumulator on the driver and update it
- val acc: AccumulatorV2[Any, Any] = AccumulatorContext.get(id) match {
- case Some(accum) => accum.asInstanceOf[AccumulatorV2[Any, Any]]
- case None =>
+ val acc: AccumulatorV2[Any, Any] = {
+ val accum = AccumulatorContext.getOrNull(id)
+ if (accum == null) {
throw SparkCoreErrors.accessNonExistentAccumulatorError(id)
+ }
+ accum.asInstanceOf[AccumulatorV2[Any, Any]]
}
acc.merge(updates.asInstanceOf[AccumulatorV2[Any, Any]])
if (acc.isInstanceOf[LastAttemptAccumulator[_, _, _]]) {
diff --git a/core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala
b/core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala
index 20b2baef0f23..688445033a44 100644
--- a/core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala
+++ b/core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala
@@ -287,9 +287,18 @@ private[spark] object AccumulatorContext extends Logging {
* Returns the [[AccumulatorV2]] registered with the given ID, if any.
*/
def get(id: Long): Option[AccumulatorV2[_, _]] = {
+ Option(getOrNull(id))
+ }
+
+ /**
+ * Same as [[get]] but returns null instead of None when the accumulator is
not registered or has
+ * been garbage collected. Used on the driver's per-task accumulator update
path to avoid
+ * allocating an Option for every lookup.
+ */
+ def getOrNull(id: Long): AccumulatorV2[_, _] = {
val ref = originals.get(id)
if (ref eq null) {
- None
+ null
} else {
// Since we are storing weak references, warn when the underlying data
is not valid.
val acc = ref.get
@@ -297,7 +306,7 @@ private[spark] object AccumulatorContext extends Logging {
logWarning(log"Attempted to access garbage collected accumulator " +
log"${MDC(ACCUMULATOR_ID, id)}")
}
- Option(acc)
+ acc
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
