This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 4b6955eb00f7 [SPARK-60571][CORE] Set the task result size accumulator
in place in TaskResultGetter
4b6955eb00f7 is described below
commit 4b6955eb00f70a7f2235dfe8cd1239ff37f0e2e3
Author: Shujing Yang <[email protected]>
AuthorDate: Tue Oct 27 16:31:18 2026 +0000
[SPARK-60571][CORE] Set the task result size accumulator in place in
TaskResultGetter
### What changes were proposed in this pull request?
When `TaskResultGetter` receives a task result, it sets the task result
size on the `RESULT_SIZE` accumulator. It did this with `map`, which rebuilds
the whole accumulator update sequence. This PR walks the updates with `foreach`
and sets the matching `LongAccumulator` in place. The name check now uses
`Option.contains` instead of comparing with a new `Some`.
### Why are the changes needed?
This code runs on the driver once per task result. The update sequence can
hold hundreds of entries when SQL metrics are involved. The old code already
mutated the matching accumulator in place, so the new sequence held the same
objects as the old one. Building it, and allocating a `Some` for every name
check, was unnecessary work.
### Does this PR introduce _any_ user-facing change?
No. `result.accumUpdates` keeps the same elements, and the result-size
accumulator gets the same value. The update still runs at the same point,
because `accumUpdates` is a strict `Seq` after deserialization
(`ArrayBuffer.toSeq` or `Seq.empty` in `DirectTaskResult.readExternal`). The
assert is unchanged.
### How was this patch tested?
No new test, because the accumulator updates and their values do not
change. Existing suites cover this path and pass locally:
```
build/sbt "core/testOnly org.apache.spark.scheduler.TaskResultGetterSuite
org.apache.spark.scheduler.TaskSchedulerImplSuite
org.apache.spark.AccumulatorSuite"
```
### 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 #59802 from shujingyang-db/task-result-accums-in-place.
Authored-by: Shujing Yang <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../main/scala/org/apache/spark/scheduler/TaskResultGetter.scala | 7 ++-----
1 file changed, 2 insertions(+), 5 deletions(-)
diff --git
a/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala
b/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala
index 70f58747506c..bde2c5705c23 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala
@@ -108,14 +108,11 @@ private[spark] class TaskResultGetter(sparkEnv: SparkEnv,
scheduler: TaskSchedul
// Set the task result size in the accumulator updates received from
the executors.
// We need to do this here on the driver because if we did this on
the executors then
// we would have to serialize the result again after updating the
size.
- result.accumUpdates = result.accumUpdates.map { a =>
- if (a.name == Some(InternalAccumulator.RESULT_SIZE)) {
+ result.accumUpdates.foreach { a =>
+ if (a.name.contains(InternalAccumulator.RESULT_SIZE)) {
val acc = a.asInstanceOf[LongAccumulator]
assert(acc.sum == 0L, "task result size should not have been set
on the executors")
acc.setValue(size)
- acc
- } else {
- a
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
