szehon-ho commented on code in PR #57323:
URL: https://github.com/apache/spark/pull/57323#discussion_r3624934293
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/metric/MetricsFailureInjectionSuite.scala:
##########
@@ -299,6 +304,71 @@ class MetricsFailureInjectionSuite
}
}
+ test("Observation metrics block failure injection") {
Review Comment:
Good point. Renamed this to "Observation metrics select attempts according
to legacy config" and now verify both last-successful-attempt and legacy
all-attempt behavior.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CollectMetricsExec.scala:
##########
@@ -36,9 +36,10 @@ case class CollectMetricsExec(
child: SparkPlan)
extends UnaryExecNode {
- private lazy val accumulator: AggregatingAccumulator = {
- val acc = AggregatingAccumulator(metricExpressions, child.output)
+ private lazy val accumulator: LastAttemptAggregatingAccumulator = {
+ val acc = AggregatingAccumulator.lastAttempt(metricExpressions,
child.output)
acc.register(sparkContext, Option(CollectMetricsExec.ACCUMULATOR_NAME))
+ acc.initializeLastAttemptAccumulator()
Review Comment:
Done. The registered overloads now take the SparkContext and name, and the
last-attempt factory also initializes its tracking state.
--
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]
