cloud-fan commented on code in PR #56597:
URL: https://github.com/apache/spark/pull/56597#discussion_r3462632909
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/metric/MetricsFailureInjectionSuite.scala:
##########
@@ -420,6 +420,63 @@ class MetricsFailureInjectionSuite
}
}
+ test("Three stage metrics block failure injection with AQE") {
+ // Same as the previous test but with AQE enabled. Under AQE each Exchange
is materialized
+ // as its own map-stage job, which exercises a different DAGScheduler path
than the
+ // AQE-disabled variant: the injection's deferred corruption must survive
across those
+ // per-shuffle jobs for the downstream FetchFailed (and thus the producer
re-run) to fire.
+ val stage1Metric = SQLMetrics.createMetric(spark.sparkContext, "stage 1
counter")
+ val stage2Metric = SQLMetrics.createMetric(spark.sparkContext, "stage 2
counter")
+ val stage3Metric = SQLMetrics.createMetric(spark.sparkContext, "stage 3
counter")
+ val stage1SLAMetric =
+ SQLLastAttemptMetrics.createMetric(spark.sparkContext, "stage 1 SLAM")
+ val stage2SLAMetric =
+ SQLLastAttemptMetrics.createMetric(spark.sparkContext, "stage 2 SLAM")
+ val stage3SLAMetric =
+ SQLLastAttemptMetrics.createMetric(spark.sparkContext, "stage 3 SLAM")
+
+ withTable("primary_table", "secondary_table") {
+ setUpTestTable("primary_table")
+ setUpTestTable("secondary_table")
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
+ withSparkContextConf(
+ config.Tests.INJECT_SHUFFLE_FETCH_FAILURES.key -> "true") {
+ val stage1MetricsExpr = incrementMetrics(Seq(stage1Metric,
stage1SLAMetric))
+ val stage1 = spark.read.table("primary_table")
+ .filter(Column(stage1MetricsExpr))
+ val stage2MetricsExpr = incrementMetrics(Seq(stage2Metric,
stage2SLAMetric))
+ val stage2 = stage1.join(
+ spark.read.table("secondary_table"),
+ usingColumn = "id",
+ joinType = "fullOuter")
+ .filter(Column(stage2MetricsExpr))
+ val stage3MetricsExpr = incrementMetrics(Seq(stage3Metric,
stage3SLAMetric))
+ val stage3 = stage2
+ .groupBy("primary_table.low_cardinality_col")
+ .count()
+ .filter(Column(stage3MetricsExpr))
+ val finalDf = stage3.as[(Int, Long)]
+ val result = finalDf.collect()
+ assert(result.toMap === (0 until 5).map(v => (v, 300 / 5)).toMap)
+
+ // The non-leaf stage 2 gets its first successful attempt corrupted
and re-runs, so its
+ // raw counter overcounts. SLAM reports only the last successful
attempt per RDD.
Review Comment:
This comment explains only stage 2, but line 464 asserts `stage1Metric.value
> 300` too — under AQE the leaf stage 1 is also a shuffle map stage (it feeds
the join shuffle), so it gets corrupted and re-runs as well. The non-AQE
sibling's comment covers both; matching it:
```suggestion
// Both the leaf stage 1 and the non-leaf stage 2 get their first
successful attempt
// corrupted and re-run, so their raw counters overcount. SLAM
reports only the last
// successful attempt per RDD.
```
--
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]
