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 388c247d0aef [SPARK-58102][SQL] Fix SLAM scope extraction bailing out
on ReusedSubqueryExec
388c247d0aef is described below
commit 388c247d0aefed39ac383c3c9ff681e30573a65a
Author: Juliusz Sompolski <[email protected]>
AuthorDate: Wed Jul 15 12:32:41 2026 +0800
[SPARK-58102][SQL] Fix SLAM scope extraction bailing out on
ReusedSubqueryExec
### What changes were proposed in this pull request?
`SQLLastAttemptAccumulator.extractStageRDDScopes` walks a Dataset's
`executedPlan` to find the
`RDDOperationScope` ids of the stages that make up its execution, so that
`SQLLastAttemptAccumulator`
can attribute "last attempt" metric updates to that execution. Its
`BaseSubqueryExec` match handles
`SubqueryExec`, `SubqueryBroadcastExec` and
`SubqueryAdaptiveBroadcastExec`, and then bails out on a
catch-all `case p =>` that records a `bailOutReason`. When extraction bails
out,
`lastAttemptValueForQueryExecution` / `lastAttemptValueForDataset` return
`None`.
`ReusedSubqueryExec` was not enumerated, so any plan containing a reused
subquery hit the catch-all
and caused `lastAttemptValueForDataset` /
`lastAttemptValueForQueryExecution` to return `None`.
Subquery reuse is enabled by default (`spark.sql.execution.reuseSubquery`),
and a
`ReusedSubqueryExec` is produced whenever the same subquery appears more
than once in a plan --
including when predicate pushdown duplicates a subquery into a scan's data
filters.
This PR adds a `ReusedSubqueryExec` case that recurses into its `child`,
mirroring the existing
`ReusedExchangeExec` handling. `ReusedSubqueryExec` is a `LeafExecNode`
whose `child` is a field
rather than a plan child, so the normal plan traversal never descends into
it; recursing collects
the reused subquery's scopes. If the original `SubqueryExec` is still
present elsewhere in the plan,
its scopes are collected twice, but that is harmless:
`lastAttemptValueForRDDScopes` deduplicates by
scope (it reduces the collected scope ids to a set and aggregates one RDD
per scope).
### Why are the changes needed?
Without this, `lastAttemptValueForDataset` /
`lastAttemptValueForQueryExecution` silently return
`None` for any query whose physical plan contains a `ReusedSubqueryExec`,
even though the metric was
tracked correctly. Because subquery reuse is on by default, this affects
common queries -- e.g. the
same scalar/`IN` subquery referenced twice, or a subquery duplicated into a
`FileScan`'s data filters
by predicate pushdown.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added a test to `SQLLastAttemptMetricPlanShapesSuite` whose physical plan
contains a
`ReusedSubqueryExec`: a scalar subquery with a nested `(SELECT MIN(id) FROM
range(5))` predicate that
gets pushed into the outer subquery's scan data filters and duplicated, so
subquery reuse rewrites one
copy to a `ReusedSubqueryExec`. The test asserts both the metric value and
(via
`testPhysicalPlanShape`) that a `ReusedSubqueryExec` is actually present in
the plan -- with a
fallback for the forced-AQE-replan variants, which bypass subquery reuse
and keep the duplicated
subquery as two separate `SubqueryExec` nodes (this mirrors the existing
`exchange - Shuffle` test's
`ReusedExchangeExec` fallback). Before the fix,
`lastAttemptValueForDataset` returned `None` for this
plan (assertion `None did not equal Some(300)`); after the fix it returns
the expected value. The
test runs across the suite's existing `useAQE` x `failureMode` x
`aqeReplans` matrix (9 variants).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57226 from juliuszsompolski/SPARK-58102-reused-subquery-slam.
Authored-by: Juliusz Sompolski <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 53d48e3cc4bbc301003100fcde06564744b47c1b)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../metric/SQLLastAttemptAccumulator.scala | 6 ++++-
.../SQLLastAttemptMetricPlanShapesSuite.scala | 27 +++++++++++++++++++++-
2 files changed, 31 insertions(+), 2 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLLastAttemptAccumulator.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLLastAttemptAccumulator.scala
index 114d3974bb0e..245e5204104c 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLLastAttemptAccumulator.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLLastAttemptAccumulator.scala
@@ -23,7 +23,7 @@ import scala.util.control.NonFatal
import org.apache.spark.SparkContext
import org.apache.spark.internal.{LogEntry, Logging}
import org.apache.spark.sql.Dataset
-import org.apache.spark.sql.execution.{BaseSubqueryExec, QueryExecution,
SparkPlan, SubqueryAdaptiveBroadcastExec, SubqueryBroadcastExec, SubqueryExec,
WholeStageCodegenExec}
+import org.apache.spark.sql.execution.{BaseSubqueryExec, QueryExecution,
ReusedSubqueryExec, SparkPlan, SubqueryAdaptiveBroadcastExec,
SubqueryBroadcastExec, SubqueryExec, WholeStageCodegenExec}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec,
BroadcastExchangeLike, ReusedExchangeExec, ShuffleExchangeExec,
ShuffleExchangeLike}
import org.apache.spark.util.{AccumulatorV2, LastAttemptAccumulator}
@@ -374,6 +374,10 @@ object SQLLastAttemptAccumulator extends Logging {
case _: SubqueryAdaptiveBroadcastExec =>
// Used by DPP filter only.
Nil
+ case r: ReusedSubqueryExec =>
+ // Reused subquery is going to reuse stuff executed in the scope of
its child,
+ // i.e. the subquery it reuses.
+ recurse(r.child)
case p =>
// Bail out if future unknown implementation is encountered.
bailOutReason = Some(s"Unsupported BaseSubqueryExec:
${p.getClass.getName}")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLLastAttemptMetricPlanShapesSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLLastAttemptMetricPlanShapesSuite.scala
index d51c8cae1caa..033a0eea5b5a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLLastAttemptMetricPlanShapesSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLLastAttemptMetricPlanShapesSuite.scala
@@ -22,7 +22,7 @@ import scala.util.Random
import org.scalatest.Tag
import org.apache.spark.internal.config
-import org.apache.spark.sql.execution.{CollectLimitExec, RDDScanExec,
SparkPlan}
+import org.apache.spark.sql.execution.{CollectLimitExec, RDDScanExec,
ReusedSubqueryExec, SparkPlan, SubqueryExec}
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
AQETestHelper, DisableAdaptiveExecutionSuite}
import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
import org.apache.spark.sql.execution.exchange._
@@ -298,6 +298,31 @@ class SQLLastAttemptMetricPlanShapesSuite
metricValueCheck = MetricValue.exactly(NUM_RECORDS)
)
+ testPhysicalPlanShape(
+ // The inner `(SELECT MIN(id) FROM range(5))` subquery gets pushed into
the outer subquery's
+ // scan DataFilters and duplicated, so subquery reuse turns one copy into a
+ // ReusedSubqueryExec. extractStageRDDScopes must handle that node instead
of bailing out
+ // (which would make lastAttemptValueForDataset return None for the whole
plan).
+ label = "subquery - scalar - reused",
+ sqlQuery =
+ s"""SELECT *
+ | FROM $TABLE_NAME
+ | WHERE id == (
+ | SELECT MAX(low_cardinality_col)
+ | FROM $TABLE_NAME
+ | WHERE increment_metric()
+ | AND low_cardinality_col >= (
+ | SELECT MIN(id)
+ | FROM range(5)))""".stripMargin,
+ metricValueCheck = MetricValue.exactly(NUM_RECORDS),
+ executedPlanCheck = PhysicalPlan.exists {
+ case _: ReusedSubqueryExec => true
+ // Forced AQE replans bypass subquery reuse, so the duplicated inner
subquery stays as two
+ // separate SubqueryExec nodes instead of a ReusedSubqueryExec.
+ case _: SubqueryExec if PhysicalPlan.hasAQEReplans => true
+ }
+ )()
+
testPhysicalPlanShape(
label = "subquery - EXISTS",
sqlQuery =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
